问题似乎是Kafka的错误，我没有看到报告。如果使用同一个consumerconnector创建多个consumeriterator（通过为其提供多个主题的Map），则这些主题会在consumeriterator中频繁切换。如果您试图查看consumeriterator，通过在调试器中暂停，它们会切换回。
下面是我的旧代码，用于创建存在错误的consumeriterators：

/**
 * @param zookeeperAddresses (includes the port number)
 * @param topics all topics to be consumed.
 * @return A list of ConsumerIterators.
 */
public List<ConsumerIterator> getConsumers(String zookeeperAddresses, List<String> topics) {
    String groupId = "client_" + topics.get(0);
    LOGGER.info("Zookeeper address = " + zookeeperAddresses + ", group id = " + groupId);
    ConsumerConnector consumer = kafka.consumer.Consumer.createJavaConsumerConnector(
            createConsumerConfig(zookeeperAddresses, groupId));
    consumers.add(consumer);
    Map<String, Integer> topicCountMap = new HashMap<>();
    for (String topic : topics) {
        topicCountMap.put(topic, Integer.valueOf(1));
    }
    Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap);
    List<ConsumerIterator> topicConsumers = new LinkedList<>();
    for (String topic : topics) {
        List<KafkaStream<byte[], byte[]>> streams = consumerMap.get(topic);
        assert(streams.size() == 1);
        ConsumerIterator<byte[], byte[]> consumerIterator = streams.get(0).iterator();
        topicConsumers.add(consumerIterator);
    }
    return topicConsumers;
}

下面是解决此错误的固定代码：

/**
 * @param zookeeperAddresses (includes the port number)
 * @param topics all topics to be consumed.
 * @return A list of ConsumerIterators.
 */
public List<ConsumerIterator> getConsumers(String zookeeperAddresses, List<String> topics) {
    String groupId = "client_" + topics.get(0);
    LOGGER.info("Zookeeper address = " + zookeeperAddresses + ", group id = " + groupId);
    List<ConsumerIterator> topicConsumers = new LinkedList<>();
    for (String topic : topics) {
        ConsumerConnector consumer = kafka.consumer.Consumer.createJavaConsumerConnector(
                createConsumerConfig(zookeeperAddresses, groupId));
        consumers.add(consumer);
        Map<String, Integer> topicCountMap = new HashMap<>();
        topicCountMap.put(topic, Integer.valueOf(1));
        Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap);
        List<KafkaStream<byte[], byte[]>> streams = consumerMap.get(topic);
        assert(streams.size() == 1);
        ConsumerIterator<byte[], byte[]> consumerIterator = streams.get(0).iterator();
        topicConsumers.add(consumerIterator);
    }
    return topicConsumers;
}