CODEWITHSUNDEEP
javaapache-kafka
bhavanak
bhavanak

Reputation: 255

NoSuchElementException kafka

I am new to kafka.I read documentation to get started and now I am trying to do hands on using embedded kafka mode.I tried a sample program for the same.

 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();
  }

}

but Iam not able fetch data for the topics.Iam getting the following exception while executing the programm

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)

Can anyone guide me?

UPDATED-

 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)

Upvotes: 4

Views: 2208

Answers (2)

Gaurav Khare
Gaurav Khare

Reputation: 2327

This error signifies that your consumer is trying to read the message even before it gets persisted to kafka logs. Ideally you should run producer and consumer as separate process. I was facing the same issue but that was due to other reason being iterator.next() was invoked twice mistakenly. Just in case someone else facing same issue.

Upvotes: 0

amethystic
amethystic

Reputation: 7089

Try to invoke producer.flush() before reading messages to ensure the produced messages indeed are persisted on disks.

Upvotes: 1

Related Questions