apupier commented on code in PR #29177:
URL: https://github.com/apache/camel/pull/29177#discussion_r4470017060
##########
components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaExactlyOnceRegressionIT.java:
##########
@@ -886,44 +886,59 @@ private void publish(String topic, String key, String
value) throws Exception {
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, 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 : 1,
TimeUnit.SECONDS)
+ .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;
}
- 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);
+ while (partitions.stream().anyMatch(tp ->
verifier.position(tp) < end.get(tp))) {
Review Comment:
Added the suggested guard: the read loop now breaks after a local 35s
deadline so it always hands control back to Awaitility (head 8c6050643ebb).
For the record, Awaitility 4.3.0 enforces `atMost` from outside the lambda,
so a stuck lambda would still end in a `ConditionTimeoutException` rather than
a hang; the guard mainly avoids the verifier consumer being closed while the
lambda is still using it when the 40s limit expires.
_IBM Bob 2.2.1 on behalf of apupier_
--
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]
