You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "lucasbru (via GitHub)" <gi...@apache.org> on 2025/06/30 09:42:30 UTC
[PR] KAFKA-19429: Deflake streams_smoke_test, again [kafka]
lucasbru opened a new pull request, #20070:
URL: https://github.com/apache/kafka/pull/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.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19429: Deflake streams_smoke_test, again [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #20070:
URL: https://github.com/apache/kafka/pull/20070#discussion_r2176921868
##########
streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestDriver.java:
##########
@@ -694,7 +700,7 @@ private static boolean verifyTAgg(final PrintStream resultStream,
expectedCount = 0L;
}
- if (!validationPredicate.test(expectedCount, entry.getValue().getLast().value())) {
+ if (entry.getValue().getLast().value().longValue() != expectedCount) {
Review Comment:
It is guaranteed to have at least one element, otherwise the key-value pair wouldn't be created, see Line 433 in this file.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19429: Deflake streams_smoke_test, again [kafka]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #20070:
URL: https://github.com/apache/kafka/pull/20070#discussion_r2174665972
##########
streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestDriver.java:
##########
@@ -580,7 +587,7 @@ private static boolean verify(final PrintStream resultStream,
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<>())),
Review Comment:
Instead of creating a new LinkedList for each missing key, consider using Collections.emptyList() to avoid unnecessary allocations.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19429: Deflake streams_smoke_test, again [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #20070:
URL: https://github.com/apache/kafka/pull/20070#discussion_r2176918934
##########
streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestDriver.java:
##########
@@ -580,7 +587,7 @@ private static boolean verify(final PrintStream resultStream,
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<>())),
Review Comment:
Collections.emptyList() and List.of() don't return a linked list.
Since this is test code, I wouldn't worry about unnecessary allocations.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19429: Deflake streams_smoke_test, again [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20070:
URL: https://github.com/apache/kafka/pull/20070#discussion_r2178408847
##########
streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestDriver.java:
##########
@@ -522,7 +522,11 @@ private static VerificationResult verifyAll(final Map<String, Set<Integer>> inpu
}
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);
Review Comment:
Ah. Got it. Thanks.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19429: Deflake streams_smoke_test, again [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20070:
URL: https://github.com/apache/kafka/pull/20070#discussion_r2176027337
##########
streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestDriver.java:
##########
@@ -522,7 +522,11 @@ private static VerificationResult verifyAll(final Map<String, Set<Integer>> inpu
}
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);
Review Comment:
But isn't this why we set a different `validationPredicate` for ALOS, and don't use equals, but lessEquals, to say `expectedCount <= actualCount` ?
##########
streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestDriver.java:
##########
@@ -694,7 +700,7 @@ private static boolean verifyTAgg(final PrintStream resultStream,
expectedCount = 0L;
}
- if (!validationPredicate.test(expectedCount, entry.getValue().getLast().value())) {
+ if (entry.getValue().getLast().value().longValue() != expectedCount) {
Review Comment:
Should be ok to not check this -- it would mean we did not get any data for a particular key, what would be quite some data-loss issue.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19429: Deflake streams_smoke_test, again [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru merged PR #20070:
URL: https://github.com/apache/kafka/pull/20070
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19429: Deflake streams_smoke_test, again [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on PR #20070:
URL: https://github.com/apache/kafka/pull/20070#issuecomment-3022770983
Test run of 160 test variations: https://confluent-open-source-kafka-branch-builder-system-test-results.s3-us-west-2.amazonaws.com/trunk/2025-06-30--001.3e0621bc-305c-4b8f-8626-108a654c96b1--1751312156--lucasbru--deflake_smoke_test--91c3fd21bb/report.html
It's roughly the limit that I can run without hitting sempahore timeouts.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19429: Deflake streams_smoke_test, again [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #20070:
URL: https://github.com/apache/kafka/pull/20070#discussion_r2176631895
##########
streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestDriver.java:
##########
@@ -522,7 +522,11 @@ private static VerificationResult verifyAll(final Map<String, Set<Integer>> inpu
}
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);
Review Comment:
See commit description. LessEqual was sufficient for sum and count, but it's not sufficient for count-by-count.
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.
So effectively, the count for key 3 decreased, and key 4 increased, compared to expectation, so lessEqual won't fix this.
We could consider opening a ticket to validate whatever you can validate on TAGG, but you'd have to model precisely what happens under duplication at various stages (before / after the repartitioning). Since we run the same test with EOSv2, not sure if it's worth it. Pretty sure it's not worth living with a flaky test.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19429: Deflake streams_smoke_test, again [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #20070:
URL: https://github.com/apache/kafka/pull/20070#discussion_r2176918934
##########
streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestDriver.java:
##########
@@ -580,7 +587,7 @@ private static boolean verify(final PrintStream resultStream,
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<>())),
Review Comment:
Collections.emptyList() and List.of() don't return a linked list.
Since this is test code, I wouldn't worry about unnecessary allocations.
I could try to refactor the test to not use LinkedLists, however that would at least require rerunning the system test for 8+ hours, so I'd only do that if you feel strongly about your comment.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19429: Deflake streams_smoke_test, again [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #20070:
URL: https://github.com/apache/kafka/pull/20070#discussion_r2175536687
##########
streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestDriver.java:
##########
@@ -694,7 +700,7 @@ private static boolean verifyTAgg(final PrintStream resultStream,
expectedCount = 0L;
}
- if (!validationPredicate.test(expectedCount, entry.getValue().getLast().value())) {
+ if (entry.getValue().getLast().value().longValue() != expectedCount) {
Review Comment:
Is the `LinkedList` guaranteed to have at least one element? Is it worth adding that check?
##########
streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestDriver.java:
##########
@@ -580,7 +587,7 @@ private static boolean verify(final PrintStream resultStream,
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<>())),
Review Comment:
or `List.of()`
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org