KafkaConsumer API Performance Challenges and Optimization Suggestions ( Kafka clients 3.6.1)

[gmungi]

Body

Scenario:

We are migrating from an old Kafka client (version 0.8.1) to a newer Kafka client (version 3.6.1). The old Kafka implementation used the following approach, which was fast and efficient:

Old Code (Kafka Client 0.8.1):java

FetchRequest req = new FetchRequestBuilder()
    .clientId(clientName)
    .addFetch(a_topic, a_partition, readOffset, fetchSize)
    .build();
FetchResponse fetchResponse = consumer.fetch(req);
ByteBufferMessageSet set = fetchResponse.messageSet(a_topic, a_partition);
for (MessageAndOffset messageAndOffset : set) {
ByteBuffer payload = messageAndOffset.message().payload();
byte[] bytes = new byte[payload.limit()];
payload.get(bytes);
message = new String(bytes, "UTF-8");
records.add(message)				
}

This approach is super fast, and we are trying to replicate this performance using the KafkaConsumer API in the newer Kafka client. However, we are encountering slowness, especially in fetching the first 500 records.

New Code (Kafka Client 3.6.1):

public static KafkaConsumer<String,String> createConsumer(String clientName,int fetchSize) {
			Properties props = new Properties();
			String kafkaBrokerStr = Config.getConsumerPropValue("kafkabrokerslist");
			String groupId = Config.getConsumerPropValue("group.id");
	        props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
	        props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
	        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);
	        props.setProperty(ConsumerConfig.CLIENT_ID_CONFIG,clientName);
	        props.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
	        props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
			props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, "1024");
			props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,"500");
			props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, "500");
			return new KafkaConsumer<String,String>(props);
   }

TopicPartition topicPartition = new TopicPartition("Test", 0);
consumer.seekToBeginning(Collections.singletonList(topicPartition));
long kafkaEarliestOffset = consumer.position(topicPartition);

try (KafkaConsumer<String, String> consumer = KafkaConsumerFactory.createConsumer(clientName, fetchSize)) {
    consumer.partitionsFor(topicName);
    consumer.assign(Collections.singletonList(topicPartition));
    consumer.seek(topicPartition, readOffset);
    
    do {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1500));
    } while (!end);
}

Observed Issue:

Slowness in First Poll: The first fetch of 500 records takes significantly longer (e.g., 1284 ms) compared to subsequent fetches (e.g., 4 ms).
Kafka Consumer Initialization: Operations like assign(), seek(), and fetching the earliest offset are taking time, and this impacts the performance.

Objective:
We need to reduce the time taken for the first poll (the first 500 records).
We are also looking for ways to efficiently perform assign(), seek(), and get the earliest offset in Kafka, without introducing unnecessary delays.

Please provide guidance on these areas to replicate the performance of the old Kafka implementation using the newer Kafka Consumer API.

Guidelines

• I have read and understood this category's guidelines before making this post.

[pwoltschk]

The performance difference you're experiencing is due to architectural changes in the Kafka client API from version 0.8.1 to 3.6.1. The newer KafkaConsumer introduces additional overhead, such as group coordination and metadata fetching, which can slow down the initial poll.

1. Disable Consumer Group Coordination:

   • Set the group.id to an empty string to prevent the consumer from joining a group, eliminating unnecessary overhead.

     props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "");

2. Optimize Consumer Configurations:

   • Increase fetch.min.bytes: Allows the broker to send larger batches, reducing network roundtrips.

     props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, "1048576"); // 1 MB

   • Adjust fetch.max.wait.ms: Lowering this value reduces the wait time for data retrieval.

     props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, "100");

   • Set max.partition.fetch.bytes: Ensures sufficient data per partition is fetched.

     props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, "1048576"); // 1 MB

3. Minimize Metadata Overhead:

   • Avoid Unnecessary Calls: Remove seekToBeginning() and partitionsFor() if not needed.
   • Fetch Earliest Offset Without Seeking:

     Map<TopicPartition, Long> offsets = consumer.beginningOffsets(Collections.singletonList(topicPartition));
     long earliestOffset = offsets.get(topicPartition);
     consumer.seek(topicPartition, readOffset);

   • Perform a Warm-Up Poll: Initiates metadata fetching without delay.

     consumer.poll(Duration.ofMillis(0));

4. Streamline Polling:

   • Use a single poll() call with an appropriate timeout instead of looping.

     ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));

[gmungi]

@pwoltschk Thankyou for your reply. Please help.

I have tried setting groupid to "" and rebalancing not happening .After warming up also the first poll taking time. I am also seeing one strange issue:

My kafka topic which have one partition contains some 5 lakh records

The starting records taking time to fetch from kafka.

For example if i give "0" as input offset and it is taking time as below (6 seconds) and not getting 500 records also,it is getting only 200 records per poll and taking lot of time...why this is happening and how to avoid this.

Poll Records Count :292 Time taken :1227 ms
Poll Records Count :292 Time taken :1181 ms
Poll Records Count :296 Time taken:1234 ms
Poll Records Count :292 Time taken:1133 ms

If i give offset as 110999 and it is getting some fast and records getting as 500 each..Why this difference please.

Poll Records Count :500 Time taken:1278 ms
Poll Records Count :500 Time taken:3 ms

After applying all the steps and warming up consumer ,still the initial poll issue is not resolved
Why the starting records (offset "0" etc) taking time.
How to avoid this please give your suggestion.

My requirement is i need to collect 1000 records and give to the user as response.that's why iam using loop.I am breaking out from loop after collecting 1000 records.

This is my updated properties and code :

try (KafkaConsumer<String,String> consumer = KafkaConsumerFactory.createConsumer()){
TopicPartition topicPartition = new TopicPartition(topicName, 0);
consumer.assign(Collections.singletonList(topicPartition));
Map<TopicPartition, Long> offsets = consumer.beginningOffsets(Collections.singletonList(topicPartition));
long kafkaEarliestOffset = offsets.get(topicPartition);
consumer.seek(topicPartition, readOffset);
consumer.poll(Duration.ofMillis(0));//warming up

do {
ConsumerRecords < String, String > records = consumer.poll(Duration.ofMillis(1500));
if (msglist.size() >= Math.round(limit
/ inputReq.getApplicationArea().getReqInfo().size())
|| (endTime - startTime) >= waitTime) {
end = true;
consumer.commitSync();
}
} while (!end);

ConsumerConfig values:
allow.auto.create.topics = true
auto.commit.interval.ms = 5000
auto.include.jmx.reporter = true
auto.offset.reset = earliest
bootstrap.servers = [ngkaf-lt1-01.cisco.com:9092, ngkaf-lt1-02.cisco.com:9092, ngkaf-lt1-03.cisco.com:9092, ngkaf-lt2-01.cisco.com:9092, ngkaf-lt2-02.cisco.com:9092, ngkaf-lt2-03.cisco.com:9092]
check.crcs = true
client.dns.lookup = use_all_dns_ips
client.id = consumer--13
client.rack =
connections.max.idle.ms = 540000
default.api.timeout.ms = 60000
enable.auto.commit = false
exclude.internal.topics = true
fetch.max.bytes = 52428800
fetch.max.wait.ms = 100
fetch.min.bytes = 1048576
group.id =
group.instance.id = null
heartbeat.interval.ms = 3000
interceptor.classes = []
internal.leave.group.on.close = true
internal.throw.on.fetch.stable.offset.unsupported = false
isolation.level = read_uncommitted
key.deserializer = class org.apache.kafka.common.serialization.StringDeserializer
max.partition.fetch.bytes = 1048576
max.poll.interval.ms = 300000
max.poll.records = 500
metadata.max.age.ms = 300000
metric.reporters = []
metrics.num.samples = 2
metrics.recording.level = INFO
metrics.sample.window.ms = 30000
partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor, class org.apache.kafka.clients.consumer.CooperativeStickyAssignor]
receive.buffer.bytes = 65536
reconnect.backoff.max.ms = 1000
reconnect.backoff.ms = 50
request.timeout.ms = 30000
retry.backoff.ms = 100
sasl.client.callback.handler.class = null
sasl.jaas.config = [hidden]
sasl.kerberos.kinit.cmd = /usr/bin/kinit
sasl.kerberos.min.time.before.relogin = 60000
sasl.kerberos.service.name = null
sasl.kerberos.ticket.renew.jitter = 0.05
sasl.kerberos.ticket.renew.window.factor = 0.8
sasl.login.callback.handler.class = null
sasl.login.class = null
sasl.login.connect.timeout.ms = null
sasl.login.read.timeout.ms = null
sasl.login.refresh.buffer.seconds = 300
sasl.login.refresh.min.period.seconds = 60
sasl.login.refresh.window.factor = 0.8
sasl.login.refresh.window.jitter = 0.05
sasl.login.retry.backoff.max.ms = 10000
sasl.login.retry.backoff.ms = 100
sasl.mechanism = PLAIN
sasl.oauthbearer.clock.skew.seconds = 30
sasl.oauthbearer.expected.audience = null
sasl.oauthbearer.expected.issuer = null
sasl.oauthbearer.jwks.endpoint.refresh.ms = 3600000
sasl.oauthbearer.jwks.endpoint.retry.backoff.max.ms = 10000
sasl.oauthbearer.jwks.endpoint.retry.backoff.ms = 100
sasl.oauthbearer.jwks.endpoint.url = null
sasl.oauthbearer.scope.claim.name = scope
sasl.oauthbearer.sub.claim.name = sub
sasl.oauthbearer.token.endpoint.url = null
security.protocol = SASL_SSL
security.providers = null
send.buffer.bytes = 131072
session.timeout.ms = 45000
socket.connection.setup.timeout.max.ms = 30000
socket.connection.setup.timeout.ms = 10000
ssl.cipher.suites = null
ssl.enabled.protocols = [TLSv1.2, TLSv1.3]
ssl.endpoint.identification.algorithm = https
ssl.engine.factory.class = null
ssl.key.password = null
ssl.keymanager.algorithm = SunX509
ssl.keystore.certificate.chain = null
ssl.keystore.key = null
ssl.keystore.location = null
ssl.keystore.password = null
ssl.keystore.type = JKS
ssl.protocol = TLSv1.3
ssl.provider = null
ssl.secure.random.implementation = null
ssl.trustmanager.algorithm = PKIX
ssl.truststore.certificates = null
ssl.truststore.location = null
ssl.truststore.password = null
ssl.truststore.type = JKS
value.deserializer = class org.apache.kafka.common.serialization.StringDeserializer

Please do help on this.