Parallel Consumer does not get records from poll after OFFSET_OUT_OF_RANGE in auto.offset.reset = earliest

[jikeshdpatel]

Hi @astubbs , this is a great library and improves the performance drastically. However, I am seeing some strange issue. I feel like it is because of the auto_offset_reset = earliest.

• Populated events in topic
• Started parallel consumer with auto_offset_reset = earliest and it started consuming from beginning.
• Stopped the process let's say at offset 5000, at partition 3 and PC was behind 1000 events. Similarly stats for other 5 partitions.
• Waited until the topic's retention period.
• Produced more records. waited for retention period and then produced more records.
• Started PC and it only consumes from a partition 5 (last partition). I am not sure whether PC was lagging behind on partition 5.
• I have started a spring kafka listener (with auto_offset_reset = earliest) for partition 3 and it started consuming
• Stopped spring kafka and started PC (with auto_offset_reset = earliest) and it now consumes from partitions 5 and 3
• Stopped PC and started it with auto_offset_reset = latest and now it starts consuming from all the partitions.

The processing order was based on KEY.

Can you please advise on how should I further debug it ?

Thank you for your help!

Below are the config:
2022-07-14 17:17:32.036 INFO [main] [org.apache.kafka.coborker01mmon.config.AbstractConfig.logAll(AbstractConfig.java:372)] - ConsumerConfig values:
allow.auto.create.topics = true
auto.commit.interval.ms = 5000
auto.offset.reset = earliest
bootstrap.servers = [broker01:9092]
check.crcs = true
client.dns.lookup = use_all_dns_ips
client.id = client-01
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 = 500
fetch.min.bytes = 1
group.id = app-group-01
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]
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.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.mechanism = PLAIN
security.protocol = SASL_SSL
security.providers = null
send.buffer.bytes = 131072
session.timeout.ms = 10000
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
2022-07-14 17:17:32.038 DEBUG [main] [org.apache.kafka.clients.consumer.KafkaConsumer.(KafkaConsumer.java:696)] - [Consumer clientId=client-01, groupId=app-group-01] Initializing the Kafka consumer
2022-07-14 17:17:32.135 INFO [main] [org.apache.kafka.common.security.authenticator.AbstractLogin.login(AbstractLogin.java:61)] - Successfully logged in.
2022-07-14 17:17:32.141 DEBUG [main] [org.apache.kafka.common.security.ssl.DefaultSslEngineFactory.createSSLContext(DefaultSslEngineFactory.java:264)] - Created SSL context with keystore null, truststore null, provider SunJSSE.
2022-07-14 17:17:32.192 INFO [main] [org.apache.kafka.common.utils.AppInfoParser$AppInfo.(AppInfoParser.java:119)] - Kafka version: 6.2.1-ccs
2022-07-14 17:17:32.192 INFO [main] [org.apache.kafka.common.utils.AppInfoParser$AppInfo.(AppInfoParser.java:120)] - Kafka commitId: fa4bec046a2df3a6
2022-07-14 17:17:32.192 INFO [main] [org.apache.kafka.common.utils.AppInfoParser$AppInfo.(AppInfoParser.java:121)] - Kafka startTimeMs: 1657833452190
2022-07-14 17:17:32.193 DEBUG [main] [org.apache.kafka.clients.consumer.KafkaConsumer.(KafkaConsumer.java:815)] - [Consumer clientId=client-01, groupId=app-group01] Kafka consumer initialized
2022-07-14 17:17:32.199 INFO [main] [io.confluent.parallelconsumer.internal.AbstractParallelEoSStreamProcessor.(AbstractParallelEoSStreamProcessor.java:231)] - Confluent Parallel Consumer initialise... Options: ParallelConsumerOptions(consumer=org.apache.kafka.clients.consumer.KafkaConsumer@6c42c856, producer=null, managedExecutorService=java:comp/DefaultManagedExecutorService, managedThreadFactory=java:comp/DefaultManagedThreadFactory, ordering=KEY, commitMode=PERIODIC_CONSUMER_ASYNCHRONOUS, maxConcurrency=100, defaultMessageRetryDelay=PT1S, retryDelayProvider=null, sendTimeout=PT10S, offsetCommitTimeout=PT10S, batchSize=1, thresholdForTimeSpendInQueueWarning=PT10S, maxFailureHistory=10)

Below are logs : OFFSET_OUT_OF_RANGE partition, it does reset the offset but does not get any records. However, there are records with the new offset value.

2022-07-14 17:17:39.407 DEBUG [pc-broker-ptest-topic-1oll] [org.apacborker01he.kafka.clients.consumer.internals.Fetcher$1.onSuccess(Fetcher.java:314)] - [Consumer clientId=client-01, groupId=app-group-01] Fetctest-topich READ_UNCOMMITTED at offset 842285 for partition test-topic-1 returned fetch data (error=OFFSET_OUT_OF_RANGE, highWaterMark=-1, lastStableOffset = -1, logStartOffset = -1, preferredReadReplica = absent, abortedTransacticlient-01ons = null, divergingEpoch =Optional.empty, recordsSizeInBytes=0) 2022-07-14 17:17:39.407 INFO [pc-broker-poll] [org.apache.kafka.clients.consumer.internals.Fetcher.handleOffsetOutOfRange(Fetcher.java:1360)] - [Consumer clientId=client-01, groupId=app-group-01] Fetch position FetchPosition{offset=842285, offsetEpoch=Optional.empty, currentLeader=LeaderAndEpoch{leader=Optional[borker01:9092 (id: 21 rack: 0)], epoch=12}} is out of range for partition test-topic-1, resetting offset 2022-07-14 17:17:39.407 DEBUG [pc-broker-poll] [org.apache.kafka.clients.consumer.internals.Fetcher.sendListOffsetRequest(Fetcher.java:983)] - [Consumer clientId=client-01, groupId=app-group-01] Sending ListOffsetRequest ListOffsetsRequestData(replicaId=-1, isolationLevel=0, topics=[ListOffsetsTopic(name='test-topic', partitions=[ListOffsetsPartition(partitionIndex=1, currentLeaderEpoch=12, timestamp=-2, maxNumOffsets=1)])]) to broker borker01:9092 (id: 21 rack: 0) 2022-07-14 17:17:39.407 DEBUG [pc-broker-poll] [org.apache.kafka.clients.NetworkClient.doSend(NetworkClient.java:522)] - [Consumer clientId=client-01, groupId=app-group-01] Sending LIST_OFFSETS request with header RequestHeader(apiKey=LIST_OFFSETS, apiVersion=6, clientId=client-01, correlationId=25) and timeout 30000 to node 21: ListOffsetsRequestData(replicaId=-1, isolationLevel=0, topics=[ListOffsetsTopic(name='test-topic', partitions=[ListOffsetsPartition(partitionIndex=1, currentLeaderEpoch=12, timestamp=-2, maxNumOffsets=1)])]) 2022-07-14 17:17:39.451 DEBUG [pc-broker-poll] [org.apache.kafka.clients.NetworkClient.handleCompletedReceives(NetworkClient.java:880)] - [Consumer clientId=client-01, groupId=app-group-01] Received LIST_OFFSETS response from node 21 for request with header RequestHeader(apiKey=LIST_OFFSETS, apiVersion=6, clientId=client-01, correlationId=25): ListOffsetsResponseData(throttleTimeMs=0, topics=[ListOffsetsTopicResponse(name='test-topic', partitions=[ListOffsetsPartitionResponse(partitionIndex=1, errorCode=0, oldStyleOffsets=[], timestamp=-1, offset=1197212, leaderEpoch=7)])]) 2022-07-14 17:17:39.451 DEBUG [pc-broker-poll] [org.apache.kafka.clients.consumer.internals.Fetcher.handleListOffsetResponse(Fetcher.java:1035)] - [Consumer clientId=client-01, groupId=app-group-01] Handling ListOffsetResponse response for test-topic-1. Fetched offset 1197212, timestamp -1 2022-07-14 17:17:39.451 DEBUG [pc-broker-poll] [org.apache.kafka.clients.Metadata.updateLastSeenEpochIfNewer(Metadata.java:183)] - [Consumer clientId=client-01, groupId=app-group-01] Not replacing existing epoch 12 with new epoch 7 for partition test-topic-1 2022-07-14 17:17:39.451 INFO [pc-broker-poll] [org.apache.kafka.clients.consumer.internals.SubscriptionState.maybeSeekUnvalidated(SubscriptionState.java:398)] - [Consumer clientId=client-01, groupId=app-group-01] Resetting offset for partition test-topic-1 to position FetchPosition{offset=1197212, offsetEpoch=Optional.empty, currentLeader=LeaderAndEpoch{leader=Optional[borker01:9092 (id: 21 rack: 0)], epoch=12}}. 2022-07-14 17:17:39.452 DEBUG [pc-broker-poll] [org.apache.kafka.clients.consumer.internals.Fetcher.prepareFetchRequests(Fetcher.java:1198)] - [Consumer clientId=client-01, groupId=app-group-01] Added READ_UNCOMMITTED fetch request for partition test-topic-1 at position FetchPosition{offset=1197212, offsetEpoch=Optional.empty, currentLeader=LeaderAndEpoch{leader=Optional[borker01:9092 (id: 21 rack: 0)], epoch=12}} to node borker01:9092 (id: 21 rack: 0) 2022-07-14 17:17:39.452 DEBUG [pc-broker-poll] [org.apache.kafka.clients.FetchSessionHandler$Builder.build(FetchSessionHandler.java:259)] - [Consumer clientId=client-01, groupId=app-group-01] Built incremental fetch (sessionId=1279947631, epoch=1) for node 21. Added 0 partition(s), altered 1 partition(s), removed 0 partition(s) out of 1 partition(s) 2022-07-14 17:17:39.452 DEBUG [pc-broker-poll] [org.apache.kafka.clients.consumer.internals.Fetcher.sendFetches(Fetcher.java:265)] - [Consumer clientId=client-01, groupId=app-group-01] Sending READ_UNCOMMITTED IncrementalFetchRequest(toSend=(test-topic-1), toForget=(), implied=()) to broker borker01:9092 (id: 21 rack: 0) 2022-07-14 17:17:39.452 DEBUG [pc-broker-poll] [org.apache.kafka.clients.NetworkClient.doSend(NetworkClient.java:522)] - [Consumer clientId=client-01, groupId=app-group-01] Sending FETCH request with header RequestHeader(apiKey=FETCH, apiVersion=12, clientId=client-01, correlationId=26) and timeout 30000 to node 21: FetchRequestData(clusterId=null, replicaId=-1, maxWaitMs=500, minBytes=1, maxBytes=52428800, isolationLevel=0, sessionId=1279947631, sessionEpoch=1, topics=[FetchTopic(topic='test-topic', partitions=[FetchPartition(partition=1, currentLeaderEpoch=12, fetchOffset=1197212, lastFetchedEpoch=-1, logStartOffset=-1, partitionMaxBytes=1048576)])], forgottenTopicsData=[], rackId='') 2022-07-14 17:17:39.608 DEBUG [pc-broker-poll] [io.confluent.parallelconsumer.internal.ConsumerManager.poll(ConsumerManager.java:55)] - Poll completed normally (after timeout of PT2S) and returned 0... 2022-07-14 17:17:39.608 DEBUG [pc-broker-poll] [io.confluent.parallelconsumer.internal.BrokerPollSystem.pollBrokerForRecords(BrokerPollSystem.java:185)] - Poll completed 2022-07-14 17:17:39.608 DEBUG [pc-broker-poll] [io.confluent.parallelconsumer.internal.BrokerPollSystem.handlePoll(BrokerPollSystem.java:142)] - Got 0 records in poll result 2022-07-14 17:17:39.608 DEBUG [pc-broker-poll] [io.confluent.parallelconsumer.internal.BrokerPollSystem.pollBrokerForRecords(BrokerPollSystem.java:176)] - Subscriptions are paused: false 2022-07-14 17:17:39.608 DEBUG [pc-broker-poll] [io.confluent.parallelconsumer.internal.BrokerPollSystem.pollBrokerForRecords(BrokerPollSystem.java:182)] - Long polling broker with timeout PT2S, might appear to sleep here if subs are paused, or no data available on broker. Run state: running

[astubbs]

Discussion continued in issue #352