kafka.network.blockingchannel.send（blockingchannel）的线程“main”java.nio.channels.closedchannelexception中出现异常。scala:100)在kafka.consumer.simpleconsumer.liftedtree1$1（simpleconsumer。scala:78)在kafka.consumer.simpleconsumer.kafka$consumer$simpleconsumer$$sendrequest（simpleconsumer。scala:68)在kafka.consumer.simpleconsumer.send（simpleconsumer。scala:91)在kafka.javaapi.consumer.simpleconsumer.send（simpleconsumer。scala:68)在cmb.sparkstream.kafka.kafkaoffsettool.gettopicofsets（kafkaoffsettool。java:47)在cmb.sparkstream.logclassify.main（logclassify。java:95)位于sun.reflect.nativemethodaccessorimpl.invoke0（本机方法）sun.reflect.nativemethodaccessorimpl.invoke（nativemethodaccessorimpl。java:57)在sun.reflect.delegatingmethodaccessorimpl.invoke（delegatingmethodaccessorimpl。java:43)在java.lang.reflect.method.invoke（方法。java:606)在org.apache.spark.deploy.sparksubmit$.org$apache$spark$deploy$sparksubmit$$runmain（sparksubmit）。scala:729)在org.apache.spark.deploy.sparksubmit$.dorunmain$1（sparksubmit。scala:185)在org.apache.spark.deploy.sparksubmit$.submit（sparksubmit。scala:210)位于org.apache.spark.deploy.sparksubmit$.main（sparksubmit.com）。scala:124)位于org.apache.spark.deploy.sparksubmit.main（sparksubmit.scala）
我的代码是：

public static Map<TopicAndPartition, Long> getTopicOffsets(String zkServers, String topic) {

  Map<TopicAndPartition, Long> retVals = new HashMap<TopicAndPartition, Long>();
  for (String zkserver : zkServers.split(",")) {
   SimpleConsumer simpleConsumer = new SimpleConsumer(zkserver.split(":")[0],
     Integer.valueOf(zkserver.split(":")[1]), Consts.getKafkaConfigBean().getDuration(), 1024,
     "consumser");
   TopicMetadataRequest topicMetadataRequest = new TopicMetadataRequest(Arrays.asList(topic));

   TopicMetadataResponse topicMetadataResponse = simpleConsumer.send(topicMetadataRequest);

   for (TopicMetadata metadata : topicMetadataResponse.topicsMetadata()) {
    for (PartitionMetadata part : metadata.partitionsMetadata()) {
     Broker leader = part.leader();
     if (leader != null) {
      TopicAndPartition topicAndPartition = new TopicAndPartition(topic, part.partitionId());

      PartitionOffsetRequestInfo partitionOffsetRequestInfo = new PartitionOffsetRequestInfo(
        kafka.api.OffsetRequest.LatestTime(), 10000);
      OffsetRequest offsetRequest = new OffsetRequest(
        ImmutableMap.of(topicAndPartition, partitionOffsetRequestInfo),
        kafka.api.OffsetRequest.CurrentVersion(), simpleConsumer.clientId());
      OffsetResponse offsetResponse = simpleConsumer.getOffsetsBefore(offsetRequest);
      if (!offsetResponse.hasError()) {
       long[] offsets = offsetResponse.offsets(topic, part.partitionId());
       retVals.put(topicAndPartition, offsets[0]);
      }
     }

    }

   }
   simpleConsumer.close();
  }
  return retVals;
 }