我有一段java代码试图向kafka队列添加消息

String msgID = UUID.randomUUID().toString();

    Properties prop = new Properties();
    prop.put("metadata.broker.list", DEFAULT_BROKER);
    prop.put("serializer.class", "kafka.serializer.StringEncoder");
    prop.put("request.required.acks", "-1");
    prop.put("producer.type", "async");

    ProducerConfig config = new ProducerConfig(prop);
    Producer<String, String> producer = new Producer<String, String>(config);

    KeyedMessage<String, String> message = new KeyedMessage<String, String>(TOPIC_NAME, msgID, "stackoverflow");

    try {
        producer.send(message);
        LOG.info("Messages sent with key" + msgID);
    } catch (Exception exception) {

    }

    producer.close();

kafka代理是由kerberos保护的远程linux机器
当我用eclispe在本地windows机器上执行上述java代码时，我得到以下日志输出

2016-01-06 15:48:57 INFO  VerifiableProperties:68 - Verifying properties
2016-01-06 15:48:57 INFO  VerifiableProperties:68 - Property metadata.broker.list is overridden to remote000001.machine.test.group:6667,remote000002.machine.test.group:6667,remote000003.machine.test.group:6667
2016-01-06 15:48:57 INFO  VerifiableProperties:68 - Property producer.type is overridden to async
2016-01-06 15:48:57 INFO  VerifiableProperties:68 - Property request.required.acks is overridden to -1
2016-01-06 15:48:57 INFO  VerifiableProperties:68 - Property serializer.class is overridden to kafka.serializer.StringEncoder
2016-01-06 15:48:58 TRACE Producer:36 - Added to send queue an event: KeyedMessage(stackoverflowTopic,2f0c8b29-46e0-4860-961b-0fea480d21f9,2f0c8b29-46e0-4860-961b-0fea480d21f9,stackoverflow)
2016-01-06 15:48:58 TRACE Producer:36 - Remaining queue size: 10000
2016-01-06 15:48:58 INFO  KafkaPro:63 - Messages sent with key2f0c8b29-46e0-4860-961b-0fea480d21f9
2016-01-06 15:48:58 INFO  Producer:68 - Shutting down producer
2016-01-06 15:48:58 TRACE ProducerSendThread:36 - Dequeued item for topic stackoverflowTopic, partition key: 2f0c8b29-46e0-4860-961b-0fea480d21f9, data: stackoverflow
2016-01-06 15:48:58 INFO  ProducerSendThread:68 - Begin shutting down ProducerSendThread
2016-01-06 15:48:58 DEBUG ProducerSendThread:52 - Handling 1 events
2016-01-06 15:48:58 DEBUG DefaultEventHandler:52 - Handling 1 events
2016-01-06 15:48:58 TRACE SyncProducer:36 - Instantiating Scala Sync Producer with properties: {metadata.broker.list=remote000001.machine.test.group:6667,remote000002.machine.test.group:6667,remote000003.machine.test.group:6667, request.required.acks=-1, port=6667, serializer.class=kafka.serializer.StringEncoder, host=remote000002.machine.test.group, producer.type=async}
2016-01-06 15:48:58 INFO  ClientUtils$:68 - Fetching metadata from broker id:1,host:remote000002.machine.test.group,port:6667 with correlation id 0 for 1 topic(s) Set(stackoverflowTopic)
2016-01-06 15:48:58 TRACE SyncProducer:36 - verifying sendbuffer of size 34
2016-01-06 15:48:58 INFO  SyncProducer:68 - Connected to remote000002.machine.test.group:6667 for producing
2016-01-06 15:48:58 INFO  SyncProducer:68 - Disconnecting from remote000002.machine.test.group:6667
2016-01-06 15:48:58 WARN  ClientUtils$:89 - Fetching topic metadata with correlation id 0 for topics [Set(stackoverflowTopic)] from broker [id:1,host:remote000002.machine.test.group,port:6667] failed
java.nio.channels.ClosedChannelException
    at kafka.network.BlockingChannel.send(BlockingChannel.scala:100)
    at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:73)
    at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:72)
    at kafka.producer.SyncProducer.send(SyncProducer.scala:113)
    at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:58)
    at kafka.producer.BrokerPartitionInfo.updateInfo(BrokerPartitionInfo.scala:82)
    at kafka.producer.async.DefaultEventHandler$$anonfun$handle$1.apply$mcV$sp(DefaultEventHandler.scala:67)
    at kafka.utils.Utils$.swallow(Utils.scala:172)
    at kafka.utils.Logging$class.swallowError(Logging.scala:106)
    at kafka.utils.Utils$.swallowError(Utils.scala:45)
    at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:67)
    at kafka.producer.async.ProducerSendThread.tryToHandle(ProducerSendThread.scala:105)
    at kafka.producer.async.ProducerSendThread.processEvents(ProducerSendThread.scala:94)
    at kafka.producer.async.ProducerSendThread.run(ProducerSendThread.scala:45)
2016-01-06 15:48:58 INFO  SyncProducer:68 - Disconnecting from remote000002.machine.test.group:6667
2016-01-06 15:48:58 TRACE SyncProducer:36 - Instantiating Scala Sync Producer with properties: {metadata.broker.list=remote000001.machine.test.group:6667,remote000002.machine.test.group:6667,remote000003.machine.test.group:6667, request.required.acks=-1, port=6667, serializer.class=kafka.serializer.StringEncoder, host=remote000001.machine.test.group, producer.type=async}
2016-01-06 15:48:58 INFO  ClientUtils$:68 - Fetching metadata from broker id:0,host:remote000001.machine.test.group,port:6667 with correlation id 0 for 1 topic(s) Set(stackoverflowTopic)
2016-01-06 15:48:58 TRACE SyncProducer:36 - verifying sendbuffer of size 34
2016-01-06 15:48:58 INFO  SyncProducer:68 - Connected to remote000001.machine.test.group:6667 for producing
2016-01-06 15:48:58 INFO  SyncProducer:68 - Disconnecting from remote000001.machine.test.group:6667
2016-01-06 15:48:58 WARN  ClientUtils$:89 - Fetching topic metadata with correlation id 0 for topics [Set(stackoverflowTopic)] from broker [id:0,host:remote000001.machine.test.group,port:6667] failed

我创建了一个kafka\u jaas.conf文件，包含以下内容

KafkaServer {
   com.sun.security.auth.module.Krb5LoginModule required
   useKeyTab=true
   keyTab="/etc/security/keytabs/kafka.service.keytab"
   storeKey=true
   useTicketCache=false
   serviceName="kafka"
   principal="myPrincipal";
};
KafkaClient {
   com.sun.security.auth.module.Krb5LoginModule required
   useTicketCache=true
   renewTicket=true
   serviceName="kafka";
};
Client {
   com.sun.security.auth.module.Krb5LoginModule required
   useKeyTab=true
   keyTab="/etc/security/keytabs/kafka.service.keytab"
   storeKey=true
   useTicketCache=false
   serviceName="zookeeper"
   principal="myPrincipal";
};

并将其位置作为jvm参数传递

-Djava.security.auth.login.config=C:\\kafka_jaas.conf

为什么这种方法不起作用？
如何执行kafka procuder，该procuder正在向kerberos保护的代理写入消息。