gnodet-bot commented on code in PR #29177:
URL: https://github.com/apache/camel/pull/29177#discussion_r4469736238
##########
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:
⚠️ **Low severity:** This inner `while` loop can block Awaitility's
`atMost(40s)` timeout indefinitely. Awaitility only checks the timeout
*between* lambda invocations — if `poll()` returns empty records while
`position < end` (e.g. network hiccup, uncommitted transaction holding back
LSO), `position` never advances and the lambda never returns. Consider adding a
local deadline or iteration cap.
--
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]
