You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@camel.apache.org by "Mike Barlotta (Jira)" <ji...@apache.org> on 2023/11/06 16:47:00 UTC
[jira] [Commented] (CAMEL-20044) camel-kafka - On rejoining consumer group Camel can set offset incorrectly causing messages to be replayed
[ https://issues.apache.org/jira/browse/CAMEL-20044?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17783278#comment-17783278 ]
Mike Barlotta commented on CAMEL-20044:
---------------------------------------
Got some time and I've added some extra logging statements to _KafkaRecordProcessor_ in the 3.21.x branch
in this run there are 3 consumer threads
* thread 1 processes partition 2
** handles offset 0 and 1 fine
** 2:2 has an error
* thread 2 processes partition 1
** handles offset 0 and 1 fine
* thread 3 processes partition 0
** handles offset 0, 1, 2 fine
** 0:3 has an error
thread 1 and 3 both unsubscribe
{code:java}
2023-11-06 | 10:23:12.774 | WARN | [Camel (camel-1) thread #3 - KafkaConsumer[foobarTopic]] | o.a.c.c.k.c.s.KafkaRecordProcessor (KafkaRecordProcessor.java:144) | Will seek consumer to offset 2 on partition 0 and start polling again.
2023-11-06 | 10:23:12.782 | TRACE | [Camel (camel-1) thread #3 - KafkaConsumer[foobarTopic]] | o.a.c.c.kafka.KafkaFetchRecords (KafkaFetchRecords.java:354) | the polling iteration had a result returned for partition 0 and offset 2
2023-11-06 | 10:23:12.784 | TRACE | [Camel (camel-1) thread #1 - KafkaConsumer[foobarTopic]] | o.a.c.c.kafka.KafkaFetchRecords (KafkaFetchRecords.java:354) | the polling iteration had a result returned for partition 2 and offset 1
2023-11-06 | 10:23:12.784 | DEBUG | [Camel (camel-1) thread #3 - KafkaConsumer[foobarTopic]] | o.a.c.c.kafka.KafkaFetchRecords (KafkaFetchRecords.java:366) | We hit an error ... setting flags to force reconnect
2023-11-06 | 10:23:12.785 | DEBUG | [Camel (camel-1) thread #1 - KafkaConsumer[foobarTopic]] | o.a.c.c.kafka.KafkaFetchRecords (KafkaFetchRecords.java:366) | We hit an error ... setting flags to force reconnect
2023-11-06 | 10:23:12.785 | DEBUG | [Camel (camel-1) thread #3 - KafkaConsumer[foobarTopic]] | o.a.c.c.kafka.KafkaFetchRecords (KafkaFetchRecords.java:382) | Not reconnecting, check whether to auto-commit or not ...
2023-11-06 | 10:23:12.785 | DEBUG | [Camel (camel-1) thread #1 - KafkaConsumer[foobarTopic]] | o.a.c.c.kafka.KafkaFetchRecords (KafkaFetchRecords.java:382) | Not reconnecting, check whether to auto-commit or not ...
2023-11-06 | 10:23:12.785 | INFO | [Camel (camel-1) thread #3 - KafkaConsumer[foobarTopic]] | o.a.c.c.k.consumer.NoopCommitManager (NoopCommitManager.java:35) | Auto commit on foobarTopic-Thread 2 from topic foobarTopic is enabled via Kafka consumer (NO-OP)
2023-11-06 | 10:23:12.785 | INFO | [Camel (camel-1) thread #1 - KafkaConsumer[foobarTopic]] | o.a.c.c.k.consumer.NoopCommitManager (NoopCommitManager.java:35) | Auto commit on foobarTopic-Thread 0 from topic foobarTopic is enabled via Kafka consumer (NO-OP)
2023-11-06 | 10:23:12.785 | DEBUG | [Camel (camel-1) thread #3 - KafkaConsumer[foobarTopic]] | o.a.c.c.kafka.KafkaFetchRecords (KafkaFetchRecords.java:468) | Unsubscribing from Kafka
2023-11-06 | 10:23:12.785 | DEBUG | [Camel (camel-1) thread #1 - KafkaConsumer[foobarTopic]] | o.a.c.c.kafka.KafkaFetchRecords (KafkaFetchRecords.java:468) | Unsubscribing from Kafka {code}
The last we heard from thread #2 is that it had processed 1:1 and was manually committing
{code:java}
2023-11-06 | 10:23:12.636 | INFO | [Camel (camel-1) thread #2 - KafkaConsumer[foobarTopic]] | c.c.k.KafkaOffsetManagerProcessor (KafkaOffsetManagerProcessor.java:49) | manually committing the offset for batch
Message consumed from foobarTopic
The Partition:Offset is 1:1
The Key is null
10 {code}
*The _lastResult_ of 1:1 on thread #2 will end up causing the problem*
When Camel resubscribes we get this
* thread 1 processes partition 2
** based on Camel logic it reprocesses 2:2
** 2:2 has an error
The other threads don't have a chance to do anything before Camel unsubscribes
{code:java}
2023-11-06 | 10:23:15.421 | WARN | [Camel (camel-1) thread #1 - KafkaConsumer[foobarTopic]] | o.a.c.c.k.c.s.KafkaRecordProcessor (KafkaRecordProcessor.java:144) | Will seek consumer to offset -1 on partition -1 and start polling again.
2023-11-06 | 10:23:15.421 | TRACE | [Camel (camel-1) thread #1 - KafkaConsumer[foobarTopic]] | o.a.c.c.kafka.KafkaFetchRecords (KafkaFetchRecords.java:354) | the polling iteration had a result returned for partition -1 and offset -1
2023-11-06 | 10:23:15.421 | DEBUG | [Camel (camel-1) thread #1 - KafkaConsumer[foobarTopic]] | o.a.c.c.kafka.KafkaFetchRecords (KafkaFetchRecords.java:366) | We hit an error ... setting flags to force reconnect
2023-11-06 | 10:23:15.421 | DEBUG | [Camel (camel-1) thread #1 - KafkaConsumer[foobarTopic]] | o.a.c.c.kafka.KafkaFetchRecords (KafkaFetchRecords.java:382) | Not reconnecting, check whether to auto-commit or not ...
2023-11-06 | 10:23:15.421 | INFO | [Camel (camel-1) thread #1 - KafkaConsumer[foobarTopic]] | o.a.c.c.k.consumer.NoopCommitManager (NoopCommitManager.java:35) | Auto commit on foobarTopic-Thread 0 from topic foobarTopic is enabled via Kafka consumer (NO-OP)
2023-11-06 | 10:23:15.422 | DEBUG | [Camel (camel-1) thread #1 - KafkaConsumer[foobarTopic]] | o.a.c.c.kafka.KafkaFetchRecords (KafkaFetchRecords.java:468) | Unsubscribing from Kafka
2023-11-06 | 10:23:15.422 | DEBUG | [Camel (camel-1) thread #1 - KafkaConsumer[foobarTopic]] | o.a.c.c.k.consumer.NoopCommitManager (NoopCommitManager.java:42) | Auto commit to offset foobarTopic-Thread 0 from topic foobarTopic is disabled (NO-OP)
2023-11-06 | 10:23:15.425 | INFO | [Camel (camel-1) thread #1 - KafkaConsumer[foobarTopic]] | o.a.k.clients.consumer.KafkaConsumer (KafkaConsumer.java:1075) | [Consumer clientId=consumer-test_group_id-5, groupId=test_group_id] Unsubscribed all topics or patterns and assigned partitions
2023-11-06 | 10:23:15.426 | DEBUG | [Camel (camel-1) thread #1 - KafkaConsumer[foobarTopic]] | o.a.c.c.kafka.KafkaFetchRecords (KafkaFetchRecords.java:470) | Done unsubscribing from Kafka
2023-11-06 | 10:23:15.426 | DEBUG | [Camel (camel-1) thread #1 - KafkaConsumer[foobarTopic]] | o.a.c.c.kafka.KafkaFetchRecords (KafkaFetchRecords.java:117) | not connected so will initialize consumer task
2023-11-06 | 10:23:15.426 | INFO | [Camel (camel-1) thread #1 - KafkaConsumer[foobarTopic]] | o.a.c.c.kafka.KafkaFetchRecords (KafkaFetchRecords.java:250) | Reconnecting Kafka consumer thread ID foobarTopic-Thread 0 with poll timeout of 5000 ms
...
2023-11-06 | 10:23:15.443 | INFO | [Camel (camel-1) thread #1 - KafkaConsumer[foobarTopic]] | o.a.c.c.kafka.KafkaFetchRecords (KafkaFetchRecords.java:297) | Subscribing foobarTopic-Thread 0 to topic foobarTopic{code}
When Camel resubscribes we get this:
* thread 2 is assigned partition 0
* {*}the problem is lastResult is 1:1 which is from the last time it ran{*}{*}{*}
** *this should have been reset to null*
* it processes 0:3 which has an error
{code:java}
2023-11-06 | 10:23:15.870 | TRACE | [Camel (camel-1) thread #2 - KafkaConsumer[foobarTopic]] | o.a.c.c.kafka.KafkaFetchRecords (KafkaFetchRecords.java:340) | this polling iteration is using lastresult with partition 1 and offset 1
...
2023-11-06 | 10:23:15.871 | DEBUG | [Camel (camel-1) thread #2 - KafkaConsumer[foobarTopic]] | o.a.c.c.k.c.s.KafkaRecordProcessor (KafkaRecordProcessor.java:68) | setting up the exchange for message from partition 0 and offset 3
...
2023-11-06 | 10:23:15.881 | DEBUG | [Camel (camel-1) thread #2 - KafkaConsumer[foobarTopic]] | o.a.c.c.k.c.s.KafkaRecordProcessor (KafkaRecordProcessor.java:118) | an exception was thrown for record at partition 0 and offset 3
2023-11-06 | 10:23:15.881 | WARN | [Camel (camel-1) thread #2 - KafkaConsumer[foobarTopic]] | o.a.c.c.k.c.s.KafkaRecordProcessor (KafkaRecordProcessor.java:122) | about to process an exception with UNEXPECTED offset 1 on partition 1 was expecting offset 3 on partition 0{code}
Then _KafkaRecordProcessor_ will call _processException_ and call _commitManager.forceCommit._ This __ uses the values in lastResult.
{code:java}
2023-11-06 | 10:23:15.881 | WARN | [Camel (camel-1) thread #2 - KafkaConsumer[foobarTopic]] | o.a.c.c.k.c.s.KafkaRecordProcessor (KafkaRecordProcessor.java:144) | Will seek consumer to offset 1 on partition 1 and start polling again. {code}
*At this point the consumer is pointing in the wrong place as* *partition 0 is moved back to offset 1*
After Camel unsubscribes and then reconnects we find this on thread #3 (which was assigned partition 0)
{code:java}
2023-11-06 | 10:23:18.553 | TRACE | [Camel (camel-1) thread #3 - KafkaConsumer[foobarTopic]] | o.a.c.c.k.c.s.KafkaRecordProcessorFacade (KafkaRecordProcessorFacade.java:140) | Partition = 0, offset = 2, key = null, value = 6
2023-11-06 | 10:23:18.553 | DEBUG | [Camel (camel-1) thread #3 - KafkaConsumer[foobarTopic]] | o.a.c.c.k.c.s.KafkaRecordProcessor (KafkaRecordProcessor.java:68) | setting up the exchange for message from partition 0 and offset 2
{code}
At this point all of the messages on partition 0 will be reprocessed
It appears that when the initial errors ({_}breakOnFirstError{_}) occurred that thread #2 never had lastResult set back to null (which seems it should have if _KafkaFetchRecords_ broke out of its loop in _startPolling._ However, I didn't see the following log for _KafkaFetchRecords_ for that thread as I did for threads 1 and 3.
thread 1
{code:java}
2023-11-06 | 10:23:12.789 | INFO | [Camel (camel-1) thread #1 - KafkaConsumer[foobarTopic]] | o.a.c.c.kafka.KafkaFetchRecords (KafkaFetchRecords.java:250) | Reconnecting Kafka consumer thread ID foobarTopic-Thread 0 with poll timeout of 5000 ms
...
2023-11-06 | 10:23:12.808 | TRACE | [Camel (camel-1) thread #1 - KafkaConsumer[foobarTopic]] | o.a.c.c.kafka.KafkaFetchRecords (KafkaFetchRecords.java:320) | Polling foobarTopic-Thread 0 from topic foobarTopic with timeout: 5000
{code}
thread 3
{code:java}
2023-11-06 | 10:23:12.789 | INFO | [Camel (camel-1) thread #3 - KafkaConsumer[foobarTopic]] | o.a.c.c.kafka.KafkaFetchRecords (KafkaFetchRecords.java:250) | Reconnecting Kafka consumer thread ID foobarTopic-Thread 2 with poll timeout of 5000 ms
...
2023-11-06 | 10:23:12.809 | TRACE | [Camel (camel-1) thread #3 - KafkaConsumer[foobarTopic]] | o.a.c.c.kafka.KafkaFetchRecords (KafkaFetchRecords.java:320) | Polling foobarTopic-Thread 2 from topic foobarTopic with timeout: 5000{code}
the only entry for thread 2 at this time was
{code:java}
2023-11-06 | 10:23:15.339 | DEBUG | [Camel (camel-1) thread #2 - KafkaConsumer[foobarTopic]] | o.a.c.c.k.consumer.NoopCommitManager (NoopCommitManager.java:42) | Auto commit to offset foobarTopic-Thread 1 from topic foobarTopic is disabled (NO-OP) {code}
this suggests it never was unsubscribed. But not totally sure.
I attached the log for this test run (camel-kafka-offset.11-06-2023.log)
I can create a PR with the extra logging.
I am willing to try and patch this, but wanted to make sure had a path forward as to what to change.
Some thoughts:
* don't use current Camel retry attempt when breakOnFirstError and manualCommit are both true
** this won't fix the problem but will let consumer handle offset committing completeley
* when UNEXPECTED occurs
** can avoid the force commit
** not sure what else to use except the current record values
* other?
** open to suggestions
> camel-kafka - On rejoining consumer group Camel can set offset incorrectly causing messages to be replayed
> ----------------------------------------------------------------------------------------------------------
>
> Key: CAMEL-20044
> URL: https://issues.apache.org/jira/browse/CAMEL-20044
> Project: Camel
> Issue Type: Bug
> Components: camel-kafka
> Affects Versions: 3.21.0
> Environment: * Rocky Linux 8.7
> * Open JDK 11.0.8
> * Camel 3.21.0
> * Spring Boot 2.7.14
> * Strimzi Kafka 0.28.0/3.0.0
> Reporter: Mike Barlotta
> Priority: Major
>
> {*}Reproducing (intermittent){*}:
> * Configure camel kafka consumer with following:
> ** autoCommitEnable = false
> ** allowManualCommit = true
> ** autoOffsetReset = earliest
> ** maxPollRecords = 1
> ** breakOnFirstError = true
> * Produce a series of records to kafka record to both partitions.
> * Throw an exception that is unhandled
> * commit the offset in the onException block
> *Expected behavior:*
> * Application should consume the record 1 more time, then move on to the next offset in the partition
> *Actual behavior:*
> * Application will often work. Occasionally will use the offset from another partition and assign that to the partition where the record failed. This can then result in the consumer replaying messages instead of moving forward.
> I put together a sample that can recreate the error. However, given its intermittent nature it may not fail on each run. I have included the logs from 3 different runs on my laptop from this test. Two of them show the error occurring. One of the them has a successful run. I have also provided more details in the README.
> * [https://github.com/CodeSmell/CamelKafkaOffset]
> This seems related to other issues with how Camel processes the _breakOnFirstError_ attribute.
> * CAMEL-14935
> * CAMEL-17925
> * CAMEL-18350
> * CAMEL-19894
--
This message was sent by Atlassian Jira
(v8.20.10#820010)