You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "majialoong (via GitHub)" <gi...@apache.org> on 2026/09/10 18:16:28 UTC
[PR] KAFKA-21028: [2/2] Remove hamcrest from org.apache.kafka.streams.processor.internals package [kafka]
majialoong opened a new pull request, #23428:
URL: https://github.com/apache/kafka/pull/23428
Remove hamcrest from a subset of tests in the `org.apache.kafka.streams.processor.internals` package by migrating to JUnit assertions.
--
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-21028: [2/2] Remove hamcrest from org.apache.kafka.streams.processor.internals package [kafka]
Posted by "majialoong (via GitHub)" <gi...@apache.org>.
majialoong commented on code in PR #23428:
URL: https://github.com/apache/kafka/pull/23428#discussion_r3999565105
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StateDirectoryTest.java:
##########
@@ -446,13 +434,13 @@ public void shouldOnlyListNonEmptyTaskDirectories() throws IOException {
final File storeDir = new File(taskDir1.file(), "store");
assertTrue(storeDir.mkdir());
- assertThat(Set.of(taskDir1, taskDir2), equalTo(new HashSet<>(directory.listAllTaskDirectories())));
- assertThat(singletonList(taskDir1), equalTo(directory.listNonEmptyTaskDirectories()));
+ assertEquals(Set.of(taskDir1, taskDir2), new HashSet<>(directory.listAllTaskDirectories()));
Review Comment:
That’s an interesting question. From a quick look at the code, I think the set conversion is mainly to avoid asserting an order that `File.listFiles()` does not guarantee. These methods seem to simply collect directory entries for iteration, so returning a `List` also seems fine.
--
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-21028: [2/2] Remove hamcrest from org.apache.kafka.streams.processor.internals package [kafka]
Posted by "clolov (via GitHub)" <gi...@apache.org>.
clolov merged PR #23428:
URL: https://github.com/apache/kafka/pull/23428
--
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-21028: [2/2] Remove hamcrest from org.apache.kafka.streams.processor.internals package [kafka]
Posted by "majialoong (via GitHub)" <gi...@apache.org>.
majialoong commented on code in PR #23428:
URL: https://github.com/apache/kafka/pull/23428#discussion_r3999568208
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignorTest.java:
##########
@@ -1636,7 +1629,7 @@ public void shouldGenerateTasksForAllCreatedPartitions(final Map<String, Object>
);
// check if we created a task for all expected topicPartitions.
- assertThat(new HashSet<>(assignment.get(client).partitions()), equalTo(new HashSet<>(expectedAssignment)));
+ assertEquals(expectedAssignment, new HashSet<>(assignment.get(client).partitions()));
Review Comment:
This also seems intended to check the assigned partitions without their order.
After reading the code, I found they are sorted using `PARTITION_COMPARATOR`, so a small test enhancement to verify that order seems reasonable. I’ve sorted `expectedAssignment` and updated the assertion to compare the lists directly.
--
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-21028: [2/2] Remove hamcrest from org.apache.kafka.streams.processor.internals package [kafka]
Posted by "majialoong (via GitHub)" <gi...@apache.org>.
majialoong commented on code in PR #23428:
URL: https://github.com/apache/kafka/pull/23428#discussion_r3999561176
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/RepartitionOptimizingTest.java:
##########
@@ -214,14 +209,14 @@ private void runTest(final String optimizationConfig, final int expectedNumberRe
assertEquals(expectedNumberRepartitionTopics, getCountOfRepartitionTopicsFound(topologyString));
// Verify the values collected by the processor
- assertThat(3, equalTo(processorValueCollector.size()));
- assertThat(processorValueCollector, equalTo(expectedCollectedProcessorValues));
+ assertEquals(3, processorValueCollector.size());
+ assertEquals(expectedCollectedProcessorValues, processorValueCollector);
// Verify the expected output
- assertThat(countOutputTopic.readKeyValuesToMap(), equalTo(keyValueListToMap(expectedCountKeyValues)));
- assertThat(aggregationOutputTopic.readKeyValuesToMap(), equalTo(keyValueListToMap(expectedAggKeyValues)));
- assertThat(reduceOutputTopic.readKeyValuesToMap(), equalTo(keyValueListToMap(expectedReduceKeyValues)));
- assertThat(joinedOutputTopic.readKeyValuesToMap(), equalTo(keyValueListToMap(expectedJoinKeyValues)));
+ assertEquals(expectedCountKeyValues, countOutputTopic.readKeyValuesToMap());
+ assertEquals(expectedAggKeyValues, aggregationOutputTopic.readKeyValuesToMap());
+ assertEquals(expectedReduceKeyValues, reduceOutputTopic.readKeyValuesToMap());
+ assertEquals(keyValueListToMap(expectedJoinKeyValues), joinedOutputTopic.readKeyValuesToMap());
Review Comment:
Thanks for the review!
I initially kept the list to minimize changes because its duplicate keys prevent a direct replacement with `Map.of`.
After looking through the history, it seems the test previously checked multiple records but now only compares the final map. I’ve updated it to use `Map.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
Re: [PR] KAFKA-21028: [2/2] Remove hamcrest from org.apache.kafka.streams.processor.internals package [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #23428:
URL: https://github.com/apache/kafka/pull/23428#discussion_r3995575465
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/SinkNodeTest.java:
##########
@@ -67,12 +65,9 @@ public void shouldThrowStreamsExceptionOnInputRecordWithInvalidTimestamp() {
sink.init(context);
// When/Then
context.setTime(-1); // ensures a negative timestamp is set for the record we send next
- try {
- illTypedSink.process(new Record<>("any key".getBytes(), "any value".getBytes(), -1));
- fail("Should have thrown StreamsException");
- } catch (final StreamsException ignored) {
- // expected
- }
+ assertThrows(StreamsException.class,
+ () -> illTypedSink.process(new Record<>("any key".getBytes(), "any value".getBytes(), -1)),
+ "Should have thrown StreamsException");
Review Comment:
I think this message is redundant. We can remove it.
--
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-21028: [2/2] Remove hamcrest from org.apache.kafka.streams.processor.internals package [kafka]
Posted by "clolov (via GitHub)" <gi...@apache.org>.
clolov commented on code in PR #23428:
URL: https://github.com/apache/kafka/pull/23428#discussion_r3989007073
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StateDirectoryTest.java:
##########
@@ -446,13 +434,13 @@ public void shouldOnlyListNonEmptyTaskDirectories() throws IOException {
final File storeDir = new File(taskDir1.file(), "store");
assertTrue(storeDir.mkdir());
- assertThat(Set.of(taskDir1, taskDir2), equalTo(new HashSet<>(directory.listAllTaskDirectories())));
- assertThat(singletonList(taskDir1), equalTo(directory.listNonEmptyTaskDirectories()));
+ assertEquals(Set.of(taskDir1, taskDir2), new HashSet<>(directory.listAllTaskDirectories()));
Review Comment:
Both here and elsewhere there are examples of seemingly extraneous wrappings into sets. Are they needed? If yes, shouldn't the underlying method return a set?
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/RepartitionOptimizingTest.java:
##########
@@ -214,14 +209,14 @@ private void runTest(final String optimizationConfig, final int expectedNumberRe
assertEquals(expectedNumberRepartitionTopics, getCountOfRepartitionTopicsFound(topologyString));
// Verify the values collected by the processor
- assertThat(3, equalTo(processorValueCollector.size()));
- assertThat(processorValueCollector, equalTo(expectedCollectedProcessorValues));
+ assertEquals(3, processorValueCollector.size());
+ assertEquals(expectedCollectedProcessorValues, processorValueCollector);
// Verify the expected output
- assertThat(countOutputTopic.readKeyValuesToMap(), equalTo(keyValueListToMap(expectedCountKeyValues)));
- assertThat(aggregationOutputTopic.readKeyValuesToMap(), equalTo(keyValueListToMap(expectedAggKeyValues)));
- assertThat(reduceOutputTopic.readKeyValuesToMap(), equalTo(keyValueListToMap(expectedReduceKeyValues)));
- assertThat(joinedOutputTopic.readKeyValuesToMap(), equalTo(keyValueListToMap(expectedJoinKeyValues)));
+ assertEquals(expectedCountKeyValues, countOutputTopic.readKeyValuesToMap());
+ assertEquals(expectedAggKeyValues, aggregationOutputTopic.readKeyValuesToMap());
+ assertEquals(expectedReduceKeyValues, reduceOutputTopic.readKeyValuesToMap());
+ assertEquals(keyValueListToMap(expectedJoinKeyValues), joinedOutputTopic.readKeyValuesToMap());
Review Comment:
You could do the same trick for expectedJoinKeyValues as you did for the other three, no?
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignorTest.java:
##########
@@ -1636,7 +1629,7 @@ public void shouldGenerateTasksForAllCreatedPartitions(final Map<String, Object>
);
// check if we created a task for all expected topicPartitions.
- assertThat(new HashSet<>(assignment.get(client).partitions()), equalTo(new HashSet<>(expectedAssignment)));
+ assertEquals(expectedAssignment, new HashSet<>(assignment.get(client).partitions()));
Review Comment:
In a similar vein as my previous comment - is the HashSet wrapper needed, because if not then I believe you can just remove it rather than change the list to a set?
--
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-21028: [2/2] Remove hamcrest from org.apache.kafka.streams.processor.internals package [kafka]
Posted by "majialoong (via GitHub)" <gi...@apache.org>.
majialoong commented on code in PR #23428:
URL: https://github.com/apache/kafka/pull/23428#discussion_r3999569637
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/SinkNodeTest.java:
##########
@@ -67,12 +65,9 @@ public void shouldThrowStreamsExceptionOnInputRecordWithInvalidTimestamp() {
sink.init(context);
// When/Then
context.setTime(-1); // ensures a negative timestamp is set for the record we send next
- try {
- illTypedSink.process(new Record<>("any key".getBytes(), "any value".getBytes(), -1));
- fail("Should have thrown StreamsException");
- } catch (final StreamsException ignored) {
- // expected
- }
+ assertThrows(StreamsException.class,
+ () -> illTypedSink.process(new Record<>("any key".getBytes(), "any value".getBytes(), -1)),
+ "Should have thrown StreamsException");
Review Comment:
Removed. Thanks for the review!
--
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