链接KAFKA异常：Authentication failed during authentication due to invalid credentials with SASL mechanism

使用带kerberos 认证的Kafka客户端链接kafka 创建topic 出现如下异常：Authentication failed during authentication due to invalid credentials with SASL mechanism。kafka server 后台只有如下异常信息：

Failed authentication with /192.168.0.1 (Authentication failed during authentication due to invalid credentials with SASL mechanism GSSAPI) (org.apache.kafka.common.network.Selector)

开始排查问题原因：

通过查看Kafka源代码定位到错误大致发生在：

org.apache.kafka.common.security.authenticator.SaslServerAuthenticator 类的handleSaslToken 方法：

private void handleSaslToken(byte[] clientToken) throws IOException {
        if (!this.enableKafkaSaslAuthenticateHeaders) {
            byte[] response = this.saslServer.evaluateResponse(clientToken);
            if (this.reauthInfo.reauthenticating() && this.saslServer.isComplete()) {
                this.reauthInfo.ensurePrincipalUnchanged(this.principal());
            }

            if (response != null) {
                this.netOutBuffer = new NetworkSend(this.connectionId, ByteBuffer.wrap(response));
                this.flushNetOutBufferAndUpdateInterestOps();
            }
        } else {
            ByteBuffer requestBuffer = ByteBuffer.wrap(clientToken);
            RequestHeader header = RequestHeader.parse(requestBuffer);
            ApiKeys apiKey = header.apiKey();
            short version = header.apiVersion();
            RequestContext requestContext = new RequestContext(header, this.connectionId, this.clientAddress(), KafkaPrincipal.ANONYMOUS, this.listenerName, this.securityProtocol);
            RequestAndSize requestAndSize = requestContext.parseRequest(requestBuffer);
            if (apiKey != ApiKeys.SASL_AUTHENTICATE) {
                IllegalSaslStateException e = new IllegalSaslStateException("Unexpected Kafka request of type " + apiKey + " during SASL authentication.");
                this.buildResponseOnAuthenticateFailure(requestContext, requestAndSize.request.getErrorResponse(e));
                throw e;
            }

            if (!apiKey.isVersionSupported(version)) {
                throw new UnsupportedVersionException("Version " + version + " is not supported for apiKey " + apiKey);
            }

            if (!this.reauthInfo.connectedClientSupportsReauthentication) {
                this.reauthInfo.connectedClientSupportsReauthentication = version > 0;
            }

            SaslAuthenticateRequest saslAuthenticateRequest = (SaslAuthenticateRequest)requestAndSize.request;

            try {
                byte[] responseToken = this.saslServer.evaluateResponse(Utils.copyArray(saslAuthenticateRequest.data().authBytes()));
                if (this.reauthInfo.reauthenticating() && this.saslServer.isComplete()) {
                    this.reauthInfo.ensurePrincipalUnchanged(this.principal());
                }

                byte[] responseBytes = responseToken == null ? new byte[0] : responseToken;
                long sessionLifetimeMs = !this.saslServer.isComplete() ? 0L : this.reauthInfo.calcCompletionTimesAndReturnSessionLifetimeMs();
                this.sendKafkaResponse(requestContext, new SaslAuthenticateResponse((new SaslAuthenticateResponseData()).setErrorCode(Errors.NONE.code()).setAuthBytes(responseBytes).setSessionLifetimeMs(sessionLifetimeMs)));
            } catch (SaslAuthenticationException var13) {
                this.buildResponseOnAuthenticateFailure(requestContext, new SaslAuthenticateResponse((new SaslAuthenticateResponseData()).setErrorCode(Errors.SASL_AUTHENTICATION_FAILED.code()).setErrorMessage(var13.getMessage())));
                throw var13;
            } catch (SaslException var14) {
                KerberosError kerberosError = KerberosError.fromException(var14);
                if (kerberosError != null && kerberosError.retriable()) {
                    throw var14;
                }

                String errorMessage = "Authentication failed during " + this.reauthInfo.authenticationOrReauthenticationText() + " due to invalid credentials with SASL mechanism " + this.saslMechanism;
                this.sendKafkaResponse(requestContext, new SaslAuthenticateResponse((new SaslAuthenticateResponseData()).setErrorCode(Errors.SASL_AUTHENTICATION_FAILED.code()).setErrorMessage(errorMessage)));
                throw new SaslAuthenticationException(errorMessage, var14);
            }
        }

    }

大概是在  saslServer.evaluateResponse 的时候抛出了异常，可是看不到具体的异常信息。

接下来使用 Arthas，查看具体的异常信息，Arthas 的使用方法略过，大家可以直接在Arthas官网查看。

使用Arthas链接上Kafka进程后 使用 sm 命令查找 org.apache.kafka.common.security.authenticator.SaslServerAuthenticator 类 和  handleSaslToken 方法。

使用 watch 命令 来查看这个方法的参数 返回值 和异常信息：

watch org.apache.kafka.common.security.authenticator.SaslServerAuthenticator handleSaslToken {params,returnObj,throwExp} -e  -x 6 

终于看到了真正导致认证失败的异常信息：

 KrbException: Clock skew too great (37)

这个异常信息的原因就是时钟不同步，经排查原来是 Kafka集群的NTP服务挂了导致的，启动NTP服务，同步时钟，搞定。