CAMEL-18024: rework the last processed offset tracking

[orpiske]

No description provided.

[github-actions]

⚠️ This PR changes Camel components and will be tested automatically.

[github-actions]

❌ Finished component verification: 1 component(s) test failed out of 1 component(s) tested

[github-actions]

❌ Finished component verification: 1 component(s) test failed out of 4 component(s) tested

[github-actions]

❌ Finished component verification: 1 component(s) test failed out of 4 component(s) tested

[github-actions]

❌ Finished component verification: 1 component(s) test failed out of 4 component(s) tested

[github-actions]

✔️ Finished component verification: 0 component(s) test failed out of 1 component(s) tested

[orpiske]

The checkstyle issue is not on this PR code, so it's safe to merge.

[ludovic-boutros]

Hi @orpiske , I just checked this PR. The async ConcurrentQueue is mandatory if you want to be able to commit offsets in an async process like an aggregation with completion timeout. this is because the commit function must be called by the consumer thread. If not you will get a ConcurrentModification exception.
The use case is for instance to index documents in a bulk manner using aggregate and commit once indexed.

[orpiske]

Hi @orpiske , I just checked this PR. The async ConcurrentQueue is mandatory if you want to be able to commit offsets in an async process like an aggregation with completion timeout. this is because the commit function must be called by the consumer thread. If not you will get a ConcurrentModification exception.
The use case is for instance to index documents in a bulk manner using aggregate and commit once indexed.

Hi, do you have a reproducer I can quickly transform into an integration test so I can take a look at it and make sure it works/won't break again?

[orpiske]

@ludovic-boutros doing a quick research regarding your comment I noticed our async test was disabled (it runs well on my own CI, but I re-enabled it again so we have regular checks on the Apache one).

If you have any suggestion about how to modify it to fit the use case/issue you mention, please, just let us know (or send a PR).

Thanks for reviewing it/suggesting/contributing!

[ludovic-boutros]

@orpiske Thank you ! I will try to find some time to make some tests with this PR and let you know.

[ludovic-boutros]

@orpiske the async commit process seems to be broken. The commit() function in DefaultKafkaManualAsyncCommit calls a forceCommit() function which is synchronous.
And when fixing it using the commit() function (async with the manager), the test throw a ConcurrentModificationException().
We need to reintroduce a ConcurrentQueue storing commits which should be process by the main kafka consumer polling loop and thread.

[orpiske]

@orpiske the async commit process seems to be broken. The commit() function in DefaultKafkaManualAsyncCommit calls a forceCommit() function which is synchronous. And when fixing it using the commit() function (async with the manager), the test throw a ConcurrentModificationException(). We need to reintroduce a ConcurrentQueue storing commits which should be process by the main kafka consumer polling loop and thread.

Please, can you please open a ticket for this or send a PR with your proposed changes?

The important requirement is that the concurrent queue must be isolated to only the async-related code. It should not be part of the other code (ie.: the concerns of the async manual commit code should be separated) nor should be present in any way in the KafkaConsumer, KafkaFetchRecords or Kafka*Processor code. This is important so that we can continue cleaning up and reducing the maintenance effort for this component.

[orpiske]

And once again: thanks for the review and testing this part of the code. It's not a scenario that is widely used/tested, so I appreciate additional eyes on it.

[orpiske]

@orpiske the async commit process seems to be broken. The commit() function in DefaultKafkaManualAsyncCommit calls a forceCommit() function which is synchronous. And when fixing it using the commit() function (async with the manager), the test throw a ConcurrentModificationException(). We need to reintroduce a ConcurrentQueue storing commits which should be process by the main kafka consumer polling loop and thread.

Please, can you please open a ticket for this or send a PR with your proposed changes?

The important requirement is that the concurrent queue must be isolated to only the async-related code. It should not be part of the other code (ie.: the concerns of the async manual commit code should be separated) nor should be present in any way in the KafkaConsumer, KafkaFetchRecords or Kafka*Processor code. This is important so that we can continue cleaning up and reducing the maintenance effort for this component.

Alternatively, on your scenario, can you also please test if changing the call from forceCommit to recordOffset leads to the correct behavior?

I believe the offset will eventually be committed by the processor after successful processing.

[orpiske]

@ludovic-boutros I've got some time to look at this: #7664. Changing to recordOffset seems to do the trick. I'd kindly appreciate if you could give the fix a try on your scenario.

Thanks.