You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by mj...@apache.org on 2025/08/25 23:09:16 UTC
(kafka) branch 4.1 updated: KAFKA-19429: Deflake streams_smoke_test, again (#20070)
This is an automated email from the ASF dual-hosted git repository.
mjsax pushed a commit to branch 4.1
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/4.1 by this push:
new 500bd70a292 KAFKA-19429: Deflake streams_smoke_test, again (#20070)
500bd70a292 is described below
commit 500bd70a29248ef72725c5d8c7f9da0e2198cb55
Author: Lucas Brutschy <lb...@confluent.io>
AuthorDate: Tue Jul 1 21:48:07 2025 +0200
KAFKA-19429: Deflake streams_smoke_test, again (#20070)
It looks like we are checking for properties that are not guaranteed
under at_least_once, for example, exact counting (not allowing for
overcounting).
This change relaxes the validation constraint:
The TAGG topic contains effectively count-by-count results. So for
example, if we have the input without duplication
0 -> 1,2,3 we will get in TAGG 3 -> 1, since 1 key had 3 values.
with duplication:
0 -> 1,1,2,3 we will get in TAGG 4 -> 1, since 1 key had 4 values.
This makes the result difficult to compare. Since we run the smoke test
also with Exactly_Once, I propose to disable validation off TAGG under
ALOS.
Similarly, the topic AVG may overcount or undercount. The test case is
extremely similar to DIF, both performing a join and two streams, the
only difference being the mathematical operation performed, so we can
also disable this validation under ALOS with minimal loss of coverage.
Finally, the change fixes a bug that would throw a NPE when validation
of a windowed stream would fail.
Reviewers: Kirk True <ki...@kirktrue.pro>, Matthias J. Sax
<ma...@confluent.io>
---
.../org/apache/kafka/streams/tests/SmokeTestDriver.java | 16 +++++++++++-----
1 file changed, 11 insertions(+), 5 deletions(-)
diff --git a/streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestDriver.java b/streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestDriver.java
index ad3cead1763..59698607912 100644
--- a/streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestDriver.java
+++ b/streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestDriver.java
@@ -522,7 +522,11 @@ public class SmokeTestDriver extends SmokeTestUtil {
}
boolean pass;
try (final PrintStream resultStream = new PrintStream(byteArrayOutputStream)) {
- pass = verifyTAgg(resultStream, inputs, events.get("tagg"), validationPredicate, printResults);
+ pass = true;
+ if (eosEnabled) {
+ // TAGG is computing "Count-by-count", which may produce keys that are not in the input data in ALOS, so we skip validation in this case.
+ pass = verifyTAgg(resultStream, inputs, events.get("tagg"), printResults);
+ }
pass &= verifySuppressed(resultStream, "min-suppressed", events, printResults);
pass &= verify(resultStream, "min-suppressed", inputs, events, windowedKey -> {
final String unwindowedKey = windowedKey.substring(1, windowedKey.length() - 1).replaceAll("@.*", "");
@@ -534,7 +538,10 @@ public class SmokeTestDriver extends SmokeTestUtil {
pass &= verify(resultStream, "dif", inputs, events, key -> getMax(key).intValue() - getMin(key).intValue(), Object::equals, printResults, eosEnabled);
pass &= verify(resultStream, "sum", inputs, events, SmokeTestDriver::getSum, validationPredicate, printResults, eosEnabled);
pass &= verify(resultStream, "cnt", inputs, events, key1 -> getMax(key1).intValue() - getMin(key1).intValue() + 1L, validationPredicate, printResults, eosEnabled);
- pass &= verify(resultStream, "avg", inputs, events, SmokeTestDriver::getAvg, validationPredicate, printResults, eosEnabled);
+ if (eosEnabled) {
+ // Average can overcount and undercount in ALOS, so we skip validation in that case.
+ pass &= verify(resultStream, "avg", inputs, events, SmokeTestDriver::getAvg, Object::equals, printResults, eosEnabled);
+ }
}
return new VerificationResult(pass, new String(byteArrayOutputStream.toByteArray(), StandardCharsets.UTF_8));
}
@@ -580,7 +587,7 @@ public class SmokeTestDriver extends SmokeTestUtil {
if (printResults) {
resultStream.printf("\t inputEvents=%n%s%n\t" +
"echoEvents=%n%s%n\tmaxEvents=%n%s%n\tminEvents=%n%s%n\tdifEvents=%n%s%n\tcntEvents=%n%s%n\ttaggEvents=%n%s%n",
- indent("\t\t", observedInputEvents.get(key)),
+ indent("\t\t", observedInputEvents.getOrDefault(key, new LinkedList<>())),
indent("\t\t", events.getOrDefault("echo", emptyMap()).getOrDefault(key, new LinkedList<>())),
indent("\t\t", events.getOrDefault("max", emptyMap()).getOrDefault(key, new LinkedList<>())),
indent("\t\t", events.getOrDefault("min", emptyMap()).getOrDefault(key, new LinkedList<>())),
@@ -662,7 +669,6 @@ public class SmokeTestDriver extends SmokeTestUtil {
private static boolean verifyTAgg(final PrintStream resultStream,
final Map<String, Set<Integer>> allData,
final Map<String, LinkedList<ConsumerRecord<String, Number>>> taggEvents,
- final BiPredicate<Number, Number> validationPredicate,
final boolean printResults) {
resultStream.println("verifying topic tagg");
if (taggEvents == null) {
@@ -694,7 +700,7 @@ public class SmokeTestDriver extends SmokeTestUtil {
expectedCount = 0L;
}
- if (!validationPredicate.test(expectedCount, entry.getValue().getLast().value())) {
+ if (entry.getValue().getLast().value().longValue() != expectedCount) {
resultStream.println("fail: key=" + key + " tagg=" + entry.getValue() + " expected=" + expectedCount);
if (printResults)