allthingssecurity commented on PR #28434: URL: https://github.com/apache/camel/pull/28434#issuecomment-6395217146
@davsclaus thanks for the review. I rebased the branch on main (4ee3a7e1dd08, which has the CAMEL-25872 fix) and added two commits: 195c07b80c75 (world and fake) and 8b5a3e0998dc (doc row). **1. The acknowledgement type** (195c07b80c75). I used your sketch. The route records `failedDeliveries` next to `completedDeliveries`. `FakeShareGroup.apply` now flags a type other than ACCEPT for a completed delivery, a type other than `onFailure` for a failed one, and REJECT for a delivery that did not fail (the existing "accepted only for a completed delivery" check stays). The archived excuse is narrower: a record that never completed passes only when the group archived it at the delivery limit, or rejected a delivery that failed while `onFailure=REJECT`. All three planted faults are detected (5000 seeds each, in `KafkaShareFetchRecords`, local only): - failed exchange → ACCEPT: 1922 of 5000 fail; - failed exchange → REJECT with `onFailure=RELEASE` (the new entry): 966 of 5000 fail, all of them `onFailure=RELEASE`, with "REJECT for failed m23#1, onFailure is RELEASE"; - `pollOnError=RECONNECT` ending the consumer thread: 377 of 5000 fail, all `RECONNECT` with poll failures, on "a started route holds one share consumer". **2. CamelContext suspend/resume and step 8** (195c07b80c75, 8b5a3e0998dc). There is now a third operator kind, `context.suspend()`, a pause, then `context.resume()`. I reverted only the main code of 20c94435789e locally (`AbstractKafkaComponent`, `KafkaShareConsumer`, `KafkaConsumer`), and the world fails on it: **2305 of 5000 seeds fail, and every one of them suspended and resumed the CamelContext**. - 1088 fail on "a started route holds one share consumer". After the resume, the consumer waits for a CamelContext start that never comes, as you expected. - 620 fail on "the route is started once the faults cleared". - 597 fail on a route resume that throws: a later start or resume of the route runs the stale queued task, and it fails with a `NullPointerException` on the executor. With the fix (main), all of them pass. Seed 3 replays with the same trace hash in two JVMs. The doc row and its catalog copy now say CAMEL-25872, and the PR description is being updated to match (it no longer calls the fix open). **3. Fake fidelity against kafka-clients 4.3.1.** There is no sources jar in my local repository, so I checked the `ShareConsumerImpl`, `ShareInFlightBatch`, `WakeupTrigger` and `Acknowledgements` bytecode with `javap`. It agrees with what you found. - [K] The callback is now invoked where `handleCompletedAcknowledgements` is called: at the start of poll, commitSync and commitAsync, before poll returns, after commitSync has its own result, and at close after the acknowledgements were sent. The Javadoc says so. - [E] I modelled the second acknowledgement as the client does. A record of the last poll can be acknowledged again until its acknowledgement is sent, and the last type wins (`Acknowledgements.add` is a `Map.put`). Once it was sent with a poll, commit or close, it throws "The record cannot be acknowledged." and counts as a violation. - [W] I added this one. A wakeup during the fetch stays pending. With no records, poll throws `WakeupException` (as `setShareFetchAction` does on the next loop). With records, the seed chooses whether they arrived. If they did, poll returns them, the stopping consumer releases them (`isRunAllowed() ? process(record) : RELEASE`), and commitSync sends the acknowledgements and then throws `WakeupException`. commitSync with nothing to send returns an empty map without a wakeup check, and commitAsync has none. A probe over 2000 seeds (not committed) reached these paths: 681 polls returned records after a wakeup, 328 commitSync calls threw after sending, and 2734 RELEASE acknowledgements went to records the consumer did not route. **4. Configuration the seed never picks** (195c07b80c75). A comment says that `pollOnError=STOP` ends the consumer thread by design, so "a started route holds one share consumer" cannot hold. `createConsumerBackoffMaxAttempts` is now seeded (0 or 3). When the fake has failed that many creations in a row and the consumer's health state is terminated with `KafkaConsumerFatalException`, the operator restarts the route once the faults cleared, one restart per give-up. In the 2000-seed probe, 106 seeds had 3 creation failures in a row, and 45 of them needed the restart. While doing this I found a false failure of the world, and fixed it. A slow exchange's locks can expire at the delivery limit and settle the group while the exchange still blocks below the finisher on the one stack, so the final stop waited for ever (step budget). The finisher now also waits until no exchange blocks. **5. Table order** (8b5a3e0998dc). The kafka-share row now comes after camel-azure-eventhubs in the page and in the catalog copy, and `cmp` says the two are identical. Runs: on main, 20000 of 20000 seeds pass, in strict mode with 0 runtime escapes. Unit tests: camel-kafka-share 1029 (29 plus 1000 seeds), camel-kafka-common 11, camel-test-dst 49, with 0 failures, errors and skips. The formatter and import sort leave no change. _Claude Code on behalf of allthingssecurity_ -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
