[
https://issues.apache.org/jira/browse/CAMEL-25910?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Work on CAMEL-25910 started by shashank.
----------------------------------------
> camel-kafka-share - deterministic simulation test of the share consumer
> -----------------------------------------------------------------------
>
> Key: CAMEL-25910
> URL: https://issues.apache.org/jira/browse/CAMEL-25910
> Project: Camel
> Issue Type: Test
> Components: camel-kafka, camel-test
> Reporter: shashank
> Assignee: shashank
> Priority: Minor
>
> camel-kafka-share (CAMEL-25029, the consumer for Kafka share groups, KIP-932)
> has unit tests against {{MockShareConsumer}}, but nothing that runs the
> consumer under faults and timing for long: expired acquisition locks, failed
> polls and commits, a share consumer that cannot be created, route restarts.
> This adds a deterministic simulation test (camel-test-dst, CAMEL-25738) for
> it, in the {{dst}} package of the camel-kafka-share tests, in the same form
> as the SQS, Event Hubs and Slack worlds.
> *What the world models*: the real consumer ({{KafkaShareConsumer}},
> {{KafkaShareFetchRecords}}) and Camel's routing run on the virtual-time loop
> of the harness; only the share group is faked ({{FakeShareGroup}}, a
> {{ShareConsumer}} as a dynamic proxy, swapped in through
> {{KafkaShareComponent.setKafkaShareClientFactory}}). The fake follows KIP-932
> and the {{KafkaShareConsumer}} Javadoc (Kafka 4.3): records acquired with a
> time-limited acquisition lock that expires, a delivery count per acquisition
> and archiving at the delivery limit, explicit acknowledgement (a poll with
> records not acknowledged, or a second acknowledgement of a record, throws
> {{IllegalStateException}} as the client does), acknowledgements sent with the
> next poll, {{commitSync}}, {{commitAsync}} or {{close}}, a stale
> acknowledgement answered with {{InvalidRecordStateException}}, {{wakeup}},
> and {{close}} releasing the acquired records that were not acknowledged. The
> seed picks {{onFailure}} (RELEASE, REJECT), {{commitMode}} (SYNC, ASYNC),
> {{pollOnError}} (ERROR_HANDLER, RECONNECT, RETRY, DISCARD),
> {{pollTimeoutMs}}, {{shutdownTimeout}}, the consumer creation backoff,
> {{maxPollRecords}}, the lock duration (5, 15, 30 s) and the delivery limit
> (5, 10), and injects share consumer creation failures, poll failures, commit
> failures, processing failures, processing slower than the lock, route
> restarts and suspend/resume. An operator starts the route again when its
> start fails, as the supervising route controller does.
> *Invariants*:
> * the share consumer is used as its Javadoc requires (no poll with records
> not acknowledged, no second or foreign acknowledgement, no call after
> {{close}});
> * a record is accepted only for a delivery the route completed;
> * every published record completes in the route at least once, or the group
> archived it (delivery limit, REJECT);
> * once the faults are cleared, the group is drained and the started route
> holds exactly one share consumer;
> * no share consumer is left open after the route stopped.
> *Seeds*: 1000 by default (about 2.5 ms per seed). On main 5000 of 5000 seeds
> pass; a seed replays with the same trace hash in two JVMs. Merged with the
> CAMEL-25872 fix, 1000 of 1000 pass.
> *Self-check*: the consumer has no past fix of this behaviour to revert, so
> two faults were planted in {{KafkaShareFetchRecords}} (locally, not
> committed):
> * a failed exchange acknowledged with ACCEPT instead of {{onFailure}}: 1953
> of 5000 seeds fail, all with "accepted ... but the route did not complete
> that delivery";
> * {{pollOnError=RECONNECT}} ending the consumer thread instead of
> reconnecting: 385 of 5000 seeds fail, all {{RECONNECT}} with poll failures,
> all with "a started route holds one share consumer".
> *Limits* (in the world's Javadoc): {{consumersCount}} is 1, because each
> consumer thread loops for ever and on the one stack of the harness a second
> looping thread nested above the first would never let it resume; processing
> takes time (and can outlast the lock) only after the route operations are
> done, because a route stop above an exchange that blocks in the route would
> wait for ever; locks do not exclude on the one stack, so a stop never waits
> for the processing lock of the consumer thread. Not modelled: several
> partitions, the batch_optimized acquire mode, lock renewal, the heartbeat and
> assignment of the group.
> *Change* (test only): the world ({{KafkaShareConsumerDstTest}},
> {{KafkaShareWorld}}, {{FakeShareGroup}}), the {{camel-test-dst}} test
> dependency, the test log of camel-kafka-share at WARN in a size-capped file,
> and the world in the list of worlds of the camel-test-dst page (and its
> catalog copy). camel-kafka-share unit tests: 1027 run (27, plus 1000 seeds),
> 0 failures.
> _Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
