kafka GSSHeader did not find the right tag

kafka 配置好kerberos后可以正常启动，也可以创建topic
但无法往topic里写消息也无法接收
kafka版本： KAFKA-2.0.1-1.2.0.1.p0.5
jdk：jdk1.7.0_67-cloudera

kerberos：krb5-server-1.10.3-57.el6.x86_64
krb5-auth-dialog-0.13-5.el6.x86_64
krb5-workstation-1.10.3-57.el6.x86_64
krb5-devel-1.10.3-57.el6.x86_64
krb5-libs-1.10.3-57.el6.x86_64

系统： centos6.7X64

# klist -e
Ticket cache: FILE:/tmp/krb5cc_0
Default principal: xiaoyu@ATM.COM

Valid starting    Expires          Service principal
05/30/16 14:00:31  05/31/16 14:00:31  krbtgt/ATM.COM@ATM.COM
renew until 06/06/16 14:00:31, Etype (skey, tkt): aes256-cts-hmac-sha1-96, aes256-cts-hmac-sha1-96

JCE文件也是从oracle官网下的 jce7 并替换了对应的文件。重启kdc 、kafka等服务
相关报错如下：
# kafka-console-consumer --topic test --zookeeper hadoop2:2181,hadoop3:2181,hadoop4:2181
[2016-05-30 15:11:24,650] WARN [console-consumer-10846_hadoop2-1464592282973-78a519fe-leader-finder-thread], Failed to find leader for Set([test,0]) (kafka.consumer.ConsumerFetcherManager$LeaderFinderThread)
kafka.common.BrokerEndPointNotAvailableException: End point PLAINTEXT not found for broker 112
at kafka.cluster.Broker.getBrokerEndPoint(Broker.scala:141)
at kafka.utils.ZkUtils$$anonfun$getAllBrokerEndPointsForChannel$1.apply(ZkUtils.scala:171)

…………………………

# kafka-consolproducer  --topic test  --broker-list hadoop4:9092,hadoop3:9092,hadoop2:9092
123
[2016-05-30 15:16:41,993] ERROR Error when sending message to topic test with key: null, value: 3 bytes with error: Failed to update metadata after 60000 ms. (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback)

………………………………
日志出现如下讯息：
2016-05-30 15:16:45,109 DEBUG org.apache.kafka.common.network.Selector: Connection with /172.16.0.6 disconnected
java.io.IOException: javax.security.sasl.SaslException: GSS initiate failed [Caused by GSSException: Defective token detected (Mechanism level: GSSHeader did not find the right tag)]
at org.apache.kafka.common.security.authenticator.SaslServerAuthenticator.authenticate(SaslServerAuthenticator.java:170)
at org.apache.kafka.common.network.KafkaChannel.prepare(KafkaChannel.java:63)
at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:318)
at org.apache.kafka.common.network.Selector.poll(Selector.java:283)
at kafka.network.Processor.run(SocketServer.scala:413)
at java.lang.Thread.run(Thread.java:745)
Caused by: javax.security.sasl.SaslException: GSS initiate failed [Caused by GSSException: Defective token detected (Mechanism level: GSSHeader did not find the right tag)]
at com.sun.security.sasl.gsskerb.GssKrb5Server.evaluateResponse(GssKrb5Server.java:177)
at org.apache.kafka.common.security.authenticator.SaslServerAuthenticator.authenticate(SaslServerAuthenticator.java:164)
... 5 more
Caused by: GSSException: Defective token detected (Mechanism level: GSSHeader did not find the right tag)
at sun.security.jgss.GSSHeader.<init>(GSSHeader.java:97)
at sun.security.jgss.GSSContextImpl.acceptSecContext(GSSContextImpl.java:306)
at sun.security.jgss.GSSContextImpl.acceptSecContext(GSSContextImpl.java:285)
at com.sun.security.sasl.gsskerb.GssKrb5Server.evaluateResponse(GssKrb5Server.java:155)
... 6 more

#####################

kerberos的相关配置如下：
krb5.conf
[logging]
  default = FILE:/var/log/krb5libs.log
  kdc = FILE:/var/log/krb5kdc.log
  admin_server = FILE:/var/log/kadmind.log

[libdefaults]
  default_realm = ATM.COM
  dns_lookup_realm = false
  dns_lookup_kdc = false
  ticket_lifetime = 24h
  renew_lifetime = 7d
  forwardable = true

[realms]
  ATM.COM = {
  kdc = hadoop5
  admin_server = hadoop5
}

[domain_realm]
  hadoop5 = ATM.COM
  hadoop4 = ATM.COM
  hadoop2 = ATM.COM
  hadoop3 = ATM.COM

# cat /var/kerberos/krb5kdc/kdc.conf
[kdcdefaults]
kdc_ports = 88
kdc_tcp_ports = 88

[realms]
ATM.COM = {
  #master_key_type = aes256-cts
  max_life = 1d
  max_renewable_life = 7d
  acl_file = /var/kerberos/krb5kdc/kadm5.acl
  dict_file = /usr/share/dict/words
  admin_keytab = /var/kerberos/krb5kdc/kadm5.keytab
  supported_enctypes = aes256-cts:normal aes128-cts:normal des3-hmac-sha1:normal arcfour-hmac:normal des-hmac-sha1:normal des-cbc-md5:normal des-cbc-crc:normal
  default_principal_flags = +renewable, +forwardable
}