请帮助我修复一些异常，而连接到Kafka经纪在一个kerberized集群。
我正在cloudera集群上运行3.0.0-1版本的kafka。kafka是作为cloudera manager（cm）的服务安装的。经纪人开局不错。我可以创建和列出主题。
但是我的控制台制作人无法连接到kafka代理主题。我提供我的Kafka客户和制片人财产如下：
使用的命令和错误

[root@local-dn-1.HADOOP.COM ~]$ /opt/cloudera/parcels/KAFKA/lib/kafka/bin/kafka-console-producer.sh --broker-list local-dn-1.HADOOP.COM:9092 --topic "Kafka-Sucker"  --producer.config /etc/kafka/conf/producer-conf/producer.properties
SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in [jar:file:/opt/cloudera/parcels/KAFKA-3.0.0-1.3.0.0.p0.40/lib/kafka/libs/slf4j-log4j12-1.7.25.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/opt/cloudera/parcels/KAFKA-3.0.0-1.3.0.0.p0.40/lib/kafka/libs/slf4j-log4j12-1.7.5.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
18/03/28 07:38:45 INFO producer.ProducerConfig: ProducerConfig values:
        acks = 1
        batch.size = 16384
        bootstrap.servers = [local-dn-1.HADOOP.COM:9092]
        buffer.memory = 33554432
        client.id = console-producer
        compression.type = none
        connections.max.idle.ms = 540000
        enable.idempotence = false
        interceptor.classes = null
        key.serializer = class org.apache.kafka.common.serialization.ByteArraySerializer
        linger.ms = 1000
        max.block.ms = 60000
        max.in.flight.requests.per.connection = 5
        max.request.size = 1048576
        metadata.max.age.ms = 300000
        metric.reporters = []
        metrics.num.samples = 2
        metrics.recording.level = INFO
        metrics.sample.window.ms = 30000
        partitioner.class = class org.apache.kafka.clients.producer.internals.DefaultPartitioner
        receive.buffer.bytes = 32768
        reconnect.backoff.max.ms = 1000
        reconnect.backoff.ms = 50
        request.timeout.ms = 1500
        retries = 3
        retry.backoff.ms = 100
        sasl.jaas.config = null
        sasl.kerberos.kinit.cmd = /usr/bin/kinit
        sasl.kerberos.min.time.before.relogin = 60000
        sasl.kerberos.service.name = "kafka"
        sasl.kerberos.ticket.renew.jitter = 0.05
        sasl.kerberos.ticket.renew.window.factor = 0.8
        sasl.mechanism = GSSAPI
        security.protocol = SASL_PLAINTEXT
        send.buffer.bytes = 102400
        ssl.cipher.suites = null
        ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
        ssl.endpoint.identification.algorithm = null
        ssl.key.password = null
        ssl.keymanager.algorithm = SunX509
        ssl.keystore.location = null
        ssl.keystore.password = null
        ssl.keystore.type = JKS
        ssl.protocol = TLS
        ssl.provider = null
        ssl.secure.random.implementation = null
        ssl.trustmanager.algorithm = PKIX
        ssl.truststore.location = null
        ssl.truststore.password = null
        ssl.truststore.type = JKS
        transaction.timeout.ms = 60000
        transactional.id = null
        value.serializer = class org.apache.kafka.common.serialization.ByteArraySerializer
18/03/28 07:38:45 DEBUG metrics.Metrics: Added sensor with name bufferpool-wait-time
18/03/28 07:38:45 DEBUG metrics.Metrics: Added sensor with name buffer-exhausted-records
18/03/28 07:38:45 DEBUG clients.Metadata: Updated cluster metadata version 1 to Cluster(id = null, nodes = [local-dn-1.HADOOP.COM:9092 (id: -1 rack: null)], partitions = [])
Java config name: null
Native config name: /etc/krb5.conf
Loaded from native config
>>> KeyTabInputStream, readName(): HADOOP.COM
>>> KeyTabInputStream, readName(): kafka-client
>>> KeyTab: load() entry length: 93; type: 18
>>> KeyTabInputStream, readName(): HADOOP.COM
>>> KeyTabInputStream, readName(): kafka-client
>>> KeyTab: load() entry length: 77; type: 17
>>> KeyTabInputStream, readName(): HADOOP.COM
>>> KeyTabInputStream, readName(): kafka-client
>>> KeyTab: load() entry length: 77; type: 23
Looking for keys for: kafka-client@HADOOP.COM
Added key: 23version: 1
Added key: 17version: 1
Added key: 18version: 1
>>> KdcAccessibility: reset
Looking for keys for: kafka-client@HADOOP.COM
Added key: 23version: 1
Added key: 17version: 1
Added key: 18version: 1
default etypes for default_tkt_enctypes: 23 17 18.
>>> KrbAsReq creating message
>>> KrbKdcReq send: kdc=ForestAD.HADOOP.COM TCP:88, timeout=3000, number of retries =3, #bytes=180
>>> KDCCommunication: kdc=ForestAD.HADOOP.COM TCP:88, timeout=3000,Attempt =1, #bytes=180
>>>DEBUG: TCPClient reading 240 bytes
>>> KrbKdcReq send: #bytes read=240
>>>Pre-Authentication Data:
         PA-DATA type = 19
         PA-ETYPE-INFO2 etype = 18, salt = HADOOP.COMkafka-client, s2kparams = null
         PA-ETYPE-INFO2 etype = 23, salt = null, s2kparams = null
>>>Pre-Authentication Data:
         PA-DATA type = 2
         PA-ENC-TIMESTAMP
>>>Pre-Authentication Data:
         PA-DATA type = 16
>>>Pre-Authentication Data:
         PA-DATA type = 15
>>> KdcAccessibility: remove hadoop.com
>>> KDCRep: init() encoding tag is 126 req type is 11
>>>KRBError:
         sTime is Wed Mar 28 07:37:50 EDT 2018 1522237070000
         suSec is 110488
         error code is 25
         error Message is Additional pre-authentication required
         sname is krbtgt/HADOOP.COM@HADOOP.COM
         eData provided.
         msgType is 30
>>>Pre-Authentication Data:
         PA-DATA type = 19
         PA-ETYPE-INFO2 etype = 18, salt = HADOOP.COMkafka-client, s2kparams = null
         PA-ETYPE-INFO2 etype = 23, salt = null, s2kparams = null
>>>Pre-Authentication Data:
         PA-DATA type = 2
         PA-ENC-TIMESTAMP
>>>Pre-Authentication Data:
         PA-DATA type = 16
>>>Pre-Authentication Data:
         PA-DATA type = 15
KrbAsReqBuilder: PREAUTH FAILED/REQ, re-send AS-REQ
default etypes for default_tkt_enctypes: 23 17 18.
Looking for keys for: kafka-client@HADOOP.COM
Added key: 23version: 1
Added key: 17version: 1
Added key: 18version: 1
Looking for keys for: kafka-client@HADOOP.COM
Added key: 23version: 1
Added key: 17version: 1
Added key: 18version: 1
default etypes for default_tkt_enctypes: 23 17 18.
>>> EType: sun.security.krb5.internal.crypto.Aes256CtsHmacSha1EType
>>> KrbAsReq creating message
>>> KrbKdcReq send: kdc=ForestAD.HADOOP.COM TCP:88, timeout=3000, number of retries =3, #bytes=269
>>> KDCCommunication: kdc=ForestAD.HADOOP.COM TCP:88, timeout=3000,Attempt =1, #bytes=269
>>>DEBUG: TCPClient reading 1678 bytes
>>> KrbKdcReq send: #bytes read=1678
>>> KdcAccessibility: remove hadoop.com
Looking for keys for: kafka-client@HADOOP.COM
Added key: 23version: 1
Added key: 17version: 1
Added key: 18version: 1
>>> EType: sun.security.krb5.internal.crypto.Aes256CtsHmacSha1EType
>>> KrbAsRep cons in KrbAsReq.getReply kafka-client
18/03/28 07:38:45 INFO authenticator.AbstractLogin: Successfully logged in.
18/03/28 07:38:45 DEBUG kerberos.KerberosLogin: [Principal=kafka-client@HADOOP.COM]: It is a Kerberos ticket
18/03/28 07:38:45 INFO kerberos.KerberosLogin: [Principal=kafka-client@HADOOP.COM]: TGT refresh thread started.
18/03/28 07:38:45 DEBUG kerberos.KerberosLogin: Found TGT with client principal 'kafka-client@HADOOP.COM' and server principal 'krbtgt/HADOOP.COM@HADOOP.COM'.
18/03/28 07:38:45 INFO kerberos.KerberosLogin: [Principal=kafka-client@HADOOP.COM]: TGT valid starting at: Wed Mar 28 07:37:50 EDT 2018
18/03/28 07:38:45 INFO kerberos.KerberosLogin: [Principal=kafka-client@HADOOP.COM]: TGT expires: Wed Mar 28 17:37:50 EDT 2018
18/03/28 07:38:45 INFO kerberos.KerberosLogin: [Principal=kafka-client@HADOOP.COM]: TGT refresh sleeping until: Wed Mar 28 15:42:00 EDT 2018
18/03/28 07:38:45 DEBUG metrics.Metrics: Added sensor with name produce-throttle-time
18/03/28 07:38:45 DEBUG metrics.Metrics: Added sensor with name connections-closed:
18/03/28 07:38:45 DEBUG metrics.Metrics: Added sensor with name connections-created:
18/03/28 07:38:45 DEBUG metrics.Metrics: Added sensor with name bytes-sent-received:
18/03/28 07:38:45 DEBUG metrics.Metrics: Added sensor with name bytes-sent:
18/03/28 07:38:45 DEBUG metrics.Metrics: Added sensor with name bytes-received:
18/03/28 07:38:45 DEBUG metrics.Metrics: Added sensor with name select-time:
18/03/28 07:38:45 DEBUG metrics.Metrics: Added sensor with name io-time:
18/03/28 07:38:45 DEBUG metrics.Metrics: Added sensor with name batch-size
18/03/28 07:38:45 DEBUG metrics.Metrics: Added sensor with name compression-rate
18/03/28 07:38:45 DEBUG metrics.Metrics: Added sensor with name queue-time
18/03/28 07:38:45 DEBUG metrics.Metrics: Added sensor with name request-time
18/03/28 07:38:45 DEBUG metrics.Metrics: Added sensor with name records-per-request
18/03/28 07:38:45 DEBUG metrics.Metrics: Added sensor with name record-retries
18/03/28 07:38:45 DEBUG metrics.Metrics: Added sensor with name errors
18/03/28 07:38:45 DEBUG metrics.Metrics: Added sensor with name record-size-max
18/03/28 07:38:45 DEBUG metrics.Metrics: Added sensor with name batch-split-rate
18/03/28 07:38:45 DEBUG internals.Sender: Starting Kafka producer I/O thread.
18/03/28 07:38:45 INFO utils.AppInfoParser: Kafka version : 0.11.0-kafka-3.0.0
18/03/28 07:38:45 INFO utils.AppInfoParser: Kafka commitId : unknown
18/03/28 07:38:45 DEBUG producer.KafkaProducer: Kafka producer with client id console-producer created
>Hello World
18/03/28 07:38:53 DEBUG clients.NetworkClient: Initialize connection to node local-dn-1.HADOOP.COM:9092 (id: -1 rack: null) for sending metadata request
18/03/28 07:38:53 DEBUG clients.NetworkClient: Initiating connection to node local-dn-1.HADOOP.COM:9092 (id: -1 rack: null)
18/03/28 07:38:53 DEBUG authenticator.SaslClientAuthenticator: Set SASL client state to SEND_HANDSHAKE_REQUEST
18/03/28 07:38:53 DEBUG authenticator.SaslClientAuthenticator: Creating SaslClient: client=kafka-client@HADOOP.COM;service="kafka";serviceHostname=local-dn-1.HADOOP.COM;mechs=[GSSAPI]
18/03/28 07:38:53 DEBUG metrics.Metrics: Added sensor with name node--1.bytes-sent
18/03/28 07:38:53 DEBUG metrics.Metrics: Added sensor with name node--1.bytes-received
18/03/28 07:38:53 DEBUG metrics.Metrics: Added sensor with name node--1.latency
18/03/28 07:38:53 DEBUG network.Selector: Created socket with SO_RCVBUF = 32768, SO_SNDBUF = 102400, SO_TIMEOUT = 0 to node -1
18/03/28 07:38:53 DEBUG authenticator.SaslClientAuthenticator: Set SASL client state to RECEIVE_HANDSHAKE_RESPONSE
18/03/28 07:38:53 DEBUG clients.NetworkClient: Completed connection to node -1.  Fetching API versions.
18/03/28 07:38:53 DEBUG authenticator.SaslClientAuthenticator: Set SASL client state to INITIAL
Found ticket for kafka-client@HADOOP.COM to go to krbtgt/HADOOP.COM@HADOOP.COM expiring on Wed Mar 28 17:37:50 EDT 2018
Entered Krb5Context.initSecContext with state=STATE_NEW
Found ticket for kafka-client@HADOOP.COM to go to krbtgt/HADOOP.COM@HADOOP.COM expiring on Wed Mar 28 17:37:50 EDT 2018
Service ticket not found in the subject
>>> Credentials acquireServiceCreds: same realm
default etypes for default_tgs_enctypes: 23 17 18.
>>> CksumType: sun.security.krb5.internal.crypto.RsaMd5CksumType
>>> EType: sun.security.krb5.internal.crypto.ArcFourHmacEType
>>> KrbKdcReq send: kdc=ForestAD.HADOOP.COM TCP:88, timeout=3000, number of retries =3, #bytes=1631
>>> KDCCommunication: kdc=ForestAD.HADOOP.COM TCP:88, timeout=3000,Attempt =1, #bytes=1631
>>>DEBUG: TCPClient reading 151 bytes
>>> KrbKdcReq send: #bytes read=151
>>> KdcAccessibility: remove hadoop.com
>>> KDCRep: init() encoding tag is 126 req type is 13
>>>KRBError:
         sTime is Wed Mar 28 07:37:59 EDT 2018 1522237079000
         suSec is 467340
         error code is 7
         error Message is Server not found in Kerberos database
         sname is "kafka"/local-dn-1.HADOOP.COM@HADOOP.COM
         msgType is 30
KrbException: Server not found in Kerberos database (7)
        at sun.security.krb5.KrbTgsRep.<init>(KrbTgsRep.java:70)
        at sun.security.krb5.KrbTgsReq.getReply(KrbTgsReq.java:251)
        at sun.security.krb5.KrbTgsReq.sendAndGetCreds(KrbTgsReq.java:262)
        at sun.security.krb5.internal.CredentialsUtil.serviceCreds(CredentialsUtil.java:308)
        at sun.security.krb5.internal.CredentialsUtil.acquireServiceCreds(CredentialsUtil.java:126)
        at sun.security.krb5.Credentials.acquireServiceCreds(Credentials.java:458)
        at sun.security.jgss.krb5.Krb5Context.initSecContext(Krb5Context.java:693)
        at sun.security.jgss.GSSContextImpl.initSecContext(GSSContextImpl.java:248)
        at sun.security.jgss.GSSContextImpl.initSecContext(GSSContextImpl.java:179)
        at com.sun.security.sasl.gsskerb.GssKrb5Client.evaluateChallenge(GssKrb5Client.java:192)
        at org.apache.kafka.common.security.authenticator.SaslClientAuthenticator$2.run(SaslClientAuthenticator.java:280)
        at org.apache.kafka.common.security.authenticator.SaslClientAuthenticator$2.run(SaslClientAuthenticator.java:278)
        at java.security.AccessController.doPrivileged(Native Method)
        at javax.security.auth.Subject.doAs(Subject.java:422)
        at org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.createSaslToken(SaslClientAuthenticator.java:278)
        at org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.sendSaslToken(SaslClientAuthenticator.java:215)
        at org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.authenticate(SaslClientAuthenticator.java:183)
        at org.apache.kafka.common.network.KafkaChannel.prepare(KafkaChannel.java:76)
        at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:376)
        at org.apache.kafka.common.network.Selector.poll(Selector.java:326)
        at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:454)
        at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:224)
        at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:162)
        at java.lang.Thread.run(Thread.java:748)
Caused by: KrbException: Identifier doesn't match expected value (906)
        at sun.security.krb5.internal.KDCRep.init(KDCRep.java:140)
        at sun.security.krb5.internal.TGSRep.init(TGSRep.java:65)
        at sun.security.krb5.internal.TGSRep.<init>(TGSRep.java:60)
        at sun.security.krb5.KrbTgsRep.<init>(KrbTgsRep.java:55)
        ... 23 more
18/03/28 07:38:53 DEBUG network.Selector: Connection with local-dn-1.HADOOP.COM/10.133.144.108 disconnected
javax.security.sasl.SaslException: An error: (java.security.PrivilegedActionException: javax.security.sasl.SaslException: GSS initiate failed [Caused by GSSException: No valid credentials provided (Mechanism level: Server not found in Kerberos database (7))]) occurred when evaluating SASL token received from the Kafka Broker. Kafka Client will go to AUTH_FAILED state. [Caused by javax.security.sasl.SaslException: GSS initiate failed [Caused by GSSException: No valid credentials provided (Mechanism level: Server not found in Kerberos database (7))]]
        at org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.createSaslToken(SaslClientAuthenticator.java:298)
        at org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.sendSaslToken(SaslClientAuthenticator.java:215)
        at org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.authenticate(SaslClientAuthenticator.java:183)
        at org.apache.kafka.common.network.KafkaChannel.prepare(KafkaChannel.java:76)
        at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:376)
        at org.apache.kafka.common.network.Selector.poll(Selector.java:326)
        at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:454)
        at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:224)
        at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:162)
        at java.lang.Thread.run(Thread.java:748)
Caused by: javax.security.sasl.SaslException: GSS initiate failed [Caused by GSSException: No valid credentials provided (Mechanism level: Server not found in Kerberos database (7))]
        at com.sun.security.sasl.gsskerb.GssKrb5Client.evaluateChallenge(GssKrb5Client.java:211)
        at org.apache.kafka.common.security.authenticator.SaslClientAuthenticator$2.run(SaslClientAuthenticator.java:280)
        at org.apache.kafka.common.security.authenticator.SaslClientAuthenticator$2.run(SaslClientAuthenticator.java:278)
        at java.security.AccessController.doPrivileged(Native Method)
        at javax.security.auth.Subject.doAs(Subject.java:422)
        at org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.createSaslToken(SaslClientAuthenticator.java:278)
        ... 9 more
Caused by: GSSException: No valid credentials provided (Mechanism level: Server not found in Kerberos database (7))
        at sun.security.jgss.krb5.Krb5Context.initSecContext(Krb5Context.java:770)
        at sun.security.jgss.GSSContextImpl.initSecContext(GSSContextImpl.java:248)
        at sun.security.jgss.GSSContextImpl.initSecContext(GSSContextImpl.java:179)
        at com.sun.security.sasl.gsskerb.GssKrb5Client.evaluateChallenge(GssKrb5Client.java:192)
        ... 14 more
Caused by: KrbException: Server not found in Kerberos database (7)
        at sun.security.krb5.KrbTgsRep.<init>(KrbTgsRep.java:70)
        at sun.security.krb5.KrbTgsReq.getReply(KrbTgsReq.java:251)
        at sun.security.krb5.KrbTgsReq.sendAndGetCreds(KrbTgsReq.java:262)
        at sun.security.krb5.internal.CredentialsUtil.serviceCreds(CredentialsUtil.java:308)
        at sun.security.krb5.internal.CredentialsUtil.acquireServiceCreds(CredentialsUtil.java:126)
        at sun.security.krb5.Credentials.acquireServiceCreds(Credentials.java:458)
        at sun.security.jgss.krb5.Krb5Context.initSecContext(Krb5Context.java:693)
        ... 17 more
Caused by: KrbException: Identifier doesn't match expected value (906)
        at sun.security.krb5.internal.KDCRep.init(KDCRep.java:140)
        at sun.security.krb5.internal.TGSRep.init(TGSRep.java:65)
        at sun.security.krb5.internal.TGSRep.<init>(TGSRep.java:60)
        at sun.security.krb5.KrbTgsRep.<init>(KrbTgsRep.java:55)
        ... 23 more
18/03/28 07:38:53 DEBUG clients.NetworkClient: Node -1 disconnected.
18/03/28 07:38:53 WARN clients.NetworkClient: Connection to node -1 terminated during authentication. This may indicate that authentication failed due to invalid credentials.
18/03/28 07:38:53 DEBUG clients.NetworkClient: Give up sending metadata request since no node is available
18/03/28 07:38:53 DEBUG clients.NetworkClient: Give up sending metadata request since no node is available
18/03/28 07:38:53 DEBUG clients.NetworkClient: Initialize connection to node local-dn-1.HADOOP.COM:9092 (id: -1 rack: null) for sending metadata request
18/03/28 07:38:53 DEBUG clients.NetworkClient: Initiating connection to node local-dn-1.HADOOP.COM:9092 (id: -1 rack: null)
18/03/28 07:38:53 DEBUG authenticator.SaslClientAuthenticator: Set SASL client state to SEND_HANDSHAKE_REQUEST
18/03/28 07:38:53 DEBUG authenticator.SaslClientAuthenticator: Creating SaslClient: client=kafka-client@HADOOP.COM;service="kafka";serviceHostname=local-dn-1.HADOOP.COM;mechs=[GSSAPI]
18/03/28 07:38:53 DEBUG network.Selector: Created socket with SO_RCVBUF = 32768, SO_SNDBUF = 102400, SO_TIMEOUT = 0 to node -1
18/03/28 07:38:53 DEBUG authenticator.SaslClientAuthenticator: Set SASL client state to RECEIVE_HANDSHAKE_RESPONSE
18/03/28 07:38:53 DEBUG clients.NetworkClient: Completed connection to node -1.  Fetching API versions.
18/03/28 07:38:53 DEBUG authenticator.SaslClientAuthenticator: Set SASL client state to INITIAL
^C18/03/28 07:38:54 INFO producer.KafkaProducer: Closing the Kafka producer with timeoutMillis = 9223372036854775807 ms.
18/03/28 07:38:54 DEBUG internals.Sender: Beginning shutdown of Kafka producer I/O thread, sending remaining records.
18/03/28 07:38:54 DEBUG metrics.Metrics: Removed sensor with name connections-closed:
18/03/28 07:38:54 DEBUG metrics.Metrics: Removed sensor with name connections-created:
18/03/28 07:38:54 DEBUG metrics.Metrics: Removed sensor with name bytes-sent-received:
18/03/28 07:38:54 DEBUG metrics.Metrics: Removed sensor with name bytes-sent:
18/03/28 07:38:54 DEBUG metrics.Metrics: Removed sensor with name bytes-received:
18/03/28 07:38:54 DEBUG metrics.Metrics: Removed sensor with name select-time:
18/03/28 07:38:54 DEBUG metrics.Metrics: Removed sensor with name io-time:
18/03/28 07:38:54 DEBUG metrics.Metrics: Removed sensor with name node--1.bytes-sent
18/03/28 07:38:54 DEBUG metrics.Metrics: Removed sensor with name node--1.bytes-received
18/03/28 07:38:54 DEBUG metrics.Metrics: Removed sensor with name node--1.latency
18/03/28 07:38:54 WARN kerberos.KerberosLogin: [Principal=kafka-client@HADOOP.COM]: TGT renewal thread has been interrupted and will exit.
18/03/28 07:38:54 DEBUG internals.Sender: Shutdown of Kafka producer I/O thread has completed.
18/03/28 07:38:54 DEBUG producer.KafkaProducer: Kafka producer with client id console-producer has been closed
[root@local-dn-1.HADOOP.COM ~]$

配置和环境变量

export KAFKA_HOME=/opt/cloudera/parcels/KAFKA-3.0.0-1.3.0.0.p0.40/lib/kafka
export JAVA_HOME=/usr/java/jdk1.8.0_131
export KAFKA_OPTS="-Djava.security.auth.login.config=/etc/kafka/conf/producer-conf/kafka-client-jaas.conf -Dsun.security.krb5.debug=true"
export JVM_ARGS="-Djava.security.krb5.conf=/etc/krb5.conf -Djava.security.auth.login.config=/etc/kafka/conf/producer-conf/kafka-client-jaas.conf"
export BROKER_JAVA_OPTS="-Djava.security.krb5.conf=/etc/krb5.conf"
``` `/etc/kafka/conf/producer-conf/kafka-client-jaas.conf` ```
KafkaServer {
   com.sun.security.auth.module.Krb5LoginModule required
   doNotPrompt=true
   useKeyTab=true
   storeKey=true
   keyTab="/etc/kafka/conf/kafka.keytab"
   principal="kafka/local-dn-1.hadoop.com@HADOOP.COM";
};
KafkaClient {
   com.sun.security.auth.module.Krb5LoginModule required
   useKeyTab=true
   storeKey=true
   useTicketCache=false
   keyTab="/etc/kafka/conf/producer-conf/kafka-client.keytab"
   principal="kafka-client@HADOOP.COM";
};
Client {
   com.sun.security.auth.module.Krb5LoginModule required
   useKeyTab=true
   storeKey=true
   useTicketCache=false
   keyTab="/etc/kafka/conf/kafka.keytab"
   principal="kafka/local-dn-1.hadoop.com.com@HADOOP.COM";
};
``` `producer.properties` ```
bootstrap.servers=local-dn-1.hadoop.com:9092
security.protocol=SASL_PLAINTEXT
sasl.kerberos.service.name="kafka"
sasl.mechanism = GSSAPI

我用来启动生产者的命令是：

/opt/cloudera/parcels/KAFKA/bin/kafka-console-producer --broker-list local-dn-1.hadoop.com:9092 --topic "Kafka-Test"  --producer.config /etc/kafka/conf/producer-conf/producer.properties