Forum Discussion

Prabaharan's avatar
Prabaharan
Contributor
5 years ago
Solved

Need groovy script to read messages from kafka topic

Hi Team,

 

I have written code in groovy script test step to post messages in kafka topic. Now I'm trying to read messages from the kafka topic. All the code samples in online have an infinite loop in it inorder to read messages from kafka topic. But I would need a groovy code to read kafka messages and the test step should exit once all the messages are read, i.e.., it should not have an infinite loop.

Appreciate if anyone can help me in resolving this.

 

 

Scripting
  • TSanchez_1's avatar
    TSanchez_1
    3 years ago

    Hello, First off thank you for posting your question in the community. I do have a question, why are you using groovy to publish the message instead of using the Kafka publish test step?

3 Replies

Sort By
  • Prabaharan's avatar
    Prabaharan
    Contributor
    5 years ago

    Hi Team,

     

    Below is the code I'm using for now.

    String topic = "sample";

    Properties props = new Properties();
    props.put("bootstrap.servers", "localhost:9092");
    props.put("group.id", "test");
    props.put("enable.auto.commit", "true");
    props.put("auto.commit.interval.ms", "1000");
    props.put("session.timeout.ms", "30000");
    props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);

    consumer.subscribe(Arrays.asList(topic));
    log.info("Subscribed to topic " + topic);

    long t = System.currentTimeMillis();
    long end = t + 9000;
    while (System.currentTimeMillis()<end)
    {
    ConsumerRecords<String, String> records = consumer.poll(9000);
    for (ConsumerRecord<String, String> record : records)
    {
    log.info ( "offset = " + record.offset() +" value = " + record.value());
    }
    consumer.commitSync();
    }
    consumer.close();

     

    The test step neither returns error message nor returns the kafka mewssage. instead iot executes for sometimes and exits without returning anything.

    Can anyone help me in resolving this query? Appreciate your help in advance.

    • dvictor's avatar
      dvictor
      Occasional Contributor
      4 years ago

      I am experiencing the same thing. What I noticed is that props is staying null and I can't figure out how to work around it or fix it

  • TSanchez_1's avatar
    TSanchez_1
    Moderator
    3 years ago

    Hello, First off thank you for posting your question in the community. I do have a question, why are you using groovy to publish the message instead of using the Kafka publish test step?