Repository navigation
Conversation
…umer thread KafkaManualCommit.commit() with DefaultKafkaManualAsyncCommitFactory records the offset again, and the consumer thread commits it before its next poll, so a commit from another thread, such as after an aggregator, no longer fails with ConcurrentModificationException. Regression from CAMEL-23997.
|
🌟 Thank you for your contribution to the Apache Camel project! 🌟 🐫 Apache Camel Committers, please review the following items:
|
gnodet-bot
left a comment
gnodet-bot
left a comment
There was a problem hiding this comment.
The ConcurrentHashMap-based deferred-commit design is correct and thread-safe. requestCommit() can be called from any thread (concurrent put() on ConcurrentHashMap is safe), and commitRequestedOffsets() runs exclusively on the consumer thread before each poll — no concurrent iteration/modification is possible.
One doc nit on the upgrade guide.
Backport labels missing. This fixes a regression introduced in 4.22.0 (CAMEL-23997) — users on 4.22.x and 4.23.x hit ConcurrentModificationException with async manual commits from a non-consumer thread (e.g. after an aggregator). Please add port/camel-4.22.x and port/camel-4.18.x labels so the fix reaches maintenance branches.
This review was generated by an AI agent, Hermès on behalf of @gnodet.
|
🧪 CI tested the following changed modules:
🔁 1 test passed only after a retry on JDK 21 (1 retried attempt) Recovered flaky tests on JDK 21 (1)
🔬 Scalpel shadow comparison — Scalpel: 9 of 726 tested, 25 compile-only — current: 9 all testedMaveniverse Scalpel detected 9 affected modules (current approach: 9). Skip-tests mode would test 9 modules (3 direct + 8 downstream), skip tests for 25 (generated code, meta-modules) Modules Scalpel would test (9)
Modules with tests skipped (25)
All tested modules (36 modules, 5m 20s total)Total reactor time: 5m 20s
Top 20 slowest modules:
|
davsclaus
left a comment
davsclaus
left a comment
There was a problem hiding this comment.
Thanks @Croway. The diagnosis is right: #24696 (CAMEL-23997) made DefaultKafkaManualAsyncCommit.commit() call commitAsync on the calling thread. After an aggregator that is the Aggregator thread, even with completionSize, because the reifier always runs the aggregate output on its own single-thread executor. Moving the commit back to the consumer thread is the right fix:
requestCommitonly writes to aConcurrentHashMap.commitRequestedOffsets()runs on the polling thread beforepoll(), outside the lock, sostop()/wakeup()is not affected.commit(partition)takes the pending request out, so the revoke on unsubscribe/close still flushes it.- The default methods keep the sync, noop and to-offset managers as they were.
The SingleThreadMockConsumer test is a good reproducer, and the docs, the catalog copy and the upgrade guide are updated.
- A request that arrives after the partition is revoked is committed for a partition this consumer no longer owns.
AsyncCommitManager.commit(TopicPartition)removes the pending request on revoke. But an aggregator that completes later callsrequestCommitagain, and the nextcommitRequestedOffsets()sends acommitAsyncfor that partition. At revoke timeClassicRebalanceListener.onPartitionsRevokedalready committed the last processed offset (from the offset cache). So this late commit is lower, and it can move the new owner's committed offset back, which means duplicates on the new owner. The polling thread can check the assignment:A test would help: inSet<TopicPartition> assigned = consumer.assignment(); for (TopicPartition partition : requestedOffsets.keySet()) { Long offset = requestedOffsets.remove(partition); if (offset == null) { continue; } if (assigned.contains(partition)) { offsets.put(partition, new OffsetAndMetadata(offset + 1)); } else { LOG.debug("Dropping the requested commit of offset {} on {}: the partition is no longer assigned to {}", offset, partition, threadId); } }
DefaultKafkaManualAsyncCommitTest,mockConsumer.rebalance(List.of())before the aggregator completes. - Optional:
requestedOffsets.putmeans the last writer wins. Two aggregation groups that complete out of order (or withparallelProcessing) can replace a higher requested offset with a lower one before the flush. 4.21 had the same semantics, so this is not a regression, butrequestedOffsets.merge(partition, partitionLastOffset, Math::max)is a cheap guard. - Question, not for this PR: on revoke and on stop,
commit(partition)prefersoffsetCache, whichKafkaRecordStreamingProcessorfills for every processed record, also withallowManualCommit=true. So records still held by an aggregator are committed on a rebalance although the route never calledcommit()for them.breakOnFirstErrordepends on that cache, so changing it is a design decision. Could you open a follow-up JIRA so it is not lost? - Backport: only
camel-4.22.xneeds this. Oncamel-4.18.x,DefaultKafkaManualAsyncCommit.commit()still only callsrecordOffset, so it is not affected.
Claude Code on behalf of davsclaus. This review was generated by an AI agent and may contain inaccuracies; please verify all suggestions before applying.
… partition was revoked is not made The consumer thread drops a requested offset whose partition is no longer assigned, so a late commit (for example by an aggregator) does not move back the offset committed by the new owner. Upgrade guide entry for 4.22.3, and the 4.24 entry split into shorter sentences.
|
@gnodet-bot Backport labels: Claude Code on behalf of Federico Mariani |
Claude Code on behalf of Federico Mariani |
gnodet-bot
left a comment
gnodet-bot
left a comment
There was a problem hiding this comment.
Following up on my earlier review — both issues are addressed.
- Backport labels:
port/camel-4.22.xadded ✓. The author explainsport/camel-4.18.xis not needed (the 4.18.xcommit()only callsrecordOffset(), nevercommitAsync()on the calling thread — the regression is not present there). Agreed. - Doc nit (upgrade guide sentence): Previous 36-word sentence ✅ rewritten. The file moved; the prior review thread is now outdated.
One new doc nit in the upgrade guides.
This review was generated by an AI agent, Hermès on behalf of @gnodet.
| that the new owner of the partition committed. | ||
|
|
There was a problem hiding this comment.
💡 Doc nit (low): This sentence runs 30 words — just over the 25-word guideline. Split suggestion:
| that the new owner of the partition committed. | |
| A commit called after the partition is revoked is not made. | |
| This prevents it from moving back the offset that the new owner of the partition committed. |
| that the new owner of the partition committed. | ||
|
|
There was a problem hiding this comment.
💡 Doc nit (low): Same sentence, same nit — 30 words. Split suggestion:
| that the new owner of the partition committed. | |
| A commit called after the partition is revoked is not made. | |
| This prevents it from moving back the offset that the new owner of the partition committed. |
davsclaus
left a comment
davsclaus
left a comment
There was a problem hiding this comment.
Thanks @Croway, the new commit covers everything from my earlier review:
- Revoked partition:
commitRequestedOffsets()now drops a request whose partition is not inconsumer.assignment(), with a DEBUG log.testCommitAfterRevokeIsNotCommittedreproduces it (revoke withrebalance(List.of()), then a latecommit()from the test thread), and waiting forpolls + 2is a clean way to be sure a flush ran, without a sleep. - Keeping last-writer-wins as in 4.21: fine for a minimal 4.22.x port.
- Thanks for opening CAMEL-26993.
- Backport: only
port/camel-4.22.xis set and the 4.22 entry is under "4.22.2 to 4.22.3". I checked the tags: 4.22.0 to 4.22.2 commit on the calling thread, 4.21.0 does not, so the version wording in both guides is right.KafkaFetchRecordsdiffers between camel-4.22.x and main, so thecommitRequestedOffsets()hook in the poll loop may need a manual port.
One small leftover on the new lines, a suggestion rather than a blocker:
AsyncCommitManager.requestCommitalso writes the late offset intooffsetCache, and whencommitRequestedOffsets()drops the request that entry stays (only a successful commit callback removes it). If the partition later comes back to this consumer and is revoked again before any of its records is processed,commit(partition)prefers that staleoffsetCachevalue and commits it, moving back the other owner's offset, which is the case the new check is there to prevent. The window is narrow, but clearing the entry in the drop branch (consumer thread) is cheap; it needs a smallOffsetCache.remove(TopicPartition):The new test could cover it with a second} else { offsetCache.remove(partition); LOG.debug("Dropping the requested commit of offset {} on {} from {}: the partition is no longer assigned", offset, partition, threadId); }
rebalance(List.of(PARTITION))andrebalance(List.of()), checking the last committed offset stays 2.
Doc: gnodet-bot's two newer nits (4_22.adoc:775, 4_24.adoc:360) on the 30-word "A commit called after the partition was revoked…" sentence are still open; the suggested split looks good to me.
Claude Code on behalf of davsclaus. This review was generated by an AI agent and may contain inaccuracies; please verify all suggestions before applying.
JIRA: CAMEL-26978
KafkaManualCommit.commit()withDefaultKafkaManualAsyncCommitFactorycalledcommitAsyncon the calling thread. After an aggregator, every commit failed withConcurrentModificationException, and nothing was committed while the route ran.commit()now records the offset, and the consumer thread commits the requested offsets before its next poll, as in 4.21 and as the docs say. Commits are still made only when the route callscommit(), as CAMEL-23997 requires.Tests:
DefaultKafkaManualAsyncCommitTest(aMockConsumerthat only the polling thread may use): a commit after an aggregator fails on main and passes now; a commit in the Kafka route still works; a commit called after the partition is revoked is not made.Upgrade guide: entry "camel-kafka - the asynchronous manual commit is done by the consumer thread" in the 4.24 guide and in the 4.22 guide (4.22.2 to 4.22.3).
Backport: camel-4.22.x only. On camel-4.18.x,
DefaultKafkaManualAsyncCommit.commit()only records the offset, so it is not affected.Follow-up: CAMEL-26993 (with manual commits, a rebalance or a stop commits the last processed record, not the last committed one).
Claude Code on behalf of Federico Mariani