This is an automated email from the ASF dual-hosted git repository.
apupier 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 51f8fb59025b CAMEL-26593: camel-kafka - reduce the duration of the
exactly-once ITs
51f8fb59025b is described below
commit 51f8fb59025b1962f582f30d3f408fc774512abc
Author: Aurélien Pupier <[email protected]>
AuthorDate: Wed Nov 4 16:30:56 2026 +0100
CAMEL-26593: camel-kafka - reduce the duration of the exactly-once ITs
- KafkaExactlyOnceIT: sedaIsTheOnlyPathToKafka and splitFanOutToKafka no
longer wait the full 30s for an offset commit that cannot happen. The
route is rejected and the source record is replayed in a loop, so the
tests now stop waiting as soon as the record has been replayed.
- KafkaExactlyOnceRegressionIT: the read_committed verifier assigns the
partitions directly instead of joining a consumer group, and uses a
shorter quiet period to detect late duplicates.
Co-authored-by: IBM Bob 2.2.1
Signed-off-by: Aurélien Pupier <[email protected]>
---
.../kafka/integration/KafkaExactlyOnceIT.java | 10 +++-
.../integration/KafkaExactlyOnceITSupport.java | 8 +++
.../integration/KafkaExactlyOnceRegressionIT.java | 70 +++++++++++++++-------
3 files changed, 63 insertions(+), 25 deletions(-)
diff --git
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaExactlyOnceIT.java
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaExactlyOnceIT.java
index 44d030a1955c..0ce64a65abe1 100644
---
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaExactlyOnceIT.java
+++
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaExactlyOnceIT.java
@@ -415,7 +415,12 @@ class KafkaExactlyOnceIT extends KafkaExactlyOnceITSupport
{
}
});
produce(in, "m0", "m1", "m2");
- boolean committed3 = awaitQuietly(() ->
Long.valueOf(3).equals(committed(id, in)), 30);
+ awaitCommittedOrReplayed(in, 3, "m0");
+ // The SEDA copy runs asynchronously: wait for its claim to be
rejected, so that a copy committing its own
+ // transaction would be visible below instead of racing with the read.
+ awaitQuietly(() -> failures.stream().anyMatch(f ->
f.contains("ClaimRejectedException"))
+ || Long.valueOf(3).equals(committed(id, in)), 30);
+ boolean committed3 = Long.valueOf(3).equals(committed(id, in));
List<String> out = readCommittedKeys(id + "-out");
LOG.info("Exactly-once {}: committedOffset={} out={} attempts={}
failures={}", id,
committed(id, in), out, attempts, sample());
@@ -438,7 +443,8 @@ class KafkaExactlyOnceIT extends KafkaExactlyOnceITSupport {
}
});
produce(in, Map.of("m0", "a,b,c"));
- boolean committed1 = awaitQuietly(() ->
Long.valueOf(1).equals(committed(id, in)), 30);
+ awaitCommittedOrReplayed(in, 1, "m0");
+ boolean committed1 = Long.valueOf(1).equals(committed(id, in));
List<String> out = readCommittedValues(id + "-out");
LOG.info(
"Exactly-once split fan-out: committed1={} committedOffset={}
out={} attempts={} failures={}",
diff --git
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaExactlyOnceITSupport.java
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaExactlyOnceITSupport.java
index ee247e46d040..19e8bbcef076 100644
---
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaExactlyOnceITSupport.java
+++
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaExactlyOnceITSupport.java
@@ -180,6 +180,14 @@ abstract class KafkaExactlyOnceITSupport {
}
}
+ /**
+ * Waits until the source offset is committed or the record with the given
key has been replayed. A replay means the
+ * route was rejected and the source transaction aborted, so waiting
longer cannot change the outcome.
+ */
+ protected void awaitCommittedOrReplayed(String topic, long offset, String
key) {
+ awaitQuietly(() -> Long.valueOf(offset).equals(committed(id, topic))
|| attempts(key) >= 2, 30);
+ }
+
protected void awaitCommitted(String topic, long offset) {
await().atMost(60, TimeUnit.SECONDS).until(() ->
Long.valueOf(offset).equals(committed(id, topic)));
}
diff --git
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaExactlyOnceRegressionIT.java
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaExactlyOnceRegressionIT.java
index 424be2ef422a..f287a953beff 100644
---
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaExactlyOnceRegressionIT.java
+++
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaExactlyOnceRegressionIT.java
@@ -33,7 +33,6 @@ import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.kafka.KafkaConstants;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.kafka.clients.consumer.ConsumerConfig;
-import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.producer.KafkaProducer;
@@ -41,6 +40,7 @@ import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.awaitility.Awaitility;
+import org.awaitility.core.ConditionTimeoutException;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Tag;
@@ -886,44 +886,68 @@ class KafkaExactlyOnceRegressionIT extends
BaseKafkaTestSupport {
producer.flush();
}
+ /**
+ * Reads every record that is visible to a read_committed consumer, i.e.
everything below the last stable offset,
+ * and waits until at least {@code expectedCount} records are visible. The
partitions are assigned directly, so
+ * there is no consumer group join to wait for. A short quiet period at
the end detects late duplicates or
+ * unexpected records.
+ */
private List<String> readCommitted(String topic, int expectedCount) {
createdTopics.add(topic);
List<String> values = new ArrayList<>();
Properties properties = new Properties();
properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
getBootstrapServers());
- properties.put(ConsumerConfig.GROUP_ID_CONFIG, PREFIX + "Verifier-" +
UUID.randomUUID());
properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
- properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
properties.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG,
"read_committed");
properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName());
properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName());
try (KafkaConsumer<String, String> verifier = new
KafkaConsumer<>(properties)) {
- verifier.subscribe(List.of(topic));
- if (expectedCount == 0) {
- long deadline = System.nanoTime() +
TimeUnit.SECONDS.toNanos(3);
- while (System.nanoTime() < deadline) {
- ConsumerRecords<String, String> records =
verifier.poll(Duration.ofMillis(250));
- assertTrue(records.isEmpty(),
- "Expected no committed output records for " +
topic + " but received " + records.count());
+ // The topic can be created lazily by the producer under test (or
by the metadata request of this
+ // verifier), so wait for its partitions to exist. When no record
is expected, a topic that never appears
+ // is a valid "nothing committed" outcome.
+ List<TopicPartition> partitions = new ArrayList<>();
+ try {
+ Awaitility.await().atMost(expectedCount > 0 ? 40 : 10,
TimeUnit.SECONDS)
+ .pollInSameThread().pollInterval(50,
TimeUnit.MILLISECONDS).ignoreExceptions()
+ .until(() -> {
+ partitions.clear();
+ verifier.partitionsFor(topic,
Duration.ofSeconds(2)).forEach(
+ p -> partitions.add(new
TopicPartition(topic, p.partition())));
+ return !partitions.isEmpty();
+ });
+ } catch (ConditionTimeoutException e) {
+ if (expectedCount > 0) {
+ throw e;
}
+ }
+ if (partitions.isEmpty()) {
+ // only reachable when expectedCount == 0: the topic was never
created, so nothing was committed
return values;
}
- Awaitility.await().atMost(40, TimeUnit.SECONDS)
- .pollInterval(250, TimeUnit.MILLISECONDS)
- .untilAsserted(() -> {
- ConsumerRecords<String, String> records =
verifier.poll(Duration.ofMillis(500));
- records.forEach(record -> values.add(record.value()));
- assertEquals(expectedCount, values.size(),
- "Unexpected committed records for " + topic +
": " + values);
- });
- // Continue polling briefly so a delayed duplicate is also
detected.
- for (int i = 0; i < 4; i++) {
- ConsumerRecords<String, String> records =
verifier.poll(Duration.ofMillis(250));
- records.forEach(record -> values.add(record.value()));
+ verifier.assign(partitions);
+ verifier.seekToBeginning(partitions);
+ if (expectedCount > 0) {
+ Awaitility.await().atMost(40,
TimeUnit.SECONDS).pollInterval(50, TimeUnit.MILLISECONDS).until(() -> {
+ // endOffsets is the last stable offset for a
read_committed consumer
+ Map<TopicPartition, Long> end =
verifier.endOffsets(partitions);
+ long loopDeadline = System.nanoTime() +
TimeUnit.SECONDS.toNanos(35);
+ while (partitions.stream().anyMatch(tp ->
verifier.position(tp) < end.get(tp))) {
+ if (System.nanoTime() > loopDeadline) {
+ break; // let Awaitility re-enter and eventually
time out
+ }
+ verifier.poll(Duration.ofMillis(100)).forEach(record
-> values.add(record.value()));
+ }
+ return values.size() >= expectedCount;
+ });
+ }
+ // Keep polling briefly so a delayed duplicate or an unexpected
committed record is also detected.
+ long deadline = System.nanoTime() +
TimeUnit.MILLISECONDS.toNanos(1500);
+ while (System.nanoTime() < deadline) {
+ verifier.poll(Duration.ofMillis(250)).forEach(record ->
values.add(record.value()));
}
}
- assertEquals(expectedCount, values.size(), "Duplicate committed output
records for " + topic + ": " + values);
+ assertEquals(expectedCount, values.size(), "Unexpected committed
records for " + topic + ": " + values);
return values;
}
