Paused consumption across multiple consumers

[rbokade-rbk]

We have observed that consumption is frequently halted or paused indefinitely until we restart the pod associated with the partition.

In production, we observe that message consumption is being paused in a partition because the PartitionStateManager marks the partition as lost and skips message processing, assuming it has been assigned to a different consumer.

Rebalance triggered : P12 assigned to same pod
2025-03-03 07:30:15.219 INFO pc-broker-poll io.confluent.parallelconsumer.state.PartitionStateManager Partitions revoked: [app-xxx.consent.consent.receipts-11]
2025-03-03 07:30:15.232 INFO pc-broker-poll io.confluent.parallelconsumer.internal.AbstractParallelEoSStreamProcessor Assigned 1 total (1 new) partition(s) [app-xxx.consent.consent.receipts-12]

PartitionStateManager declared partition Lost
2025-03-03 12:07:42.765 INFO pc-broker-poll io.confluent.parallelconsumer.state.PartitionStateManager Lost partitions: [app-xxx.consent.consent.receipts-12]

No work in mailbox though we have enough lag to process and No Inflight messages , last commit was on 23rd , which was paused until 24th .

Logs : Partition is assigned to this pod but as per logs it claims that its assigned to different consumer .

2025-03-03 12:07:42.765 INFO pc-broker-poll io.confluent.parallelconsumer.state.PartitionStateManager Lost partitions: [app-xxx.consent.consent.receipts-12]
2025-03-03 12:07:43.003 INFO (tenant-config-input-consumer,app-xxx.ds-preference-tenant-config.retry).retry-worker-4 com.onetrust.messaging.consumer.KafkaConsumer Partitions assigned: []
2025-03-03 12:07:44.303 INFO pc-broker-poll io.confluent.parallelconsumer.internal.AbstractParallelEoSStreamProcessor Assigned 1 total (1 new) partition(s) [app-xxx.consent.consent.receipts-12]
2025-03-04 00:00:19.334 DEBUG pc-broker-poll io.confluent.parallelconsumer.internal.ConsumerManager Poll completed normally (after timeout of PT2S) and returned 0...
2025-03-04 00:00:19.335 DEBUG pc-broker-poll io.confluent.parallelconsumer.internal.BrokerPollSystem Poll completed
2025-03-04 00:00:19.335 DEBUG pc-broker-poll io.confluent.parallelconsumer.internal.BrokerPollSystem Got 0 records in poll result
2025-03-04 00:00:19.335 DEBUG pc-broker-poll io.confluent.parallelconsumer.internal.BrokerPollSystem Subscriptions are paused: true
2025-03-04 00:00:19.335 DEBUG pc-broker-poll io.confluent.parallelconsumer.internal.BrokerPollSystem Long polling broker with timeout PT2S, might appear to sleep here if subs are paused, or no data available on broker. Run state: RUNNING
2025-03-04 00:00:19.335 DEBUG pc-broker-poll io.confluent.parallelconsumer.internal.ConsumerManager Poll starting with timeout: PT2S
2025-03-04 00:00:21.336 DEBUG pc-broker-poll io.confluent.parallelconsumer.internal.ConsumerManager Poll completed normally (after timeout of PT2S) and returned 0...
2025-03-04 00:00:21.336 DEBUG pc-broker-poll io.confluent.parallelconsumer.internal.BrokerPollSystem Poll completed
2025-03-04 00:00:21.336 DEBUG pc-broker-poll io.confluent.parallelconsumer.internal.BrokerPollSystem Got 0 records in poll result
2025-03-04 00:00:21.337 DEBUG pc-broker-poll io.confluent.parallelconsumer.internal.BrokerPollSystem Subscriptions are paused: true
2025-03-04 00:00:21.337 DEBUG pc-broker-poll io.confluent.parallelconsumer.internal.BrokerPollSystem Long polling broker with timeout PT2S, might appear to sleep here if subs are paused, or no data available on broker. Run state: RUNNING
2025-03-04 00:00:21.337 DEBUG pc-broker-poll io.confluent.parallelconsumer.internal.ConsumerManager Poll starting with timeout: PT2S
2025-03-04 00:00:22.440 DEBUG pc-control io.confluent.parallelconsumer.internal.AbstractParallelEoSStreamProcessor Mailbox results returned null, indicating timeToBlockFor elapsed (which was set as PT4.99999484S)
2025-03-04 00:00:22.440 DEBUG pc-control io.confluent.parallelconsumer.internal.AbstractParallelEoSStreamProcessor isPoolQueueLow()? workAmountBelowTarget true 0 vs 10;
2025-03-04 00:00:22.440 DEBUG pc-control io.confluent.parallelconsumer.internal.AbstractParallelEoSStreamProcessor Will try to get work - target: 20, current queue size: 0, requesting: 20, loading factor: 2
2025-03-04 00:00:22.441 DEBUG pc-control io.confluent.parallelconsumer.state.PartitionState Epoch mismatch 0 vs 2 for record WorkContainer(tp:app-xxx.consent.consent.receipts-12:o:771160:k:16ffceee). Skipping message - it's partition has already assigned to a different consumer.
2025-03-04 00:00:22.441 DEBUG pc-control io.confluent.parallelconsumer.state.PartitionState Work is in queue with stale epoch or no longer assigned. Skipping. Shard it came from will/was removed during partition revocation. WC: WorkContainer(tp:app-xxx.consent.consent.receipts-12:o:771160:k:16ffceee)
2025-03-04 00:00:22.441 DEBUG pc-control io.confluent.parallelconsumer.state.PartitionState Epoch mismatch 0 vs 2 for record WorkContainer(tp:app-xxx.consent.consent.receipts-12:o:771169:k:3a1af6a2). Skipping message - it's partition has already assigned to a different consumer.
2025-03-04 00:00:22.441 DEBUG pc-control io.confluent.parallelconsumer.state.PartitionState Work is in queue with stale epoch or no longer assigned. Skipping. Shard it came from will/was removed during partition revocation. WC: WorkContainer(tp:app-xxx.consent.consent.receipts-12:o:771169:k:3a1af6a2)
2025-03-04 00:00:22.441 DEBUG pc-control io.confluent.parallelconsumer.state.PartitionState Epoch mismatch 0 vs 2 for record WorkContainer(tp:app-xxx.consent.consent.receipts-12:o:771170:k:fc145cb0). Skipping message - it's partition has already assigned to a different consumer.
2025-03-04 00:00:22.441 DEBUG pc-control io.confluent.parallelconsumer.state.PartitionState Work is in queue with stale epoch or no longer assigned. Skipping. Shard it came from will/was removed during partition revocation. WC: WorkContainer(tp:app-xxx.consent.consent.receipts-12:o:771170:k:fc145cb0)
2025-03-04 00:00:22.441 DEBUG pc-control io.confluent.parallelconsumer.state.PartitionState Epoch mismatch 0 vs 2 for record WorkContainer(tp:app-xxx.consent.consent.receipts-12:o:771215:k:a7bbbfff). Skipping message - it's partition has already assigned to a different consumer.
2025-03-04 00:00:22.441 DEBUG pc-control io.confluent.parallelconsumer.state.PartitionState Work is in queue with stale epoch or no longer assigned. Skipping. Shard it came from will/was removed during partition revocation. WC: WorkContainer(tp:app-xxx.consent.consent.receipts-12:o:771215:k:a7bbbfff)
2025-03-04 00:00:22.441 DEBUG pc-control io.confluent.parallelconsumer.state.PartitionState Epoch mismatch 0 vs 2 for record WorkContainer(tp:app-xxx.consent.consent.receipts-12:o:771189:k:3c5ce0b3). Skipping message - it's partition has already assigned to a different consumer.
2025-03-04 00:00:22.441 DEBUG pc-control io.confluent.parallelconsumer.state.PartitionState Work is in queue with stale epoch or no longer assigned. Skipping. Shard it came from will/was removed during partition revocation. WC: WorkContainer(tp:app-xxx.consent.consent.receipts-12:o:771189:k:3c5ce0b3)
2025-03-04 00:00:22.441 DEBUG pc-control io.confluent.parallelconsumer.state.PartitionState Epoch mismatch 0 vs 2 for record WorkContainer(tp:app-xxx.consent.consent.receipts-12:o:771243:k:8e803207). Skipping message - it's partition has already assigned to a different consumer.
2025-03-04 00:00:22.441 DEBUG pc-control io.confluent.parallelconsumer.state.PartitionState Work is in queue with stale epoch or no longer assigned. Skipping. Shard it came from will/was removed during partition revocation. WC: WorkContainer(tp:app-xxx.consent.consent.receipts-12:o:771243:k:8e803207)
2025-03-04 00:00:22.441 DEBUG pc-control io.confluent.parallelconsumer.state.PartitionState Epoch mismatch 0 vs 2 for record WorkContainer(tp:app-xxx.consent.consent.receipts-12:o:771168:k:d50689cd). Skipping message - it's partition has already assigned to a different consumer.
2025-03-04 00:00:22.441 DEBUG pc-control io.confluent.parallelconsumer.state.PartitionState Work is in queue with stale epoch or no longer assigned. Skipping. Shard it came from will/was removed during partition revocation. WC: WorkContainer(tp:app-xxx.consent.consent.receipts-12:o:771168:k:d50689cd)
2025-03-04 00:00:22.442 DEBUG pc-control io.confluent.parallelconsumer.state.PartitionState Epoch mismatch 0 vs 2 for record WorkContainer(tp:app-xxx.consent.consent.receipts-12:o:771171:k:20d5aeca). Skipping message - it's partition has already assigned to a different consumer.

On reviewing thread dump , PC threads are in wait state

"(dsCacheSync-in-0,ds-preference-cache-sync).pc-pool-13-thread-1" - Thread t@139
java.lang.Thread.State: WAITING
at java.base@17.0.14/jdk.internal.misc.Unsafe.park(Native Method)
- parking to wait for <70bbade3> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
at java.base@17.0.14/java.util.concurrent.locks.LockSupport.park(Unknown Source)
at java.base@17.0.14/java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionNode.block(Unknown Source)
at java.base@17.0.14/java.util.concurrent.ForkJoinPool.unmanagedBlock(Unknown Source)
at java.base@17.0.14/java.util.concurrent.ForkJoinPool.managedBlock(Unknown Source)
at java.base@17.0.14/java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(Unknown Source)
at java.base@17.0.14/java.util.concurrent.LinkedBlockingQueue.take(Unknown Source)
at java.base@17.0.14/java.util.concurrent.ThreadPoolExecutor.getTask(Unknown Source)
at java.base@17.0.14/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
at java.base@17.0.14/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
at java.base@17.0.14/java.lang.Thread.run(Unknown Source)

PC Version in production : 0.5.2.8
Upgrade in progress : 0.5.3.2

[rbokade-rbk]

Paused_consumer.txt

[sangreal]

rbokade-rbk , I think the partitions sub is paused according to the log and that is why you could not get any new records even you assigned the partition.
And I highly suspect that this bug is due to : failed to clear the stale container which I fixed in 2023, but somehow the fix doesn't work.

I will check on this one. But if you could reproduce, could u please tune the log level to TRACE and catch again? so I could get more information to confirm.

[netroute-js]

We've stumbled upon probably the same kind of issue. Sometimes when there is a redeployment and Kafka rebalancing kicks in (under heavy load) we may end up in a situation where it seems like consumer is not processing any messages. However there are messages not processed which is visible via a lag.

I've also tried to use Kafka Incremental Cooperative Rebalancing and this issue is even more visible.

Martyn Ye (@sangreal) any news about potential fix?

[rbokade-rbk]

Martyn Ye (@sangreal) , could you please see if this can be prioritized , we are widely noticing this issue

[netroute-js]

Martyn Ye (@sangreal) or Antony Stubbs (@astubbs), any ETA when this issue could be prioritized and fixed? It's really something that impacts our flow.

[sangreal]

netroute-js rbokade-rbk please provide a trace level log when this issue happens, I will take a look.

[rbokade-rbk]

It. cannot be reproduced in non production envs , it does halt in production , we enabled the logs after the fact and have attached to the ticket (#857 (comment)) , but cannot enable the logs before the fact as we are unsure how long it would take to issue to reoccur in an env/consumer

[sangreal]

rbokade-rbk the log is too short, I checked again. I don't have enough information. My fix is introduced in 0.5.2.7. So supposedly you should not have the issue. Because the issue happens where partition revoked and assign to the same consumer again(high possibility if you enable Kafka Incremental Cooperative Rebalancing). And after I fixed, our system never encounter this issue again. So it is weird.
What I need is :

1. give me the log contains when rebalancing happens
2. It will be very helpful if you enable TRACE level log for para-consumer.

And please make sure of that you are using 0.5.2.8.

Another thought is you may try to remove the StickyCoorperativeAssignor to check if this issue is gone or much less.

[netroute-js]

We've stumbled upon a lag of more than 700 000 records in one topic and consumer completely stack on that one until the consumer is terminated.

We are using 0.5.3.2. I will try to enable TRACE logging tommorow and reproduce it in a testing environment.

[sangreal]

netroute-js please set TRACE level logging limited to the io.confluent package to avoid too many other irrelevant logging.

[netroute-js]

Martyn Ye (@sangreal), I was able to reproduce the issue in our testing environment. Below you can find the information about the environment:

• PC version: 0.5.3.2
• Kafka topic has 30 partitions,
• Traffic TPS: couple of thousands of events per sec
• Enabled Kafka Incremental Rebalancing. However we've stumbled upon this issue in production as well (in prod, we don't have Incremental Rebalancing)
• During the test, I was basically adding/removing k8s pods (like 2-3 in single go) to force rebalancing and at some point the issue kicked in and made our consumer to stop processing events. In that case lag grows very quickly,
• the only way to move forward is to re-assign the impacted partition to a different consumer

In the attachment you can find logs from the test. It's quite a lot of them and our service consumes from multiple different topics so I've replaced the topics names in the output file:

• LAG_TOPIC, that one is of the interest here,
• NOT_INTERESTING_TOPIC, you can skip that one.

I've also gathered some PC metrics for that specific consumer and partition:

• In AKQH, I see that this partition has offset set to 8037030, Lag set to 4989
• Highest completed offset is 8042018
• Highest seen offset is 8042018
• Sequentially succeded offset is 8042018
• Incomplete offset is 0
• Latest commited offset is 8042019

It's important to note that at 10:34:00, it seems that the issue has already happened because the lag is there but consumption stopped more or less.

If you need any more information, let me know please.

pc_logs.txt

[rbokade-rbk]

Martyn Ye (@sangreal), please let know if we are able to identify issue ?

[netroute-js]

Martyn Ye (@sangreal) any news? We've stumbled upon this issue in production again. The bad news is that redeploying doesn't help now.

UPDATE:
We've had a lag of 5 messages in a retry topic. Once I've restarted the pods, 3 of those messages were consumed but the two remain as lag - no matter how many times I redeploy. I've checked the logs and I've found the log that says:

Fetch position FetchPosition{offset=8, ....} is out for range for partition
Resetting offset for partition to position FetchPosition{offset=10}
There are no new messages but somehow the consumer is stack at 8 as committed offset seen in AKHQ and offset 7 in PC metrics.
Once I redeploy this pod, it starts from the offset 8 and loop continues.
I think, this is a different issue but maybe that will help you guys to troubleshoot whats going on.