cadonna commented on code in PR #13927:
URL: https://github.com/apache/kafka/pull/13927#discussion_r1246652052
##########
streams/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java:
##########
@@ -843,15 +846,19 @@ private static <K, V, T> List<T>
waitUntilFinalKeyValueRecordsReceived(final Pro
// still need to check that for each key, the ordering is
expected
final Map<K, List<T>> finalAccumData = new HashMap<>();
for (final T kv : accumulatedActual) {
- finalAccumData.computeIfAbsent(
- withTimestamp ? ((KeyValueTimestamp<K, V>) kv).key() :
((KeyValue<K, V>) kv).key,
- key -> new ArrayList<>()).add(kv);
+ final K key = withTimestamp ? ((KeyValueTimestamp<K, V>)
kv).key() : ((KeyValue<K, V>) kv).key;
+ final List<T> records =
finalAccumData.computeIfAbsent(key, k -> new ArrayList<>());
+ if (!records.contains(kv)) {
+ records.add(kv);
+ }
}
final Map<K, List<T>> finalExpected = new HashMap<>();
for (final T kv : expectedRecords) {
- finalExpected.computeIfAbsent(
- withTimestamp ? ((KeyValueTimestamp<K, V>) kv).key() :
((KeyValue<K, V>) kv).key,
- key -> new ArrayList<>()).add(kv);
+ final K key = withTimestamp ? ((KeyValueTimestamp<K, V>)
kv).key() : ((KeyValue<K, V>) kv).key;
+ final List<T> records = finalExpected.computeIfAbsent(key,
k -> new ArrayList<>());
+ if (!records.contains(kv)) {
+ records.add(kv);
+ }
Review Comment:
These changes do not consider duplicate record during the comparison. A test
that used this verification triggered a failure and verified then that the
expected records were in the output topic. However, without the state updater
no records were written to the output topic before the failure. With the state
updater some records were written to the output topics.
From a correctness point of view, both is correct since no commit happens
before the simulated failure and so Streams reads again all input records after
the failover.
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
