This is an automated email from the ASF dual-hosted git repository.
Croway pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 65ca4a521f4f CAMEL-25910: camel-kafka-share - deterministic simulation
test of the share consumer (#28434)
65ca4a521f4f is described below
commit 65ca4a521f4fcc701e220336018f1f155e729699
Author: allthingssecurity <[email protected]>
AuthorDate: Mon Oct 26 19:31:37 2026 +0530
CAMEL-25910: camel-kafka-share - deterministic simulation test of the share
consumer (#28434)
* CAMEL-25910: camel-kafka-share - deterministic simulation test of the
share consumer
Adds a camel-test-dst world for the camel-kafka-share consumer, in the
dst package of its tests: KafkaShareConsumerDstTest (1000 seeds),
KafkaShareWorld and FakeShareGroup. The real consumer runs against a fake
share group (KIP-932) behind the ShareConsumer interface, swapped in
through KafkaShareComponent.setKafkaShareClientFactory: acquisition locks
that expire, delivery counts and the delivery limit, explicit
acknowledgement, acknowledgements sent with poll, commit and close, the
acknowledgement commit callback called with the offsets at a later poll
or commit (as kafka-clients 4.3.1 does), wakeup. The seed picks onFailure,
commitMode, pollOnError, the poll and
shutdown timeouts, the backoff, maxPollRecords, the lock duration and the
delivery limit, and injects consumer creation, poll and commit failures,
slow processing, route restarts and suspend/resume.
Invariants: the share consumer is used as its Javadoc requires; a record
is accepted only for a delivery the route completed; every published
record completes at least once or is archived by the group; the group is
drained and a started route holds one share consumer once the faults
cleared; no share consumer is left open after the route stopped.
It passes on main (5000 of 5000 seeds). There is no past fix of this
consumer to revert, so it was checked with two planted faults: a failed
exchange accepted instead of onFailure fails 1953 of 5000 seeds, and
pollOnError=RECONNECT ending the consumer thread fails 385 of 5000.
Limits (consumersCount=1, slow processing only after the operations) are
in the world's Javadoc. The test log of camel-kafka-share is at WARN and
capped in size. The world is added to the list in the camel-test-dst
page (and its catalog copy), whose intro no longer counts the worlds.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* CAMEL-25910: camel-kafka-share - the DST world checks the acknowledgement
type and suspends the CamelContext
Review of 2026-10-25:
- The acknowledgement type is checked for every delivery the group takes:
the world records the failed deliveries next to the completed ones, and
FakeShareGroup flags a type other than ACCEPT for a completed delivery,
a type other than onFailure for a failed one, and REJECT for one that
did not fail. A record that never completed is excused only when the
group archived it at the delivery limit, or rejected a delivery that
failed with onFailure=REJECT (before, any archived record was excused).
- A third operator kind suspends and resumes the CamelContext. With the
main code of CAMEL-25872 (20c94435789e) reverted, 2305 of 5000 seeds
fail, all of them with a CamelContext suspend: the resumed consumer
waits for a CamelContext start that never comes, so the route has no
share consumer, and a later start or resume of the route runs the stale
task and fails (NullPointerException on the executor).
- The fake follows kafka-clients 4.3.1 more closely: the acknowledgement
commit callback is also called before poll returns, after commitSync
has its own result, and at close; a record can be acknowledged again
until its acknowledgement is sent, and the last type wins
(ShareInFlightBatch.acknowledge); a wakeup during the fetch stays
pending, and the seed chooses whether poll still returns the records,
so the consumer releases the records it no longer routes and commitSync
sends the acknowledgements, then throws WakeupException; commitSync
with nothing to send returns an empty map.
- createConsumerBackoffMaxAttempts is seeded (0 or 3). After that many
creation failures in a row the consumer thread gives up, and once the
faults cleared the operator restarts the route. A comment says why
pollOnError=STOP is not in the default set.
- The finisher stops the route only when no exchange blocks in the
route: the locks of a slow exchange can expire and settle the group
while it still blocks below on the one stack, and a stop above it
waits for ever.
On main: 20000 of 20000 seeds pass. Planted faults (not committed): a
failed exchange acknowledged with ACCEPT fails 1922 of 5000 seeds, with
REJECT while onFailure=RELEASE 966 of 5000, and pollOnError=RECONNECT
ending the consumer thread 377 of 5000.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* CAMEL-25910: camel-kafka-share - the DST world row in alphabetical order,
as the regression test for CAMEL-25872
The kafka-share row of the camel-test-dst page (and its catalog copy,
byte-identical) comes after camel-azure-eventhubs, and names CAMEL-25872:
the world fails on the code before that fix.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* CAMEL-25910: camel-kafka-share - the DST fake flags a RELEASE of a
delivery the consumer did not route while it was not stopping
KafkaShareFetchRecords releases a record without routing it only when it
stops, and stop() always wakes up the share consumer. The fake share
consumer now remembers that it was woken up (never cleared), and a
RELEASE of a delivery that the route neither completed nor failed before
that is a violation. Without it, a consumer that never routes passed: it
released every record until the group archived it at the delivery limit,
which the world excuses.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* CAMEL-25910: camel-kafka-share - the DST world notes that its RELEASE
check relies on the one-stack lock limit
Review follow-up, comments only: the fake flags a RELEASE of a delivery the
route did not route before the first
wakeup of the share consumer. That holds in the world because a stop never
waits for the processing lock on the one
stack, so the stop wakes up the share consumer before the consumer thread
releases the rest of its batch. With real
threads, KafkaShareFetchRecords.stop() waits for that lock, and the rest of
the batch is released before the wakeup.
The contract bullet of FakeShareGroup and the limits in the KafkaShareWorld
Javadoc now say so, so that a later change
that models the lock wait does not read this as a consumer bug.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* CAMEL-25910: camel-kafka-share - say in the fake share group which
kafka-clients version its rules follow
Review follow-up: the class Javadoc of FakeShareGroup now says that the
rules were written against
kafka-clients 4.3.1 and should be checked again when kafka-clients is
upgraded.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---------
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../org/apache/camel/catalog/docs/test-dst.adoc | 2 +
components/camel-kafka-share/pom.xml | 5 +
.../component/kafka/share/dst/FakeShareGroup.java | 581 +++++++++++++++++++++
.../kafka/share/dst/KafkaShareConsumerDstTest.java | 41 ++
.../component/kafka/share/dst/KafkaShareWorld.java | 398 ++++++++++++++
.../src/test/resources/log4j2.properties | 10 +-
.../camel-test-dst/src/main/docs/test-dst.adoc | 2 +
7 files changed, 1037 insertions(+), 2 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/test-dst.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/test-dst.adoc
index 16bf1c96d140..9c0d809a57ad 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/test-dst.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/test-dst.adoc
@@ -60,6 +60,8 @@ These worlds are in the Camel code base and are the examples
to copy:
|camel-aws2-sqs |`SqsRedeliveryDstTest` (the extender while Camel's error
handler waits to redeliver, in virtual
time) |-
|camel-azure-eventhubs |`EventHubsCheckpointDstTest` (checkpointing by batch
size and timeout) |CAMEL-25683
+|camel-kafka-share |`KafkaShareConsumerDstTest` (acknowledgements, acquisition
locks, reconnects, and suspend and resume
+of the route and of the CamelContext) |CAMEL-25872
|camel-pulsar |`PulsarConsumerDstTest` (subscribe failures, broker and route
restarts while the route starts and
consumes) |CAMEL-25869
|camel-slack |`SlackSuspendDstTest` (route suspend in the middle of a poll)
|CAMEL-25684
diff --git a/components/camel-kafka-share/pom.xml
b/components/camel-kafka-share/pom.xml
index 43f005b567c9..4e854da76914 100644
--- a/components/camel-kafka-share/pom.xml
+++ b/components/camel-kafka-share/pom.xml
@@ -59,6 +59,11 @@
<version>${project.version}</version>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-test-dst</artifactId>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.assertj</groupId>
<artifactId>assertj-core</artifactId>
diff --git
a/components/camel-kafka-share/src/test/java/org/apache/camel/component/kafka/share/dst/FakeShareGroup.java
b/components/camel-kafka-share/src/test/java/org/apache/camel/component/kafka/share/dst/FakeShareGroup.java
new file mode 100644
index 000000000000..2d079d22856f
--- /dev/null
+++
b/components/camel-kafka-share/src/test/java/org/apache/camel/component/kafka/share/dst/FakeShareGroup.java
@@ -0,0 +1,581 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.kafka.share.dst;
+
+import java.lang.reflect.InvocationHandler;
+import java.lang.reflect.Method;
+import java.lang.reflect.Proxy;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.TreeSet;
+
+import org.apache.camel.test.dst.Dst;
+import org.apache.camel.test.dst.DstLedger;
+import org.apache.kafka.clients.consumer.AcknowledgeType;
+import org.apache.kafka.clients.consumer.AcknowledgementCommitCallback;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.ShareConsumer;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.TopicIdPartition;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.errors.InvalidRecordStateException;
+import org.apache.kafka.common.errors.WakeupException;
+import org.apache.kafka.common.header.internals.RecordHeaders;
+import org.apache.kafka.common.record.TimestampType;
+
+/**
+ * A fake Kafka share group (KIP-932) for one topic with one partition, and
fake share consumers behind the
+ * {@link ShareConsumer} interface (dynamic proxies).
+ * <p>
+ * Written against kafka-clients 4.3.1: when kafka-clients is upgraded, check
the rules below ([A] to [X]) against the
+ * new ShareConsumerImpl and ShareInFlightBatch.
+ * <p>
+ * Rules (KIP-932 and the KafkaShareConsumer Javadoc, Kafka 4.3):
+ * <ul>
+ * <li>[A] "records are acquired for a consumer ... with a time-limited
acquisition lock"
+ * (group.share.record.lock.duration.ms). "If the lock times out, the record
becomes available again". Every acquisition
+ * increments the delivery count; "once the delivery count reaches
group.share.delivery.count.limit the record is
+ * archived" instead of becoming available.</li>
+ * <li>[E] explicit acknowledgement: "each record ... must be acknowledged ...
using acknowledge(ConsumerRecord,
+ * AcknowledgeType) before the next call to poll", otherwise poll throws
IllegalStateException ("All records must be
+ * acknowledged in explicit acknowledgement mode."). A record that is not
waiting for an acknowledgement throws
+ * IllegalStateException ("The record cannot be acknowledged."). As
kafka-clients 4.3.1 ShareInFlightBatch.acknowledge
+ * does, a record of the last poll can be acknowledged again until its
acknowledgement is sent, and the last type wins;
+ * once it was sent (by a poll, commit or close), it is no longer waiting and
throws.</li>
+ * <li>[C] the acknowledgements are sent with the next poll, commitSync,
commitAsync or close. commitSync returns the
+ * result per partition; an acknowledgement of a record whose lock expired
fails (InvalidRecordStateException).</li>
+ * <li>[K] the acknowledgement commit callback gets the result of the
acknowledgements that were sent, with their
+ * offsets, where kafka-clients 4.3.1 ShareConsumerImpl calls
handleCompletedAcknowledgements: at the start of poll,
+ * commitSync and commitAsync, before poll returns (the acknowledgements sent
with that poll have their result by then
+ * in the fake), after commitSync has its own result, and at close after the
acknowledgements were sent. Not from within
+ * commitAsync.</li>
+ * <li>[W] wakeup: "the thread which is blocking in an operation will throw
WakeupException. If no thread is blocking
+ * ... the next call will raise it". As in kafka-clients 4.3.1: poll checks
for a wakeup before it fetches; a wakeup
+ * while poll waits for the fetch wakes the fetch buffer and stays pending, so
poll throws WakeupException when no
+ * records were fetched, and otherwise the seed chooses whether the records
arrived (poll returns them) or not (poll
+ * throws). commitSync with acknowledgements to send sends them, then throws a
pending WakeupException; commitSync with
+ * nothing to send returns an empty map and leaves the wakeup pending.
commitAsync does not check for a wakeup.</li>
+ * <li>[X] close: the pending acknowledgements are sent; acquired records that
were not acknowledged are released (the
+ * delivery count is kept).</li>
+ * <li>ACCEPT: done. RELEASE: available again (or archived at the delivery
limit). REJECT: archived.</li>
+ * <li>Camel's contract (KafkaShareFetchRecords, the route never sets the
acknowledge header): RELEASE of a delivery
+ * that the route neither completed nor failed only once the consumer stops,
and it always wakes up its share consumer
+ * when it stops. Before the first wakeup of a share consumer such a RELEASE
is a violation. This check relies on the
+ * stop not waiting for the processing lock (see the limits in {@link
KafkaShareWorld}): with real threads,
+ * KafkaShareFetchRecords.stop() waits for that lock, and the consumer thread
releases the rest of the batch before the
+ * wakeup.</li>
+ * </ul>
+ * Not modelled: several partitions, the batch_optimized acquire mode (a poll
returns at most max.poll.records), lock
+ * renewal (RENEW), the heartbeat and assignment of the group.
+ * <p>
+ * Timing (one-stack rule): poll waits first and acquires at the end of its
wait; acknowledgements take effect at once;
+ * commit and close do not block (the consumer calls commit while it holds its
lock).
+ */
+final class FakeShareGroup {
+
+ static final String TOPIC = "dst";
+ static final Uuid TOPIC_ID = new Uuid(4711L, 42L);
+ static final TopicIdPartition PARTITION = new TopicIdPartition(TOPIC_ID,
0, TOPIC);
+
+ enum State {
+ AVAILABLE,
+ ACQUIRED,
+ ACKNOWLEDGED,
+ ARCHIVED
+ }
+
+ /** The result of acknowledgements that were sent, for the acknowledgement
commit callback. */
+ record CompletedAcks(Map<TopicIdPartition, Set<Long>> offsets, Exception
error) {
+ }
+
+ static final class Rec {
+ final long offset;
+ final String id;
+ State state = State.AVAILABLE;
+ int deliveries;
+ // the delivery that was rejected (archived by REJECT), 0 if none
+ int rejected;
+ long lockUntil;
+ Fake owner;
+
+ Rec(long offset, String id) {
+ this.offset = offset;
+ this.id = id;
+ }
+ }
+
+ static final class Delivery {
+ final Rec rec;
+ final int number;
+ final ConsumerRecord<Object, Object> record;
+ AcknowledgeType ack;
+ // the acknowledgement was sent (with a poll, commit or close)
+ boolean sent;
+
+ Delivery(Rec rec, int number, ConsumerRecord<Object, Object> record) {
+ this.rec = rec;
+ this.number = number;
+ this.record = record;
+ }
+ }
+
+ final class Fake implements InvocationHandler {
+ final String name;
+ final List<Delivery> lastBatch = new ArrayList<>();
+ final List<Delivery> pendingAcks = new ArrayList<>();
+ boolean closed;
+ // a wakeup that is pending (cleared when it is thrown)
+ boolean wakeup;
+ // woken up at least once (never cleared): the consumer stops it with
a wakeup, so only from then on it may
+ // release a delivery that it did not route
+ boolean wokenUp;
+ AcknowledgementCommitCallback callback;
+ // [K] the results of the acknowledgements sent, for the callback at
the next poll or commit
+ final List<CompletedAcks> completedAcks = new ArrayList<>();
+
+ Fake(String name) {
+ this.name = name;
+ }
+
+ @Override
+ public Object invoke(Object proxy, Method method, Object[] args) {
+ String m = method.getName();
+ switch (m) {
+ case "toString":
+ return "FakeShareConsumer[" + name + "]";
+ case "hashCode":
+ return System.identityHashCode(this);
+ case "equals":
+ return proxy == args[0];
+ case "wakeup":
+ dst.trace("fake: " + name + " wakeup");
+ wakeup = true;
+ wokenUp = true;
+ return null;
+ case "close":
+ close();
+ return null;
+ default:
+ break;
+ }
+ checkOpen(m);
+ switch (m) {
+ case "subscribe":
+ @SuppressWarnings("unchecked")
+ Collection<String> topics = (Collection<String>) args[0];
+ if (!topics.contains(TOPIC)) {
+ violations.add(name + " subscribed to " + topics);
+ }
+ return null;
+ case "unsubscribe":
+ return null;
+ case "subscription":
+ return Set.of(TOPIC);
+ case "setAcknowledgementCommitCallback":
+ callback = (AcknowledgementCommitCallback) args[0];
+ return null;
+ case "poll":
+ return poll((Duration) args[0]);
+ case "acknowledge":
+ if (args.length != 2) {
+ violations.add(name + " acknowledge" + args.length);
+ throw new IllegalStateException("not modelled");
+ }
+ @SuppressWarnings("unchecked")
+ ConsumerRecord<Object, Object> record =
(ConsumerRecord<Object, Object>) args[0];
+ acknowledge(record, (AcknowledgeType) args[1]);
+ return null;
+ case "commitSync":
+ return commitSync();
+ case "commitAsync":
+ commitAsync();
+ return null;
+ case "acquisitionLockTimeoutMs":
+ return Optional.of((int) lockDurationMillis);
+ case "metrics":
+ return Map.of();
+ default:
+ violations.add(name + " called " + m + " (not modelled)");
+ throw new UnsupportedOperationException(m);
+ }
+ }
+
+ private void checkOpen(String m) {
+ if (closed) {
+ violations.add(name + " used after close: " + m);
+ throw new IllegalStateException("This consumer has already
been closed.");
+ }
+ }
+
+ /** [K] the callback gets the results of the acknowledgements sent
before this call. */
+ private void handleCompletedAcknowledgements() {
+ List<CompletedAcks> completed = new ArrayList<>(completedAcks);
+ completedAcks.clear();
+ if (callback != null) {
+ for (CompletedAcks c : completed) {
+ callback.onComplete(c.offsets(), c.error());
+ }
+ }
+ }
+
+ private ConsumerRecords<Object, Object> poll(Duration timeout) {
+ handleCompletedAcknowledgements();
+ // [E]
+ for (Delivery d : lastBatch) {
+ if (d.ack == null) {
+ violations.add(name + " polled before acknowledging " +
d.rec.id + "#" + d.number);
+ dst.block(10, "poll refused");
+ throw new IllegalStateException("All records must be
acknowledged in explicit acknowledgement mode.");
+ }
+ }
+ // [W] a pending wakeup, before the fetch
+ if (wakeup) {
+ wakeup = false;
+ throw new WakeupException();
+ }
+ lastBatch.clear();
+ // [C] the acknowledgements go with the poll
+ send(false);
+ if (faults && dst.chance(pollFailRate)) {
+ dst.block(dst.between(1, 50), "poll fails");
+ ledger.record("pollFault", "c", name);
+ throw Dst.injected(new KafkaException("dst: poll failed"));
+ }
+ long waitMillis = Math.max(1, timeout.toMillis());
+ boolean ready = dst.awaitUntil(() -> wakeup || closed ||
hasAvailable(), waitMillis, name + " poll");
+ if (ready && !wakeup) {
+ // reads take effect at the end of their latency
+ dst.block(dst.between(1, 20), name + " fetch");
+ }
+ if (closed) {
+ violations.add(name + " closed while polling");
+ throw new IllegalStateException("This consumer has already
been closed.");
+ }
+ if (wakeup) {
+ // [W] the wakeup woke the fetch and stays pending: without
records the next fetch throws it, and
+ // with records the seed chooses whether they were fetched
before the wakeup
+ if (!hasAvailable() || dst.chance(0.5)) {
+ wakeup = false;
+ throw new WakeupException();
+ }
+ dst.trace("fake: " + name + " woken up during the fetch, the
records arrived, the wakeup stays pending");
+ }
+ expire();
+ Map<TopicPartition, List<ConsumerRecord<Object, Object>>> map =
new LinkedHashMap<>();
+ List<ConsumerRecord<Object, Object>> list = new ArrayList<>();
+ for (Rec r : log) {
+ if (list.size() >= maxPollRecords) {
+ break;
+ }
+ if (r.state == State.AVAILABLE) {
+ r.state = State.ACQUIRED;
+ r.deliveries++;
+ r.lockUntil = dst.now() + lockDurationMillis;
+ r.owner = this;
+ ConsumerRecord<Object, Object> cr = new ConsumerRecord<>(
+ TOPIC, 0, r.offset, Dst.EPOCH_MILLIS + dst.now(),
TimestampType.CREATE_TIME, -1, -1,
+ null, r.id, new RecordHeaders(), Optional.empty(),
Optional.of((short) r.deliveries));
+ lastBatch.add(new Delivery(r, r.deliveries, cr));
+ list.add(cr);
+ ledger.record("delivered", "t", r.id);
+ }
+ }
+ if (!list.isEmpty()) {
+ map.put(new TopicPartition(TOPIC, 0), list);
+ dst.trace("fake: " + name + " poll returns " + list.size() + "
records");
+ }
+ // [K] before poll returns
+ handleCompletedAcknowledgements();
+ return new ConsumerRecords<>(map, Map.of());
+ }
+
+ private void acknowledge(ConsumerRecord<Object, Object> record,
AcknowledgeType type) {
+ for (Delivery d : lastBatch) {
+ if (d.rec.offset == record.offset()) {
+ // [E] no longer waiting for an acknowledgement once it
was sent
+ if (d.sent) {
+ violations.add(name + " acknowledged " + d.rec.id +
"#" + d.number + " after its " + d.ack.name()
+ + " was sent");
+ throw new IllegalStateException("The record cannot be
acknowledged.");
+ }
+ // RELEASE is for a delivery that the stopping consumer
did not route, and the consumer always
+ // wakes up its share consumer when it stops
+ String key = d.rec.id + "#" + d.number;
+ if (type == AcknowledgeType.RELEASE && !wokenUp &&
!completedDeliveries.contains(key)
+ && !failedDeliveries.contains(key)) {
+ violations.add(name + " released " + key + " without
routing it, and it was not stopping");
+ }
+ if (d.ack == null) {
+ pendingAcks.add(d);
+ } else {
+ // [E] not sent yet: the last type wins
+ dst.trace("fake: " + name + " " + d.rec.id + "#" +
d.number + " acknowledged again, "
+ + d.ack.name() + " -> " + type.name());
+ }
+ d.ack = type;
+ return;
+ }
+ }
+ violations.add(name + " acknowledged a record not in its last
poll: offset " + record.offset());
+ throw new IllegalStateException("The record cannot be
acknowledged.");
+ }
+
+ private Map<TopicIdPartition, Optional<KafkaException>> commitSync() {
+ handleCompletedAcknowledgements();
+ if (pendingAcks.isEmpty()) {
+ // [W] nothing to send: no wakeup check
+ return Map.of();
+ }
+ KafkaException error = send(true);
+ // [W] the acknowledgements were sent before the wait for their
result is woken up
+ if (wakeup) {
+ wakeup = false;
+ throw new WakeupException();
+ }
+ Map<TopicIdPartition, Optional<KafkaException>> result = new
LinkedHashMap<>();
+ result.put(PARTITION, Optional.ofNullable(error));
+ // [K] after its own result
+ handleCompletedAcknowledgements();
+ return result;
+ }
+
+ private void commitAsync() {
+ handleCompletedAcknowledgements();
+ // the result reaches the callback at a later poll or commit
+ send(true);
+ }
+
+ /** Sends the pending acknowledgements. */
+ private KafkaException send(boolean mayFail) {
+ if (pendingAcks.isEmpty()) {
+ return null;
+ }
+ List<Delivery> acks = new ArrayList<>(pendingAcks);
+ pendingAcks.clear();
+ Set<Long> offsets = new TreeSet<>();
+ for (Delivery d : acks) {
+ d.sent = true;
+ offsets.add(d.rec.offset);
+ }
+ if (mayFail && faults && dst.chance(commitFailRate)) {
+ for (Delivery d : acks) {
+ ledger.record("ackLost", "t", d.rec.id);
+ }
+ dst.trace("fake: " + name + " commit fails, " + acks.size() +
" acknowledgements lost");
+ KafkaException failure = Dst.injected(new KafkaException("dst:
commit failed"));
+ completedAcks.add(new CompletedAcks(Map.of(PARTITION,
offsets), failure));
+ return failure;
+ }
+ expire();
+ KafkaException error = null;
+ for (Delivery d : acks) {
+ Rec r = d.rec;
+ if (r.state != State.ACQUIRED || r.deliveries != d.number ||
r.owner != this) {
+ ledger.record("ackStale", "t", r.id);
+ error = new InvalidRecordStateException("dst: the lock of
" + r.id + "#" + d.number + " expired");
+ continue;
+ }
+ apply(r, d.number, d.ack);
+ }
+ completedAcks.add(new CompletedAcks(Map.of(PARTITION, offsets),
error));
+ return error;
+ }
+
+ private void close() {
+ if (closed) {
+ return;
+ }
+ dst.trace("fake: " + name + " close");
+ send(false);
+ expire();
+ // [X]
+ for (Delivery d : lastBatch) {
+ Rec r = d.rec;
+ if (d.ack == null && r.state == State.ACQUIRED && r.deliveries
== d.number && r.owner == this) {
+ release(r, "released on close");
+ }
+ }
+ lastBatch.clear();
+ closed = true;
+ open.remove(this);
+ // [K] at close, after the acknowledgements were sent
+ handleCompletedAcknowledgements();
+ }
+ }
+
+ private final Dst dst;
+ private final DstLedger ledger;
+ final List<Rec> log = new ArrayList<>();
+ final List<String> violations = new ArrayList<>();
+ final Set<Fake> open = new HashSet<>();
+ // the deliveries (id#deliveryCount) whose exchange completed, and those
whose exchange failed
+ final Set<String> completedDeliveries = new HashSet<>();
+ final Set<String> failedDeliveries = new HashSet<>();
+ // the acknowledgement type of a failed exchange (the onFailure option of
the endpoint)
+ AcknowledgeType expectedOnFailure = AcknowledgeType.RELEASE;
+ long lockDurationMillis = 30000;
+ int deliveryLimit = 5;
+ int maxPollRecords = 500;
+ double createFailRate;
+ // the createConsumerBackoffMaxAttempts of the component (0: retry for
ever), and how often the creation failed that
+ // many times in a row
+ int createConsumerBackoffMaxAttempts;
+ int createGiveUps;
+ private int createFaultsInARow;
+ double pollFailRate;
+ double commitFailRate;
+ boolean faults = true;
+ private int created;
+
+ FakeShareGroup(Dst dst, DstLedger ledger) {
+ this.dst = dst;
+ this.ledger = ledger;
+ }
+
+ /** KafkaShareClientFactory.getShareConsumer. */
+ @SuppressWarnings("unchecked")
+ ShareConsumer<Object, Object> newConsumer() {
+ if (faults && dst.chance(createFailRate)) {
+ ledger.record("createFault", "c", "c" + created);
+ createFaultsInARow++;
+ if (createConsumerBackoffMaxAttempts > 0 && createFaultsInARow ==
createConsumerBackoffMaxAttempts) {
+ // as many failed attempts in a row as the consumer makes
before it gives up
+ createGiveUps++;
+ createFaultsInARow = 0;
+ dst.trace("fake: " + createConsumerBackoffMaxAttempts + "
creation failures in a row");
+ }
+ throw Dst.injected(new KafkaException("dst: Failed to construct
kafka share consumer"));
+ }
+ createFaultsInARow = 0;
+ Fake fake = new Fake("c" + created++);
+ open.add(fake);
+ dst.trace("fake: created " + fake.name);
+ return (ShareConsumer<Object, Object>) Proxy.newProxyInstance(
+ ShareConsumer.class.getClassLoader(), new Class<?>[] {
ShareConsumer.class }, fake);
+ }
+
+ void publish(String id) {
+ log.add(new Rec(log.size(), id));
+ ledger.record("published", "t", id);
+ }
+
+ boolean hasAvailable() {
+ expire();
+ for (Rec r : log) {
+ if (r.state == State.AVAILABLE) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /** [A] locks that timed out. */
+ void expire() {
+ long now = dst.now();
+ for (Rec r : log) {
+ if (r.state == State.ACQUIRED && now >= r.lockUntil) {
+ ledger.record("lockExpired", "t", r.id);
+ release(r, "lock expired");
+ }
+ }
+ }
+
+ /**
+ * Applies an acknowledgement that the group took. The route never sets
the acknowledge header, so the type is
+ * exact: ACCEPT for a delivery the route completed, onFailure for one
that failed, and RELEASE only for one the
+ * consumer did not route (it was stopping; checked in acknowledge,
against the wakeup).
+ */
+ private void apply(Rec r, int number, AcknowledgeType type) {
+ String key = r.id + "#" + number;
+ if (type == AcknowledgeType.ACCEPT &&
!completedDeliveries.contains(key)) {
+ violations.add("accepted " + key + " but the route did not
complete that delivery");
+ }
+ if (type != AcknowledgeType.ACCEPT &&
completedDeliveries.contains(key)) {
+ violations.add(type.name() + " for " + key + " which the route
completed");
+ }
+ if (failedDeliveries.contains(key) && type != expectedOnFailure) {
+ violations.add(type.name() + " for failed " + key + ", onFailure
is " + expectedOnFailure.name());
+ }
+ if (type == AcknowledgeType.REJECT && !failedDeliveries.contains(key))
{
+ violations.add("rejected " + key + " which did not fail");
+ }
+ if (type == AcknowledgeType.ACCEPT) {
+ r.state = State.ACKNOWLEDGED;
+ r.owner = null;
+ ledger.record("accepted", "t", r.id);
+ } else if (type == AcknowledgeType.REJECT) {
+ r.state = State.ARCHIVED;
+ r.rejected = number;
+ r.owner = null;
+ ledger.record("rejected", "t", r.id);
+ } else if (type == AcknowledgeType.RELEASE) {
+ release(r, "released");
+ } else {
+ violations.add("acknowledgement type " + type + " not modelled");
+ }
+ }
+
+ private void release(Rec r, String why) {
+ r.owner = null;
+ if (r.deliveries >= deliveryLimit) {
+ r.state = State.ARCHIVED;
+ ledger.record("archived", "t", r.id);
+ } else {
+ r.state = State.AVAILABLE;
+ }
+ dst.trace("fake: " + r.id + " " + why + " -> " + r.state);
+ }
+
+ boolean settled() {
+ expire();
+ for (Rec r : log) {
+ if (r.state != State.ACKNOWLEDGED && r.state != State.ARCHIVED) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ String unsettled() {
+ StringBuilder sb = new StringBuilder();
+ for (Rec r : log) {
+ if (r.state != State.ACKNOWLEDGED && r.state != State.ARCHIVED) {
+
sb.append(r.id).append('=').append(r.state).append("/d").append(r.deliveries)
+ .append(r.owner != null ? "@" + r.owner.name :
"").append(' ');
+ }
+ }
+ return sb.toString().trim();
+ }
+
+ String openNames() {
+ List<String> names = new ArrayList<>();
+ for (Fake f : open) {
+ names.add(f.name);
+ }
+ names.sort(null);
+ return names.toString();
+ }
+}
diff --git
a/components/camel-kafka-share/src/test/java/org/apache/camel/component/kafka/share/dst/KafkaShareConsumerDstTest.java
b/components/camel-kafka-share/src/test/java/org/apache/camel/component/kafka/share/dst/KafkaShareConsumerDstTest.java
new file mode 100644
index 000000000000..0985328de40f
--- /dev/null
+++
b/components/camel-kafka-share/src/test/java/org/apache/camel/component/kafka/share/dst/KafkaShareConsumerDstTest.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.kafka.share.dst;
+
+import org.apache.camel.test.dst.Dst;
+import org.apache.camel.test.dst.DstInstrument;
+import org.apache.camel.test.dst.DstTest;
+
+/**
+ * Deterministic simulation test of the camel-kafka-share consumer against a
fake share group, with consumer creation
+ * failures, poll and commit failures, acquisition locks that expire under
slow processing, route restarts, suspend and
+ * resume of the route and of the CamelContext: every published record
completes in the route at least once or is
+ * archived by the group as it should be, and each delivery is acknowledged
with the type its exchange calls for (ACCEPT
+ * when it completed, onFailure when it failed). A failing seed is replayed
with {@code -Ddst.seed=<seed>};
+ * {@code -Ddst.seeds=5000} runs more seeds; {@code
-Ddst.share.pollOnError=RECONNECT} runs one pollOnError strategy.
+ */
+@DstInstrument(value = "org.apache.camel.component.kafka.share",
+ alsoRewrite = {
+ "org.apache.camel.util.StopWatch",
"org.apache.camel.impl.engine.DefaultShutdownStrategy",
+ "org.apache.camel.support.task" })
+class KafkaShareConsumerDstTest {
+
+ @DstTest(seeds = 1000)
+ void everyRecordCompletesOrIsArchivedAndOnlyCompletedOnesAreAccepted(Dst
dst) throws Exception {
+ new KafkaShareWorld(dst).run();
+ }
+}
diff --git
a/components/camel-kafka-share/src/test/java/org/apache/camel/component/kafka/share/dst/KafkaShareWorld.java
b/components/camel-kafka-share/src/test/java/org/apache/camel/component/kafka/share/dst/KafkaShareWorld.java
new file mode 100644
index 000000000000..5ec8f75b2665
--- /dev/null
+++
b/components/camel-kafka-share/src/test/java/org/apache/camel/component/kafka/share/dst/KafkaShareWorld.java
@@ -0,0 +1,398 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.kafka.share.dst;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Properties;
+
+import org.apache.camel.ServiceStatus;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.kafka.KafkaConsumerFatalException;
+import org.apache.camel.component.kafka.TaskHealthState;
+import org.apache.camel.component.kafka.share.KafkaShareClientFactory;
+import org.apache.camel.component.kafka.share.KafkaShareComponent;
+import org.apache.camel.component.kafka.share.KafkaShareConfiguration;
+import org.apache.camel.component.kafka.share.KafkaShareConstants;
+import org.apache.camel.component.kafka.share.KafkaShareConsumer;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.test.dst.Dst;
+import org.apache.camel.test.dst.DstCamel;
+import org.apache.camel.test.dst.DstExecutor;
+import org.apache.camel.test.dst.DstLedger;
+import org.apache.camel.test.dst.Invariants;
+import org.apache.kafka.clients.consumer.AcknowledgeType;
+import org.apache.kafka.clients.consumer.ShareConsumer;
+
+/**
+ * The world around a camel-kafka-share consumer: a fake share group ({@link
FakeShareGroup}), a producer, a route whose
+ * processing is sometimes slow (longer than the acquisition lock) and
sometimes fails, and an operator that starts,
+ * restarts, suspends and resumes the route, and suspends and resumes the
CamelContext. Faults: the share consumer
+ * cannot be created, polls fail, commits fail, locks expire, a wakeup arrives
while a poll fetches.
+ * <p>
+ * consumersCount is 1: a consumer thread loops for ever, and with the
one-stack model a second looping thread nested
+ * above the first would never let it resume. For the same reason a route stop
that runs while the consumer thread waits
+ * (in poll or in the route) cannot see it finish: the stop waits for its
timeouts and the thread ends later. A stop
+ * above an exchange that blocks in the route would wait for ever, so the
processing takes time (and can outlast the
+ * acquisition lock) only once the operations are done. The locks of the
consumer do not exclude on the one stack, so a
+ * stop never waits for the processing lock of the consumer thread. The
RELEASE check of the fake relies on this: with
+ * real threads, the stop waits for that lock, and the consumer thread
releases the rest of the batch before the wakeup.
+ * <p>
+ * Properties: every published record completes in the route at least once,
unless the share group archived it at the
+ * delivery limit, or rejected a delivery that failed with onFailure=REJECT;
the acknowledgement type is exact (ACCEPT
+ * for a delivery the route completed, onFailure for one that failed, REJECT
for no other, and RELEASE of a delivery the
+ * route did not route only once the consumer woke up its share consumer to
stop, so a consumer that does not route
+ * cannot pass by releasing every record until the group archives it); the
group is drained once the faults cleared; a
+ * started route holds one open share consumer, a stopped one none; the
consumer uses the client as the Javadoc
+ * requires.
+ * <p>
+ * When the seed sets createConsumerBackoffMaxAttempts and the creation failed
that many times in a row, the consumer
+ * thread gives up and the route has no share consumer: once the faults
cleared, the operator restarts the route, once
+ * for each such run of failures.
+ */
+final class KafkaShareWorld {
+
+ static final String ROUTE = "share";
+
+ private final Dst dst;
+ private final DstLedger ledger;
+ private final FakeShareGroup group;
+ private final String onFailure;
+ private final String commitMode;
+ private final String pollOnError;
+ private final long pollTimeoutMs;
+ private final long shutdownTimeout;
+ private final long backoffInterval;
+ private final int backoffMaxAttempts;
+ private int restartsAfterGiveUp;
+ private final List<String> violations = new ArrayList<>();
+ private boolean faults = true;
+ private int sendsLeft;
+ private int opsLeft;
+ // exchanges that block in the route
+ private int blocked;
+ private boolean finished;
+ private String finalStatus;
+ private String finalOpen;
+ private int finalOpenCount;
+ private boolean finalSettled;
+
+ KafkaShareWorld(Dst dst) {
+ this.dst = dst;
+ this.ledger = new DstLedger(dst);
+ this.group = new FakeShareGroup(dst, ledger);
+ this.onFailure = dst.pick(List.of("RELEASE", "REJECT"));
+ group.expectedOnFailure = AcknowledgeType.valueOf(onFailure);
+ this.commitMode = dst.pick(List.of("SYNC", "ASYNC"));
+ // not STOP: it ends the consumer thread by design, so "a started
route holds one share consumer" cannot hold
+ this.pollOnError = dst.pick(Arrays.asList(
+ System.getProperty("dst.share.pollOnError",
"ERROR_HANDLER,RECONNECT,RETRY,DISCARD").split(",")));
+ this.pollTimeoutMs = dst.pick(List.of(100L, 1000L, 5000L));
+ this.shutdownTimeout = dst.pick(List.of(1000L, 30000L));
+ this.backoffInterval = dst.pick(List.of(500L, 5000L));
+ this.backoffMaxAttempts = dst.pick(List.of(0, 3));
+ group.createConsumerBackoffMaxAttempts = backoffMaxAttempts;
+ group.maxPollRecords = dst.pick(List.of(1, 3, 10));
+ group.lockDurationMillis = dst.pick(List.of(5000L, 15000L, 30000L));
+ group.deliveryLimit = dst.pick(List.of(5, 10));
+ group.createFailRate = dst.pick(List.of(0.0, 0.2, 0.5));
+ group.pollFailRate = dst.pick(List.of(0.0, 0.02, 0.1));
+ group.commitFailRate = dst.pick(List.of(0.0, 0.05));
+ dst.trace("world: " + config());
+ }
+
+ private String config() {
+ return "onFailure=" + onFailure + " commitMode=" + commitMode + "
pollOnError=" + pollOnError
+ + " pollTimeoutMs=" + pollTimeoutMs + " shutdownTimeout=" +
shutdownTimeout + " backoff=" + backoffInterval
+ + " backoffMaxAttempts=" + backoffMaxAttempts
+ + " maxPollRecords=" + group.maxPollRecords + " lock=" +
group.lockDurationMillis + " deliveryLimit="
+ + group.deliveryLimit + " createFail=" + group.createFailRate +
" pollFail=" + group.pollFailRate
+ + " commitFail=" + group.commitFailRate;
+ }
+
+ void run() throws Exception {
+ DefaultCamelContext context = DstCamel.newContext(dst);
+ KafkaShareComponent component = new KafkaShareComponent();
+ component.setCreateConsumerBackoffInterval(backoffInterval);
+ component.setCreateConsumerBackoffMaxAttempts(backoffMaxAttempts);
+ component.setKafkaShareClientFactory(new KafkaShareClientFactory() {
+ @Override
+ public ShareConsumer<Object, Object> getShareConsumer(Properties
kafkaProps) {
+ return group.newConsumer();
+ }
+
+ @Override
+ public String getBrokers(KafkaShareConfiguration configuration) {
+ return configuration.getBrokers();
+ }
+ });
+ context.addComponent("kafka-share", component);
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("kafka-share:" + FakeShareGroup.TOPIC +
"?brokers=127.0.0.1:9092&groupId=g&consumersCount=1"
+ + "&pollTimeoutMs=" + pollTimeoutMs + "&maxPollRecords="
+ group.maxPollRecords
+ + "&acquireMode=record_limit&onFailure=" + onFailure +
"&commitMode=" + commitMode
+ + "&pollOnError=" + pollOnError + "&shutdownTimeout=" +
shutdownTimeout)
+ .routeId(ROUTE).autoStartup(false)
+ .process(exchange -> {
+ String id =
exchange.getMessage().getBody(String.class);
+ Object d =
exchange.getMessage().getHeader(KafkaShareConstants.DELIVERY_COUNT);
+ ledger.record("routed", "t", id);
+ // a route stop cannot see an exchange that blocks
below it on the one stack finish (the
+ // shutdown strategy waits for it for ever), so
the processing takes time only once the
+ // operations are done, and the finisher stops the
route only when no exchange blocks
+ if (faults && opsLeft == 0) {
+ long latency = dst.chance(0.1)
+ ? group.lockDurationMillis +
dst.between(0, group.lockDurationMillis)
+ : dst.between(0, 300);
+ blocked++;
+ try {
+ dst.block(latency, "process " + id + "#" +
d);
+ } finally {
+ blocked--;
+ }
+ }
+ if (faults && dst.chance(0.05)) {
+ group.failedDeliveries.add(id + "#" + d);
+ ledger.record("failed", "t", id);
+ throw Dst.injected(new
IllegalStateException("dst: processing failed for " + id));
+ }
+ group.completedDeliveries.add(id + "#" + d);
+ ledger.record("completed", "t", id);
+ });
+ }
+ });
+ context.start();
+ DstExecutor operator = dst.executor("operator", 1);
+ try {
+ operator.schedule("start route", 0, () -> startRoute(context,
operator));
+ planSends();
+ planOperations(context, operator);
+ operator.schedule("finisher", 1000, () -> finisher(context,
operator, -1));
+ dst.runUntil(() -> finished, "the finisher");
+ // the consumer thread ends after the stop; let it close its share
consumer
+ dst.runFor(2 * shutdownTimeout + 10_000);
+ dst.check(
+ Invariants.of("the consumer uses the share consumer as
documented",
+ () -> group.violations.isEmpty() &&
violations.isEmpty()
+ ? null : group.violations + " " +
violations + " (" + config() + ")"),
+ Invariants.of("the route is started once the faults
cleared",
+ () -> "Started".equals(finalStatus) ? null :
"route is " + finalStatus + " (" + config() + ")"),
+ Invariants.of("a started route holds one share consumer",
+ () -> finalOpenCount == 1
+ ? null : "open share consumers " +
finalOpen + " (" + config() + ")"),
+ Invariants.of("every published record completes at least
once, or the group archived it as it should",
+ this::everyRecordCompletedOrArchived),
+ Invariants.of("the share group is drained once the faults
cleared",
+ () -> finalSettled ? null : "not settled at the
end (" + config() + ")"),
+ Invariants.of("no share consumer is left open after the
route stopped",
+ () -> group.open.isEmpty()
+ ? null : "open after stop: " +
group.openNames() + " (" + config() + ")"));
+ } finally {
+ dst.dropPending();
+ context.stop();
+ }
+ }
+
+ private String everyRecordCompletedOrArchived() {
+ List<String> missing = new ArrayList<>();
+ List<String> completed =
ledger.of("completed").stream().map(DstLedger.Entry::key).toList();
+ for (FakeShareGroup.Rec r : group.log) {
+ if (!completed.contains(r.id) && !archivedAsItShould(r)) {
+ missing.add(r.id + "=" + r.state + "/d" + r.deliveries +
(r.rejected > 0 ? "/rejected#" + r.rejected : ""));
+ }
+ }
+ return missing.isEmpty() ? null : "never completed: " + missing + " ("
+ config() + ")";
+ }
+
+ /**
+ * The excuse for a record that never completed: the group archived it at
the delivery limit (the deliveries failed
+ * with onFailure=RELEASE, or their locks expired, or they were released
when the consumer stopped), or it rejected
+ * a delivery that failed, with onFailure=REJECT.
+ */
+ private boolean archivedAsItShould(FakeShareGroup.Rec r) {
+ if (r.state != FakeShareGroup.State.ARCHIVED) {
+ return false;
+ }
+ if (r.rejected == 0) {
+ return r.deliveries >= group.deliveryLimit;
+ }
+ return "REJECT".equals(onFailure) &&
group.failedDeliveries.contains(r.id + "#" + r.rejected);
+ }
+
+ /**
+ * Whether the share consumer thread of the route gave up creating its
share consumer, after as many creation
+ * failures in a row as createConsumerBackoffMaxAttempts, and the operator
has not restarted the route for that yet.
+ */
+ private boolean consumerGaveUp(DefaultCamelContext context) {
+ if (restartsAfterGiveUp >= group.createGiveUps ||
context.getRoute(ROUTE) == null
+ || !(context.getRoute(ROUTE).getConsumer() instanceof
KafkaShareConsumer consumer)) {
+ return false;
+ }
+ for (TaskHealthState state : consumer.healthStates()) {
+ if (state.isTerminated() && state.getLastError() instanceof
KafkaConsumerFatalException) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Waits for the sends and operations, switches the faults off, waits for
the group to drain (or a deadline), takes
+ * the final state and stops the route.
+ */
+ private void finisher(DefaultCamelContext context, DstExecutor operator,
long deadline) {
+ if (deadline == -1 && (sendsLeft > 0 || opsLeft > 0)) {
+ operator.schedule("finisher", 500, () -> finisher(context,
operator, -1));
+ return;
+ }
+ if (deadline == -1) {
+ long slow = dst.between(0, 3 * group.lockDurationMillis);
+ dst.trace("world: operations done, slow processing for " + slow +
" ms");
+ operator.schedule("finisher", slow, () -> finisher(context,
operator, -2));
+ return;
+ }
+ if (deadline == -2) {
+ faults = false;
+ group.faults = false;
+ dst.trace("world: faults cleared");
+ long d = dst.now() + 5 * 60_000L + 3 * group.lockDurationMillis;
+ operator.schedule("finisher", 500, () -> finisher(context,
operator, d));
+ return;
+ }
+ ServiceStatus status =
context.getRouteController().getRouteStatus(ROUTE);
+ if (status == ServiceStatus.Started && group.open.isEmpty() &&
consumerGaveUp(context)) {
+ restartsAfterGiveUp++;
+ dst.trace("world: the consumer gave up creating its share
consumer, restart the route");
+ try {
+ context.getRouteController().stopRoute(ROUTE);
+ ledger.record("routeStopped", "r", ROUTE);
+ } catch (Exception e) {
+ violations.add("restart after the give up failed: " + e);
+ }
+ startRoute(context, operator);
+ operator.schedule("finisher", 500, () -> finisher(context,
operator, deadline));
+ return;
+ }
+ boolean settled = group.settled();
+ // the locks of a slow exchange can expire and settle the group
(delivery limit) while it still blocks below
+ if ((!settled || status != ServiceStatus.Started || group.open.size()
!= 1 || blocked > 0)
+ && dst.now() < deadline) {
+ operator.schedule("finisher", 500, () -> finisher(context,
operator, deadline));
+ return;
+ }
+ finalStatus = String.valueOf(status);
+ finalSettled = settled;
+ finalOpenCount = group.open.size();
+ finalOpen = group.openNames();
+ if (!settled) {
+ dst.trace("world: unsettled " + group.unsettled());
+ }
+ try {
+ context.getRouteController().stopRoute(ROUTE);
+ } catch (Exception e) {
+ violations.add("final stop failed: " + e);
+ }
+ finished = true;
+ }
+
+ /** The operator starts the route; a failed start is retried, as the
supervising route controller does. */
+ private void startRoute(DefaultCamelContext context, DstExecutor operator)
{
+ try {
+ context.getRouteController().startRoute(ROUTE);
+ ledger.record("routeStarted", "r", ROUTE);
+ } catch (Exception e) {
+ ledger.record("routeStartFailed", "r", ROUTE);
+ dst.trace("world: start failed (" + e.getMessage() + "), retry in
2 s");
+ operator.schedule("retry start route", 2000, () ->
startRoute(context, operator));
+ }
+ }
+
+ private void planSends() {
+ DstExecutor producer = dst.executor("producer", 1);
+ long t = dst.between(0, 500);
+ int n = 0;
+ int bursts = (int) dst.between(1, 4);
+ for (int b = 0; b < bursts; b++) {
+ int size = (int) dst.between(1, 12);
+ for (int i = 0; i < size; i++) {
+ String id = "m" + n++;
+ sendsLeft++;
+ producer.schedule("publish " + id, t, () -> {
+ group.publish(id);
+ sendsLeft--;
+ });
+ t += dst.between(0, 100);
+ }
+ t += dst.between(0, 8000);
+ }
+ }
+
+ private void planOperations(DefaultCamelContext context, DstExecutor
operator) {
+ int ops = (int) dst.between(0, 4);
+ long t = dst.between(500, 3000);
+ for (int i = 0; i < ops; i++) {
+ int kind = dst.nextInt(3);
+ opsLeft++;
+ if (kind == 0) {
+ operator.schedule("restart route", t, () -> {
+ try {
+ if (context.getRouteController().getRouteStatus(ROUTE)
== ServiceStatus.Started) {
+ context.getRouteController().stopRoute(ROUTE);
+ ledger.record("routeStopped", "r", ROUTE);
+ startRoute(context, operator);
+ }
+ } catch (Exception e) {
+ violations.add("restart failed: " + e);
+ }
+ opsLeft--;
+ });
+ } else if (kind == 1) {
+ operator.schedule("suspend and resume route", t, () -> {
+ try {
+ if (context.getRouteController().getRouteStatus(ROUTE)
== ServiceStatus.Started) {
+ context.getRouteController().suspendRoute(ROUTE);
+ dst.block(dst.between(100, 3000), "route
suspended");
+ context.getRouteController().resumeRoute(ROUTE);
+ }
+ } catch (Exception e) {
+ violations.add("suspend/resume failed: " + e);
+ }
+ opsLeft--;
+ });
+ } else {
+ operator.schedule("suspend and resume the CamelContext", t, ()
-> {
+ try {
+ if (context.getRouteController().getRouteStatus(ROUTE)
== ServiceStatus.Started) {
+ context.suspend();
+ ledger.record("contextSuspended", "r", ROUTE);
+ dst.block(dst.between(100, 3000), "CamelContext
suspended");
+ context.resume();
+ ledger.record("contextResumed", "r", ROUTE);
+ }
+ } catch (Exception e) {
+ violations.add("CamelContext suspend/resume failed: "
+ e);
+ }
+ opsLeft--;
+ });
+ }
+ t += dst.between(500, 8000);
+ }
+ }
+}
diff --git a/components/camel-kafka-share/src/test/resources/log4j2.properties
b/components/camel-kafka-share/src/test/resources/log4j2.properties
index bf65f4a08864..d8615f48d12d 100644
--- a/components/camel-kafka-share/src/test/resources/log4j2.properties
+++ b/components/camel-kafka-share/src/test/resources/log4j2.properties
@@ -15,9 +15,15 @@
## limitations under the License.
## ---------------------------------------------------------------------------
-appender.out.type = File
+appender.out.type = RollingFile
appender.out.name = out
appender.out.fileName = target/camel-kafka-share-test.log
+appender.out.filePattern = target/camel-kafka-share-test.%i.log
+appender.out.policies.type = Policies
+appender.out.policies.size.type = SizeBasedTriggeringPolicy
+appender.out.policies.size.size = 50MB
+appender.out.strategy.type = DefaultRolloverStrategy
+appender.out.strategy.max = 1
appender.out.layout.type = PatternLayout
appender.out.layout.pattern = %d [%-15.15t] %-5p %-30.30c{1} - %m%n
appender.stdout.type = Console
@@ -25,7 +31,7 @@ appender.stdout.name = stdout
appender.stdout.layout.type = PatternLayout
appender.stdout.layout.pattern = %d [%-15.15t] %-5p %-30.30c{1} - %m%n
-rootLogger.level = INFO
+rootLogger.level = WARN
rootLogger.appenderRef.out.ref = out
logger.kafka.name = org.apache.kafka
diff --git a/components/camel-test/camel-test-dst/src/main/docs/test-dst.adoc
b/components/camel-test/camel-test-dst/src/main/docs/test-dst.adoc
index 16bf1c96d140..9c0d809a57ad 100644
--- a/components/camel-test/camel-test-dst/src/main/docs/test-dst.adoc
+++ b/components/camel-test/camel-test-dst/src/main/docs/test-dst.adoc
@@ -60,6 +60,8 @@ These worlds are in the Camel code base and are the examples
to copy:
|camel-aws2-sqs |`SqsRedeliveryDstTest` (the extender while Camel's error
handler waits to redeliver, in virtual
time) |-
|camel-azure-eventhubs |`EventHubsCheckpointDstTest` (checkpointing by batch
size and timeout) |CAMEL-25683
+|camel-kafka-share |`KafkaShareConsumerDstTest` (acknowledgements, acquisition
locks, reconnects, and suspend and resume
+of the route and of the CamelContext) |CAMEL-25872
|camel-pulsar |`PulsarConsumerDstTest` (subscribe failures, broker and route
restarts while the route starts and
consumes) |CAMEL-25869
|camel-slack |`SlackSuspendDstTest` (route suspend in the middle of a poll)
|CAMEL-25684
