NoSuchElementException kafka

Question

我是kafka的新手。我阅读了文档来开始使用，现在我正在尝试使用嵌入式kafka模式进行动手操作。我尝试了一个示例程序来进行同样的操作。

 public static void main(String args[]) throws InterruptedException, IOException {

    // setup Zookeeper
    EmbeddedZookeeper zkServer = new EmbeddedZookeeper();
    String zkConnect = ZKHOST + ":" + zkServer.port();
    ZkClient zkClient = new ZkClient(zkConnect, 30000, 30000, ZKStringSerializer$.MODULE$);
    ZkUtils zkUtils = ZkUtils.apply(zkClient, false);

    // setup Broker
    Properties brokerProps = new Properties();
    brokerProps.setProperty("zookeeper.connect", zkConnect);
    brokerProps.setProperty("broker.id", "0");
    brokerProps.setProperty("log.dirs", Files.createTempDirectory("kafka-").toAbsolutePath().toString());
    brokerProps.setProperty("listeners", "PLAINTEXT://" + BROKERHOST +":" + BROKERPORT);
    KafkaConfig config = new KafkaConfig(brokerProps);
    Time mock = new MockTime();
    KafkaServer kafkaServer = TestUtils.createServer(config, mock);

    // create topic
    AdminUtils.createTopic(zkUtils, TOPIC, 1, 1, new Properties(), RackAwareMode.Disabled$.MODULE$);

    // setup producer
    Properties producerProps = new Properties();
    producerProps.setProperty("bootstrap.servers", BROKERHOST + ":" + BROKERPORT);
    producerProps.setProperty("key.serializer","org.apache.kafka.common.serialization.IntegerSerializer");
    producerProps.setProperty("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
    KafkaProducer<Integer, byte[]> producer = new KafkaProducer<Integer, byte[]>(producerProps);
    List<PartitionInfo> partitionInfo = producer.partitionsFor("test");
    System.out.println(partitionInfo);
    // setup consumer
    Properties consumerProps = new Properties();
    consumerProps.setProperty("bootstrap.servers", BROKERHOST + ":" + BROKERPORT);
    consumerProps.setProperty("group.id", "group0");
    consumerProps.setProperty("client.id", "consumer0");
    consumerProps.setProperty("key.deserializer","org.apache.kafka.common.serialization.IntegerDeserializer");
    consumerProps.setProperty("value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");
    consumerProps.put("auto.offset.reset", "earliest");  // to make sure the consumer starts from the beginning of the topic
    KafkaConsumer<Integer, byte[]> consumer = new KafkaConsumer<>(consumerProps);
    consumer.subscribe(Arrays.asList(TOPIC));

    // send message
    ProducerRecord<Integer, byte[]> data = new ProducerRecord<>(TOPIC, 42, "test-message".getBytes(StandardCharsets.UTF_8));
    producer.send(data);
    producer.close();

    // starting consumer
    ConsumerRecords<Integer, byte[]> records = consumer.poll(1000);

    Iterator<ConsumerRecord<Integer, byte[]>> recordIterator = records.iterator();
    ConsumerRecord<Integer, byte[]> record = recordIterator.next();
    System.out.printf("offset = %d, key = %s, value = %s", record.offset(), record.key(), record.value());


    kafkaServer.shutdown();
    zkClient.close();
    zkServer.shutdown();
  }

}

但我无法为topics.Iam获取数据，在执行程序时出现以下异常

java.util.NoSuchElementException
at      org.apache.kafka.common.utils.AbstractIterator.next(AbstractIterator.java:52)
at com.nuwaza.evlauation.embedded.kafka.EmbeddedKafka.main(EmbeddedKafka.java:105)

有人能给我指路吗？

更新-

 WARN [main] (Logging.scala#warn:83) - No meta.properties file under dir    C:\Users\bhavanak\AppData\Local\Temp\kafka-1238324273778000675\meta.properties
 WARN [main] (Logging.scala#warn:83) - No meta.properties file under dir C:\Users\bhavanak\AppData\Local\Temp\kafka-1238324273778000675\meta.properties
 WARN [kafka-producer-network-thread | producer-1] (NetworkClient.java#handleResponse:600) - Error while fetching metadata with correlation id 0 : {test=LEADER_NOT_AVAILABLE}
WARN [kafka-producer-network-thread | producer-1] (NetworkClient.java#handleResponse:600) - Error while fetching metadata with correlation id 1 : {test=LEADER_NOT_AVAILABLE}
WARN [kafka-producer-network-thread | producer-1]  (NetworkClient.java#handleResponse:600) - Error while fetching metadata with  correlation id 2 : {test=LEADER_NOT_AVAILABLE}
  [Partition(topic = test, partition = 0, leader = 0, replicas = [0,], isr = [0,]]
 ERROR [main] (NIOServerCnxnFactory.java#uncaughtException:44) - Thread  Thread[main,5,main] died
java.util.NoSuchElementException
at     org.apache.kafka.common.utils.AbstractIterator.next(AbstractIterator.java:52)
at    com.nuwaza.evlauation.embedded.kafka.EmbeddedKafka.main(EmbeddedKafka.java:105)

Answer 1

在读取消息之前尝试调用producer.flush()，以确保生成的消息确实保存在磁盘上。

Answer 2

此错误表示您的消费者甚至在消息被持久化到kafka日志之前就试图读取该消息。理想情况下，您应该将生产者和消费者作为单独的进程运行。我也面临着同样的问题，但这是由于其他原因，iterator.next()被错误地调用了两次。以防其他人遇到同样的问题。