Unable the consume messages from cloud kafka broker

[vamsidarbhamulla]

Hi,

I am trying to publish/consume messages from a Kafka cluster from a confluent cloud deployed and maintained by a different team to build a load test script using xk6-kafka.

I'm able to produce messages successfully but observe timeout errors while consuming messages from any of the topics.

I can publish/subscribe messages using the Kafka java client successfully to confirm I have the right configuration.

I did follow all the suggestions from the below two closed issues but still not successful:

1. Q: Unable to read messages #180
2. Unable to send messages to the kafka topics #185

reader script snippet:

import { Reader } from "k6/x/kafka";

import { kafkaBroker, saslConfig, tlsConfig } from "../libs/kafka_config.js";

export const options = { 
    vus: 1,
    duration : "1m",
  };

const topic = "test.topic";

const reader = new Reader({
    brokers: [kafkaBroker()],
    sasl: saslConfig(),
    tls: tlsConfig(),
    groupID: "test-1",
    groupTopics: [topic],
    heartbeatInterval: 200,
    maxWait:900000 ,
    readBatchTimeout:3000000,
    connectLogger: true,
    // joinGroupBackoff: 1000,
    // offset: offset,
    // topic: topic,
    // partition: partition,
    // startOffset: "38",
    // maxWait:1000,
    // sessionTimeout: 10,
    // rebalanceTimeout: 900000,
    // retentionTimeout: 900000,
    // readBatchTimeout:1000,   
});

export default function () {
    let messages = reader.consume({ limit: 2 });
    console.log(messages);
}

export function teardown(data) {
    reader.close();
}

[vamsidarbhamulla]

Below is the log:

INFO[0002] subscribed to topics and partitions: map[]   
INFO[0002] started commit for group test-1              terations
INFO[0002] subscribed to topics and partitions: map[{topic:test.topic partition:0}:-2] 
INFO[0002] started commit for group test-1              
INFO[0002] started heartbeat for group, test-1 [200ns]  
INFO[0002] started heartbeat for group, test-1 [200ns]  
INFO[0002] initializing kafka reader for partition 0 of test.topic starting at offset first offset 
INFO[0003] the kafka reader for partition 0 of test.topic is seeking to offset 37 
INFO[0003] the kafka reader got an unknown error reading partition 0 of test.topic at offset 37: read tcp host:9092: i/o timeout ed iterations
INFO[0003] initializing kafka reader for partition 0 of test.topic starting at offset 37 
INFO[0040] the kafka reader got an unknown error reading partition 0 of test.topic at offset 37: read tcp host:9092: i/o timeout ed iterations
INFO[0040] initializing kafka reader for partition 0 of test.topic starting at offset 37 
ERRO[0040] Unable to read messages.                      error="Unable to read messages."

[mostafa]

Hey @vamsidarbhamulla,

Can you try the timeout values mentioned here?
#197 (comment)

[vamsidarbhamulla]

Hi @mostafa - Thanks for the quick response. It did work after updating the session timeout.

The delay in responding is due to the reasons

• To find the Kafka cluster min and max session timeout values.
• I assumed the timeout values default to milliseconds in the script.

I still observe the reader consumes data only on certain attempts and the rest of the time I see the unable-to-read-data error might be with the fact I'm running the reader as a separate process and also for a small duration as of now. I will try to find the optimal approach. Thanks again.