You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ch...@apache.org on 2026/09/06 11:39:37 UTC
(kafka) branch trunk updated: KAFKA-21005 Remove hamcrest from org.apache.kafka.streams.kstream.internals package (#23370)
This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 0eacd545c53 KAFKA-21005 Remove hamcrest from org.apache.kafka.streams.kstream.internals package (#23370)
0eacd545c53 is described below
commit 0eacd545c53acf92067ce82e8710eba7866ec951
Author: majialong <ma...@gmail.com>
AuthorDate: Sun Sep 6 19:39:29 2026 +0800
KAFKA-21005 Remove hamcrest from org.apache.kafka.streams.kstream.internals package (#23370)
Remove hamcrest from the `org.apache.kafka.streams.kstream.internals`
package by migrating its assertions to JUnit.
Reviewers: Chia-Ping Tsai <ch...@gmail.com>
---
.../kstream/internals/ChangedSerdeTest.java | 15 +-
.../internals/CogroupedKStreamImplTest.java | 99 +++--
.../kstream/internals/FullChangeSerdeTest.java | 47 +--
.../internals/InternalStreamsBuilderTest.java | 39 +-
.../kstream/internals/KGroupedStreamImplTest.java | 82 ++--
.../kstream/internals/KGroupedTableImplTest.java | 26 +-
.../kstream/internals/KStreamFlatMapTest.java | 4 +-
.../internals/KStreamGlobalKTableJoinTest.java | 30 +-
.../internals/KStreamGlobalKTableLeftJoinTest.java | 30 +-
.../streams/kstream/internals/KStreamImplTest.java | 457 ++++++++++-----------
.../kstream/internals/KStreamKStreamJoinTest.java | 47 +--
.../internals/KStreamKStreamOuterJoinTest.java | 8 +-
.../kstream/internals/KStreamKTableJoinTest.java | 45 +-
.../streams/kstream/internals/KStreamMapTest.java | 4 +-
.../kstream/internals/KStreamRepartitionTest.java | 7 +-
...KStreamSessionWindowAggregateProcessorTest.java | 46 +--
.../KStreamSlidingWindowAggregateTest.java | 69 ++--
.../internals/KStreamWindowAggregateTest.java | 102 ++---
.../kstream/internals/KTableFilterTest.java | 16 +-
.../streams/kstream/internals/KTableImplTest.java | 8 +-
.../KTableKTableForeignKeyJoinScenarioTest.java | 12 +-
.../internals/KTableKTableInnerJoinTest.java | 12 +-
.../internals/KTableKTableLeftJoinTest.java | 19 +-
.../internals/KTableKTableOuterJoinTest.java | 12 +-
.../internals/KTableKTableRightJoinTest.java | 14 +-
.../kstream/internals/KTableMapValuesTest.java | 11 +-
.../kstream/internals/KTableSourceTest.java | 22 +-
.../internals/KTableTransformValuesTest.java | 94 +++--
.../internals/MaterializedInternalTest.java | 22 +-
.../SessionWindowedCogroupedKStreamImplTest.java | 10 +-
.../internals/SessionWindowedKStreamImplTest.java | 59 ++-
.../SlidingWindowedCogroupedKStreamImplTest.java | 9 +-
.../internals/SlidingWindowedKStreamImplTest.java | 255 ++++++------
.../kstream/internals/SuppressScenarioTest.java | 5 +-
.../kstream/internals/SuppressTopologyTest.java | 11 +-
.../TimeWindowedCogroupedKStreamImplTest.java | 10 +-
.../internals/TimeWindowedKStreamImplTest.java | 74 ++--
37 files changed, 857 insertions(+), 975 deletions(-)
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/ChangedSerdeTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/ChangedSerdeTest.java
index f95530ab42d..b7fe35ffda1 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/ChangedSerdeTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/ChangedSerdeTest.java
@@ -34,9 +34,8 @@ import java.nio.ByteBuffer;
import java.util.Collections;
import java.util.Map;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.is;
-import static org.hamcrest.Matchers.notNullValue;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
@@ -62,9 +61,9 @@ public class ChangedSerdeTest {
private static <T> void checkRoundTrip(final T data, final Serializer<T> serializer, final Deserializer<T> deserializer) {
final byte[] serialized = serializer.serialize(TOPIC, HEADERS, data);
- assertThat(serialized, is(notNullValue()));
+ assertNotNull(serialized);
final T deserialized = deserializer.deserialize(TOPIC, HEADERS, serialized);
- assertThat(deserialized, is(data));
+ assertEquals(data, deserialized);
}
@Test
@@ -110,7 +109,7 @@ public class ChangedSerdeTest {
public void shouldThrowErrorIfEncountersAnUnknownByteValueForOldNewFlag() {
final Change<String> data = new Change<>(null, nonNullOldValue);
final byte[] serialized = CHANGED_STRING_SERIALIZER.serialize(TOPIC, HEADERS, data);
- assertThat(serialized, is(notNullValue()));
+ assertNotNull(serialized);
// mutate the serialized array to replace OLD_NEW_FLAG with an unsupported byte value
final ByteBuffer buffer = ByteBuffer.wrap(serialized);
@@ -182,9 +181,9 @@ public class ChangedSerdeTest {
private static void checkRoundTripForReservedVersion(final Change<String> data) {
final byte[] serialized = serializeVersions3Through5(TOPIC, data);
- assertThat(serialized, is(notNullValue()));
+ assertNotNull(serialized);
final Change<String> deserialized = CHANGED_STRING_DESERIALIZER.deserialize(TOPIC, HEADERS, serialized);
- assertThat(deserialized, is(data));
+ assertEquals(data, deserialized);
}
@Test
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/CogroupedKStreamImplTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/CogroupedKStreamImplTest.java
index 5ca2619ee80..f3db687c7fb 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/CogroupedKStreamImplTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/CogroupedKStreamImplTest.java
@@ -55,8 +55,7 @@ import org.junit.jupiter.params.provider.ValueSource;
import java.util.Properties;
-import static org.hamcrest.CoreMatchers.equalTo;
-import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -204,9 +203,8 @@ public class CogroupedKStreamImplTest {
final String topologyDescription = builder.build().describe().toString();
- assertThat(
- topologyDescription,
- equalTo("Topologies:\n" +
+ assertEquals(
+ "Topologies:\n" +
" Sub-topology: 0\n" +
" Source: KSTREAM-SOURCE-0000000000 (topics: [one])\n" +
" --> test-cogroup-agg-0\n" +
@@ -225,7 +223,8 @@ public class CogroupedKStreamImplTest {
" --> KSTREAM-SINK-0000000006\n" +
" <-- test-cogroup-merge\n" +
" Sink: KSTREAM-SINK-0000000006 (topic: output)\n" +
- " <-- KTABLE-TOSTREAM-0000000005\n\n"));
+ " <-- KTABLE-TOSTREAM-0000000005\n\n",
+ topologyDescription);
}
@ParameterizedTest
@@ -249,9 +248,8 @@ public class CogroupedKStreamImplTest {
final String topologyDescription = builder.build().describe().toString();
- assertThat(
- topologyDescription,
- equalTo("Topologies:\n" +
+ assertEquals(
+ "Topologies:\n" +
" Sub-topology: 0\n" +
" Source: KSTREAM-SOURCE-0000000000 (topics: [one])\n" +
" --> KSTREAM-MAP-0000000002\n" +
@@ -281,7 +279,8 @@ public class CogroupedKStreamImplTest {
" --> KSTREAM-SINK-0000000011\n" +
" <-- COGROUPKSTREAM-MERGE-0000000009\n" +
" Sink: KSTREAM-SINK-0000000011 (topic: output)\n" +
- " <-- KTABLE-TOSTREAM-0000000010\n\n"));
+ " <-- KTABLE-TOSTREAM-0000000010\n\n",
+ topologyDescription);
}
@ParameterizedTest
@@ -304,9 +303,8 @@ public class CogroupedKStreamImplTest {
final String topologyDescription = builder.build().describe().toString();
- assertThat(
- topologyDescription,
- equalTo("Topologies:\n" +
+ assertEquals(
+ "Topologies:\n" +
" Sub-topology: 0\n" +
" Source: KSTREAM-SOURCE-0000000000 (topics: [one])\n" +
" --> KSTREAM-MAP-0000000002\n" +
@@ -336,7 +334,8 @@ public class CogroupedKStreamImplTest {
" --> KSTREAM-SINK-0000000010\n" +
" <-- test-cogroup-merge\n" +
" Sink: KSTREAM-SINK-0000000010 (topic: output)\n" +
- " <-- KTABLE-TOSTREAM-0000000009\n\n"));
+ " <-- KTABLE-TOSTREAM-0000000009\n\n",
+ topologyDescription);
}
@ParameterizedTest
@@ -366,9 +365,8 @@ public class CogroupedKStreamImplTest {
final String topologyDescription = builder.build().describe().toString();
- assertThat(
- topologyDescription,
- equalTo("Topologies:\n" +
+ assertEquals(
+ "Topologies:\n" +
" Sub-topology: 0\n" +
" Source: KSTREAM-SOURCE-0000000000 (topics: [one])\n" +
" --> KSTREAM-MAP-0000000002\n" +
@@ -419,7 +417,8 @@ public class CogroupedKStreamImplTest {
" Sink: KSTREAM-SINK-0000000018 (topic: output)\n" +
" <-- KTABLE-TOSTREAM-0000000017\n" +
" Sink: KSTREAM-SINK-0000000020 (topic: OUTPUT2)\n" +
- " <-- KTABLE-TOSTREAM-0000000019\n\n"));
+ " <-- KTABLE-TOSTREAM-0000000019\n\n",
+ topologyDescription);
}
@ParameterizedTest
@@ -450,9 +449,8 @@ public class CogroupedKStreamImplTest {
final String topologyDescription = builder.build(props).describe().toString();
- assertThat(
- topologyDescription,
- equalTo("Topologies:\n" +
+ assertEquals(
+ "Topologies:\n" +
" Sub-topology: 0\n" +
" Source: KSTREAM-SOURCE-0000000000 (topics: [one])\n" +
" --> KSTREAM-MAP-0000000002\n" +
@@ -496,7 +494,8 @@ public class CogroupedKStreamImplTest {
" Sink: KSTREAM-SINK-0000000018 (topic: output)\n" +
" <-- KTABLE-TOSTREAM-0000000017\n" +
" Sink: KSTREAM-SINK-0000000020 (topic: OUTPUT2)\n" +
- " <-- KTABLE-TOSTREAM-0000000019\n\n"));
+ " <-- KTABLE-TOSTREAM-0000000019\n\n",
+ topologyDescription);
}
@ParameterizedTest
@@ -523,9 +522,8 @@ public class CogroupedKStreamImplTest {
final String topologyDescription = builder.build().describe().toString();
- assertThat(
- topologyDescription,
- equalTo("Topologies:\n" +
+ assertEquals(
+ "Topologies:\n" +
" Sub-topology: 0\n" +
" Source: KSTREAM-SOURCE-0000000000 (topics: [one])\n" +
" --> KSTREAM-MAP-0000000003\n" +
@@ -569,7 +567,8 @@ public class CogroupedKStreamImplTest {
" <-- KSTREAM-SOURCE-0000000002\n" +
" Processor: COGROUPKSTREAM-MERGE-0000000010 (stores: [])\n" +
" --> none\n" +
- " <-- COGROUPKSTREAM-AGGREGATE-0000000008, COGROUPKSTREAM-AGGREGATE-0000000009\n\n"));
+ " <-- COGROUPKSTREAM-AGGREGATE-0000000008, COGROUPKSTREAM-AGGREGATE-0000000009\n\n",
+ topologyDescription);
}
@ParameterizedTest
@@ -599,9 +598,8 @@ public class CogroupedKStreamImplTest {
final String topologyDescription = builder.build(properties).describe().toString();
- assertThat(
- topologyDescription,
- equalTo("Topologies:\n" +
+ assertEquals(
+ "Topologies:\n" +
" Sub-topology: 0\n" +
" Source: KSTREAM-SOURCE-0000000000 (topics: [one])\n" +
" --> KSTREAM-MAP-0000000003\n" +
@@ -637,7 +635,8 @@ public class CogroupedKStreamImplTest {
" <-- COGROUPKSTREAM-AGGREGATE-0000000008, COGROUPKSTREAM-AGGREGATE-0000000009\n" +
" Processor: COGROUPKSTREAM-MERGE-0000000017 (stores: [])\n" +
" --> none\n" +
- " <-- COGROUPKSTREAM-AGGREGATE-0000000015, COGROUPKSTREAM-AGGREGATE-0000000016\n\n"));
+ " <-- COGROUPKSTREAM-AGGREGATE-0000000015, COGROUPKSTREAM-AGGREGATE-0000000016\n\n",
+ topologyDescription);
}
@ParameterizedTest
@@ -660,9 +659,8 @@ public class CogroupedKStreamImplTest {
final String topologyDescription = builder.build().describe().toString();
- assertThat(
- topologyDescription,
- equalTo("Topologies:\n" +
+ assertEquals(
+ "Topologies:\n" +
" Sub-topology: 0\n" +
" Source: KSTREAM-SOURCE-0000000000 (topics: [one])\n" +
" --> KSTREAM-MAP-0000000002\n" +
@@ -698,7 +696,8 @@ public class CogroupedKStreamImplTest {
" --> KSTREAM-AGGREGATE-0000000011\n" +
" Processor: KSTREAM-AGGREGATE-0000000011 (stores: [KSTREAM-AGGREGATE-STATE-STORE-0000000010])\n" +
" --> none\n" +
- " <-- KSTREAM-SOURCE-0000000014\n\n"));
+ " <-- KSTREAM-SOURCE-0000000014\n\n",
+ topologyDescription);
}
@ParameterizedTest
@@ -724,9 +723,8 @@ public class CogroupedKStreamImplTest {
final String topologyDescription = builder.build(properties).describe().toString();
- assertThat(
- topologyDescription,
- equalTo("Topologies:\n" +
+ assertEquals(
+ "Topologies:\n" +
" Sub-topology: 0\n" +
" Source: KSTREAM-SOURCE-0000000000 (topics: [one])\n" +
" --> KSTREAM-MAP-0000000002\n" +
@@ -754,7 +752,8 @@ public class CogroupedKStreamImplTest {
" <-- COGROUPKSTREAM-AGGREGATE-0000000007, COGROUPKSTREAM-AGGREGATE-0000000008\n" +
" Processor: KSTREAM-AGGREGATE-0000000011 (stores: [KSTREAM-AGGREGATE-STATE-STORE-0000000010])\n" +
" --> none\n" +
- " <-- COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition-source\n\n"));
+ " <-- COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition-source\n\n",
+ topologyDescription);
}
@ParameterizedTest
@@ -787,9 +786,8 @@ public class CogroupedKStreamImplTest {
final String topologyDescription = builder.build(properties).describe().toString();
- assertThat(
- topologyDescription,
- equalTo("Topologies:\n" +
+ assertEquals(
+ "Topologies:\n" +
" Sub-topology: 0\n" +
" Source: KSTREAM-SOURCE-0000000000 (topics: [one])\n" +
" --> KSTREAM-MAP-0000000003, KSTREAM-MAP-0000000004\n" +
@@ -836,7 +834,8 @@ public class CogroupedKStreamImplTest {
" <-- COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000012-repartition-source\n" +
" Processor: COGROUPKSTREAM-MERGE-0000000018 (stores: [])\n" +
" --> none\n" +
- " <-- COGROUPKSTREAM-AGGREGATE-0000000016, COGROUPKSTREAM-AGGREGATE-0000000017\n\n"));
+ " <-- COGROUPKSTREAM-AGGREGATE-0000000016, COGROUPKSTREAM-AGGREGATE-0000000017\n\n",
+ topologyDescription);
}
@ParameterizedTest
@@ -857,9 +856,8 @@ public class CogroupedKStreamImplTest {
final String topologyDescription = builder.build(properties).describe().toString();
- assertThat(
- topologyDescription,
- equalTo("Topologies:\n" +
+ assertEquals(
+ "Topologies:\n" +
" Sub-topology: 0\n" +
" Source: KSTREAM-SOURCE-0000000000 (topics: [one])\n" +
" --> KSTREAM-MAP-0000000001\n" +
@@ -885,7 +883,8 @@ public class CogroupedKStreamImplTest {
" <-- COGROUPKSTREAM-AGGREGATE-0000000006\n" +
" Processor: COGROUPKSTREAM-MERGE-0000000013 (stores: [])\n" +
" --> none\n" +
- " <-- COGROUPKSTREAM-AGGREGATE-0000000012\n\n"));
+ " <-- COGROUPKSTREAM-AGGREGATE-0000000012\n\n",
+ topologyDescription);
}
@ParameterizedTest
@@ -1316,17 +1315,13 @@ public class CogroupedKStreamImplTest {
final String expectedKey,
final String expectedValue,
final long expectedTimestamp) {
- assertThat(
- outputTopic.readRecord(),
- equalTo(new TestRecord<>(expectedKey, expectedValue, null, expectedTimestamp)));
+ assertEquals(new TestRecord<>(expectedKey, expectedValue, null, expectedTimestamp), outputTopic.readRecord());
}
private void assertOutputKeyValueTimestamp(final TestOutputTopic<String, Integer> outputTopic,
final String expectedKey,
final Integer expectedValue,
final long expectedTimestamp) {
- assertThat(
- outputTopic.readRecord(),
- equalTo(new TestRecord<>(expectedKey, expectedValue, null, expectedTimestamp)));
+ assertEquals(new TestRecord<>(expectedKey, expectedValue, null, expectedTimestamp), outputTopic.readRecord());
}
}
\ No newline at end of file
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/FullChangeSerdeTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/FullChangeSerdeTest.java
index 50074f22dff..b9d366c6b0e 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/FullChangeSerdeTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/FullChangeSerdeTest.java
@@ -29,9 +29,8 @@ import org.junit.jupiter.api.Test;
import java.nio.ByteBuffer;
-import static org.hamcrest.CoreMatchers.nullValue;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.core.Is.is;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
@@ -69,29 +68,23 @@ public class FullChangeSerdeTest {
@Test
public void shouldRoundTripNull() {
- assertThat(serde.serializeParts(null, null, null), nullValue());
- assertThat(mergeChangeArraysIntoSingleLegacyFormattedArray(null), nullValue());
- assertThat(FullChangeSerde.decomposeLegacyFormattedArrayIntoChangeArrays(null), nullValue());
- assertThat(serde.deserializeParts(null, null, null), nullValue());
+ assertNull(serde.serializeParts(null, null, null));
+ assertNull(mergeChangeArraysIntoSingleLegacyFormattedArray(null));
+ assertNull(FullChangeSerde.decomposeLegacyFormattedArrayIntoChangeArrays(null));
+ assertNull(serde.deserializeParts(null, null, null));
}
@Test
public void shouldRoundTripNullChange() {
- assertThat(
- serde.serializeParts(null, null, new Change<>(null, null)),
- is(new Change<byte[]>(null, null))
- );
+ assertEquals(new Change<byte[]>(null, null), serde.serializeParts(null, null, new Change<>(null, null)));
- assertThat(
- serde.deserializeParts(null, null, new Change<>(null, null)),
- is(new Change<String>(null, null))
- );
+ assertEquals(new Change<String>(null, null), serde.deserializeParts(null, null, new Change<>(null, null)));
final byte[] legacyFormat = mergeChangeArraysIntoSingleLegacyFormattedArray(new Change<>(null, null));
- assertThat(
- FullChangeSerde.decomposeLegacyFormattedArrayIntoChangeArrays(legacyFormat),
- is(new Change<byte[]>(null, null))
+ assertEquals(
+ new Change<byte[]>(null, null),
+ FullChangeSerde.decomposeLegacyFormattedArrayIntoChangeArrays(legacyFormat)
);
}
@@ -100,9 +93,9 @@ public class FullChangeSerdeTest {
final Change<byte[]> serialized = serde.serializeParts(null, new RecordHeaders(), new Change<>("new", null));
final byte[] legacyFormat = mergeChangeArraysIntoSingleLegacyFormattedArray(serialized);
final Change<byte[]> decomposedLegacyFormat = FullChangeSerde.decomposeLegacyFormattedArrayIntoChangeArrays(legacyFormat);
- assertThat(
- serde.deserializeParts(null, new RecordHeaders(), decomposedLegacyFormat),
- is(new Change<>("new", null))
+ assertEquals(
+ new Change<>("new", null),
+ serde.deserializeParts(null, new RecordHeaders(), decomposedLegacyFormat)
);
}
@@ -111,9 +104,9 @@ public class FullChangeSerdeTest {
final Change<byte[]> serialized = serde.serializeParts(null, new RecordHeaders(), new Change<>(null, "old"));
final byte[] legacyFormat = mergeChangeArraysIntoSingleLegacyFormattedArray(serialized);
final Change<byte[]> decomposedLegacyFormat = FullChangeSerde.decomposeLegacyFormattedArrayIntoChangeArrays(legacyFormat);
- assertThat(
- serde.deserializeParts(null, new RecordHeaders(), decomposedLegacyFormat),
- is(new Change<>(null, "old"))
+ assertEquals(
+ new Change<>(null, "old"),
+ serde.deserializeParts(null, new RecordHeaders(), decomposedLegacyFormat)
);
}
@@ -122,9 +115,9 @@ public class FullChangeSerdeTest {
final Change<byte[]> serialized = serde.serializeParts(null, new RecordHeaders(), new Change<>("new", "old"));
final byte[] legacyFormat = mergeChangeArraysIntoSingleLegacyFormattedArray(serialized);
final Change<byte[]> decomposedLegacyFormat = FullChangeSerde.decomposeLegacyFormattedArrayIntoChangeArrays(legacyFormat);
- assertThat(
- serde.deserializeParts(null, new RecordHeaders(), decomposedLegacyFormat),
- is(new Change<>("new", "old"))
+ assertEquals(
+ new Change<>("new", "old"),
+ serde.deserializeParts(null, new RecordHeaders(), decomposedLegacyFormat)
);
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/InternalStreamsBuilderTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/InternalStreamsBuilderTest.java
index 0d12532a800..c8a55ad0dac 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/InternalStreamsBuilderTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/InternalStreamsBuilderTest.java
@@ -68,9 +68,6 @@ import java.util.regex.Pattern;
import static java.time.Duration.ofMillis;
import static java.util.Arrays.asList;
-import static org.hamcrest.CoreMatchers.equalTo;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.core.IsInstanceOf.instanceOf;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
@@ -297,7 +294,7 @@ public class InternalStreamsBuilderTest {
builder.stream(Collections.singleton(topicName), consumed);
builder.buildAndOptimizeTopology();
- assertThat(builder.internalTopologyBuilder.offsetResetStrategy(topicName), equalTo(AutoOffsetResetStrategy.NONE));
+ assertEquals(AutoOffsetResetStrategy.NONE, builder.internalTopologyBuilder.offsetResetStrategy(topicName));
}
@Test
@@ -307,7 +304,7 @@ public class InternalStreamsBuilderTest {
builder.stream(Collections.singleton(topicName), consumed);
builder.buildAndOptimizeTopology();
- assertThat(builder.internalTopologyBuilder.offsetResetStrategy(topicName), equalTo(AutoOffsetResetStrategy.EARLIEST));
+ assertEquals(AutoOffsetResetStrategy.EARLIEST, builder.internalTopologyBuilder.offsetResetStrategy(topicName));
}
@Test
@@ -317,7 +314,7 @@ public class InternalStreamsBuilderTest {
final ConsumedInternal<String, String> consumed = new ConsumedInternal<>(Consumed.with(AutoOffsetReset.latest()));
builder.stream(Collections.singleton(topicName), consumed);
builder.buildAndOptimizeTopology();
- assertThat(builder.internalTopologyBuilder.offsetResetStrategy(topicName), equalTo(AutoOffsetResetStrategy.LATEST));
+ assertEquals(AutoOffsetResetStrategy.LATEST, builder.internalTopologyBuilder.offsetResetStrategy(topicName));
}
@Test
@@ -327,8 +324,8 @@ public class InternalStreamsBuilderTest {
final ConsumedInternal<String, String> consumed = new ConsumedInternal<>(Consumed.with(new AutoOffsetResetInternal(AutoOffsetReset.byDuration(Duration.ofSeconds(42L)))));
builder.stream(Collections.singleton(topicName), consumed);
builder.buildAndOptimizeTopology();
- assertThat(builder.internalTopologyBuilder.offsetResetStrategy(topicName).type(), equalTo(AutoOffsetResetStrategy.StrategyType.BY_DURATION));
- assertThat(builder.internalTopologyBuilder.offsetResetStrategy(topicName).duration().get().toSeconds(), equalTo(42L));
+ assertEquals(AutoOffsetResetStrategy.StrategyType.BY_DURATION, builder.internalTopologyBuilder.offsetResetStrategy(topicName).type());
+ assertEquals(42L, builder.internalTopologyBuilder.offsetResetStrategy(topicName).duration().get().toSeconds());
}
@Test
@@ -336,7 +333,7 @@ public class InternalStreamsBuilderTest {
final String topicName = "topic-1";
builder.table(topicName, new ConsumedInternal<>(Consumed.with(AutoOffsetReset.none())), materialized);
builder.buildAndOptimizeTopology();
- assertThat(builder.internalTopologyBuilder.offsetResetStrategy(topicName), equalTo(AutoOffsetResetStrategy.NONE));
+ assertEquals(AutoOffsetResetStrategy.NONE, builder.internalTopologyBuilder.offsetResetStrategy(topicName));
}
@Test
@@ -344,7 +341,7 @@ public class InternalStreamsBuilderTest {
final String topicName = "topic-1";
builder.table(topicName, new ConsumedInternal<>(Consumed.with(AutoOffsetReset.earliest())), materialized);
builder.buildAndOptimizeTopology();
- assertThat(builder.internalTopologyBuilder.offsetResetStrategy(topicName), equalTo(AutoOffsetResetStrategy.EARLIEST));
+ assertEquals(AutoOffsetResetStrategy.EARLIEST, builder.internalTopologyBuilder.offsetResetStrategy(topicName));
}
@Test
@@ -352,7 +349,7 @@ public class InternalStreamsBuilderTest {
final String topicName = "topic-1";
builder.table(topicName, new ConsumedInternal<>(Consumed.with(AutoOffsetReset.latest())), materialized);
builder.buildAndOptimizeTopology();
- assertThat(builder.internalTopologyBuilder.offsetResetStrategy(topicName), equalTo(AutoOffsetResetStrategy.LATEST));
+ assertEquals(AutoOffsetResetStrategy.LATEST, builder.internalTopologyBuilder.offsetResetStrategy(topicName));
}
@Test
@@ -360,8 +357,8 @@ public class InternalStreamsBuilderTest {
final String topicName = "topic-1";
builder.table(topicName, new ConsumedInternal<>(Consumed.with(AutoOffsetResetInternal.byDuration(Duration.ofSeconds(42L)))), materialized);
builder.buildAndOptimizeTopology();
- assertThat(builder.internalTopologyBuilder.offsetResetStrategy(topicName).type(), equalTo(AutoOffsetResetStrategy.StrategyType.BY_DURATION));
- assertThat(builder.internalTopologyBuilder.offsetResetStrategy(topicName).duration().get().toSeconds(), equalTo(42L));
+ assertEquals(AutoOffsetResetStrategy.StrategyType.BY_DURATION, builder.internalTopologyBuilder.offsetResetStrategy(topicName).type());
+ assertEquals(42L, builder.internalTopologyBuilder.offsetResetStrategy(topicName).duration().get().toSeconds());
}
@Test
@@ -371,7 +368,7 @@ public class InternalStreamsBuilderTest {
builder.table(topicName, consumed, materialized);
builder.buildAndOptimizeTopology();
- assertThat(builder.internalTopologyBuilder.offsetResetStrategy(topicName), equalTo(null));
+ assertNull(builder.internalTopologyBuilder.offsetResetStrategy(topicName));
}
@Test
@@ -382,7 +379,7 @@ public class InternalStreamsBuilderTest {
builder.stream(topicPattern, consumed);
builder.buildAndOptimizeTopology();
- assertThat(builder.internalTopologyBuilder.offsetResetStrategy(topic), equalTo(null));
+ assertNull(builder.internalTopologyBuilder.offsetResetStrategy(topic));
}
@Test
@@ -393,7 +390,7 @@ public class InternalStreamsBuilderTest {
builder.stream(topicPattern, new ConsumedInternal<>(Consumed.with(AutoOffsetReset.earliest())));
builder.buildAndOptimizeTopology();
- assertThat(builder.internalTopologyBuilder.offsetResetStrategy(topicTwo), equalTo(AutoOffsetResetStrategy.EARLIEST));
+ assertEquals(AutoOffsetResetStrategy.EARLIEST, builder.internalTopologyBuilder.offsetResetStrategy(topicTwo));
}
@Test
@@ -404,7 +401,7 @@ public class InternalStreamsBuilderTest {
builder.stream(topicPattern, new ConsumedInternal<>(Consumed.with(AutoOffsetReset.latest())));
builder.buildAndOptimizeTopology();
- assertThat(builder.internalTopologyBuilder.offsetResetStrategy(topicTwo), equalTo(AutoOffsetResetStrategy.LATEST));
+ assertEquals(AutoOffsetResetStrategy.LATEST, builder.internalTopologyBuilder.offsetResetStrategy(topicTwo));
}
@Test
@@ -415,8 +412,8 @@ public class InternalStreamsBuilderTest {
builder.stream(topicPattern, new ConsumedInternal<>(Consumed.with(AutoOffsetResetInternal.byDuration(Duration.ofSeconds(42L)))));
builder.buildAndOptimizeTopology();
- assertThat(builder.internalTopologyBuilder.offsetResetStrategy(topicTwo).type(), equalTo(AutoOffsetResetStrategy.StrategyType.BY_DURATION));
- assertThat(builder.internalTopologyBuilder.offsetResetStrategy(topicTwo).duration().get().toSeconds(), equalTo(42L));
+ assertEquals(AutoOffsetResetStrategy.StrategyType.BY_DURATION, builder.internalTopologyBuilder.offsetResetStrategy(topicTwo).type());
+ assertEquals(42L, builder.internalTopologyBuilder.offsetResetStrategy(topicTwo).duration().get().toSeconds());
}
@Test
@@ -436,7 +433,7 @@ public class InternalStreamsBuilderTest {
final ProcessorTopology processorTopology = builder.internalTopologyBuilder
.rewriteTopology(new StreamsConfig(StreamsTestUtils.getStreamsConfig(APP_ID)))
.buildTopology();
- assertThat(processorTopology.source("topic").timestampExtractor(), instanceOf(MockTimestampExtractor.class));
+ assertInstanceOf(MockTimestampExtractor.class, processorTopology.source("topic").timestampExtractor());
}
@Test
@@ -457,7 +454,7 @@ public class InternalStreamsBuilderTest {
final ProcessorTopology processorTopology = builder.internalTopologyBuilder
.rewriteTopology(new StreamsConfig(StreamsTestUtils.getStreamsConfig(APP_ID)))
.buildTopology();
- assertThat(processorTopology.source("topic").timestampExtractor(), instanceOf(MockTimestampExtractor.class));
+ assertInstanceOf(MockTimestampExtractor.class, processorTopology.source("topic").timestampExtractor());
}
@Test
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KGroupedStreamImplTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KGroupedStreamImplTest.java
index c5ca1ec5f01..7292b612e5a 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KGroupedStreamImplTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KGroupedStreamImplTest.java
@@ -51,18 +51,16 @@ import org.junit.jupiter.api.Test;
import java.time.Duration;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.Comparator;
+import java.util.List;
import java.util.Map;
import java.util.Properties;
import static java.time.Duration.ofMillis;
-import static org.hamcrest.CoreMatchers.equalTo;
-import static org.hamcrest.CoreMatchers.hasItem;
-import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
public class KGroupedStreamImplTest {
@@ -246,7 +244,7 @@ public class KGroupedStreamImplTest {
final ArrayList<KeyValueTimestamp<Windowed<String>, Long>> actual = supplier.theCapturedProcessor().processed();
actual.sort(comparator);
- assertThat(actual, equalTo(Arrays.asList(
+ assertEquals(List.of(
// processing A@500
new KeyValueTimestamp<>(new Windowed<>("1", new TimeWindow(0L, 500L)), 1L, 500L),
// processing A@600
@@ -313,7 +311,7 @@ public class KGroupedStreamImplTest {
new KeyValueTimestamp<>(new Windowed<>("3", new TimeWindow(100L, 600L)), 2L, 600L),
// processing C@600
new KeyValueTimestamp<>(new Windowed<>("3", new TimeWindow(502L, 1002L)), 1L, 600L)
- )));
+ ), actual);
}
private void doAggregateSessionWindows(final MockApiProcessorSupplier<Windowed<String>, Integer, Void, Void> supplier) {
@@ -600,16 +598,16 @@ public class KGroupedStreamImplTest {
{
final KeyValueStore<String, Long> count = driver.getKeyValueStore("count");
- assertThat(count.get("1"), equalTo(3L));
- assertThat(count.get("2"), equalTo(1L));
- assertThat(count.get("3"), equalTo(2L));
+ assertEquals(3L, count.get("1"));
+ assertEquals(1L, count.get("2"));
+ assertEquals(2L, count.get("3"));
}
{
final KeyValueStore<String, ValueAndTimestamp<Long>> count = driver.getTimestampedKeyValueStore("count");
- assertThat(count.get("1"), equalTo(ValueAndTimestamp.make(3L, 10L)));
- assertThat(count.get("2"), equalTo(ValueAndTimestamp.make(1L, 1L)));
- assertThat(count.get("3"), equalTo(ValueAndTimestamp.make(2L, 9L)));
+ assertEquals(ValueAndTimestamp.make(3L, 10L), count.get("1"));
+ assertEquals(ValueAndTimestamp.make(1L, 1L), count.get("2"));
+ assertEquals(ValueAndTimestamp.make(2L, 9L), count.get("3"));
}
}
}
@@ -623,11 +621,8 @@ public class KGroupedStreamImplTest {
processData(driver);
- assertThat(
- appender.getMessages(),
- hasItem("Skipping record due to null key or value. topic=[topic] partition=[0] "
- + "offset=[6]")
- );
+ assertTrue(appender.getMessages().contains(
+ "Skipping record due to null key or value. topic=[topic] partition=[0] offset=[6]"));
}
}
@@ -645,16 +640,16 @@ public class KGroupedStreamImplTest {
{
final KeyValueStore<String, String> reduced = driver.getKeyValueStore("reduce");
- assertThat(reduced.get("1"), equalTo("A+C+D"));
- assertThat(reduced.get("2"), equalTo("B"));
- assertThat(reduced.get("3"), equalTo("E+F"));
+ assertEquals("A+C+D", reduced.get("1"));
+ assertEquals("B", reduced.get("2"));
+ assertEquals("E+F", reduced.get("3"));
}
{
final KeyValueStore<String, ValueAndTimestamp<String>> reduced = driver.getTimestampedKeyValueStore("reduce");
- assertThat(reduced.get("1"), equalTo(ValueAndTimestamp.make("A+C+D", 10L)));
- assertThat(reduced.get("2"), equalTo(ValueAndTimestamp.make("B", 1L)));
- assertThat(reduced.get("3"), equalTo(ValueAndTimestamp.make("E+F", 9L)));
+ assertEquals(ValueAndTimestamp.make("A+C+D", 10L), reduced.get("1"));
+ assertEquals(ValueAndTimestamp.make("B", 1L), reduced.get("2"));
+ assertEquals(ValueAndTimestamp.make("E+F", 9L), reduced.get("3"));
}
}
}
@@ -673,11 +668,8 @@ public class KGroupedStreamImplTest {
processData(driver);
- assertThat(
- appender.getMessages(),
- hasItem("Skipping record due to null key or value. topic=[topic] partition=[0] "
- + "offset=[6]")
- );
+ assertTrue(appender.getMessages().contains(
+ "Skipping record due to null key or value. topic=[topic] partition=[0] offset=[6]"));
}
}
@@ -696,16 +688,16 @@ public class KGroupedStreamImplTest {
{
final KeyValueStore<String, String> aggregate = driver.getKeyValueStore("aggregate");
- assertThat(aggregate.get("1"), equalTo("0+A+C+D"));
- assertThat(aggregate.get("2"), equalTo("0+B"));
- assertThat(aggregate.get("3"), equalTo("0+E+F"));
+ assertEquals("0+A+C+D", aggregate.get("1"));
+ assertEquals("0+B", aggregate.get("2"));
+ assertEquals("0+E+F", aggregate.get("3"));
}
{
final KeyValueStore<String, ValueAndTimestamp<String>> aggregate = driver.getTimestampedKeyValueStore("aggregate");
- assertThat(aggregate.get("1"), equalTo(ValueAndTimestamp.make("0+A+C+D", 10L)));
- assertThat(aggregate.get("2"), equalTo(ValueAndTimestamp.make("0+B", 1L)));
- assertThat(aggregate.get("3"), equalTo(ValueAndTimestamp.make("0+E+F", 9L)));
+ assertEquals(ValueAndTimestamp.make("0+A+C+D", 10L), aggregate.get("1"));
+ assertEquals(ValueAndTimestamp.make("0+B", 1L), aggregate.get("2"));
+ assertEquals(ValueAndTimestamp.make("0+E+F", 9L), aggregate.get("3"));
}
}
}
@@ -721,15 +713,15 @@ public class KGroupedStreamImplTest {
try (final TopologyTestDriver driver = new TopologyTestDriverBuilder(builder.build()).withConfig(props).build()) {
processData(driver);
- assertThat(
- supplier.theCapturedProcessor().lastValueAndTimestampPerKey().get("1"),
- equalTo(ValueAndTimestamp.make("0+A+C+D", 10L)));
- assertThat(
- supplier.theCapturedProcessor().lastValueAndTimestampPerKey().get("2"),
- equalTo(ValueAndTimestamp.make("0+B", 1L)));
- assertThat(
- supplier.theCapturedProcessor().lastValueAndTimestampPerKey().get("3"),
- equalTo(ValueAndTimestamp.make("0+E+F", 9L)));
+ assertEquals(
+ ValueAndTimestamp.make("0+A+C+D", 10L),
+ supplier.theCapturedProcessor().lastValueAndTimestampPerKey().get("1"));
+ assertEquals(
+ ValueAndTimestamp.make("0+B", 1L),
+ supplier.theCapturedProcessor().lastValueAndTimestampPerKey().get("2"));
+ assertEquals(
+ ValueAndTimestamp.make("0+E+F", 9L),
+ supplier.theCapturedProcessor().lastValueAndTimestampPerKey().get("3"));
}
}
@@ -762,7 +754,7 @@ public class KGroupedStreamImplTest {
inputTopic.pipeInput("2", "B", 500L);
inputTopic.pipeInput("3", "B", 100L);
}
- assertThat(supplier.theCapturedProcessor().processed(), equalTo(Arrays.asList(
+ assertEquals(List.of(
new KeyValueTimestamp<>(new Windowed<>("1", new TimeWindow(0L, 500L)), 1L, 0L),
new KeyValueTimestamp<>(new Windowed<>("1", new TimeWindow(0L, 500L)), 2L, 499L),
new KeyValueTimestamp<>(new Windowed<>("1", new TimeWindow(0L, 500L)), 3L, 499L),
@@ -775,7 +767,7 @@ public class KGroupedStreamImplTest {
new KeyValueTimestamp<>(new Windowed<>("2", new TimeWindow(500L, 1000L)), 1L, 500L),
new KeyValueTimestamp<>(new Windowed<>("2", new TimeWindow(500L, 1000L)), 2L, 500L),
new KeyValueTimestamp<>(new Windowed<>("3", new TimeWindow(0L, 500L)), 2L, 100L)
- )));
+ ), supplier.theCapturedProcessor().processed());
}
@Test
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KGroupedTableImplTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KGroupedTableImplTest.java
index 4029bd9d922..d925c404166 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KGroupedTableImplTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KGroupedTableImplTest.java
@@ -47,8 +47,6 @@ import org.junit.jupiter.api.Test;
import java.util.Map;
import java.util.Properties;
-import static org.hamcrest.CoreMatchers.equalTo;
-import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -225,13 +223,13 @@ public class KGroupedTableImplTest {
assertReduced(supplier.theCapturedProcessor().lastValueAndTimestampPerKey(), topic, driver);
{
final KeyValueStore<String, Integer> reduce = driver.getKeyValueStore("reduce");
- assertThat(reduce.get("A"), equalTo(5));
- assertThat(reduce.get("B"), equalTo(6));
+ assertEquals(5, reduce.get("A"));
+ assertEquals(6, reduce.get("B"));
}
{
final KeyValueStore<String, ValueAndTimestamp<Integer>> reduce = driver.getTimestampedKeyValueStore("reduce");
- assertThat(reduce.get("A"), equalTo(ValueAndTimestamp.make(5, 50L)));
- assertThat(reduce.get("B"), equalTo(ValueAndTimestamp.make(6, 30L)));
+ assertEquals(ValueAndTimestamp.make(5, 50L), reduce.get("A"));
+ assertEquals(ValueAndTimestamp.make(6, 30L), reduce.get("B"));
}
}
}
@@ -254,13 +252,13 @@ public class KGroupedTableImplTest {
processData(topic, driver);
{
final KeyValueStore<String, Long> counts = driver.getKeyValueStore("count");
- assertThat(counts.get("1"), equalTo(3L));
- assertThat(counts.get("2"), equalTo(2L));
+ assertEquals(3L, counts.get("1"));
+ assertEquals(2L, counts.get("2"));
}
{
final KeyValueStore<String, ValueAndTimestamp<Long>> counts = driver.getTimestampedKeyValueStore("count");
- assertThat(counts.get("1"), equalTo(ValueAndTimestamp.make(3L, 50L)));
- assertThat(counts.get("2"), equalTo(ValueAndTimestamp.make(2L, 60L)));
+ assertEquals(ValueAndTimestamp.make(3L, 50L), counts.get("1"));
+ assertEquals(ValueAndTimestamp.make(2L, 60L), counts.get("2"));
}
}
}
@@ -287,13 +285,13 @@ public class KGroupedTableImplTest {
{
{
final KeyValueStore<String, String> aggregate = driver.getKeyValueStore("aggregate");
- assertThat(aggregate.get("1"), equalTo("0+1+1+1"));
- assertThat(aggregate.get("2"), equalTo("0+2+2"));
+ assertEquals("0+1+1+1", aggregate.get("1"));
+ assertEquals("0+2+2", aggregate.get("2"));
}
{
final KeyValueStore<String, ValueAndTimestamp<String>> aggregate = driver.getTimestampedKeyValueStore("aggregate");
- assertThat(aggregate.get("1"), equalTo(ValueAndTimestamp.make("0+1+1+1", 50L)));
- assertThat(aggregate.get("2"), equalTo(ValueAndTimestamp.make("0+2+2", 60L)));
+ assertEquals(ValueAndTimestamp.make("0+1+1+1", 50L), aggregate.get("1"));
+ assertEquals(ValueAndTimestamp.make("0+2+2", 60L), aggregate.get("2"));
}
}
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamFlatMapTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamFlatMapTest.java
index d52857e5ed5..fe9fb45f6e5 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamFlatMapTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamFlatMapTest.java
@@ -39,8 +39,6 @@ import java.time.Instant;
import java.util.ArrayList;
import java.util.Properties;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.core.Is.is;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -95,6 +93,6 @@ public class KStreamFlatMapTest {
final KStreamFlatMap<String, Integer, String, Integer> supplier = new KStreamFlatMap<>((key, value) -> null);
final Throwable throwable = assertThrows(NullPointerException.class,
() -> supplier.get().process(new Record<>("K", 0, 0L)));
- assertThat(throwable.getMessage(), is("The provided KeyValueMapper returned null which is not allowed."));
+ assertEquals("The provided KeyValueMapper returned null which is not allowed.", throwable.getMessage());
}
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoinTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoinTest.java
index 9e253ebb0de..d323519fd83 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoinTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoinTest.java
@@ -53,10 +53,8 @@ import java.util.Set;
import static org.apache.kafka.common.utils.Utils.mkEntry;
import static org.apache.kafka.common.utils.Utils.mkMap;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.containsString;
-import static org.hamcrest.Matchers.is;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
public class KStreamGlobalKTableJoinTest {
private static final KeyValueTimestamp[] EMPTY = new KeyValueTimestamp[0];
@@ -308,7 +306,8 @@ public class KStreamGlobalKTableJoinTest {
pushToStream(4, "XXX", false, false);
processor.checkAndClearProcessResult(EMPTY);
- assertThat(
+ assertEquals(
+ 4.0,
driver.metrics().get(
new MetricName(
"dropped-records-total",
@@ -319,8 +318,7 @@ public class KStreamGlobalKTableJoinTest {
mkEntry("task-id", "0_0")
)
))
- .metricValue(),
- is(4.0)
+ .metricValue()
);
}
@@ -337,7 +335,8 @@ public class KStreamGlobalKTableJoinTest {
pushToStream(4, "XXX", false, true);
processor.checkAndClearProcessResult(EMPTY);
- assertThat(
+ assertEquals(
+ 4.0,
driver.metrics().get(
new MetricName(
"dropped-records-total",
@@ -348,8 +347,7 @@ public class KStreamGlobalKTableJoinTest {
mkEntry("task-id", "0_0")
)
))
- .metricValue(),
- is(4.0)
+ .metricValue()
);
}
@@ -368,7 +366,8 @@ public class KStreamGlobalKTableJoinTest {
new KeyValueTimestamp<>(1, "X1,FKey1+Y1", 1)
);
- assertThat(
+ assertEquals(
+ 0.0,
driver.metrics().get(
new MetricName(
"dropped-records-total",
@@ -379,8 +378,7 @@ public class KStreamGlobalKTableJoinTest {
mkEntry("task-id", "0_0")
)
))
- .metricValue(),
- is(0.0)
+ .metricValue()
);
}
@@ -412,7 +410,8 @@ public class KStreamGlobalKTableJoinTest {
new KeyValueTimestamp<>(1, "FKey1|1|X1,FKey1|Y1", 1)
);
- assertThat(
+ assertEquals(
+ 1.0,
driver.metrics().get(
new MetricName(
"dropped-records-total",
@@ -423,8 +422,7 @@ public class KStreamGlobalKTableJoinTest {
mkEntry("task-id", "0_0")
)
))
- .metricValue(),
- is(1.0)
+ .metricValue()
);
}
@@ -440,7 +438,7 @@ public class KStreamGlobalKTableJoinTest {
(ValueJoinerWithStreamAndMappedKey<Integer, String, String, String, String>)
(streamKey, mappedKey, v1, v2) -> v1 + v2,
Named.as("join-table"));
- assertThat(builder.build().describe().toString(), containsString("Processor: join-table"));
+ assertTrue(builder.build().describe().toString().contains("Processor: join-table"));
}
/**
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableLeftJoinTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableLeftJoinTest.java
index 8a8f88e10f5..02b48eb021e 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableLeftJoinTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableLeftJoinTest.java
@@ -53,10 +53,8 @@ import java.util.Set;
import static org.apache.kafka.common.utils.Utils.mkEntry;
import static org.apache.kafka.common.utils.Utils.mkMap;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.containsString;
-import static org.hamcrest.Matchers.is;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
public class KStreamGlobalKTableLeftJoinTest {
private static final KeyValueTimestamp[] EMPTY = new KeyValueTimestamp[0];
@@ -325,7 +323,8 @@ public class KStreamGlobalKTableLeftJoinTest {
new KeyValueTimestamp<>(3, "XXX3+null", 3)
);
- assertThat(
+ assertEquals(
+ 0.0,
driver.metrics().get(
new MetricName(
"dropped-records-total",
@@ -336,8 +335,7 @@ public class KStreamGlobalKTableLeftJoinTest {
mkEntry("task-id", "0_0")
)
))
- .metricValue(),
- is(0.0)
+ .metricValue()
);
}
@@ -360,7 +358,8 @@ public class KStreamGlobalKTableLeftJoinTest {
new KeyValueTimestamp<>(3, "XXX3+null", 3)
);
- assertThat(
+ assertEquals(
+ 0.0,
driver.metrics().get(
new MetricName(
"dropped-records-total",
@@ -371,8 +370,7 @@ public class KStreamGlobalKTableLeftJoinTest {
mkEntry("task-id", "0_0")
)
))
- .metricValue(),
- is(0.0)
+ .metricValue()
);
}
@@ -422,7 +420,8 @@ public class KStreamGlobalKTableLeftJoinTest {
new KeyValueTimestamp<>(1, "FKey1|1|X1,FKey1|Y1", 1)
);
- assertThat(
+ assertEquals(
+ 1.0,
driver.metrics().get(
new MetricName(
"dropped-records-total",
@@ -433,8 +432,7 @@ public class KStreamGlobalKTableLeftJoinTest {
mkEntry("task-id", "0_0")
)
))
- .metricValue(),
- is(1.0)
+ .metricValue()
);
}
@@ -481,7 +479,8 @@ public class KStreamGlobalKTableLeftJoinTest {
new KeyValueTimestamp<>(1, "null|1|XXX1|null", 1)
);
- assertThat(
+ assertEquals(
+ 0.0,
driver.metrics().get(
new MetricName(
"dropped-records-total",
@@ -492,8 +491,7 @@ public class KStreamGlobalKTableLeftJoinTest {
mkEntry("task-id", "0_0")
)
))
- .metricValue(),
- is(0.0)
+ .metricValue()
);
}
@@ -509,6 +507,6 @@ public class KStreamGlobalKTableLeftJoinTest {
(ValueJoinerWithStreamAndMappedKey<Integer, String, String, String, String>) (streamKey, mappedKey, v1, v2) -> v1 + v2,
Named.as("left-join-table"));
- assertThat(builder.build().describe().toString(), containsString("Processor: left-join-table"));
+ assertTrue(builder.build().describe().toString().contains("Processor: left-join-table"));
}
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamImplTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamImplTest.java
index ad48e6deb86..e2df45d6de1 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamImplTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamImplTest.java
@@ -93,16 +93,11 @@ import java.util.regex.Pattern;
import static java.time.Duration.ofMillis;
import static java.util.Arrays.asList;
-import static java.util.Collections.emptyMap;
import static org.apache.kafka.common.utils.Utils.mkEntry;
import static org.apache.kafka.common.utils.Utils.mkMap;
-import static org.hamcrest.CoreMatchers.containsString;
-import static org.hamcrest.CoreMatchers.equalTo;
-import static org.hamcrest.CoreMatchers.is;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.core.IsInstanceOf.instanceOf;
-import static org.hamcrest.core.IsNull.notNullValue;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -135,7 +130,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.filter(null));
- assertThat(exception.getMessage(), equalTo("predicate cannot be null"));
+ assertEquals("predicate cannot be null", exception.getMessage());
}
@Test
@@ -143,7 +138,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.filter(null, Named.as("filter")));
- assertThat(exception.getMessage(), equalTo("predicate cannot be null"));
+ assertEquals("predicate cannot be null", exception.getMessage());
}
@Test
@@ -151,7 +146,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.filter((k, v) -> true, null));
- assertThat(exception.getMessage(), equalTo("named cannot be null"));
+ assertEquals("named cannot be null", exception.getMessage());
}
@Test
@@ -159,7 +154,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.filterNot(null));
- assertThat(exception.getMessage(), equalTo("predicate cannot be null"));
+ assertEquals("predicate cannot be null", exception.getMessage());
}
@Test
@@ -167,7 +162,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.filterNot(null, Named.as("filter")));
- assertThat(exception.getMessage(), equalTo("predicate cannot be null"));
+ assertEquals("predicate cannot be null", exception.getMessage());
}
@Test
@@ -175,7 +170,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.filterNot((k, v) -> true, null));
- assertThat(exception.getMessage(), equalTo("named cannot be null"));
+ assertEquals("named cannot be null", exception.getMessage());
}
@Test
@@ -183,7 +178,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.selectKey(null));
- assertThat(exception.getMessage(), equalTo("mapper cannot be null"));
+ assertEquals("mapper cannot be null", exception.getMessage());
}
@Test
@@ -191,7 +186,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.selectKey(null, Named.as("keySelector")));
- assertThat(exception.getMessage(), equalTo("mapper cannot be null"));
+ assertEquals("mapper cannot be null", exception.getMessage());
}
@Test
@@ -199,7 +194,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.selectKey((k, v) -> k, null));
- assertThat(exception.getMessage(), equalTo("named cannot be null"));
+ assertEquals("named cannot be null", exception.getMessage());
}
@Test
@@ -207,7 +202,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.map(null));
- assertThat(exception.getMessage(), equalTo("mapper cannot be null"));
+ assertEquals("mapper cannot be null", exception.getMessage());
}
@Test
@@ -215,7 +210,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.map(null, Named.as("map")));
- assertThat(exception.getMessage(), equalTo("mapper cannot be null"));
+ assertEquals("mapper cannot be null", exception.getMessage());
}
@Test
@@ -223,7 +218,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.map(KeyValue::pair, null));
- assertThat(exception.getMessage(), equalTo("named cannot be null"));
+ assertEquals("named cannot be null", exception.getMessage());
}
@Test
@@ -231,7 +226,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.mapValues((ValueMapper<Object, Object>) null));
- assertThat(exception.getMessage(), equalTo("mapper cannot be null"));
+ assertEquals("mapper cannot be null", exception.getMessage());
}
@Test
@@ -239,7 +234,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.mapValues((ValueMapperWithKey<Object, Object, Object>) null));
- assertThat(exception.getMessage(), equalTo("mapper cannot be null"));
+ assertEquals("mapper cannot be null", exception.getMessage());
}
@Test
@@ -247,7 +242,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.mapValues((ValueMapper<Object, Object>) null, Named.as("valueMapper")));
- assertThat(exception.getMessage(), equalTo("mapper cannot be null"));
+ assertEquals("mapper cannot be null", exception.getMessage());
}
@Test
@@ -257,7 +252,7 @@ public class KStreamImplTest {
() -> testStream.mapValues(
(ValueMapperWithKey<Object, Object, Object>) null,
Named.as("valueMapperWithKey")));
- assertThat(exception.getMessage(), equalTo("mapper cannot be null"));
+ assertEquals("mapper cannot be null", exception.getMessage());
}
@Test
@@ -265,7 +260,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.mapValues(v -> v, null));
- assertThat(exception.getMessage(), equalTo("named cannot be null"));
+ assertEquals("named cannot be null", exception.getMessage());
}
@Test
@@ -273,7 +268,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.mapValues((k, v) -> v, null));
- assertThat(exception.getMessage(), equalTo("named cannot be null"));
+ assertEquals("named cannot be null", exception.getMessage());
}
@Test
@@ -281,7 +276,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.flatMap(null));
- assertThat(exception.getMessage(), equalTo("mapper cannot be null"));
+ assertEquals("mapper cannot be null", exception.getMessage());
}
@Test
@@ -289,7 +284,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.flatMap(null, Named.as("flatMapper")));
- assertThat(exception.getMessage(), equalTo("mapper cannot be null"));
+ assertEquals("mapper cannot be null", exception.getMessage());
}
@Test
@@ -297,7 +292,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.flatMap((k, v) -> Collections.singleton(new KeyValue<>(k, v)), null));
- assertThat(exception.getMessage(), equalTo("named cannot be null"));
+ assertEquals("named cannot be null", exception.getMessage());
}
@Test
@@ -305,7 +300,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.flatMapValues((ValueMapper<Object, Iterable<Object>>) null));
- assertThat(exception.getMessage(), equalTo("mapper cannot be null"));
+ assertEquals("mapper cannot be null", exception.getMessage());
}
@Test
@@ -313,7 +308,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.flatMapValues((ValueMapperWithKey<Object, Object, ? extends Iterable<Object>>) null));
- assertThat(exception.getMessage(), equalTo("mapper cannot be null"));
+ assertEquals("mapper cannot be null", exception.getMessage());
}
@Test
@@ -323,7 +318,7 @@ public class KStreamImplTest {
() -> testStream.flatMapValues(
(ValueMapper<Object, Iterable<Object>>) null,
Named.as("flatValueMapper")));
- assertThat(exception.getMessage(), equalTo("mapper cannot be null"));
+ assertEquals("mapper cannot be null", exception.getMessage());
}
@Test
@@ -333,7 +328,7 @@ public class KStreamImplTest {
() -> testStream.flatMapValues(
(ValueMapperWithKey<Object, Object, ? extends Iterable<Object>>) null,
Named.as("flatValueMapperWithKey")));
- assertThat(exception.getMessage(), equalTo("mapper cannot be null"));
+ assertEquals("mapper cannot be null", exception.getMessage());
}
@Test
@@ -341,7 +336,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.flatMapValues(v -> Collections.emptyList(), null));
- assertThat(exception.getMessage(), equalTo("named cannot be null"));
+ assertEquals("named cannot be null", exception.getMessage());
}
@Test
@@ -349,7 +344,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.flatMapValues((k, v) -> Collections.emptyList(), null));
- assertThat(exception.getMessage(), equalTo("named cannot be null"));
+ assertEquals("named cannot be null", exception.getMessage());
}
@Test
@@ -357,7 +352,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.print(null));
- assertThat(exception.getMessage(), equalTo("printed cannot be null"));
+ assertEquals("printed cannot be null", exception.getMessage());
}
@Test
@@ -365,7 +360,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.foreach(null));
- assertThat(exception.getMessage(), equalTo("action cannot be null"));
+ assertEquals("action cannot be null", exception.getMessage());
}
@Test
@@ -373,7 +368,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.foreach(null, Named.as("foreach")));
- assertThat(exception.getMessage(), equalTo("action cannot be null"));
+ assertEquals("action cannot be null", exception.getMessage());
}
@Test
@@ -381,7 +376,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.foreach((k, v) -> { }, null));
- assertThat(exception.getMessage(), equalTo("named cannot be null"));
+ assertEquals("named cannot be null", exception.getMessage());
}
@Test
@@ -389,7 +384,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.peek(null));
- assertThat(exception.getMessage(), equalTo("action cannot be null"));
+ assertEquals("action cannot be null", exception.getMessage());
}
@Test
@@ -397,7 +392,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.peek(null, Named.as("peek")));
- assertThat(exception.getMessage(), equalTo("action cannot be null"));
+ assertEquals("action cannot be null", exception.getMessage());
}
@Test
@@ -405,7 +400,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.peek((k, v) -> { }, null));
- assertThat(exception.getMessage(), equalTo("named cannot be null"));
+ assertEquals("named cannot be null", exception.getMessage());
}
@Test
@@ -413,7 +408,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.merge(null));
- assertThat(exception.getMessage(), equalTo("otherStream cannot be null"));
+ assertEquals("otherStream cannot be null", exception.getMessage());
}
@Test
@@ -421,7 +416,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.merge(null, Named.as("merge")));
- assertThat(exception.getMessage(), equalTo("otherStream cannot be null"));
+ assertEquals("otherStream cannot be null", exception.getMessage());
}
@Test
@@ -429,7 +424,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.merge(testStream, null));
- assertThat(exception.getMessage(), equalTo("named cannot be null"));
+ assertEquals("named cannot be null", exception.getMessage());
}
@Test
@@ -437,7 +432,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.to((String) null));
- assertThat(exception.getMessage(), equalTo("topic cannot be null"));
+ assertEquals("topic cannot be null", exception.getMessage());
}
@Test
@@ -445,7 +440,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.repartition(null));
- assertThat(exception.getMessage(), equalTo("repartitioned cannot be null"));
+ assertEquals("repartitioned cannot be null", exception.getMessage());
}
@Test
@@ -453,7 +448,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.to((TopicNameExtractor<String, String>) null));
- assertThat(exception.getMessage(), equalTo("topicExtractor cannot be null"));
+ assertEquals("topicExtractor cannot be null", exception.getMessage());
}
@Test
@@ -461,7 +456,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.to((String) null, Produced.as("to")));
- assertThat(exception.getMessage(), equalTo("topic cannot be null"));
+ assertEquals("topic cannot be null", exception.getMessage());
}
@Test
@@ -469,7 +464,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.to((TopicNameExtractor<String, String>) null, Produced.as("to")));
- assertThat(exception.getMessage(), equalTo("topicExtractor cannot be null"));
+ assertEquals("topicExtractor cannot be null", exception.getMessage());
}
@Test
@@ -477,7 +472,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.to("topic", null));
- assertThat(exception.getMessage(), equalTo("produced cannot be null"));
+ assertEquals("produced cannot be null", exception.getMessage());
}
@Test
@@ -485,7 +480,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.to((k, v, ctx) -> "topic", null));
- assertThat(exception.getMessage(), equalTo("produced cannot be null"));
+ assertEquals("produced cannot be null", exception.getMessage());
}
@Test
@@ -493,7 +488,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.groupBy(null));
- assertThat(exception.getMessage(), equalTo("keySelector cannot be null"));
+ assertEquals("keySelector cannot be null", exception.getMessage());
}
@Test
@@ -501,7 +496,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.groupBy(null, Grouped.as("name")));
- assertThat(exception.getMessage(), equalTo("keySelector cannot be null"));
+ assertEquals("keySelector cannot be null", exception.getMessage());
}
@Test
@@ -509,7 +504,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.groupBy((k, v) -> k, null));
- assertThat(exception.getMessage(), equalTo("grouped cannot be null"));
+ assertEquals("grouped cannot be null", exception.getMessage());
}
@Test
@@ -517,7 +512,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.groupByKey(null));
- assertThat(exception.getMessage(), equalTo("grouped cannot be null"));
+ assertEquals("grouped cannot be null", exception.getMessage());
}
@Test
@@ -525,7 +520,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.toTable((Named) null));
- assertThat(exception.getMessage(), equalTo("named cannot be null"));
+ assertEquals("named cannot be null", exception.getMessage());
}
@Test
@@ -533,7 +528,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.toTable((Materialized<String, String, KeyValueStore<Bytes, byte[]>>) null));
- assertThat(exception.getMessage(), equalTo("materialized cannot be null"));
+ assertEquals("materialized cannot be null", exception.getMessage());
}
@Test
@@ -541,7 +536,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.toTable(null, Materialized.with(null, null)));
- assertThat(exception.getMessage(), equalTo("named cannot be null"));
+ assertEquals("named cannot be null", exception.getMessage());
}
@Test
@@ -549,7 +544,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.toTable(Named.as("name"), null));
- assertThat(exception.getMessage(), equalTo("materialized cannot be null"));
+ assertEquals("materialized cannot be null", exception.getMessage());
}
@SuppressWarnings("deprecation")
@@ -558,7 +553,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.join(null, MockValueJoiner.TOSTRING_JOINER, JoinWindows.of(ofMillis(10))));
- assertThat(exception.getMessage(), equalTo("otherStream cannot be null"));
+ assertEquals("otherStream cannot be null", exception.getMessage());
}
@SuppressWarnings("deprecation")
@@ -571,7 +566,7 @@ public class KStreamImplTest {
MockValueJoiner.TOSTRING_JOINER,
JoinWindows.of(ofMillis(10)),
StreamJoined.as("name")));
- assertThat(exception.getMessage(), equalTo("otherStream cannot be null"));
+ assertEquals("otherStream cannot be null", exception.getMessage());
}
@SuppressWarnings("deprecation")
@@ -580,7 +575,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.join(testStream, (ValueJoiner<? super String, ? super String, ?>) null, JoinWindows.of(ofMillis(10))));
- assertThat(exception.getMessage(), equalTo("joiner cannot be null"));
+ assertEquals("joiner cannot be null", exception.getMessage());
}
@SuppressWarnings("deprecation")
@@ -589,7 +584,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.join(testStream, (ValueJoinerWithKey<? super String, ? super String, ? super String, ?>) null, JoinWindows.of(ofMillis(10))));
- assertThat(exception.getMessage(), equalTo("joiner cannot be null"));
+ assertEquals("joiner cannot be null", exception.getMessage());
}
@SuppressWarnings("deprecation")
@@ -602,7 +597,7 @@ public class KStreamImplTest {
(ValueJoiner<? super String, ? super String, ?>) null,
JoinWindows.of(ofMillis(10)),
StreamJoined.as("name")));
- assertThat(exception.getMessage(), equalTo("joiner cannot be null"));
+ assertEquals("joiner cannot be null", exception.getMessage());
}
@SuppressWarnings("deprecation")
@@ -615,7 +610,7 @@ public class KStreamImplTest {
(ValueJoinerWithKey<? super String, ? super String, ? super String, ?>) null,
JoinWindows.of(ofMillis(10)),
StreamJoined.as("name")));
- assertThat(exception.getMessage(), equalTo("joiner cannot be null"));
+ assertEquals("joiner cannot be null", exception.getMessage());
}
@Test
@@ -623,7 +618,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.join(testStream, MockValueJoiner.TOSTRING_JOINER, null));
- assertThat(exception.getMessage(), equalTo("windows cannot be null"));
+ assertEquals("windows cannot be null", exception.getMessage());
}
@Test
@@ -635,7 +630,7 @@ public class KStreamImplTest {
MockValueJoiner.TOSTRING_JOINER,
null,
StreamJoined.as("name")));
- assertThat(exception.getMessage(), equalTo("windows cannot be null"));
+ assertEquals("windows cannot be null", exception.getMessage());
}
@SuppressWarnings("deprecation")
@@ -648,7 +643,7 @@ public class KStreamImplTest {
MockValueJoiner.TOSTRING_JOINER,
JoinWindows.of(ofMillis(10)),
null));
- assertThat(exception.getMessage(), equalTo("streamJoined cannot be null"));
+ assertEquals("streamJoined cannot be null", exception.getMessage());
}
@SuppressWarnings("deprecation")
@@ -657,7 +652,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.leftJoin(null, MockValueJoiner.TOSTRING_JOINER, JoinWindows.of(ofMillis(10))));
- assertThat(exception.getMessage(), equalTo("otherStream cannot be null"));
+ assertEquals("otherStream cannot be null", exception.getMessage());
}
@SuppressWarnings("deprecation")
@@ -670,7 +665,7 @@ public class KStreamImplTest {
MockValueJoiner.TOSTRING_JOINER,
JoinWindows.of(ofMillis(10)),
StreamJoined.as("name")));
- assertThat(exception.getMessage(), equalTo("otherStream cannot be null"));
+ assertEquals("otherStream cannot be null", exception.getMessage());
}
@SuppressWarnings("deprecation")
@@ -679,7 +674,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.leftJoin(testStream, (ValueJoiner<? super String, ? super String, ?>) null, JoinWindows.of(ofMillis(10))));
- assertThat(exception.getMessage(), equalTo("joiner cannot be null"));
+ assertEquals("joiner cannot be null", exception.getMessage());
}
@SuppressWarnings("deprecation")
@@ -688,7 +683,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.leftJoin(testStream, (ValueJoinerWithKey<? super String, ? super String, ? super String, ?>) null, JoinWindows.of(ofMillis(10))));
- assertThat(exception.getMessage(), equalTo("joiner cannot be null"));
+ assertEquals("joiner cannot be null", exception.getMessage());
}
@SuppressWarnings("deprecation")
@@ -701,7 +696,7 @@ public class KStreamImplTest {
(ValueJoiner<? super String, ? super String, ?>) null,
JoinWindows.of(ofMillis(10)),
StreamJoined.as("name")));
- assertThat(exception.getMessage(), equalTo("joiner cannot be null"));
+ assertEquals("joiner cannot be null", exception.getMessage());
}
@SuppressWarnings("deprecation")
@@ -714,7 +709,7 @@ public class KStreamImplTest {
(ValueJoinerWithKey<? super String, ? super String, ? super String, ?>) null,
JoinWindows.of(ofMillis(10)),
StreamJoined.as("name")));
- assertThat(exception.getMessage(), equalTo("joiner cannot be null"));
+ assertEquals("joiner cannot be null", exception.getMessage());
}
@@ -723,7 +718,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.leftJoin(testStream, MockValueJoiner.TOSTRING_JOINER, null));
- assertThat(exception.getMessage(), equalTo("windows cannot be null"));
+ assertEquals("windows cannot be null", exception.getMessage());
}
@Test
@@ -735,7 +730,7 @@ public class KStreamImplTest {
MockValueJoiner.TOSTRING_JOINER,
null,
StreamJoined.as("name")));
- assertThat(exception.getMessage(), equalTo("windows cannot be null"));
+ assertEquals("windows cannot be null", exception.getMessage());
}
@SuppressWarnings("deprecation")
@@ -748,7 +743,7 @@ public class KStreamImplTest {
MockValueJoiner.TOSTRING_JOINER,
JoinWindows.of(ofMillis(10)),
null));
- assertThat(exception.getMessage(), equalTo("streamJoined cannot be null"));
+ assertEquals("streamJoined cannot be null", exception.getMessage());
}
@SuppressWarnings("deprecation")
@@ -757,7 +752,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.outerJoin(null, MockValueJoiner.TOSTRING_JOINER, JoinWindows.of(ofMillis(10))));
- assertThat(exception.getMessage(), equalTo("otherStream cannot be null"));
+ assertEquals("otherStream cannot be null", exception.getMessage());
}
@SuppressWarnings("deprecation")
@@ -770,7 +765,7 @@ public class KStreamImplTest {
MockValueJoiner.TOSTRING_JOINER,
JoinWindows.of(ofMillis(10)),
StreamJoined.as("name")));
- assertThat(exception.getMessage(), equalTo("otherStream cannot be null"));
+ assertEquals("otherStream cannot be null", exception.getMessage());
}
@SuppressWarnings("deprecation")
@@ -779,7 +774,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.outerJoin(testStream, (ValueJoiner<? super String, ? super String, ?>) null, JoinWindows.of(ofMillis(10))));
- assertThat(exception.getMessage(), equalTo("joiner cannot be null"));
+ assertEquals("joiner cannot be null", exception.getMessage());
}
@SuppressWarnings("deprecation")
@@ -788,7 +783,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.outerJoin(testStream, (ValueJoinerWithKey<? super String, ? super String, ? super String, ?>) null, JoinWindows.of(ofMillis(10))));
- assertThat(exception.getMessage(), equalTo("joiner cannot be null"));
+ assertEquals("joiner cannot be null", exception.getMessage());
}
@SuppressWarnings("deprecation")
@@ -801,7 +796,7 @@ public class KStreamImplTest {
(ValueJoiner<? super String, ? super String, ?>) null,
JoinWindows.of(ofMillis(10)),
StreamJoined.as("name")));
- assertThat(exception.getMessage(), equalTo("joiner cannot be null"));
+ assertEquals("joiner cannot be null", exception.getMessage());
}
@SuppressWarnings("deprecation")
@@ -814,7 +809,7 @@ public class KStreamImplTest {
(ValueJoinerWithKey<? super String, ? super String, ? super String, ?>) null,
JoinWindows.of(ofMillis(10)),
StreamJoined.as("name")));
- assertThat(exception.getMessage(), equalTo("joiner cannot be null"));
+ assertEquals("joiner cannot be null", exception.getMessage());
}
@Test
@@ -822,7 +817,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.outerJoin(testStream, MockValueJoiner.TOSTRING_JOINER, null));
- assertThat(exception.getMessage(), equalTo("windows cannot be null"));
+ assertEquals("windows cannot be null", exception.getMessage());
}
@Test
@@ -834,7 +829,7 @@ public class KStreamImplTest {
MockValueJoiner.TOSTRING_JOINER,
null,
StreamJoined.as("name")));
- assertThat(exception.getMessage(), equalTo("windows cannot be null"));
+ assertEquals("windows cannot be null", exception.getMessage());
}
@SuppressWarnings("deprecation")
@@ -847,7 +842,7 @@ public class KStreamImplTest {
MockValueJoiner.TOSTRING_JOINER,
JoinWindows.of(ofMillis(10)),
null));
- assertThat(exception.getMessage(), equalTo("streamJoined cannot be null"));
+ assertEquals("streamJoined cannot be null", exception.getMessage());
}
@Test
@@ -855,7 +850,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.join(null, MockValueJoiner.TOSTRING_JOINER));
- assertThat(exception.getMessage(), equalTo("table cannot be null"));
+ assertEquals("table cannot be null", exception.getMessage());
}
@Test
@@ -863,7 +858,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.join(null, MockValueJoiner.TOSTRING_JOINER, Joined.as("name")));
- assertThat(exception.getMessage(), equalTo("table cannot be null"));
+ assertEquals("table cannot be null", exception.getMessage());
}
@Test
@@ -871,7 +866,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.join(testTable, (ValueJoiner<? super String, ? super String, ?>) null));
- assertThat(exception.getMessage(), equalTo("joiner cannot be null"));
+ assertEquals("joiner cannot be null", exception.getMessage());
}
@Test
@@ -879,7 +874,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.join(testTable, (ValueJoinerWithKey<? super String, ? super String, ? super String, ?>) null));
- assertThat(exception.getMessage(), equalTo("joiner cannot be null"));
+ assertEquals("joiner cannot be null", exception.getMessage());
}
@Test
@@ -887,7 +882,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.join(testTable, (ValueJoiner<? super String, ? super String, ?>) null, Joined.as("name")));
- assertThat(exception.getMessage(), equalTo("joiner cannot be null"));
+ assertEquals("joiner cannot be null", exception.getMessage());
}
@Test
@@ -895,7 +890,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.join(testTable, (ValueJoinerWithKey<? super String, ? super String, ? super String, ?>) null, Joined.as("name")));
- assertThat(exception.getMessage(), equalTo("joiner cannot be null"));
+ assertEquals("joiner cannot be null", exception.getMessage());
}
@Test
@@ -903,7 +898,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.join(testTable, MockValueJoiner.TOSTRING_JOINER, null));
- assertThat(exception.getMessage(), equalTo("joined cannot be null"));
+ assertEquals("joined cannot be null", exception.getMessage());
}
@Test
@@ -911,7 +906,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.leftJoin(null, MockValueJoiner.TOSTRING_JOINER));
- assertThat(exception.getMessage(), equalTo("table cannot be null"));
+ assertEquals("table cannot be null", exception.getMessage());
}
@Test
@@ -919,7 +914,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.leftJoin(null, MockValueJoiner.TOSTRING_JOINER, Joined.as("name")));
- assertThat(exception.getMessage(), equalTo("table cannot be null"));
+ assertEquals("table cannot be null", exception.getMessage());
}
@Test
@@ -927,7 +922,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.leftJoin(testTable, (ValueJoiner<? super String, ? super String, ?>) null));
- assertThat(exception.getMessage(), equalTo("joiner cannot be null"));
+ assertEquals("joiner cannot be null", exception.getMessage());
}
@Test
@@ -935,7 +930,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.leftJoin(testTable, (ValueJoinerWithKey<? super String, ? super String, ? super String, ?>) null));
- assertThat(exception.getMessage(), equalTo("joiner cannot be null"));
+ assertEquals("joiner cannot be null", exception.getMessage());
}
@Test
@@ -943,7 +938,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.leftJoin(testTable, (ValueJoiner<? super String, ? super String, ?>) null, Joined.as("name")));
- assertThat(exception.getMessage(), equalTo("joiner cannot be null"));
+ assertEquals("joiner cannot be null", exception.getMessage());
}
@Test
@@ -951,7 +946,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.leftJoin(testTable, (ValueJoinerWithKey<? super String, ? super String, ? super String, ?>) null, Joined.as("name")));
- assertThat(exception.getMessage(), equalTo("joiner cannot be null"));
+ assertEquals("joiner cannot be null", exception.getMessage());
}
@Test
@@ -959,7 +954,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.leftJoin(testTable, MockValueJoiner.TOSTRING_JOINER, null));
- assertThat(exception.getMessage(), equalTo("joined cannot be null"));
+ assertEquals("joined cannot be null", exception.getMessage());
}
@Test
@@ -967,7 +962,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.join(null, MockMapper.selectValueMapper(), MockValueJoiner.TOSTRING_JOINER));
- assertThat(exception.getMessage(), equalTo("globalTable cannot be null"));
+ assertEquals("globalTable cannot be null", exception.getMessage());
}
@Test
@@ -979,7 +974,7 @@ public class KStreamImplTest {
MockMapper.selectValueMapper(),
MockValueJoiner.TOSTRING_JOINER,
Named.as("name")));
- assertThat(exception.getMessage(), equalTo("globalTable cannot be null"));
+ assertEquals("globalTable cannot be null", exception.getMessage());
}
@Test
@@ -987,7 +982,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.join(testGlobalTable, null, MockValueJoiner.TOSTRING_JOINER));
- assertThat(exception.getMessage(), equalTo("keySelector cannot be null"));
+ assertEquals("keySelector cannot be null", exception.getMessage());
}
@Test
@@ -999,7 +994,7 @@ public class KStreamImplTest {
null,
MockValueJoiner.TOSTRING_JOINER,
Named.as("name")));
- assertThat(exception.getMessage(), equalTo("keySelector cannot be null"));
+ assertEquals("keySelector cannot be null", exception.getMessage());
}
@Test
@@ -1007,7 +1002,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.join(testGlobalTable, MockMapper.selectValueMapper(), (ValueJoiner<? super String, ? super String, ?>) null));
- assertThat(exception.getMessage(), equalTo("joiner cannot be null"));
+ assertEquals("joiner cannot be null", exception.getMessage());
}
@Test
@@ -1015,7 +1010,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.join(testGlobalTable, MockMapper.selectValueMapper(), (ValueJoinerWithKey<? super String, ? super String, ? super String, ?>) null));
- assertThat(exception.getMessage(), equalTo("joiner cannot be null"));
+ assertEquals("joiner cannot be null", exception.getMessage());
}
@Test
@@ -1027,7 +1022,7 @@ public class KStreamImplTest {
MockMapper.selectValueMapper(),
(ValueJoiner<? super String, ? super String, ?>) null,
Named.as("name")));
- assertThat(exception.getMessage(), equalTo("joiner cannot be null"));
+ assertEquals("joiner cannot be null", exception.getMessage());
}
@Test
@@ -1039,7 +1034,7 @@ public class KStreamImplTest {
MockMapper.selectValueMapper(),
(ValueJoiner<? super String, ? super String, ?>) null,
Named.as("name")));
- assertThat(exception.getMessage(), equalTo("joiner cannot be null"));
+ assertEquals("joiner cannot be null", exception.getMessage());
}
@Test
@@ -1047,7 +1042,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.leftJoin(null, MockMapper.selectValueMapper(), MockValueJoiner.TOSTRING_JOINER));
- assertThat(exception.getMessage(), equalTo("globalTable cannot be null"));
+ assertEquals("globalTable cannot be null", exception.getMessage());
}
@Test
@@ -1059,7 +1054,7 @@ public class KStreamImplTest {
MockMapper.selectValueMapper(),
MockValueJoiner.TOSTRING_JOINER,
Named.as("name")));
- assertThat(exception.getMessage(), equalTo("globalTable cannot be null"));
+ assertEquals("globalTable cannot be null", exception.getMessage());
}
@Test
@@ -1067,7 +1062,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.leftJoin(testGlobalTable, null, MockValueJoiner.TOSTRING_JOINER));
- assertThat(exception.getMessage(), equalTo("keySelector cannot be null"));
+ assertEquals("keySelector cannot be null", exception.getMessage());
}
@Test
@@ -1079,7 +1074,7 @@ public class KStreamImplTest {
null,
MockValueJoiner.TOSTRING_JOINER,
Named.as("name")));
- assertThat(exception.getMessage(), equalTo("keySelector cannot be null"));
+ assertEquals("keySelector cannot be null", exception.getMessage());
}
@Test
@@ -1087,7 +1082,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.leftJoin(testGlobalTable, MockMapper.selectValueMapper(), (ValueJoiner<? super String, ? super String, ?>) null));
- assertThat(exception.getMessage(), equalTo("joiner cannot be null"));
+ assertEquals("joiner cannot be null", exception.getMessage());
}
@Test
@@ -1095,7 +1090,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.leftJoin(testGlobalTable, MockMapper.selectValueMapper(), (ValueJoinerWithKey<? super String, ? super String, ? super String, ?>) null));
- assertThat(exception.getMessage(), equalTo("joiner cannot be null"));
+ assertEquals("joiner cannot be null", exception.getMessage());
}
@Test
@@ -1107,7 +1102,7 @@ public class KStreamImplTest {
MockMapper.selectValueMapper(),
(ValueJoiner<? super String, ? super String, ?>) null,
Named.as("name")));
- assertThat(exception.getMessage(), equalTo("joiner cannot be null"));
+ assertEquals("joiner cannot be null", exception.getMessage());
}
@Test
@@ -1119,7 +1114,7 @@ public class KStreamImplTest {
MockMapper.selectValueMapper(),
(ValueJoinerWithKey<? super String, ? super String, ? super String, ?>) null,
Named.as("name")));
- assertThat(exception.getMessage(), equalTo("joiner cannot be null"));
+ assertEquals("joiner cannot be null", exception.getMessage());
}
@SuppressWarnings({"rawtypes", "deprecation"}) // specifically testing the deprecated variant
@@ -1219,7 +1214,7 @@ public class KStreamImplTest {
stream2.repartition(Repartitioned.as("topic-6"));
final ProcessorTopology processorTopology = TopologyWrapper.getInternalTopologyBuilder(builder.build()).setApplicationId("X").buildTopology();
- assertThat(processorTopology.source("X-topic-6-repartition").timestampExtractor(), instanceOf(FailOnInvalidTimestamp.class));
+ assertInstanceOf(FailOnInvalidTimestamp.class, processorTopology.source("X-topic-6-repartition").timestampExtractor());
assertNull(processorTopology.source("topic-4").timestampExtractor());
assertNull(processorTopology.source("topic-3").timestampExtractor());
assertNull(processorTopology.source("topic-2").timestampExtractor());
@@ -1238,7 +1233,7 @@ public class KStreamImplTest {
driver.createInputTopic(input, new StringSerializer(), new StringSerializer(), Instant.ofEpochMilli(0L), Duration.ZERO);
inputTopic.pipeInput("a", "b");
}
- assertThat(processorSupplier.theCapturedProcessor().processed(), equalTo(Collections.singletonList(new KeyValueTimestamp<>("a", "b", 0))));
+ assertEquals(List.of(new KeyValueTimestamp<>("a", "b", 0)), processorSupplier.theCapturedProcessor().processed());
}
@Test
@@ -1254,7 +1249,7 @@ public class KStreamImplTest {
driver.createInputTopic(input, new StringSerializer(), new StringSerializer(), Instant.ofEpochMilli(0L), Duration.ZERO);
inputTopic.pipeInput("e", "f");
}
- assertThat(processorSupplier.theCapturedProcessor().processed(), equalTo(Collections.singletonList(new KeyValueTimestamp<>("e", "f", 0))));
+ assertEquals(List.of(new KeyValueTimestamp<>("e", "f", 0)), processorSupplier.theCapturedProcessor().processed());
}
@Test
@@ -1275,9 +1270,11 @@ public class KStreamImplTest {
inputTopic.pipeInput("b", "v1");
}
final List<MockApiProcessor<String, String, Void, Void>> mockProcessors = processorSupplier.capturedProcessors(2);
- assertThat(mockProcessors.get(0).processed(), equalTo(asList(new KeyValueTimestamp<>("a", "v1", 0),
- new KeyValueTimestamp<>("a", "v2", 0))));
- assertThat(mockProcessors.get(1).processed(), equalTo(Collections.singletonList(new KeyValueTimestamp<>("b", "v1", 0))));
+ assertEquals(
+ List.of(new KeyValueTimestamp<>("a", "v1", 0), new KeyValueTimestamp<>("a", "v2", 0)),
+ mockProcessors.get(0).processed()
+ );
+ assertEquals(List.of(new KeyValueTimestamp<>("b", "v1", 0)), mockProcessors.get(1).processed());
}
@SuppressWarnings("deprecation")
@@ -1303,7 +1300,7 @@ public class KStreamImplTest {
if (sourceNode.name().equals(originalSourceNode.name())) {
assertNull(sourceNode.timestampExtractor());
} else {
- assertThat(sourceNode.timestampExtractor(), instanceOf(FailOnInvalidTimestamp.class));
+ assertInstanceOf(FailOnInvalidTimestamp.class, sourceNode.timestampExtractor());
}
}
}
@@ -1333,7 +1330,7 @@ public class KStreamImplTest {
if (sourceNode.name().equals(originalSourceNode.name())) {
assertNull(sourceNode.timestampExtractor());
} else {
- assertThat(sourceNode.timestampExtractor(), instanceOf(FailOnInvalidTimestamp.class));
+ assertInstanceOf(FailOnInvalidTimestamp.class, sourceNode.timestampExtractor());
}
}
}
@@ -1354,7 +1351,7 @@ public class KStreamImplTest {
final Matcher matcher = repartitionTopicPattern.matcher(topology);
assertTrue(matcher.find());
final String match = matcher.group();
- assertThat(match, notNullValue());
+ assertNotNull(match);
assertTrue(match.endsWith("repartition"));
}
@@ -1511,7 +1508,7 @@ public class KStreamImplTest {
IllegalArgumentException.class,
() -> testStream.process(() -> processor)
);
- assertThat(exception.getMessage(), containsString("#get() must return a new object each time it is called."));
+ assertTrue(exception.getMessage().contains("#get() must return a new object each time it is called."));
}
@Test
@@ -1522,7 +1519,7 @@ public class KStreamImplTest {
IllegalArgumentException.class,
() -> testStream.process(() -> processor, "storeName")
);
- assertThat(exception.getMessage(), containsString("#get() must return a new object each time it is called."));
+ assertTrue(exception.getMessage().contains("#get() must return a new object each time it is called."));
}
@Test
@@ -1533,7 +1530,7 @@ public class KStreamImplTest {
IllegalArgumentException.class,
() -> testStream.process(() -> processor, Named.as("processor"))
);
- assertThat(exception.getMessage(), containsString("#get() must return a new object each time it is called."));
+ assertTrue(exception.getMessage().contains("#get() must return a new object each time it is called."));
}
@Test
@@ -1544,7 +1541,7 @@ public class KStreamImplTest {
IllegalArgumentException.class,
() -> testStream.process(() -> processor, Named.as("processor"), "storeName")
);
- assertThat(exception.getMessage(), containsString("#get() must return a new object each time it is called."));
+ assertTrue(exception.getMessage().contains("#get() must return a new object each time it is called."));
}
@Test
@@ -1555,7 +1552,7 @@ public class KStreamImplTest {
IllegalArgumentException.class,
() -> testStream.processValues(() -> processor)
);
- assertThat(exception.getMessage(), containsString("#get() must return a new object each time it is called."));
+ assertTrue(exception.getMessage().contains("#get() must return a new object each time it is called."));
}
@Test
@@ -1566,7 +1563,7 @@ public class KStreamImplTest {
IllegalArgumentException.class,
() -> testStream.processValues(() -> processor, "storeName")
);
- assertThat(exception.getMessage(), containsString("#get() must return a new object each time it is called."));
+ assertTrue(exception.getMessage().contains("#get() must return a new object each time it is called."));
}
@Test
@@ -1577,7 +1574,7 @@ public class KStreamImplTest {
IllegalArgumentException.class,
() -> testStream.processValues(() -> processor, Named.as("processor"))
);
- assertThat(exception.getMessage(), containsString("#get() must return a new object each time it is called."));
+ assertTrue(exception.getMessage().contains("#get() must return a new object each time it is called."));
}
@Test
@@ -1588,7 +1585,7 @@ public class KStreamImplTest {
IllegalArgumentException.class,
() -> testStream.processValues(() -> processor, Named.as("processor"), "storeName")
);
- assertThat(exception.getMessage(), containsString("#get() must return a new object each time it is called."));
+ assertTrue(exception.getMessage().contains("#get() must return a new object each time it is called."));
}
@Test
@@ -1596,7 +1593,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.process((ProcessorSupplier<? super String, ? super String, Void, Void>) null));
- assertThat(exception.getMessage(), equalTo("processorSupplier cannot be null"));
+ assertEquals("processorSupplier cannot be null", exception.getMessage());
}
@Test
@@ -1605,7 +1602,7 @@ public class KStreamImplTest {
NullPointerException.class,
() -> testStream.process((ProcessorSupplier<? super String, ? super String, Void, Void>) null,
"storeName"));
- assertThat(exception.getMessage(), equalTo("processorSupplier cannot be null"));
+ assertEquals("processorSupplier cannot be null", exception.getMessage());
}
@Test
@@ -1614,7 +1611,7 @@ public class KStreamImplTest {
NullPointerException.class,
() -> testStream.process((ProcessorSupplier<? super String, ? super String, Void, Void>) null,
Named.as("processor")));
- assertThat(exception.getMessage(), equalTo("processorSupplier cannot be null"));
+ assertEquals("processorSupplier cannot be null", exception.getMessage());
}
@Test
@@ -1623,7 +1620,7 @@ public class KStreamImplTest {
NullPointerException.class,
() -> testStream.process((ProcessorSupplier<? super String, ? super String, Void, Void>) null,
Named.as("processor"), "stateStore"));
- assertThat(exception.getMessage(), equalTo("processorSupplier cannot be null"));
+ assertEquals("processorSupplier cannot be null", exception.getMessage());
}
@Test
@@ -1631,7 +1628,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.process(processorSupplier, (String[]) null));
- assertThat(exception.getMessage(), equalTo("stateStoreNames cannot be a null array"));
+ assertEquals("stateStoreNames cannot be a null array", exception.getMessage());
}
@Test
@@ -1639,7 +1636,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.process(processorSupplier, (String) null));
- assertThat(exception.getMessage(), equalTo("state store name cannot be null"));
+ assertEquals("state store name cannot be null", exception.getMessage());
}
@Test
@@ -1647,7 +1644,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.process(processorSupplier, Named.as("processor"), (String[]) null));
- assertThat(exception.getMessage(), equalTo("stateStoreNames cannot be a null array"));
+ assertEquals("stateStoreNames cannot be a null array", exception.getMessage());
}
@Test
@@ -1655,7 +1652,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.process(processorSupplier, Named.as("processor"), (String) null));
- assertThat(exception.getMessage(), equalTo("state store name cannot be null"));
+ assertEquals("state store name cannot be null", exception.getMessage());
}
@Test
@@ -1663,7 +1660,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.process(processorSupplier, (Named) null));
- assertThat(exception.getMessage(), equalTo("named cannot be null"));
+ assertEquals("named cannot be null", exception.getMessage());
}
@Test
@@ -1671,7 +1668,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.process(processorSupplier, (Named) null, "storeName"));
- assertThat(exception.getMessage(), equalTo("named cannot be null"));
+ assertEquals("named cannot be null", exception.getMessage());
}
@Test
@@ -1679,7 +1676,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.processValues((FixedKeyProcessorSupplier<? super String, ? super String, Void>) null));
- assertThat(exception.getMessage(), equalTo("processorSupplier cannot be null"));
+ assertEquals("processorSupplier cannot be null", exception.getMessage());
}
@Test
@@ -1688,7 +1685,7 @@ public class KStreamImplTest {
NullPointerException.class,
() -> testStream.processValues((FixedKeyProcessorSupplier<? super String, ? super String, Void>) null,
"storeName"));
- assertThat(exception.getMessage(), equalTo("processorSupplier cannot be null"));
+ assertEquals("processorSupplier cannot be null", exception.getMessage());
}
@Test
@@ -1697,7 +1694,7 @@ public class KStreamImplTest {
NullPointerException.class,
() -> testStream.process((ProcessorSupplier<? super String, ? super String, Void, Void>) null,
Named.as("processor")));
- assertThat(exception.getMessage(), equalTo("processorSupplier cannot be null"));
+ assertEquals("processorSupplier cannot be null", exception.getMessage());
}
@Test
@@ -1706,7 +1703,7 @@ public class KStreamImplTest {
NullPointerException.class,
() -> testStream.process((ProcessorSupplier<? super String, ? super String, Void, Void>) null,
Named.as("processor"), "stateStore"));
- assertThat(exception.getMessage(), equalTo("processorSupplier cannot be null"));
+ assertEquals("processorSupplier cannot be null", exception.getMessage());
}
@Test
@@ -1714,7 +1711,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.processValues(fixedKeyProcessorSupplier, (String[]) null));
- assertThat(exception.getMessage(), equalTo("stateStoreNames cannot be a null array"));
+ assertEquals("stateStoreNames cannot be a null array", exception.getMessage());
}
@Test
@@ -1722,7 +1719,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.processValues(fixedKeyProcessorSupplier, (String) null));
- assertThat(exception.getMessage(), equalTo("state store name cannot be null"));
+ assertEquals("state store name cannot be null", exception.getMessage());
}
@Test
@@ -1730,7 +1727,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.processValues(fixedKeyProcessorSupplier, Named.as("processor"), (String[]) null));
- assertThat(exception.getMessage(), equalTo("stateStoreNames cannot be a null array"));
+ assertEquals("stateStoreNames cannot be a null array", exception.getMessage());
}
@Test
@@ -1738,7 +1735,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.processValues(fixedKeyProcessorSupplier, Named.as("processor"), (String) null));
- assertThat(exception.getMessage(), equalTo("state store name cannot be null"));
+ assertEquals("state store name cannot be null", exception.getMessage());
}
@Test
@@ -1746,7 +1743,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.processValues(fixedKeyProcessorSupplier, (Named) null));
- assertThat(exception.getMessage(), equalTo("named cannot be null"));
+ assertEquals("named cannot be null", exception.getMessage());
}
@Test
@@ -1754,7 +1751,7 @@ public class KStreamImplTest {
final NullPointerException exception = assertThrows(
NullPointerException.class,
() -> testStream.processValues(fixedKeyProcessorSupplier, (Named) null, "storeName"));
- assertThat(exception.getMessage(), equalTo("named cannot be null"));
+ assertEquals("named cannot be null", exception.getMessage());
}
@Test
@@ -1770,9 +1767,8 @@ public class KStreamImplTest {
final String topologyDescription = builder.build().describe().toString();
- assertThat(
- topologyDescription,
- equalTo("Topologies:\n" +
+ assertEquals(
+ "Topologies:\n" +
" Sub-topology: 0\n" +
" Source: KSTREAM-SOURCE-0000000000 (topics: [input])\n" +
" --> KSTREAM-TOTABLE-0000000001\n" +
@@ -1783,7 +1779,8 @@ public class KStreamImplTest {
" --> KSTREAM-SINK-0000000004\n" +
" <-- KSTREAM-TOTABLE-0000000001\n" +
" Sink: KSTREAM-SINK-0000000004 (topic: output)\n" +
- " <-- KTABLE-TOSTREAM-0000000003\n\n")
+ " <-- KTABLE-TOSTREAM-0000000003\n\n",
+ topologyDescription
);
try (final TopologyTestDriver driver = new TopologyTestDriverBuilder(builder.build()).withConfig(props).build()) {
@@ -1854,15 +1851,15 @@ public class KStreamImplTest {
final String topologyDescription = builder.build().describe().toString();
- assertThat(
- topologyDescription,
- equalTo("Topologies:\n"
+ assertEquals(
+ "Topologies:\n"
+ " Sub-topology: 0\n"
+ " Source: KSTREAM-SOURCE-0000000000 (topics: [input])\n"
+ " --> p\n"
+ " Processor: p (stores: [sum])\n"
+ " --> none\n"
- + " <-- KSTREAM-SOURCE-0000000000\n\n")
+ + " <-- KSTREAM-SOURCE-0000000000\n\n",
+ topologyDescription
);
try (final TopologyTestDriver driver = new TopologyTestDriverBuilder(builder.build()).withConfig(props).build()) {
@@ -1942,15 +1939,15 @@ public class KStreamImplTest {
final String topologyDescription = builder.build().describe().toString();
- assertThat(
- topologyDescription,
- equalTo("Topologies:\n"
+ assertEquals(
+ "Topologies:\n"
+ " Sub-topology: 0\n"
+ " Source: KSTREAM-SOURCE-0000000000 (topics: [input])\n"
+ " --> p\n"
+ " Processor: p (stores: [sum])\n"
+ " --> none\n"
- + " <-- KSTREAM-SOURCE-0000000000\n\n")
+ + " <-- KSTREAM-SOURCE-0000000000\n\n",
+ topologyDescription
);
try (final TopologyTestDriver driver = new TopologyTestDriverBuilder(builder.build()).withConfig(props).build()) {
@@ -1996,9 +1993,8 @@ public class KStreamImplTest {
final String topologyDescription = builder.build().describe().toString();
- assertThat(
- topologyDescription,
- equalTo("Topologies:\n" +
+ assertEquals(
+ "Topologies:\n" +
" Sub-topology: 0\n" +
" Source: KSTREAM-SOURCE-0000000000 (topics: [input])\n" +
" --> fkp\n" +
@@ -2006,7 +2002,8 @@ public class KStreamImplTest {
" --> KSTREAM-SINK-0000000001\n" +
" <-- KSTREAM-SOURCE-0000000000\n" +
" Sink: KSTREAM-SINK-0000000001 (topic: output)\n" +
- " <-- fkp\n\n")
+ " <-- fkp\n\n",
+ topologyDescription
);
try (final TopologyTestDriver driver = new TopologyTestDriverBuilder(builder.build()).withConfig(props).build()) {
@@ -2056,15 +2053,15 @@ public class KStreamImplTest {
final Topology topology = builder.build();
final String topologyDescription = topology.describe().toString();
- assertThat(
- topologyDescription,
- equalTo("Topologies:\n" +
+ assertEquals(
+ "Topologies:\n" +
" Sub-topology: 0\n" +
" Source: KSTREAM-SOURCE-0000000000 (topics: [input])\n" +
" --> KSTREAM-TOTABLE-0000000001\n" +
" Processor: KSTREAM-TOTABLE-0000000001 (stores: [store])\n" +
" --> none\n" +
- " <-- KSTREAM-SOURCE-0000000000\n\n")
+ " <-- KSTREAM-SOURCE-0000000000\n\n",
+ topologyDescription
);
try (final TopologyTestDriver driver = new TopologyTestDriverBuilder(topology).withConfig(props).build()) {
@@ -2077,7 +2074,7 @@ public class KStreamImplTest {
inputTopic.pipeInput("A", "03");
final Map<String, String> expectedStore = mkMap(mkEntry("A", "03"), mkEntry("B", "02"));
- assertThat(asMap(store), is(expectedStore));
+ assertEquals(expectedStore, asMap(store));
}
}
@@ -2098,9 +2095,8 @@ public class KStreamImplTest {
final Topology topology = builder.build();
final String topologyDescription = topology.describe().toString();
- assertThat(
- topologyDescription,
- equalTo("Topologies:\n" +
+ assertEquals(
+ "Topologies:\n" +
" Sub-topology: 0\n" +
" Source: KSTREAM-SOURCE-0000000000 (topics: [input])\n" +
" --> KSTREAM-MAP-0000000001\n" +
@@ -2123,7 +2119,8 @@ public class KStreamImplTest {
" --> KSTREAM-SINK-0000000008\n" +
" <-- KSTREAM-TOTABLE-0000000002\n" +
" Sink: KSTREAM-SINK-0000000008 (topic: output)\n" +
- " <-- KTABLE-TOSTREAM-0000000007\n\n")
+ " <-- KTABLE-TOSTREAM-0000000007\n\n",
+ topologyDescription
);
try (final TopologyTestDriver driver = new TopologyTestDriverBuilder(topology).withConfig(props).build()) {
@@ -2172,9 +2169,8 @@ public class KStreamImplTest {
final String topologyDescription = topology.describe().toString();
- assertThat(
- topologyDescription,
- equalTo("Topologies:\n" +
+ assertEquals(
+ "Topologies:\n" +
" Sub-topology: 0\n" +
" Source: KTABLE-SOURCE-0000000016 (topics: [KTABLE-FK-JOIN-SUBSCRIPTION-RESPONSE-0000000014-topic])\n" +
" --> KTABLE-FK-JOIN-SUBSCRIPTION-RESPONSE-RESOLVER-PROCESSOR-0000000017\n" +
@@ -2218,7 +2214,8 @@ public class KStreamImplTest {
" --> KTABLE-SINK-0000000015\n" +
" <-- KSTREAM-TOTABLE-0000000004\n" +
" Sink: KTABLE-SINK-0000000015 (topic: KTABLE-FK-JOIN-SUBSCRIPTION-RESPONSE-0000000014-topic)\n" +
- " <-- KTABLE-FK-JOIN-SUBSCRIPTION-PROCESSOR-0000000012, KTABLE-FK-JOIN-SUBSCRIPTION-PROCESSOR-0000000013\n\n")
+ " <-- KTABLE-FK-JOIN-SUBSCRIPTION-PROCESSOR-0000000012, KTABLE-FK-JOIN-SUBSCRIPTION-PROCESSOR-0000000013\n\n",
+ topologyDescription
);
@@ -2232,7 +2229,7 @@ public class KStreamImplTest {
right.pipeInput("rhs2", "rhsValue2");
right.pipeInput("rhs3", "rhsValue3"); // this unreferenced FK won't show up in any results
- assertThat(outputTopic.readKeyValuesToMap(), is(emptyMap()));
+ assertTrue(outputTopic.readKeyValuesToMap().isEmpty());
left.pipeInput("lhs1", "lhsValue1|rhs1");
left.pipeInput("lhs2", "lhsValue2|rhs2");
@@ -2241,25 +2238,15 @@ public class KStreamImplTest {
mkEntry("lhs1", "(lhsValue1|rhs1,rhsValue1)"),
mkEntry("lhs2", "(lhsValue2|rhs2,rhsValue2)")
);
- assertThat(outputTopic.readKeyValuesToMap(), is(expected));
+ assertEquals(expected, outputTopic.readKeyValuesToMap());
// Add another reference to an existing FK
left.pipeInput("lhs3", "lhsValue3|rhs1");
- assertThat(
- outputTopic.readKeyValuesToMap(),
- is(mkMap(
- mkEntry("lhs3", "(lhsValue3|rhs1,rhsValue1)")
- ))
- );
+ assertEquals(mkMap(mkEntry("lhs3", "(lhsValue3|rhs1,rhsValue1)")), outputTopic.readKeyValuesToMap());
left.pipeInput("lhs1", (String) null);
- assertThat(
- outputTopic.readKeyValuesToMap(),
- is(mkMap(
- mkEntry("lhs1", null)
- ))
- );
+ assertEquals(mkMap(mkEntry("lhs1", null)), outputTopic.readKeyValuesToMap());
}
}
@@ -2280,9 +2267,8 @@ public class KStreamImplTest {
final Topology topology = builder.build(props);
final String topologyDescription = topology.describe().toString();
- assertThat(
- topologyDescription,
- equalTo("Topologies:\n" +
+ assertEquals(
+ "Topologies:\n" +
" Sub-topology: 0\n" +
" Source: KSTREAM-SOURCE-0000000000 (topics: [left])\n" +
" --> KSTREAM-TOTABLE-0000000001\n" +
@@ -2307,7 +2293,9 @@ public class KStreamImplTest {
" --> KSTREAM-SINK-0000000010\n" +
" <-- KTABLE-MERGE-0000000006\n" +
" Sink: KSTREAM-SINK-0000000010 (topic: output)\n" +
- " <-- KTABLE-TOSTREAM-0000000009\n\n"));
+ " <-- KTABLE-TOSTREAM-0000000009\n\n",
+ topologyDescription
+ );
try (final TopologyTestDriver driver = new TopologyTestDriverBuilder(topology).withConfig(props).build()) {
final TestInputTopic<String, String> left = driver.createInputTopic(leftTopic, new StringSerializer(), new StringSerializer());
@@ -2318,7 +2306,7 @@ public class KStreamImplTest {
right.pipeInput("rhs2", "rhsValue2");
right.pipeInput("lhs3", "rhsValue3");
- assertThat(output.readKeyValuesToMap(), is(emptyMap()));
+ assertTrue(output.readKeyValuesToMap().isEmpty());
left.pipeInput("lhs1", "lhsValue1");
left.pipeInput("lhs2", "lhsValue2");
@@ -2327,27 +2315,14 @@ public class KStreamImplTest {
mkEntry("lhs1", "lhsValue1+rhsValue1")
);
- assertThat(
- output.readKeyValuesToMap(),
- is(expected)
- );
+ assertEquals(expected, output.readKeyValuesToMap());
left.pipeInput("lhs3", "lhsValue3");
- assertThat(
- output.readKeyValuesToMap(),
- is(mkMap(
- mkEntry("lhs3", "lhsValue3+rhsValue3")
- ))
- );
+ assertEquals(mkMap(mkEntry("lhs3", "lhsValue3+rhsValue3")), output.readKeyValuesToMap());
left.pipeInput("lhs1", "lhsValue4");
- assertThat(
- output.readKeyValuesToMap(),
- is(mkMap(
- mkEntry("lhs1", "lhsValue4+rhsValue1")
- ))
- );
+ assertEquals(mkMap(mkEntry("lhs1", "lhsValue4+rhsValue1")), output.readKeyValuesToMap());
}
}
@@ -2368,9 +2343,8 @@ public class KStreamImplTest {
final Topology topology = builder.build(props);
final String topologyDescription = topology.describe().toString();
- assertThat(
- topologyDescription,
- equalTo("Topologies:\n" +
+ assertEquals(
+ "Topologies:\n" +
" Sub-topology: 0\n" +
" Source: KSTREAM-SOURCE-0000000000 (topics: [streamTopic])\n" +
" --> KSTREAM-JOIN-0000000004\n" +
@@ -2383,7 +2357,9 @@ public class KStreamImplTest {
" <-- KSTREAM-JOIN-0000000004\n" +
" Processor: KSTREAM-TOTABLE-0000000002 (stores: [KSTREAM-TOTABLE-STATE-STORE-0000000003])\n" +
" --> none\n" +
- " <-- KSTREAM-SOURCE-0000000001\n\n"));
+ " <-- KSTREAM-SOURCE-0000000001\n\n",
+ topologyDescription
+ );
try (final TopologyTestDriver driver = new TopologyTestDriverBuilder(topology).withConfig(props).build()) {
final TestInputTopic<String, String> left = driver.createInputTopic(streamTopic, new StringSerializer(), new StringSerializer());
@@ -2394,7 +2370,7 @@ public class KStreamImplTest {
right.pipeInput("rhs2", "rhsValue2");
right.pipeInput("lhs3", "rhsValue3");
- assertThat(output.readKeyValuesToMap(), is(emptyMap()));
+ assertTrue(output.readKeyValuesToMap().isEmpty());
left.pipeInput("lhs1", "lhsValue1");
left.pipeInput("lhs2", "lhsValue2");
@@ -2403,27 +2379,14 @@ public class KStreamImplTest {
mkEntry("lhs1", "lhsValue1+rhsValue1")
);
- assertThat(
- output.readKeyValuesToMap(),
- is(expected)
- );
+ assertEquals(expected, output.readKeyValuesToMap());
left.pipeInput("lhs3", "lhsValue3");
- assertThat(
- output.readKeyValuesToMap(),
- is(mkMap(
- mkEntry("lhs3", "lhsValue3+rhsValue3")
- ))
- );
+ assertEquals(mkMap(mkEntry("lhs3", "lhsValue3+rhsValue3")), output.readKeyValuesToMap());
left.pipeInput("lhs1", "lhsValue4");
- assertThat(
- output.readKeyValuesToMap(),
- is(mkMap(
- mkEntry("lhs1", "lhsValue4+rhsValue1")
- ))
- );
+ assertEquals(mkMap(mkEntry("lhs1", "lhsValue4+rhsValue1")), output.readKeyValuesToMap());
}
}
@@ -2446,9 +2409,8 @@ public class KStreamImplTest {
final Topology topology = builder.build(props);
final String topologyDescription = topology.describe().toString();
- assertThat(
- topologyDescription,
- equalTo("Topologies:\n" +
+ assertEquals(
+ "Topologies:\n" +
" Sub-topology: 0\n" +
" Source: KSTREAM-SOURCE-0000000000 (topics: [input])\n" +
" --> KSTREAM-TOTABLE-0000000001\n" +
@@ -2471,7 +2433,9 @@ public class KStreamImplTest {
" --> KSTREAM-SINK-0000000009\n" +
" <-- KTABLE-AGGREGATE-0000000007\n" +
" Sink: KSTREAM-SINK-0000000009 (topic: output)\n" +
- " <-- KTABLE-TOSTREAM-0000000008\n\n"));
+ " <-- KTABLE-TOSTREAM-0000000008\n\n",
+ topologyDescription
+ );
try (
final TopologyTestDriver driver = new TopologyTestDriverBuilder(topology).withConfig(props).build()) {
@@ -2520,9 +2484,8 @@ public class KStreamImplTest {
final Topology topology = builder.build(props);
final String topologyDescription = topology.describe().toString();
- assertThat(
- topologyDescription,
- equalTo("Topologies:\n" +
+ assertEquals(
+ "Topologies:\n" +
" Sub-topology: 0\n" +
" Source: KSTREAM-SOURCE-0000000000 (topics: [input])\n" +
" --> KSTREAM-TOTABLE-0000000001\n" +
@@ -2536,7 +2499,9 @@ public class KStreamImplTest {
" --> KSTREAM-SINK-0000000005\n" +
" <-- KTABLE-MAPVALUES-0000000003\n" +
" Sink: KSTREAM-SINK-0000000005 (topic: output)\n" +
- " <-- KTABLE-TOSTREAM-0000000004\n\n"));
+ " <-- KTABLE-TOSTREAM-0000000004\n\n",
+ topologyDescription
+ );
try (
final TopologyTestDriver driver = new TopologyTestDriverBuilder(topology).withConfig(props).build()) {
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamJoinTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamJoinTest.java
index 79d041a29a2..ea0aac5386c 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamJoinTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamJoinTest.java
@@ -85,11 +85,8 @@ import java.util.stream.StreamSupport;
import static java.time.Duration.ofHours;
import static java.time.Duration.ofMillis;
import static org.apache.kafka.streams.processor.internals.assignment.AssignmentTestUtils.SUBTOPOLOGY_0;
-import static org.hamcrest.CoreMatchers.equalTo;
-import static org.hamcrest.CoreMatchers.hasItem;
-import static org.hamcrest.CoreMatchers.is;
-import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -127,10 +124,9 @@ public class KStreamKStreamJoinTest {
driver.createInputTopic("left", new StringSerializer(), new IntegerSerializer());
inputTopic.pipeInput("A", null);
- assertThat(
- appender.getMessages(),
- hasItem("Skipping record due to null key or value. topic=[left] partition=[0] offset=[0]")
- );
+ assertTrue(appender.getMessages().contains(
+ "Skipping record due to null key or value. topic=[left] partition=[0] offset=[0]"
+ ));
}
}
@@ -193,8 +189,8 @@ public class KStreamKStreamJoinTest {
final Topology topology = builder.build();
final InternalTopologyBuilder internalTopologyBuilder = TopologyWrapper.getInternalTopologyBuilder(topology);
- assertThat(internalTopologyBuilder.stateStores().get("store-this-join-store").loggingEnabled(), equalTo(false));
- assertThat(internalTopologyBuilder.stateStores().get("store-other-join-store").loggingEnabled(), equalTo(false));
+ assertFalse(internalTopologyBuilder.stateStores().get("store-this-join-store").loggingEnabled());
+ assertFalse(internalTopologyBuilder.stateStores().get("store-other-join-store").loggingEnabled());
}
@ParameterizedTest
@@ -223,14 +219,11 @@ public class KStreamKStreamJoinTest {
internalTopologyBuilder.buildSubtopology(0);
- assertThat(internalTopologyBuilder.stateStores().get("store-this-join-store").loggingEnabled(), equalTo(true));
- assertThat(internalTopologyBuilder.stateStores().get("store-other-join-store").loggingEnabled(), equalTo(true));
- assertThat(internalTopologyBuilder.subtopologyToTopicsInfo().get(SUBTOPOLOGY_0).stateChangelogTopics.size(), equalTo(2));
+ assertTrue(internalTopologyBuilder.stateStores().get("store-this-join-store").loggingEnabled());
+ assertTrue(internalTopologyBuilder.stateStores().get("store-other-join-store").loggingEnabled());
+ assertEquals(2, internalTopologyBuilder.subtopologyToTopicsInfo().get(SUBTOPOLOGY_0).stateChangelogTopics.size());
for (final InternalTopicConfig config : internalTopologyBuilder.subtopologyToTopicsInfo().get(SUBTOPOLOGY_0).stateChangelogTopics.values()) {
- assertThat(
- config.properties(Collections.emptyMap(), 0).get("test"),
- equalTo("property")
- );
+ assertEquals("property", config.properties(Collections.emptyMap(), 0).get("test"));
}
}
@@ -426,16 +419,16 @@ public class KStreamKStreamJoinTest {
// neither side is supplied explicitly
runJoin(streamJoined.withDslStoreSuppliers(dslStoreSuppliers), joinWindows);
- assertThat(TrackingDslStoreSuppliers.NUM_CALLS.get(), is(2));
+ assertEquals(2, TrackingDslStoreSuppliers.NUM_CALLS.get());
// one side is supplied explicitly, so we only increment once
runJoin(streamJoined.withDslStoreSuppliers(dslStoreSuppliers).withThisStoreSupplier(thisStoreSupplier), joinWindows);
- assertThat(TrackingDslStoreSuppliers.NUM_CALLS.get(), is(3));
+ assertEquals(3, TrackingDslStoreSuppliers.NUM_CALLS.get());
// both sides are supplied explicitly, so we don't increment further
runJoin(streamJoined.withDslStoreSuppliers(dslStoreSuppliers)
.withThisStoreSupplier(thisStoreSupplier).withOtherStoreSupplier(otherStoreSupplier), joinWindows);
- assertThat(TrackingDslStoreSuppliers.NUM_CALLS.get(), is(3));
+ assertEquals(3, TrackingDslStoreSuppliers.NUM_CALLS.get());
}
@@ -452,7 +445,7 @@ public class KStreamKStreamJoinTest {
// neither side is supplied explicitly, so we call the dsl supplier twice
runJoin(streamJoined, joinWindows);
- assertThat(TrackingDslStoreSuppliers.NUM_CALLS.get(), is(2));
+ assertEquals(2, TrackingDslStoreSuppliers.NUM_CALLS.get());
}
public static class CapturingStoreSuppliers extends BuiltInDslStoreSuppliers.RocksDBDslStoreSuppliers {
@@ -480,11 +473,15 @@ public class KStreamKStreamJoinTest {
runJoin(streamJoined, joinWindows);
if (withHeaders) {
- assertThat("Expected stream joined to supply builders that create headers stores",
- WrappedStateStore.isHeadersAware(storeSuppliers.capture.get().get()));
+ assertTrue(
+ WrappedStateStore.isHeadersAware(storeSuppliers.capture.get().get()),
+ "Expected stream joined to supply builders that create headers stores"
+ );
} else {
- assertThat("Expected stream joined to supply builders that create non-timestamped stores",
- !WrappedStateStore.isTimestamped(storeSuppliers.capture.get().get()));
+ assertFalse(
+ WrappedStateStore.isTimestamped(storeSuppliers.capture.get().get()),
+ "Expected stream joined to supply builders that create non-timestamped stores"
+ );
}
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamOuterJoinTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamOuterJoinTest.java
index dd17dab0c66..b1d3763ac33 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamOuterJoinTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamOuterJoinTest.java
@@ -59,8 +59,8 @@ import java.util.concurrent.atomic.AtomicReference;
import static java.time.Duration.ZERO;
import static java.time.Duration.ofMillis;
-import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
public class KStreamKStreamOuterJoinTest {
private final String topic1 = "topic1";
@@ -1481,8 +1481,10 @@ public class KStreamKStreamOuterJoinTest {
// create a TTD so that the topology gets built
try (final TopologyTestDriver ignored = new TopologyTestDriverBuilder(builder.build(PROPS)).withConfig(PROPS).build()) {
- assertThat("Expected stream joined to supply builders that create non-timestamped stores",
- !WrappedStateStore.isTimestamped(suppliers.capture.get().get()));
+ assertFalse(
+ WrappedStateStore.isTimestamped(suppliers.capture.get().get()),
+ "Expected stream joined to supply builders that create non-timestamped stores"
+ );
}
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinTest.java
index 256893d8334..4fa8dfd3fd1 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinTest.java
@@ -63,11 +63,9 @@ import java.util.Set;
import static org.apache.kafka.common.utils.Utils.mkEntry;
import static org.apache.kafka.common.utils.Utils.mkMap;
-import static org.hamcrest.CoreMatchers.hasItem;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.is;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
public class KStreamKTableJoinTest {
private static final KeyValueTimestamp<?, ?>[] EMPTY = new KeyValueTimestamp[0];
@@ -175,10 +173,7 @@ public class KStreamKTableJoinTest {
final IllegalArgumentException exception = assertThrows(IllegalArgumentException.class,
() -> streamA.join(tableB, (value1, value2) -> value1 + value2, Joined.with(Serdes.String(), Serdes.String(), Serdes.String(), "first-join", Duration.ofMillis(6))).to("out-one"));
- assertThat(
- exception.getMessage(),
- is("KTable must be versioned to use a grace period in a stream table join.")
- );
+ assertEquals("KTable must be versioned to use a grace period in a stream table join.", exception.getMessage());
}
@ParameterizedTest
@@ -197,10 +192,7 @@ public class KStreamKTableJoinTest {
streamA.join(tableB, (value1, value2) -> value1 + value2, Joined.with(Serdes.String(), Serdes.String(), Serdes.String(), "first-join", Duration.ofMillis(6))).to("out-one");
final IllegalArgumentException exception = assertThrows(IllegalArgumentException.class, builder::build);
- assertThat(
- exception.getMessage(),
- is("KTable must be versioned to use a grace period in a stream table join.")
- );
+ assertEquals("KTable must be versioned to use a grace period in a stream table join.", exception.getMessage());
}
@ParameterizedTest
@@ -236,7 +228,7 @@ public class KStreamKTableJoinTest {
streamA.join(tableB, (value1, value2) -> value1 + value2, Joined.with(Serdes.String(), Serdes.String(), Serdes.String(), "first-join", Duration.ofMinutes(6))).to("out-one");
final IllegalArgumentException exception = assertThrows(IllegalArgumentException.class, () -> builder.build(props));
- assertThat(exception.getMessage(), is("History retention must be at least grace period."));
+ assertEquals("History retention must be at least grace period.", exception.getMessage());
}
@ParameterizedTest
@@ -255,7 +247,7 @@ public class KStreamKTableJoinTest {
streamA.join(tableB, (value1, value2) -> value1 + value2, Joined.with(Serdes.String(), Serdes.String(), Serdes.String(), "first-join", Duration.ofMillis(6))).to("out-one");
final IllegalArgumentException exception = assertThrows(IllegalArgumentException.class, () -> builder.build(props));
- assertThat(exception.getMessage(), is("History retention must be at least grace period."));
+ assertEquals("History retention must be at least grace period.", exception.getMessage());
}
@@ -544,13 +536,13 @@ public class KStreamKTableJoinTest {
driver.createInputTopic(streamTopic, new IntegerSerializer(), new StringSerializer());
inputTopic.pipeInput(null, "A");
- assertThat(
- appender.getMessages(),
- hasItem("Skipping record due to null join key or value. topic=[streamTopic] partition=[0] "
- + "offset=[0]"));
+ assertTrue(appender.getMessages().contains(
+ "Skipping record due to null join key or value. topic=[streamTopic] partition=[0] offset=[0]"
+ ));
}
- assertThat(
+ assertEquals(
+ 1.0,
driver.metrics().get(
new MetricName(
"dropped-records-total",
@@ -561,8 +553,7 @@ public class KStreamKTableJoinTest {
mkEntry("task-id", "0_0")
)
))
- .metricValue(),
- is(1.0)
+ .metricValue()
);
}
@@ -575,14 +566,13 @@ public class KStreamKTableJoinTest {
driver.createInputTopic(streamTopic, new IntegerSerializer(), new StringSerializer());
inputTopic.pipeInput(1, null);
- assertThat(
- appender.getMessages(),
- hasItem("Skipping record due to null join key or value. topic=[streamTopic] partition=[0] "
- + "offset=[0]")
- );
+ assertTrue(appender.getMessages().contains(
+ "Skipping record due to null join key or value. topic=[streamTopic] partition=[0] offset=[0]"
+ ));
}
- assertThat(
+ assertEquals(
+ 1.0,
driver.metrics().get(
new MetricName(
"dropped-records-total",
@@ -593,8 +583,7 @@ public class KStreamKTableJoinTest {
mkEntry("task-id", "0_0")
)
))
- .metricValue(),
- is(1.0)
+ .metricValue()
);
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamMapTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamMapTest.java
index b7cf2d107be..4de8a4fbcfe 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamMapTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamMapTest.java
@@ -37,8 +37,6 @@ import java.time.Duration;
import java.time.Instant;
import java.util.Properties;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.core.Is.is;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -78,7 +76,7 @@ public class KStreamMapTest {
final KStreamMap<String, Integer, String, Integer> supplier = new KStreamMap<>((key, value) -> null);
final Throwable throwable = assertThrows(NullPointerException.class,
() -> supplier.get().process(new Record<>("K", 0, 0L)));
- assertThat(throwable.getMessage(), is("The provided KeyValueMapper returned null which is not allowed."));
+ assertEquals("The provided KeyValueMapper returned null which is not allowed.", throwable.getMessage());
}
@Test
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamRepartitionTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamRepartitionTest.java
index ecb733a2a7d..b6ce5d4105e 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamRepartitionTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamRepartitionTest.java
@@ -53,8 +53,7 @@ import java.util.Optional;
import java.util.Properties;
import java.util.TreeMap;
-import static org.hamcrest.CoreMatchers.equalTo;
-import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -115,8 +114,8 @@ public class KStreamRepartitionTest {
testInputTopic.pipeInput(expectedKeys[i], "X" + expectedKeys[i], i + 10);
}
- assertThat(testOutputTopic.readRecord(), equalTo(new TestRecord<>(0, "X0", Instant.ofEpochMilli(10))));
- assertThat(testOutputTopic.readRecord(), equalTo(new TestRecord<>(1, "X1", Instant.ofEpochMilli(11))));
+ assertEquals(new TestRecord<>(0, "X0", Instant.ofEpochMilli(10)), testOutputTopic.readRecord());
+ assertEquals(new TestRecord<>(1, "X1", Instant.ofEpochMilli(11)), testOutputTopic.readRecord());
assertTrue(testOutputTopic.readRecordsToList().isEmpty());
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregateProcessorTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregateProcessorTest.java
index 1e945b52f3a..52a29c8f145 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregateProcessorTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregateProcessorTest.java
@@ -21,7 +21,6 @@ import org.apache.kafka.common.header.internals.RecordHeaders;
import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.LogCaptureAppender;
-import org.apache.kafka.common.utils.LogCaptureAppender.Event;
import org.apache.kafka.common.utils.MockTime;
import org.apache.kafka.common.utils.internals.LogContext;
import org.apache.kafka.streams.KeyValue;
@@ -61,7 +60,6 @@ import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Properties;
-import java.util.stream.Collectors;
import java.util.stream.Stream;
import static java.time.Duration.ofMillis;
@@ -69,10 +67,6 @@ import static org.apache.kafka.common.utils.Utils.mkEntry;
import static org.apache.kafka.common.utils.Utils.mkMap;
import static org.apache.kafka.streams.utils.TestUtils.mockStoreFactory;
import static org.apache.kafka.test.StreamsTestUtils.getMetricByName;
-import static org.hamcrest.CoreMatchers.hasItem;
-import static org.hamcrest.CoreMatchers.is;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.greaterThan;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -490,13 +484,8 @@ public class KStreamSessionWindowAggregateProcessorTest {
processor.process(new Record<>(null, "1", 0L));
- assertThat(
- appender.getEvents().stream()
- .filter(e -> e.getLevel().equals("WARN"))
- .map(Event::getMessage)
- .collect(Collectors.toList()),
- hasItem("Skipping record due to null key. topic=[topic] partition=[-3] offset=[-2]")
- );
+ assertTrue(appender.getMessages("WARN").contains(
+ "Skipping record due to null key. topic=[topic] partition=[-3] offset=[-2]"));
}
assertEquals(
@@ -538,11 +527,10 @@ public class KStreamSessionWindowAggregateProcessorTest {
mockContext.setRecordContext(new ProcessorRecordContext(0, -2, -3, "topic", new RecordHeaders()));
processor.process(new Record<>("Late1", "1", 0L));
- assertThat(
- appender.getMessages(),
- hasItem("Skipping record for expired window." +
- " topic=[topic] partition=[-3] offset=[-2] timestamp=[0] window=[0,0] expiration=[1] streamTime=[11]")
- );
+ assertTrue(appender.getMessages().contains(
+ "Skipping record for expired window."
+ + " topic=[topic] partition=[-3] offset=[-2] timestamp=[0] window=[0,0] expiration=[1] streamTime=[11]"
+ ));
}
final MetricName dropTotal;
@@ -565,11 +553,8 @@ public class KStreamSessionWindowAggregateProcessorTest {
mkEntry("task-id", "0_0")
)
);
- assertThat(metrics.metrics().get(dropTotal).metricValue(), is(1.0));
- assertThat(
- (Double) metrics.metrics().get(dropRate).metricValue(),
- greaterThan(0.0)
- );
+ assertEquals(1.0, metrics.metrics().get(dropTotal).metricValue());
+ assertTrue((Double) metrics.metrics().get(dropRate).metricValue() > 0.0);
}
@ParameterizedTest
@@ -613,11 +598,10 @@ public class KStreamSessionWindowAggregateProcessorTest {
mockContext.setRecordContext(new ProcessorRecordContext(0, -2, -3, "topic", new RecordHeaders()));
processor.process(new Record<>("Late1", "1", 0L));
- assertThat(
- appender.getMessages(),
- hasItem("Skipping record for expired window." +
- " topic=[topic] partition=[-3] offset=[-2] timestamp=[0] window=[0,0] expiration=[1] streamTime=[12]")
- );
+ assertTrue(appender.getMessages().contains(
+ "Skipping record for expired window."
+ + " topic=[topic] partition=[-3] offset=[-2] timestamp=[0] window=[0,0] expiration=[1] streamTime=[12]"
+ ));
}
final MetricName dropTotal;
@@ -641,9 +625,7 @@ public class KStreamSessionWindowAggregateProcessorTest {
)
);
- assertThat(metrics.metrics().get(dropTotal).metricValue(), is(1.0));
- assertThat(
- (Double) metrics.metrics().get(dropRate).metricValue(),
- greaterThan(0.0));
+ assertEquals(1.0, metrics.metrics().get(dropTotal).metricValue());
+ assertTrue((Double) metrics.metrics().get(dropRate).metricValue() > 0.0);
}
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamSlidingWindowAggregateTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamSlidingWindowAggregateTest.java
index 87490146136..f32345e6fc0 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamSlidingWindowAggregateTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamSlidingWindowAggregateTest.java
@@ -22,7 +22,6 @@ import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.common.utils.LogCaptureAppender;
-import org.apache.kafka.common.utils.LogCaptureAppender.Event;
import org.apache.kafka.streams.KeyValueTimestamp;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
@@ -56,7 +55,6 @@ import org.apache.kafka.test.MockInitializer;
import org.apache.kafka.test.MockReducer;
import org.apache.kafka.test.StreamsTestUtils;
-import org.hamcrest.Matcher;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
@@ -73,20 +71,14 @@ import java.util.Map;
import java.util.Properties;
import java.util.Random;
import java.util.Set;
-import java.util.stream.Collectors;
import java.util.stream.Stream;
import static java.time.Duration.ofMillis;
import static java.util.Arrays.asList;
import static org.apache.kafka.common.utils.Utils.mkEntry;
import static org.apache.kafka.common.utils.Utils.mkMap;
-import static org.hamcrest.CoreMatchers.equalTo;
-import static org.hamcrest.CoreMatchers.hasItem;
-import static org.hamcrest.CoreMatchers.hasItems;
-import static org.hamcrest.CoreMatchers.is;
-import static org.hamcrest.CoreMatchers.not;
-import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
public class KStreamSlidingWindowAggregateTest {
@@ -1490,13 +1482,8 @@ public class KStreamSlidingWindowAggregateTest {
final TestInputTopic<String, String> inputTopic =
driver.createInputTopic(topic, new StringSerializer(), new StringSerializer());
inputTopic.pipeInput(null, "1");
- assertThat(
- appender.getEvents().stream()
- .filter(e -> e.getLevel().equals("WARN"))
- .map(Event::getMessage)
- .collect(Collectors.toList()),
- hasItem("Skipping record due to null key or value. topic=[topic] partition=[0] offset=[0]")
- );
+ assertTrue(appender.getMessages("WARN").contains(
+ "Skipping record due to null key or value. topic=[topic] partition=[0] offset=[0]"));
}
}
@@ -1540,9 +1527,9 @@ public class KStreamSlidingWindowAggregateTest {
inputTopic.pipeInput("k", "101", 300L);
inputTopic.pipeInput("k", "102", 400L);
- assertLatenessMetrics(driver, is(7.0), is(185.0), is(77.0));
+ assertLatenessMetrics(driver, 7.0, 185.0, 77.0);
- assertThat(appender.getMessages(), hasItems(
+ assertTrue(appender.getMessages().containsAll(List.of(
// left window for k@100
"Skipping record for expired window. topic=[topic] partition=[0] offset=[1] timestamp=[100] window=[90,100] expiration=[110] streamTime=[200]",
// left window for k@101
@@ -1557,23 +1544,33 @@ public class KStreamSlidingWindowAggregateTest {
"Skipping record for expired window. topic=[topic] partition=[0] offset=[6] timestamp=[105] window=[95,105] expiration=[110] streamTime=[200]",
// left window for k@15
"Skipping record for expired window. topic=[topic] partition=[0] offset=[7] timestamp=[15] window=[15,25] expiration=[110] streamTime=[200]"
- ));
+ )));
final TestOutputTopic<Windowed<String>, String> outputTopic =
driver.createOutputTopic("output", new TimeWindowedDeserializer<>(new StringDeserializer(), 10L), new StringDeserializer());
if (emitFinal) {
- assertThat(outputTopic.readRecord(), equalTo(
- new TestRecord<>(new Windowed<>("k", new TimeWindow(190, 200)), "0+100", null, 200L)));
- assertThat(outputTopic.readRecord(), equalTo(
- new TestRecord<>(new Windowed<>("k", new TimeWindow(290, 300)), "0+101", null, 300L)));
+ assertEquals(
+ new TestRecord<>(new Windowed<>("k", new TimeWindow(190, 200)), "0+100", null, 200L),
+ outputTopic.readRecord()
+ );
+ assertEquals(
+ new TestRecord<>(new Windowed<>("k", new TimeWindow(290, 300)), "0+101", null, 300L),
+ outputTopic.readRecord()
+ );
assertTrue(outputTopic.isEmpty());
} else {
- assertThat(outputTopic.readRecord(), equalTo(
- new TestRecord<>(new Windowed<>("k", new TimeWindow(190, 200)), "0+100", null, 200L)));
- assertThat(outputTopic.readRecord(), equalTo(
- new TestRecord<>(new Windowed<>("k", new TimeWindow(290, 300)), "0+101", null, 300L)));
- assertThat(outputTopic.readRecord(), equalTo(
- new TestRecord<>(new Windowed<>("k", new TimeWindow(390, 400)), "0+102", null, 400L)));
+ assertEquals(
+ new TestRecord<>(new Windowed<>("k", new TimeWindow(190, 200)), "0+100", null, 200L),
+ outputTopic.readRecord()
+ );
+ assertEquals(
+ new TestRecord<>(new Windowed<>("k", new TimeWindow(290, 300)), "0+101", null, 300L),
+ outputTopic.readRecord()
+ );
+ assertEquals(
+ new TestRecord<>(new Windowed<>("k", new TimeWindow(390, 400)), "0+102", null, 400L),
+ outputTopic.readRecord()
+ );
assertTrue(outputTopic.isEmpty());
}
}
@@ -1723,9 +1720,9 @@ public class KStreamSlidingWindowAggregateTest {
}
private void assertLatenessMetrics(final TopologyTestDriver driver,
- final Matcher<Object> dropTotal,
- final Matcher<Object> maxLateness,
- final Matcher<Object> avgLateness) {
+ final double expectedDropTotal,
+ final double expectedMaxLateness,
+ final double expectedAvgLateness) {
final MetricName dropTotalMetric;
final MetricName dropRateMetric;
@@ -1769,10 +1766,10 @@ public class KStreamSlidingWindowAggregateTest {
mkEntry("task-id", "0_0")
)
);
- assertThat(driver.metrics().get(dropTotalMetric).metricValue(), dropTotal);
- assertThat(driver.metrics().get(dropRateMetric).metricValue(), not(0.0));
- assertThat(driver.metrics().get(latenessMaxMetric).metricValue(), maxLateness);
- assertThat(driver.metrics().get(latenessAvgMetric).metricValue(), avgLateness);
+ assertEquals(expectedDropTotal, driver.metrics().get(dropTotalMetric).metricValue());
+ assertNotEquals(0.0, driver.metrics().get(dropRateMetric).metricValue());
+ assertEquals(expectedMaxLateness, driver.metrics().get(latenessMaxMetric).metricValue());
+ assertEquals(expectedAvgLateness, driver.metrics().get(latenessAvgMetric).metricValue());
}
private WindowBytesStoreSupplier setupWindowBytesStoreSupplier(final int index) {
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamWindowAggregateTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamWindowAggregateTest.java
index cdd16c50d1b..7bf360d0e92 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamWindowAggregateTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamWindowAggregateTest.java
@@ -63,7 +63,6 @@ import org.apache.kafka.test.MockInternalProcessorContext;
import org.apache.kafka.test.StreamsTestUtils;
import org.apache.kafka.test.TestUtils;
-import org.hamcrest.Matcher;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
@@ -80,13 +79,8 @@ import static java.util.Arrays.asList;
import static org.apache.kafka.common.utils.Utils.mkEntry;
import static org.apache.kafka.common.utils.Utils.mkMap;
import static org.apache.kafka.streams.utils.TestUtils.mockStoreFactory;
-import static org.hamcrest.CoreMatchers.equalTo;
-import static org.hamcrest.CoreMatchers.hasItem;
-import static org.hamcrest.CoreMatchers.hasItems;
-import static org.hamcrest.CoreMatchers.is;
-import static org.hamcrest.CoreMatchers.not;
-import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -499,7 +493,7 @@ public class KStreamWindowAggregateTest {
driver.createInputTopic(topic, new StringSerializer(), new StringSerializer());
inputTopic.pipeInput(null, "1");
- assertThat(appender.getMessages(), hasItem("Skipping record due to null key. topic=[topic] partition=[0] offset=[0]"));
+ assertTrue(appender.getMessages().contains("Skipping record due to null key. topic=[topic] partition=[0] offset=[0]"));
}
}
@@ -544,12 +538,12 @@ public class KStreamWindowAggregateTest {
assertLatenessMetrics(
driver,
- is(7.0), // how many events get dropped
- is(100.0), // k:0 is 100ms late, since its time is 0, but it arrives at stream time 100.
- is(67.9) // (0 + 100 + 99 + 98 + 97 + 96 + 95 + 94 + 0) / 10
+ 7.0, // how many events get dropped
+ 100.0, // k:0 is 100ms late, since its time is 0, but it arrives at stream time 100.
+ 67.9 // (0 + 100 + 99 + 98 + 97 + 96 + 95 + 94 + 0) / 10
);
- assertThat(appender.getMessages(), hasItems(
+ assertTrue(appender.getMessages().containsAll(List.of(
"Skipping record for expired window. topic=[topic] partition=[0] offset=[1] timestamp=[0] window=[0,10) expiration=[10] streamTime=[100]",
"Skipping record for expired window. topic=[topic] partition=[0] offset=[2] timestamp=[1] window=[0,10) expiration=[10] streamTime=[100]",
"Skipping record for expired window. topic=[topic] partition=[0] offset=[3] timestamp=[2] window=[0,10) expiration=[10] streamTime=[100]",
@@ -557,33 +551,24 @@ public class KStreamWindowAggregateTest {
"Skipping record for expired window. topic=[topic] partition=[0] offset=[5] timestamp=[4] window=[0,10) expiration=[10] streamTime=[100]",
"Skipping record for expired window. topic=[topic] partition=[0] offset=[6] timestamp=[5] window=[0,10) expiration=[10] streamTime=[100]",
"Skipping record for expired window. topic=[topic] partition=[0] offset=[7] timestamp=[6] window=[0,10) expiration=[10] streamTime=[100]"
- ));
+ )));
final TestOutputTopic<String, String> outputTopic =
driver.createOutputTopic("output", new StringDeserializer(), new StringDeserializer());
if (emitFinal) {
// Window close time is 15 when timestamp is 105
- assertThat(outputTopic.readRecord(),
- equalTo(new TestRecord<>("[k@5/15]", "+5+6", null, 6L)));
- assertEmittedMetrics(driver, is(1.0));
+ assertEquals(new TestRecord<>("[k@5/15]", "+5+6", null, 6L), outputTopic.readRecord());
+ assertEmittedMetrics(driver, 1.0);
} else {
- assertThat(outputTopic.readRecord(),
- equalTo(new TestRecord<>("[k@95/105]", "+100", null, 100L)));
- assertThat(outputTopic.readRecord(),
- equalTo(new TestRecord<>("[k@100/110]", "+100", null, 100L)));
- assertThat(outputTopic.readRecord(),
- equalTo(new TestRecord<>("[k@5/15]", "+5", null, 5L)));
- assertThat(outputTopic.readRecord(),
- equalTo(new TestRecord<>("[k@5/15]", "+5+6", null, 6L)));
- assertThat(outputTopic.readRecord(),
- equalTo(new TestRecord<>("[k@100/110]", "+100+105", null, 105L)));
- assertThat(outputTopic.readRecord(),
- equalTo(new TestRecord<>("[k@105/115]", "+105", null, 105L)));
- assertThat(outputTopic.readRecord(),
- equalTo(new TestRecord<>("[k@100/110]", "+100+105+106", null, 106L)));
- assertThat(outputTopic.readRecord(),
- equalTo(new TestRecord<>("[k@105/115]", "+105+106", null, 106L)));
+ assertEquals(new TestRecord<>("[k@95/105]", "+100", null, 100L), outputTopic.readRecord());
+ assertEquals(new TestRecord<>("[k@100/110]", "+100", null, 100L), outputTopic.readRecord());
+ assertEquals(new TestRecord<>("[k@5/15]", "+5", null, 5L), outputTopic.readRecord());
+ assertEquals(new TestRecord<>("[k@5/15]", "+5+6", null, 6L), outputTopic.readRecord());
+ assertEquals(new TestRecord<>("[k@100/110]", "+100+105", null, 105L), outputTopic.readRecord());
+ assertEquals(new TestRecord<>("[k@105/115]", "+105", null, 105L), outputTopic.readRecord());
+ assertEquals(new TestRecord<>("[k@100/110]", "+100+105+106", null, 106L), outputTopic.readRecord());
+ assertEquals(new TestRecord<>("[k@105/115]", "+105+106", null, 106L), outputTopic.readRecord());
}
assertTrue(outputTopic.isEmpty());
}
@@ -623,9 +608,9 @@ public class KStreamWindowAggregateTest {
inputTopic.pipeInput("k", "5", 105L);
inputTopic.pipeInput("k", "6", 6L);
- assertLatenessMetrics(driver, is(7.0), is(194.0), is(97.375));
+ assertLatenessMetrics(driver, 7.0, 194.0, 97.375);
- assertThat(appender.getMessages(), hasItems(
+ assertTrue(appender.getMessages().containsAll(List.of(
"Skipping record for expired window. topic=[topic] partition=[0] offset=[1] timestamp=[100] window=[100,110) expiration=[110] streamTime=[200]",
"Skipping record for expired window. topic=[topic] partition=[0] offset=[2] timestamp=[101] window=[100,110) expiration=[110] streamTime=[200]",
"Skipping record for expired window. topic=[topic] partition=[0] offset=[3] timestamp=[102] window=[100,110) expiration=[110] streamTime=[200]",
@@ -633,14 +618,13 @@ public class KStreamWindowAggregateTest {
"Skipping record for expired window. topic=[topic] partition=[0] offset=[5] timestamp=[104] window=[100,110) expiration=[110] streamTime=[200]",
"Skipping record for expired window. topic=[topic] partition=[0] offset=[6] timestamp=[105] window=[100,110) expiration=[110] streamTime=[200]",
"Skipping record for expired window. topic=[topic] partition=[0] offset=[7] timestamp=[6] window=[0,10) expiration=[110] streamTime=[200]"
- ));
+ )));
if (!emitFinal) {
final TestOutputTopic<String, String> outputTopic =
driver.createOutputTopic("output", new StringDeserializer(),
new StringDeserializer());
- assertThat(outputTopic.readRecord(),
- equalTo(new TestRecord<>("[k@200/210]", "+100", null, 200L)));
+ assertEquals(new TestRecord<>("[k@200/210]", "+100", null, 200L), outputTopic.readRecord());
assertTrue(outputTopic.isEmpty());
}
}
@@ -696,7 +680,7 @@ public class KStreamWindowAggregateTest {
new Change<>("0+3", null), 15))
);
}
- assertThat(forwarded, is(expected));
+ assertEquals(expected, forwarded);
context.resetForwards();
processor.process(new Record<>("D", "4", 15));
@@ -711,7 +695,7 @@ public class KStreamWindowAggregateTest {
new CapturedForward<>(new Record<>(new Windowed<>("D", new TimeWindow(15, 25)),
new Change<>("0+4", null), 15))
);
- assertThat(forwarded, is(expected));
+ assertEquals(expected, forwarded);
}
context.resetForwards();
@@ -727,7 +711,7 @@ public class KStreamWindowAggregateTest {
new CapturedForward<>(new Record<>(new Windowed<>("E", new TimeWindow(15, 25)),
new Change<>("0+5", null), 19))
);
- assertThat(forwarded, is(expected));
+ assertEquals(expected, forwarded);
}
context.getStateStore(WINDOW_STORE_NAME).close();
@@ -798,7 +782,7 @@ public class KStreamWindowAggregateTest {
new Change<>("0+4", null), 20))
);
}
- assertThat(forwarded, is(expected));
+ assertEquals(expected, forwarded);
context.getStateStore(WINDOW_STORE_NAME).close();
} finally {
Utils.delete(stateDir);
@@ -853,7 +837,7 @@ public class KStreamWindowAggregateTest {
new CapturedForward<>(new Record<>(new Windowed<>("C", new TimeWindow(15, 25)),
new Change<>("0+3", null), 15))
);
- assertThat(forwarded, is(expected));
+ assertEquals(expected, forwarded);
}
context.resetForwards();
@@ -880,7 +864,7 @@ public class KStreamWindowAggregateTest {
new Change<>("0+4", null), 20))
);
}
- assertThat(forwarded, is(expected));
+ assertEquals(expected, forwarded);
context.resetForwards();
// Progress
@@ -897,7 +881,7 @@ public class KStreamWindowAggregateTest {
new CapturedForward<>(new Record<>(new Windowed<>("E", new TimeWindow(40, 50)),
new Change<>("0+5", null), 40))
);
- assertThat(forwarded, is(expected));
+ assertEquals(expected, forwarded);
}
context.getStateStore(WINDOW_STORE_NAME).close();
@@ -956,7 +940,7 @@ public class KStreamWindowAggregateTest {
new Change<>("0+3", null), 15))
);
}
- assertThat(forwarded, is(expected));
+ assertEquals(expected, forwarded);
context.resetForwards();
final Processor<String, String, Windowed<String>, Change<String>> newProcessor = processorSupplier.get();
@@ -979,7 +963,7 @@ public class KStreamWindowAggregateTest {
new Change<>("0+4", null), 25))
);
}
- assertThat(forwarded, is(expected));
+ assertEquals(expected, forwarded);
context.resetForwards();
context.getStateStore(WINDOW_STORE_NAME).close();
@@ -1001,8 +985,10 @@ public class KStreamWindowAggregateTest {
MockInitializer.STRING_INIT,
MockAggregator.TOSTRING_ADDER)
);
- assertThat(e.getMessage(), is("ON_WINDOW_CLOSE strategy is only supported for "
- + "TimeWindows and SlidingWindows for TimeWindowedKStream"));
+ assertEquals(
+ "ON_WINDOW_CLOSE strategy is only supported for TimeWindows and SlidingWindows for TimeWindowedKStream",
+ e.getMessage()
+ );
} else {
new KStreamWindowAggregate<>(
UnlimitedWindows.of(),
@@ -1057,9 +1043,9 @@ public class KStreamWindowAggregateTest {
}
private void assertLatenessMetrics(final TopologyTestDriver driver,
- final Matcher<Object> dropTotal,
- final Matcher<Object> maxLateness,
- final Matcher<Object> avgLateness) {
+ final double expectedDropTotal,
+ final double expectedMaxLateness,
+ final double expectedAvgLateness) {
final MetricName dropTotalMetric;
final MetricName dropRateMetric;
@@ -1104,14 +1090,14 @@ public class KStreamWindowAggregateTest {
)
);
- assertThat(driver.metrics().get(dropTotalMetric).metricValue(), dropTotal);
- assertThat(driver.metrics().get(dropRateMetric).metricValue(), not(0.0));
- assertThat(driver.metrics().get(latenessMaxMetric).metricValue(), maxLateness);
- assertThat(driver.metrics().get(latenessAvgMetric).metricValue(), avgLateness);
+ assertEquals(expectedDropTotal, driver.metrics().get(dropTotalMetric).metricValue());
+ assertNotEquals(0.0, driver.metrics().get(dropRateMetric).metricValue());
+ assertEquals(expectedMaxLateness, driver.metrics().get(latenessMaxMetric).metricValue());
+ assertEquals(expectedAvgLateness, driver.metrics().get(latenessAvgMetric).metricValue());
}
private void assertEmittedMetrics(final TopologyTestDriver driver,
- final Matcher<Object> emittedTotal) {
+ final double expectedEmittedTotal) {
final MetricName emittedTotalMetric;
final MetricName emittedRateMetric;
@@ -1136,8 +1122,8 @@ public class KStreamWindowAggregateTest {
)
);
- assertThat(driver.metrics().get(emittedTotalMetric).metricValue(), emittedTotal);
- assertThat(driver.metrics().get(emittedRateMetric).metricValue(), not(0.0));
+ assertEquals(expectedEmittedTotal, driver.metrics().get(emittedTotalMetric).metricValue());
+ assertNotEquals(0.0, driver.metrics().get(emittedRateMetric).metricValue());
}
private <K, V, S extends StateStore> Materialized<K, V, S> setMaterializedCache(final Materialized<K, V, S> materialized) {
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableFilterTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableFilterTest.java
index 0532782201e..16e4ab4c405 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableFilterTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableFilterTest.java
@@ -52,10 +52,10 @@ import java.time.Instant;
import java.util.List;
import java.util.Properties;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.is;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
@SuppressWarnings("unchecked")
public class KTableFilterTest {
@@ -412,8 +412,8 @@ public class KTableFilterTest {
table2.enableSendingOldValues(true);
- assertThat(table1.sendingOldValueEnabled(), is(true));
- assertThat(table2.sendingOldValueEnabled(), is(true));
+ assertTrue(table1.sendingOldValueEnabled());
+ assertTrue(table2.sendingOldValueEnabled());
doTestSendingOldValue(builder, table1, table2, topic1);
}
@@ -432,8 +432,8 @@ public class KTableFilterTest {
table2.enableSendingOldValues(true);
- assertThat(table1.sendingOldValueEnabled(), is(false));
- assertThat(table2.sendingOldValueEnabled(), is(true));
+ assertFalse(table1.sendingOldValueEnabled());
+ assertTrue(table2.sendingOldValueEnabled());
doTestSendingOldValue(builder, table1, table2, topic1);
}
@@ -452,8 +452,8 @@ public class KTableFilterTest {
table2.enableSendingOldValues(false);
- assertThat(table1.sendingOldValueEnabled(), is(true));
- assertThat(table2.sendingOldValueEnabled(), is(true));
+ assertTrue(table1.sendingOldValueEnabled());
+ assertTrue(table2.sendingOldValueEnabled());
doTestSendingOldValue(builder, table1, table2, topic1);
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableImplTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableImplTest.java
index 2b481c9b79b..c7cf5a6e2e9 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableImplTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableImplTest.java
@@ -68,13 +68,13 @@ import java.util.List;
import java.util.Properties;
import static java.util.Arrays.asList;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.is;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.mock;
@ExtendWith(MockitoExtension.class)
@@ -412,7 +412,7 @@ public class KTableImplTest {
final var table = assertInstanceOf(KTableImpl.class, builder.table("topic1", consumed));
table.enableSendingOldValues(false);
- assertThat(table.sendingOldValueEnabled(), is(false));
+ assertFalse(table.sendingOldValueEnabled());
}
@ParameterizedTest
@@ -424,7 +424,7 @@ public class KTableImplTest {
final var table = assertInstanceOf(KTableImpl.class, builder.table("topic1", consumed));
table.enableSendingOldValues(true);
- assertThat(table.sendingOldValueEnabled(), is(true));
+ assertTrue(table.sendingOldValueEnabled());
}
private void assertTopologyContainsProcessor(final Topology topology, final String processorName) {
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableForeignKeyJoinScenarioTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableForeignKeyJoinScenarioTest.java
index 3f1ef503115..a832bd2b911 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableForeignKeyJoinScenarioTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableForeignKeyJoinScenarioTest.java
@@ -58,8 +58,6 @@ import java.util.concurrent.atomic.AtomicBoolean;
import static org.apache.kafka.common.utils.Utils.mkEntry;
import static org.apache.kafka.common.utils.Utils.mkMap;
import static org.apache.kafka.common.utils.Utils.mkProperties;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.is;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -236,7 +234,7 @@ public class KTableKTableForeignKeyJoinScenarioTest {
}
// verifying primarily that no extra pseudo-topics were used, but it's nice to also verify the rest of the
// topics our serdes serialize data for
- assertThat(serdeScope.registeredTopics(), is(Set.of(
+ assertEquals(Set.of(
// expected pseudo-topics
applicationId + "-KTABLE-FK-JOIN-SUBSCRIPTION-REGISTRATION-0000000006-topic-fk--key",
applicationId + "-KTABLE-FK-JOIN-SUBSCRIPTION-REGISTRATION-0000000006-topic-pk--key",
@@ -252,7 +250,7 @@ public class KTableKTableForeignKeyJoinScenarioTest {
// output topics
"output-topic--key",
"output-topic--value"
- )));
+ ), serdeScope.registeredTopics());
}
@ParameterizedTest
@@ -313,7 +311,7 @@ public class KTableKTableForeignKeyJoinScenarioTest {
expectedOutput.put("producer1:product2", "(producer1|product2-data,producer1-data)");
expectedOutput.put("producer2:product1", "(producer2|product1-data,producer2-data)");
- assertThat(output.readKeyValuesToMap(), is(expectedOutput));
+ assertEquals(expectedOutput, output.readKeyValuesToMap());
}
}
@@ -374,7 +372,7 @@ public class KTableKTableForeignKeyJoinScenarioTest {
expectedOutput.put("producer1:product2", "(product2-data,producer1-data)");
expectedOutput.put("producer2:product1", "(product1-data,producer2-data)");
- assertThat(output.readKeyValuesToMap(), is(expectedOutput));
+ assertEquals(expectedOutput, output.readKeyValuesToMap());
}
}
@@ -402,7 +400,7 @@ public class KTableKTableForeignKeyJoinScenarioTest {
aTopic.pipeInput(1, "999-alpha");
bTopic.pipeInput(999, "beta");
final Map<Integer, String> x = output.readKeyValuesToMap();
- assertThat(x, is(Collections.singletonMap(1, "(999-alpha,(999-alpha,beta))")));
+ assertEquals(Map.of(1, "(999-alpha,(999-alpha,beta))"), x);
}
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableInnerJoinTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableInnerJoinTest.java
index bce6d5bd181..fac2e2ef070 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableInnerJoinTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableInnerJoinTest.java
@@ -49,9 +49,6 @@ import java.util.Collection;
import java.util.Properties;
import java.util.Set;
-import static org.hamcrest.CoreMatchers.equalTo;
-import static org.hamcrest.CoreMatchers.hasItem;
-import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -275,10 +272,9 @@ public class KTableKTableInnerJoinTest {
try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister(KTableKTableInnerJoin.class)) {
join.process(new Record<>(null, new Change<>("new", "old"), 0));
- assertThat(
- appender.getMessages(),
- hasItem("Skipping record due to null key. topic=[left] partition=[-1] offset=[-2]")
- );
+ assertTrue(appender.getMessages().contains(
+ "Skipping record due to null key. topic=[left] partition=[-1] offset=[-2]"
+ ));
}
}
@@ -489,7 +485,7 @@ public class KTableKTableInnerJoinTest {
final Integer expectedKey,
final String expectedValue,
final long expectedTimestamp) {
- assertThat(outputTopic.readRecord(), equalTo(new TestRecord<>(expectedKey, expectedValue, null, expectedTimestamp)));
+ assertEquals(new TestRecord<>(expectedKey, expectedValue, null, expectedTimestamp), outputTopic.readRecord());
}
/**
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableLeftJoinTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableLeftJoinTest.java
index 41b922ee7c0..09ac3c2d9f8 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableLeftJoinTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableLeftJoinTest.java
@@ -58,10 +58,6 @@ import java.util.Properties;
import java.util.Random;
import java.util.Set;
-import static org.hamcrest.CoreMatchers.equalTo;
-import static org.hamcrest.CoreMatchers.hasItem;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.is;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -336,9 +332,9 @@ public class KTableKTableLeftJoinTest {
((KTableImpl<?, ?, ?>) joined).enableSendingOldValues(true);
- assertThat(((KTableImpl<?, ?, ?>) table1).sendingOldValueEnabled(), is(true));
- assertThat(((KTableImpl<?, ?, ?>) table2).sendingOldValueEnabled(), is(true));
- assertThat(((KTableImpl<?, ?, ?>) joined).sendingOldValueEnabled(), is(true));
+ assertTrue(((KTableImpl<?, ?, ?>) table1).sendingOldValueEnabled());
+ assertTrue(((KTableImpl<?, ?, ?>) table2).sendingOldValueEnabled());
+ assertTrue(((KTableImpl<?, ?, ?>) joined).sendingOldValueEnabled());
final Topology topology = builder.build().addProcessor("proc", supplier, ((KTableImpl<?, ?, ?>) joined).name);
@@ -527,10 +523,9 @@ public class KTableKTableLeftJoinTest {
try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister(KTableKTableLeftJoin.class)) {
join.process(new Record<>(null, new Change<>("new", "old"), 0));
- assertThat(
- appender.getMessages(),
- hasItem("Skipping record due to null key. topic=[left] partition=[-1] offset=[-2]")
- );
+ assertTrue(appender.getMessages().contains(
+ "Skipping record due to null key. topic=[left] partition=[-1] offset=[-2]"
+ ));
}
}
@@ -538,6 +533,6 @@ public class KTableKTableLeftJoinTest {
final Integer expectedKey,
final String expectedValue,
final long expectedTimestamp) {
- assertThat(outputTopic.readRecord(), equalTo(new TestRecord<>(expectedKey, expectedValue, null, expectedTimestamp)));
+ assertEquals(new TestRecord<>(expectedKey, expectedValue, null, expectedTimestamp), outputTopic.readRecord());
}
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableOuterJoinTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableOuterJoinTest.java
index 2df75c2b22b..b7b1ff61595 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableOuterJoinTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableOuterJoinTest.java
@@ -47,9 +47,6 @@ import java.util.Collection;
import java.util.Properties;
import java.util.Set;
-import static org.hamcrest.CoreMatchers.equalTo;
-import static org.hamcrest.CoreMatchers.hasItem;
-import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -432,10 +429,9 @@ public class KTableKTableOuterJoinTest {
try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister(KTableKTableOuterJoin.class)) {
join.process(new Record<>(null, new Change<>("new", "old"), 0));
- assertThat(
- appender.getMessages(),
- hasItem("Skipping record due to null key. topic=[left] partition=[-1] offset=[-2]")
- );
+ assertTrue(appender.getMessages().contains(
+ "Skipping record due to null key. topic=[left] partition=[-1] offset=[-2]"
+ ));
}
}
@@ -443,7 +439,7 @@ public class KTableKTableOuterJoinTest {
final Integer expectedKey,
final String expectedValue,
final long expectedTimestamp) {
- assertThat(outputTopic.readRecord(), equalTo(new TestRecord<>(expectedKey, expectedValue, null, expectedTimestamp)));
+ assertEquals(new TestRecord<>(expectedKey, expectedValue, null, expectedTimestamp), outputTopic.readRecord());
}
/**
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableRightJoinTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableRightJoinTest.java
index 416968bf8a4..3a1bf1a303e 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableRightJoinTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableRightJoinTest.java
@@ -18,7 +18,6 @@ package org.apache.kafka.streams.kstream.internals;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.LogCaptureAppender;
-import org.apache.kafka.common.utils.LogCaptureAppender.Event;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.Consumed;
@@ -31,10 +30,8 @@ import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
import java.util.Properties;
-import java.util.stream.Collectors;
-import static org.hamcrest.CoreMatchers.hasItem;
-import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertTrue;
public class KTableKTableRightJoinTest {
@@ -61,13 +58,8 @@ public class KTableKTableRightJoinTest {
try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister(KTableKTableRightJoin.class)) {
join.process(new Record<>(null, new Change<>("new", "old"), 0));
- assertThat(
- appender.getEvents().stream()
- .filter(e -> e.getLevel().equals("WARN"))
- .map(Event::getMessage)
- .collect(Collectors.toList()),
- hasItem("Skipping record due to null key. topic=[left] partition=[-1] offset=[-2]")
- );
+ assertTrue(appender.getMessages("WARN").contains(
+ "Skipping record due to null key. topic=[left] partition=[-1] offset=[-2]"));
}
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableMapValuesTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableMapValuesTest.java
index 8c4aa1e4b6a..abdf135fa68 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableMapValuesTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableMapValuesTest.java
@@ -47,11 +47,10 @@ import java.time.Instant;
import java.util.Properties;
import static java.util.Arrays.asList;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.is;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
@SuppressWarnings("unchecked")
public class KTableMapValuesTest {
@@ -268,8 +267,8 @@ public class KTableMapValuesTest {
table2.enableSendingOldValues(true);
- assertThat(table1.sendingOldValueEnabled(), is(true));
- assertThat(table2.sendingOldValueEnabled(), is(true));
+ assertTrue(table1.sendingOldValueEnabled());
+ assertTrue(table2.sendingOldValueEnabled());
testSendingOldValues(builder, topic1, table2);
}
@@ -291,8 +290,8 @@ public class KTableMapValuesTest {
table2.enableSendingOldValues(true);
- assertThat(table1.sendingOldValueEnabled(), is(false));
- assertThat(table2.sendingOldValueEnabled(), is(true));
+ assertFalse(table1.sendingOldValueEnabled());
+ assertTrue(table2.sendingOldValueEnabled());
testSendingOldValues(builder, topic1, table2);
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableSourceTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableSourceTest.java
index c8342d8ef2e..4ef5ad45878 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableSourceTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableSourceTest.java
@@ -23,7 +23,6 @@ import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.common.utils.LogCaptureAppender;
-import org.apache.kafka.common.utils.LogCaptureAppender.Event;
import org.apache.kafka.streams.KeyValueTimestamp;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.TestInputTopic;
@@ -49,12 +48,9 @@ import org.junit.jupiter.api.Test;
import java.time.Duration;
import java.time.Instant;
import java.util.Properties;
-import java.util.stream.Collectors;
import static java.util.Arrays.asList;
import static org.apache.kafka.test.StreamsTestUtils.getMetricByName;
-import static org.hamcrest.CoreMatchers.hasItem;
-import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -152,13 +148,8 @@ public class KTableSourceTest {
);
inputTopic.pipeInput(null, "value");
- assertThat(
- appender.getEvents().stream()
- .filter(e -> e.getLevel().equals("WARN"))
- .map(Event::getMessage)
- .collect(Collectors.toList()),
- hasItem("Skipping record due to null key. topic=[topic] partition=[0] offset=[0]")
- );
+ assertTrue(appender.getMessages("WARN").contains(
+ "Skipping record due to null key. topic=[topic] partition=[0] offset=[0]"));
}
}
@@ -182,13 +173,8 @@ public class KTableSourceTest {
inputTopic.pipeInput("key", "value", 10L);
inputTopic.pipeInput("key", "value", 5L);
- assertThat(
- appender.getEvents().stream()
- .filter(e -> e.getLevel().equals("WARN"))
- .map(Event::getMessage)
- .collect(Collectors.toList()),
- hasItem("Detected out-of-order KTable update for store, old timestamp=[10] new timestamp=[5]. topic=[topic] partition=[0] offset=[1].")
- );
+ assertTrue(appender.getMessages("WARN").contains(
+ "Detected out-of-order KTable update for store, old timestamp=[10] new timestamp=[5]. topic=[topic] partition=[0] offset=[1]."));
}
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableTransformValuesTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableTransformValuesTest.java
index d89f4659de8..87171f041bd 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableTransformValuesTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableTransformValuesTest.java
@@ -69,12 +69,11 @@ import java.util.List;
import java.util.Properties;
import java.util.Set;
-import static org.hamcrest.CoreMatchers.equalTo;
-import static org.hamcrest.CoreMatchers.hasItems;
-import static org.hamcrest.CoreMatchers.is;
-import static org.hamcrest.CoreMatchers.isA;
-import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.Mockito.doNothing;
@@ -161,7 +160,7 @@ public class KTableTransformValuesTest {
processor.init(context);
- assertThat(transformer.context, isA((Class) ForwardingDisabledProcessorContext.class));
+ assertInstanceOf(ForwardingDisabledProcessorContext.class, transformer.context);
}
@Test
@@ -237,7 +236,7 @@ public class KTableTransformValuesTest {
final String result = getter.get("Key").value();
- assertThat(result, is("Key->Value!"));
+ assertEquals("Key->Value!", result);
}
@Test
@@ -273,8 +272,8 @@ public class KTableTransformValuesTest {
final ValueTimestampHeaders<String> result = getter.get("Key");
- assertThat(result.value(), is("Key->null!"));
- assertThat(result.headers(), is(contextHeaders));
+ assertEquals("Key->null!", result.value());
+ assertEquals(contextHeaders, result.headers());
}
@Test
@@ -290,7 +289,7 @@ public class KTableTransformValuesTest {
final String result = getter.get("Key").value();
- assertThat(result, is("something"));
+ assertEquals("something", result);
}
@Test
@@ -303,7 +302,7 @@ public class KTableTransformValuesTest {
final String[] storeNames = transformValues.view().storeNames();
- assertThat(storeNames, is(new String[]{"store1", "store2"}));
+ assertArrayEquals(new String[]{"store1", "store2"}, storeNames);
}
@Test
@@ -313,7 +312,7 @@ public class KTableTransformValuesTest {
final String[] storeNames = transformValues.view().storeNames();
- assertThat(storeNames, is(new String[]{QUERYABLE_NAME}));
+ assertArrayEquals(new String[]{QUERYABLE_NAME}, storeNames);
}
@Test
@@ -379,10 +378,11 @@ public class KTableTransformValuesTest {
inputTopic.pipeInput("D", null, 15L);
- assertThat(output(), hasItems(new KeyValueTimestamp<>("A", "A->a!", 5),
- new KeyValueTimestamp<>("B", "B->b!", 10),
- new KeyValueTimestamp<>("D", "D->null!", 15)
- ));
+ assertTrue(output().containsAll(List.of(
+ new KeyValueTimestamp<>("A", "A->a!", 5),
+ new KeyValueTimestamp<>("B", "B->b!", 10),
+ new KeyValueTimestamp<>("D", "D->null!", 15)
+ )));
assertNull(driver.getKeyValueStore(QUERYABLE_NAME), "Store should not be materialized");
}
@@ -408,21 +408,23 @@ public class KTableTransformValuesTest {
inputTopic.pipeInput("B", "b", 10L);
inputTopic.pipeInput("C", null, 15L);
- assertThat(output(), hasItems(new KeyValueTimestamp<>("A", "A->a!", 5),
- new KeyValueTimestamp<>("B", "B->b!", 10),
- new KeyValueTimestamp<>("C", "C->null!", 15)));
+ assertTrue(output().containsAll(List.of(
+ new KeyValueTimestamp<>("A", "A->a!", 5),
+ new KeyValueTimestamp<>("B", "B->b!", 10),
+ new KeyValueTimestamp<>("C", "C->null!", 15)
+ )));
{
final KeyValueStore<String, String> keyValueStore = driver.getKeyValueStore(QUERYABLE_NAME);
- assertThat(keyValueStore.get("A"), is("A->a!"));
- assertThat(keyValueStore.get("B"), is("B->b!"));
- assertThat(keyValueStore.get("C"), is("C->null!"));
+ assertEquals("A->a!", keyValueStore.get("A"));
+ assertEquals("B->b!", keyValueStore.get("B"));
+ assertEquals("C->null!", keyValueStore.get("C"));
}
{
final KeyValueStore<String, ValueAndTimestamp<String>> keyValueStore = driver.getTimestampedKeyValueStore(QUERYABLE_NAME);
- assertThat(keyValueStore.get("A"), is(ValueAndTimestamp.make("A->a!", 5L)));
- assertThat(keyValueStore.get("B"), is(ValueAndTimestamp.make("B->b!", 10L)));
- assertThat(keyValueStore.get("C"), is(ValueAndTimestamp.make("C->null!", 15L)));
+ assertEquals(ValueAndTimestamp.make("A->a!", 5L), keyValueStore.get("A"));
+ assertEquals(ValueAndTimestamp.make("B->b!", 10L), keyValueStore.get("B"));
+ assertEquals(ValueAndTimestamp.make("C->null!", 15L), keyValueStore.get("C"));
}
}
@@ -450,14 +452,18 @@ public class KTableTransformValuesTest {
inputTopic.pipeInput("A", "ignored1", 15L);
inputTopic.pipeInput("A", "ignored2", 10L);
- assertThat(output(), equalTo(Arrays.asList(new KeyValueTimestamp<>("A", "1", 5),
- new KeyValueTimestamp<>("A", "2", 15),
- new KeyValueTimestamp<>("A", "3", 15))));
+ assertEquals(List.of(
+ new KeyValueTimestamp<>("A", "1", 5),
+ new KeyValueTimestamp<>("A", "2", 15),
+ new KeyValueTimestamp<>("A", "3", 15)
+ ), output());
final KeyValueStore<String, Integer> keyValueStore = driver.getKeyValueStore(QUERYABLE_NAME);
- assertThat(keyValueStore.get("A"), is(3));
- assertThat(driver.getAllStateStores().keySet(),
- equalTo(Set.of(QUERYABLE_NAME, "KTABLE-AGGREGATE-STATE-STORE-0000000005")));
+ assertEquals(3, keyValueStore.get("A"));
+ assertEquals(
+ Set.of(QUERYABLE_NAME, "KTABLE-AGGREGATE-STATE-STORE-0000000005"),
+ driver.getAllStateStores().keySet()
+ );
}
@ParameterizedTest
@@ -480,11 +486,15 @@ public class KTableTransformValuesTest {
inputTopic.pipeInput("A", "aa", 15L);
inputTopic.pipeInput("A", "aaa", 10);
- assertThat(output(), equalTo(Arrays.asList(new KeyValueTimestamp<>("A", "1", 5),
- new KeyValueTimestamp<>("A", "2", 15),
- new KeyValueTimestamp<>("A", "3", 15))));
- assertThat(driver.getAllStateStores().keySet(),
- equalTo(Set.of("inputTopic-STATE-STORE-0000000000", "KTABLE-AGGREGATE-STATE-STORE-0000000005")));
+ assertEquals(List.of(
+ new KeyValueTimestamp<>("A", "1", 5),
+ new KeyValueTimestamp<>("A", "2", 15),
+ new KeyValueTimestamp<>("A", "3", 15)
+ ), output());
+ assertEquals(
+ Set.of("inputTopic-STATE-STORE-0000000000", "KTABLE-AGGREGATE-STATE-STORE-0000000005"),
+ driver.getAllStateStores().keySet()
+ );
}
@ParameterizedTest
@@ -509,11 +519,15 @@ public class KTableTransformValuesTest {
inputTopic.pipeInput("A", "aa", 15L);
inputTopic.pipeInput("A", "aaa", 10);
- assertThat(output(), equalTo(Arrays.asList(new KeyValueTimestamp<>("A", "1", 5),
+ assertEquals(List.of(
+ new KeyValueTimestamp<>("A", "1", 5),
new KeyValueTimestamp<>("A", "2", 15),
- new KeyValueTimestamp<>("A", "3", 15))));
- assertThat(driver.getAllStateStores().keySet(),
- equalTo(Set.of("inputTopic-STATE-STORE-0000000000", "KTABLE-AGGREGATE-STATE-STORE-0000000005")));
+ new KeyValueTimestamp<>("A", "3", 15)
+ ), output());
+ assertEquals(
+ Set.of("inputTopic-STATE-STORE-0000000000", "KTABLE-AGGREGATE-STATE-STORE-0000000005"),
+ driver.getAllStateStores().keySet()
+ );
}
private ArrayList<KeyValueTimestamp<String, String>> output() {
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/MaterializedInternalTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/MaterializedInternalTest.java
index a3eaadd27db..b0e0c0bf423 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/MaterializedInternalTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/MaterializedInternalTest.java
@@ -42,10 +42,10 @@ import org.mockito.quality.Strictness;
import java.util.Optional;
import java.util.Properties;
-import static org.hamcrest.CoreMatchers.equalTo;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.instanceOf;
-import static org.hamcrest.Matchers.is;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -65,7 +65,7 @@ public class MaterializedInternalTest {
final MaterializedInternal<Object, Object, StateStore> materialized =
new MaterializedInternal<>(Materialized.with(null, null), nameProvider, prefix);
- assertThat(materialized.storeName(), equalTo(generatedName));
+ assertEquals(generatedName, materialized.storeName());
}
@Test
@@ -73,7 +73,7 @@ public class MaterializedInternalTest {
final String storeName = "store-name";
final MaterializedInternal<Object, Object, StateStore> materialized =
new MaterializedInternal<>(Materialized.as(storeName), nameProvider, prefix);
- assertThat(materialized.storeName(), equalTo(storeName));
+ assertEquals(storeName, materialized.storeName());
}
@Test
@@ -82,7 +82,7 @@ public class MaterializedInternalTest {
when(supplier.name()).thenReturn(storeName);
final MaterializedInternal<Object, Object, KeyValueStore<Bytes, byte[]>> materialized =
new MaterializedInternal<>(Materialized.as(supplier), nameProvider, prefix);
- assertThat(materialized.storeName(), equalTo(storeName));
+ assertEquals(storeName, materialized.storeName());
}
@SuppressWarnings("deprecation")
@@ -99,7 +99,7 @@ public class MaterializedInternalTest {
final MaterializedInternal<Object, Object, KeyValueStore<Bytes, byte[]>> materialized =
new MaterializedInternal<>(Materialized.as(supplier), internalStreamsBuilder, prefix);
- assertThat(materialized.dslStoreSuppliers(), equalTo(Optional.of(Materialized.StoreType.IN_MEMORY)));
+ assertEquals(Optional.of(Materialized.StoreType.IN_MEMORY), materialized.dslStoreSuppliers());
}
@SuppressWarnings("deprecation")
@@ -117,8 +117,8 @@ public class MaterializedInternalTest {
final MaterializedInternal<Object, Object, KeyValueStore<Bytes, byte[]>> materialized =
new MaterializedInternal<>(Materialized.as(supplier), internalStreamsBuilder, prefix);
- assertThat(materialized.dslStoreSuppliers().isPresent(), is(true));
- assertThat(materialized.dslStoreSuppliers().get(), instanceOf(TestStoreSupplier.class));
+ assertTrue(materialized.dslStoreSuppliers().isPresent());
+ assertInstanceOf(TestStoreSupplier.class, materialized.dslStoreSuppliers().get());
}
@Test
@@ -133,7 +133,7 @@ public class MaterializedInternalTest {
final MaterializedInternal<Object, Object, KeyValueStore<Bytes, byte[]>> materialized =
new MaterializedInternal<>(Materialized.as(supplier), internalStreamsBuilder, prefix);
- assertThat(materialized.dslStoreSuppliers().isPresent(), is(false));
+ assertFalse(materialized.dslStoreSuppliers().isPresent());
}
public static class TestStoreSupplier implements DslStoreSuppliers {
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SessionWindowedCogroupedKStreamImplTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SessionWindowedCogroupedKStreamImplTest.java
index a61526e1f6c..c70a51c3675 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SessionWindowedCogroupedKStreamImplTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SessionWindowedCogroupedKStreamImplTest.java
@@ -53,8 +53,7 @@ import java.util.Properties;
import static java.time.Duration.ofDays;
import static java.time.Duration.ofMillis;
-import static org.hamcrest.CoreMatchers.equalTo;
-import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
public class SessionWindowedCogroupedKStreamImplTest {
@@ -158,7 +157,7 @@ public class SessionWindowedCogroupedKStreamImplTest {
.windowedBy(SessionWindows.ofInactivityGapWithNoGrace(ofMillis(1)))
.aggregate(MockInitializer.STRING_INIT, sessionMerger, Named.as("foo"));
- assertThat(builder.build().describe().toString(), equalTo(
+ assertEquals(
"Topologies:\n" +
" Sub-topology: 0\n" +
" Source: KSTREAM-SOURCE-0000000000 (topics: [topic])\n" +
@@ -168,7 +167,8 @@ public class SessionWindowedCogroupedKStreamImplTest {
" <-- KSTREAM-SOURCE-0000000000\n" +
" Processor: foo-cogroup-merge (stores: [])\n" +
" --> none\n" +
- " <-- foo-cogroup-agg-0\n\n"));
+ " <-- foo-cogroup-agg-0\n\n",
+ builder.build().describe().toString());
}
@ParameterizedTest
@@ -361,7 +361,7 @@ public class SessionWindowedCogroupedKStreamImplTest {
final TestRecord<String, String> nonWindowedRecord = new TestRecord<>(
realRecord.getKey().key(), realRecord.getValue(), null, realRecord.timestamp());
final TestRecord<String, String> testRecord = new TestRecord<>(expectedKey, expectedValue, null, expectedTimestamp);
- assertThat(nonWindowedRecord, equalTo(testRecord));
+ assertEquals(testRecord, nonWindowedRecord);
}
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SessionWindowedKStreamImplTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SessionWindowedKStreamImplTest.java
index 14267e7264e..f255007ee24 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SessionWindowedKStreamImplTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SessionWindowedKStreamImplTest.java
@@ -56,7 +56,6 @@ import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Properties;
@@ -64,10 +63,8 @@ import java.util.stream.Stream;
import static java.time.Duration.ofMillis;
import static java.util.Arrays.asList;
-import static org.hamcrest.CoreMatchers.equalTo;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.core.IsInstanceOf.instanceOf;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertThrows;
public class SessionWindowedKStreamImplTest {
@@ -245,18 +242,18 @@ public class SessionWindowedKStreamImplTest {
final SessionStore<String, Long> store = driver.getSessionStore("count-store");
final List<KeyValue<Windowed<String>, Long>> data = unwrapAggregations(store.fetch("1", "2"));
if (!emitFinal) {
- assertThat(
- data,
- equalTo(Arrays.asList(
+ assertEquals(
+ List.of(
KeyValue.pair(new Windowed<>("1", new SessionWindow(10, 15)), 2L),
KeyValue.pair(new Windowed<>("1", new SessionWindow(600, 600)), 1L),
- KeyValue.pair(new Windowed<>("2", new SessionWindow(599, 600)), 2L))));
+ KeyValue.pair(new Windowed<>("2", new SessionWindow(599, 600)), 2L)),
+ data);
} else {
- assertThat(
- data,
- equalTo(Arrays.asList(
+ assertEquals(
+ List.of(
KeyValue.pair(new Windowed<>("1", new SessionWindow(600, 600)), 1L),
- KeyValue.pair(new Windowed<>("2", new SessionWindow(599, 600)), 2L))));
+ KeyValue.pair(new Windowed<>("2", new SessionWindow(599, 600)), 2L)),
+ data);
}
}
@@ -274,18 +271,18 @@ public class SessionWindowedKStreamImplTest {
final List<KeyValue<Windowed<String>, String>> data = unwrapAggregations(sessionStore.fetch("1", "2"));
if (!emitFinal) {
- assertThat(
- data,
- equalTo(Arrays.asList(
+ assertEquals(
+ List.of(
KeyValue.pair(new Windowed<>("1", new SessionWindow(10, 15)), "1+2"),
KeyValue.pair(new Windowed<>("1", new SessionWindow(600, 600)), "3"),
- KeyValue.pair(new Windowed<>("2", new SessionWindow(599, 600)), "1+2"))));
+ KeyValue.pair(new Windowed<>("2", new SessionWindow(599, 600)), "1+2")),
+ data);
} else {
- assertThat(
- data,
- equalTo(Arrays.asList(
+ assertEquals(
+ List.of(
KeyValue.pair(new Windowed<>("1", new SessionWindow(600, 600)), "3"),
- KeyValue.pair(new Windowed<>("2", new SessionWindow(599, 600)), "1+2"))));
+ KeyValue.pair(new Windowed<>("2", new SessionWindow(599, 600)), "1+2")),
+ data);
}
}
@@ -306,18 +303,18 @@ public class SessionWindowedKStreamImplTest {
final SessionStore<String, String> sessionStore = driver.getSessionStore("aggregated");
final List<KeyValue<Windowed<String>, String>> data = unwrapAggregations(sessionStore.fetch("1", "2"));
if (!emitFinal) {
- assertThat(
- data,
- equalTo(Arrays.asList(
+ assertEquals(
+ List.of(
KeyValue.pair(new Windowed<>("1", new SessionWindow(10, 15)), "0+0+1+2"),
KeyValue.pair(new Windowed<>("1", new SessionWindow(600, 600)), "0+3"),
- KeyValue.pair(new Windowed<>("2", new SessionWindow(599, 600)), "0+0+1+2"))));
+ KeyValue.pair(new Windowed<>("2", new SessionWindow(599, 600)), "0+0+1+2")),
+ data);
} else {
- assertThat(
- data,
- equalTo(Arrays.asList(
+ assertEquals(
+ List.of(
KeyValue.pair(new Windowed<>("1", new SessionWindow(600, 600)), "0+3"),
- KeyValue.pair(new Windowed<>("2", new SessionWindow(599, 600)), "0+0+1+2"))));
+ KeyValue.pair(new Windowed<>("2", new SessionWindow(599, 600)), "0+0+1+2")),
+ data);
}
}
@@ -429,11 +426,11 @@ public class SessionWindowedKStreamImplTest {
try (final TopologyTestDriver driver = new TopologyTestDriverBuilder(builder.build()).withConfig(props).build()) {
final StateStore store = driver.getAllStateStores().get("aggregated");
final WrappedStateStore<?, ?, ?> changeLogging = (WrappedStateStore<?, ?, ?>) ((WrappedStateStore<?, ?, ?>) store).wrapped();
- assertThat(store, instanceOf(MeteredSessionStore.class));
+ assertInstanceOf(MeteredSessionStore.class, store);
if (withHeaders) {
- assertThat(changeLogging, instanceOf(ChangeLoggingSessionBytesStoreWithHeaders.class));
+ assertInstanceOf(ChangeLoggingSessionBytesStoreWithHeaders.class, changeLogging);
} else {
- assertThat(changeLogging, instanceOf(ChangeLoggingSessionBytesStore.class));
+ assertInstanceOf(ChangeLoggingSessionBytesStore.class, changeLogging);
}
}
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SlidingWindowedCogroupedKStreamImplTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SlidingWindowedCogroupedKStreamImplTest.java
index 431adca2b4e..41d58681803 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SlidingWindowedCogroupedKStreamImplTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SlidingWindowedCogroupedKStreamImplTest.java
@@ -53,8 +53,6 @@ import java.util.List;
import java.util.Properties;
import static java.time.Duration.ofMillis;
-import static org.hamcrest.CoreMatchers.equalTo;
-import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -159,7 +157,7 @@ public class SlidingWindowedCogroupedKStreamImplTest {
.windowedBy(SlidingWindows.ofTimeDifferenceAndGrace(ofMillis(WINDOW_SIZE_MS), ofMillis(2000L)))
.aggregate(MockInitializer.STRING_INIT, Named.as("foo"));
- assertThat(builder.build().describe().toString(), equalTo(
+ assertEquals(
"Topologies:\n" +
" Sub-topology: 0\n" +
" Source: KSTREAM-SOURCE-0000000000 (topics: [topic])\n" +
@@ -169,7 +167,8 @@ public class SlidingWindowedCogroupedKStreamImplTest {
" <-- KSTREAM-SOURCE-0000000000\n" +
" Processor: foo-cogroup-merge (stores: [])\n" +
" --> none\n" +
- " <-- foo-cogroup-agg-0\n\n"));
+ " <-- foo-cogroup-agg-0\n\n",
+ builder.build().describe().toString());
}
@ParameterizedTest
@@ -289,6 +288,6 @@ public class SlidingWindowedCogroupedKStreamImplTest {
final TestRecord<String, String> nonWindowedRecord = new TestRecord<>(
realRecord.getKey().key(), realRecord.getValue(), null, realRecord.timestamp());
final TestRecord<String, String> testRecord = new TestRecord<>(expectedKey, expectedValue, null, expectedTimestamp);
- assertThat(nonWindowedRecord, equalTo(testRecord));
+ assertEquals(testRecord, nonWindowedRecord);
}
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SlidingWindowedKStreamImplTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SlidingWindowedKStreamImplTest.java
index 340eed676e5..0b5cdbae408 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SlidingWindowedKStreamImplTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SlidingWindowedKStreamImplTest.java
@@ -47,14 +47,12 @@ import org.apache.kafka.test.StreamsTestUtils;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
-import java.util.Arrays;
import java.util.List;
import java.util.Properties;
import static java.time.Duration.ofMillis;
import static java.time.Instant.ofEpochMilli;
-import static org.hamcrest.CoreMatchers.equalTo;
-import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
public class SlidingWindowedKStreamImplTest {
@@ -89,34 +87,34 @@ public class SlidingWindowedKStreamImplTest {
try (final TopologyTestDriver driver = new TopologyTestDriverBuilder(builder.build()).withConfig(props).build()) {
processData(driver);
}
- assertThat(
+ assertEquals(
+ ValueAndTimestamp.make(1L, 100L),
supplier.theCapturedProcessor().lastValueAndTimestampPerKey()
- .get(new Windowed<>("1", new TimeWindow(0L, 100L))),
- equalTo(ValueAndTimestamp.make(1L, 100L)));
- assertThat(
+ .get(new Windowed<>("1", new TimeWindow(0L, 100L))));
+ assertEquals(
+ ValueAndTimestamp.make(1L, 150L),
supplier.theCapturedProcessor().lastValueAndTimestampPerKey()
- .get(new Windowed<>("1", new TimeWindow(101L, 201L))),
- equalTo(ValueAndTimestamp.make(1L, 150L)));
- assertThat(
+ .get(new Windowed<>("1", new TimeWindow(101L, 201L))));
+ assertEquals(
+ ValueAndTimestamp.make(2L, 150L),
supplier.theCapturedProcessor().lastValueAndTimestampPerKey()
- .get(new Windowed<>("1", new TimeWindow(50L, 150L))),
- equalTo(ValueAndTimestamp.make(2L, 150L)));
- assertThat(
+ .get(new Windowed<>("1", new TimeWindow(50L, 150L))));
+ assertEquals(
+ ValueAndTimestamp.make(1L, 500L),
supplier.theCapturedProcessor().lastValueAndTimestampPerKey()
- .get(new Windowed<>("1", new TimeWindow(400L, 500L))),
- equalTo(ValueAndTimestamp.make(1L, 500L)));
- assertThat(
+ .get(new Windowed<>("1", new TimeWindow(400L, 500L))));
+ assertEquals(
+ ValueAndTimestamp.make(2L, 200L),
supplier.theCapturedProcessor().lastValueAndTimestampPerKey()
- .get(new Windowed<>("2", new TimeWindow(100L, 200L))),
- equalTo(ValueAndTimestamp.make(2L, 200L)));
- assertThat(
+ .get(new Windowed<>("2", new TimeWindow(100L, 200L))));
+ assertEquals(
+ ValueAndTimestamp.make(1L, 150L),
supplier.theCapturedProcessor().lastValueAndTimestampPerKey()
- .get(new Windowed<>("2", new TimeWindow(50L, 150L))),
- equalTo(ValueAndTimestamp.make(1L, 150L)));
- assertThat(
+ .get(new Windowed<>("2", new TimeWindow(50L, 150L))));
+ assertEquals(
+ ValueAndTimestamp.make(1L, 200L),
supplier.theCapturedProcessor().lastValueAndTimestampPerKey()
- .get(new Windowed<>("2", new TimeWindow(151L, 251L))),
- equalTo(ValueAndTimestamp.make(1L, 200L)));
+ .get(new Windowed<>("2", new TimeWindow(151L, 251L))));
}
@ParameterizedTest
@@ -132,34 +130,34 @@ public class SlidingWindowedKStreamImplTest {
try (final TopologyTestDriver driver = new TopologyTestDriverBuilder(builder.build()).withConfig(props).build()) {
processData(driver);
}
- assertThat(
+ assertEquals(
+ ValueAndTimestamp.make("1", 100L),
supplier.theCapturedProcessor().lastValueAndTimestampPerKey()
- .get(new Windowed<>("1", new TimeWindow(0L, 100L))),
- equalTo(ValueAndTimestamp.make("1", 100L)));
- assertThat(
+ .get(new Windowed<>("1", new TimeWindow(0L, 100L))));
+ assertEquals(
+ ValueAndTimestamp.make("2", 150L),
supplier.theCapturedProcessor().lastValueAndTimestampPerKey()
- .get(new Windowed<>("1", new TimeWindow(101L, 201L))),
- equalTo(ValueAndTimestamp.make("2", 150L)));
- assertThat(
+ .get(new Windowed<>("1", new TimeWindow(101L, 201L))));
+ assertEquals(
+ ValueAndTimestamp.make("1+2", 150L),
supplier.theCapturedProcessor().lastValueAndTimestampPerKey()
- .get(new Windowed<>("1", new TimeWindow(50L, 150L))),
- equalTo(ValueAndTimestamp.make("1+2", 150L)));
- assertThat(
+ .get(new Windowed<>("1", new TimeWindow(50L, 150L))));
+ assertEquals(
+ ValueAndTimestamp.make("3", 500L),
supplier.theCapturedProcessor().lastValueAndTimestampPerKey()
- .get(new Windowed<>("1", new TimeWindow(400L, 500L))),
- equalTo(ValueAndTimestamp.make("3", 500L)));
- assertThat(
+ .get(new Windowed<>("1", new TimeWindow(400L, 500L))));
+ assertEquals(
+ ValueAndTimestamp.make("10+20", 200L),
supplier.theCapturedProcessor().lastValueAndTimestampPerKey()
- .get(new Windowed<>("2", new TimeWindow(100L, 200L))),
- equalTo(ValueAndTimestamp.make("10+20", 200L)));
- assertThat(
+ .get(new Windowed<>("2", new TimeWindow(100L, 200L))));
+ assertEquals(
+ ValueAndTimestamp.make("20", 150L),
supplier.theCapturedProcessor().lastValueAndTimestampPerKey()
- .get(new Windowed<>("2", new TimeWindow(50L, 150L))),
- equalTo(ValueAndTimestamp.make("20", 150L)));
- assertThat(
+ .get(new Windowed<>("2", new TimeWindow(50L, 150L))));
+ assertEquals(
+ ValueAndTimestamp.make("10", 200L),
supplier.theCapturedProcessor().lastValueAndTimestampPerKey()
- .get(new Windowed<>("2", new TimeWindow(151L, 251L))),
- equalTo(ValueAndTimestamp.make("10", 200L)));
+ .get(new Windowed<>("2", new TimeWindow(151L, 251L))));
}
@ParameterizedTest
@@ -178,34 +176,34 @@ public class SlidingWindowedKStreamImplTest {
try (final TopologyTestDriver driver = new TopologyTestDriverBuilder(builder.build()).withConfig(props).build()) {
processData(driver);
}
- assertThat(
+ assertEquals(
+ ValueAndTimestamp.make("0+1", 100L),
supplier.theCapturedProcessor().lastValueAndTimestampPerKey()
- .get(new Windowed<>("1", new TimeWindow(0L, 100L))),
- equalTo(ValueAndTimestamp.make("0+1", 100L)));
- assertThat(
+ .get(new Windowed<>("1", new TimeWindow(0L, 100L))));
+ assertEquals(
+ ValueAndTimestamp.make("0+2", 150L),
supplier.theCapturedProcessor().lastValueAndTimestampPerKey()
- .get(new Windowed<>("1", new TimeWindow(101L, 201L))),
- equalTo(ValueAndTimestamp.make("0+2", 150L)));
- assertThat(
+ .get(new Windowed<>("1", new TimeWindow(101L, 201L))));
+ assertEquals(
+ ValueAndTimestamp.make("0+1+2", 150L),
supplier.theCapturedProcessor().lastValueAndTimestampPerKey()
- .get(new Windowed<>("1", new TimeWindow(50L, 150L))),
- equalTo(ValueAndTimestamp.make("0+1+2", 150L)));
- assertThat(
+ .get(new Windowed<>("1", new TimeWindow(50L, 150L))));
+ assertEquals(
+ ValueAndTimestamp.make("0+3", 500L),
supplier.theCapturedProcessor().lastValueAndTimestampPerKey()
- .get(new Windowed<>("1", new TimeWindow(400L, 500L))),
- equalTo(ValueAndTimestamp.make("0+3", 500L)));
- assertThat(
+ .get(new Windowed<>("1", new TimeWindow(400L, 500L))));
+ assertEquals(
+ ValueAndTimestamp.make("0+10+20", 200L),
supplier.theCapturedProcessor().lastValueAndTimestampPerKey()
- .get(new Windowed<>("2", new TimeWindow(100L, 200L))),
- equalTo(ValueAndTimestamp.make("0+10+20", 200L)));
- assertThat(
+ .get(new Windowed<>("2", new TimeWindow(100L, 200L))));
+ assertEquals(
+ ValueAndTimestamp.make("0+20", 150L),
supplier.theCapturedProcessor().lastValueAndTimestampPerKey()
- .get(new Windowed<>("2", new TimeWindow(50L, 150L))),
- equalTo(ValueAndTimestamp.make("0+20", 150L)));
- assertThat(
+ .get(new Windowed<>("2", new TimeWindow(50L, 150L))));
+ assertEquals(
+ ValueAndTimestamp.make("0+10", 200L),
supplier.theCapturedProcessor().lastValueAndTimestampPerKey()
- .get(new Windowed<>("2", new TimeWindow(151L, 251L))),
- equalTo(ValueAndTimestamp.make("0+10", 200L)));
+ .get(new Windowed<>("2", new TimeWindow(151L, 251L))));
}
@ParameterizedTest
@@ -224,28 +222,33 @@ public class SlidingWindowedKStreamImplTest {
final List<KeyValue<Windowed<String>, Long>> data =
StreamsTestUtils.toListAndCloseIterator(windowStore.fetch("1", "2", ofEpochMilli(0), ofEpochMilli(1000L)));
- assertThat(data, equalTo(Arrays.asList(
- KeyValue.pair(new Windowed<>("1", new TimeWindow(0, 100)), 1L),
- KeyValue.pair(new Windowed<>("1", new TimeWindow(50, 150)), 2L),
- KeyValue.pair(new Windowed<>("1", new TimeWindow(101, 201)), 1L),
- KeyValue.pair(new Windowed<>("1", new TimeWindow(400, 500)), 1L),
- KeyValue.pair(new Windowed<>("2", new TimeWindow(50, 150)), 1L),
- KeyValue.pair(new Windowed<>("2", new TimeWindow(100, 200)), 2L),
- KeyValue.pair(new Windowed<>("2", new TimeWindow(151, 251)), 1L))));
+ assertEquals(
+ List.of(
+ KeyValue.pair(new Windowed<>("1", new TimeWindow(0, 100)), 1L),
+ KeyValue.pair(new Windowed<>("1", new TimeWindow(50, 150)), 2L),
+ KeyValue.pair(new Windowed<>("1", new TimeWindow(101, 201)), 1L),
+ KeyValue.pair(new Windowed<>("1", new TimeWindow(400, 500)), 1L),
+ KeyValue.pair(new Windowed<>("2", new TimeWindow(50, 150)), 1L),
+ KeyValue.pair(new Windowed<>("2", new TimeWindow(100, 200)), 2L),
+ KeyValue.pair(new Windowed<>("2", new TimeWindow(151, 251)), 1L)),
+ data);
}
{
final WindowStore<String, ValueAndTimestamp<Long>> windowStore =
driver.getTimestampedWindowStore("count-store");
final List<KeyValue<Windowed<String>, ValueAndTimestamp<Long>>> data =
StreamsTestUtils.toListAndCloseIterator(windowStore.fetch("1", "2", ofEpochMilli(0), ofEpochMilli(1000L)));
- assertThat(data, equalTo(Arrays.asList(
- KeyValue.pair(new Windowed<>("1", new TimeWindow(0, 100)), ValueAndTimestamp.make(1L, 100L)),
- KeyValue.pair(new Windowed<>("1", new TimeWindow(50, 150)), ValueAndTimestamp.make(2L, 150L)),
- KeyValue.pair(new Windowed<>("1", new TimeWindow(101, 201)), ValueAndTimestamp.make(1L, 150L)),
- KeyValue.pair(new Windowed<>("1", new TimeWindow(400, 500)), ValueAndTimestamp.make(1L, 500L)),
- KeyValue.pair(new Windowed<>("2", new TimeWindow(50, 150)), ValueAndTimestamp.make(1L, 150L)),
- KeyValue.pair(new Windowed<>("2", new TimeWindow(100, 200)), ValueAndTimestamp.make(2L, 200L)),
- KeyValue.pair(new Windowed<>("2", new TimeWindow(151, 251)), ValueAndTimestamp.make(1L, 200L))))); }
+ assertEquals(
+ List.of(
+ KeyValue.pair(new Windowed<>("1", new TimeWindow(0, 100)), ValueAndTimestamp.make(1L, 100L)),
+ KeyValue.pair(new Windowed<>("1", new TimeWindow(50, 150)), ValueAndTimestamp.make(2L, 150L)),
+ KeyValue.pair(new Windowed<>("1", new TimeWindow(101, 201)), ValueAndTimestamp.make(1L, 150L)),
+ KeyValue.pair(new Windowed<>("1", new TimeWindow(400, 500)), ValueAndTimestamp.make(1L, 500L)),
+ KeyValue.pair(new Windowed<>("2", new TimeWindow(50, 150)), ValueAndTimestamp.make(1L, 150L)),
+ KeyValue.pair(new Windowed<>("2", new TimeWindow(100, 200)), ValueAndTimestamp.make(2L, 200L)),
+ KeyValue.pair(new Windowed<>("2", new TimeWindow(151, 251)), ValueAndTimestamp.make(1L, 200L))),
+ data);
+ }
}
}
@@ -265,28 +268,32 @@ public class SlidingWindowedKStreamImplTest {
final WindowStore<String, String> windowStore = driver.getWindowStore("reduced");
final List<KeyValue<Windowed<String>, String>> data =
StreamsTestUtils.toListAndCloseIterator(windowStore.fetch("1", "2", ofEpochMilli(0), ofEpochMilli(1000L)));
- assertThat(data, equalTo(Arrays.asList(
- KeyValue.pair(new Windowed<>("1", new TimeWindow(0, 100)), "1"),
- KeyValue.pair(new Windowed<>("1", new TimeWindow(50, 150)), "1+2"),
- KeyValue.pair(new Windowed<>("1", new TimeWindow(101, 201)), "2"),
- KeyValue.pair(new Windowed<>("1", new TimeWindow(400, 500)), "3"),
- KeyValue.pair(new Windowed<>("2", new TimeWindow(50, 150)), "20"),
- KeyValue.pair(new Windowed<>("2", new TimeWindow(100, 200)), "10+20"),
- KeyValue.pair(new Windowed<>("2", new TimeWindow(151, 251)), "10"))));
+ assertEquals(
+ List.of(
+ KeyValue.pair(new Windowed<>("1", new TimeWindow(0, 100)), "1"),
+ KeyValue.pair(new Windowed<>("1", new TimeWindow(50, 150)), "1+2"),
+ KeyValue.pair(new Windowed<>("1", new TimeWindow(101, 201)), "2"),
+ KeyValue.pair(new Windowed<>("1", new TimeWindow(400, 500)), "3"),
+ KeyValue.pair(new Windowed<>("2", new TimeWindow(50, 150)), "20"),
+ KeyValue.pair(new Windowed<>("2", new TimeWindow(100, 200)), "10+20"),
+ KeyValue.pair(new Windowed<>("2", new TimeWindow(151, 251)), "10")),
+ data);
}
{
final WindowStore<String, ValueAndTimestamp<Long>> windowStore =
driver.getTimestampedWindowStore("reduced");
final List<KeyValue<Windowed<String>, ValueAndTimestamp<Long>>> data =
StreamsTestUtils.toListAndCloseIterator(windowStore.fetch("1", "2", ofEpochMilli(0), ofEpochMilli(1000L)));
- assertThat(data, equalTo(Arrays.asList(
- KeyValue.pair(new Windowed<>("1", new TimeWindow(0, 100)), ValueAndTimestamp.make("1", 100L)),
- KeyValue.pair(new Windowed<>("1", new TimeWindow(50, 150)), ValueAndTimestamp.make("1+2", 150L)),
- KeyValue.pair(new Windowed<>("1", new TimeWindow(101, 201)), ValueAndTimestamp.make("2", 150L)),
- KeyValue.pair(new Windowed<>("1", new TimeWindow(400, 500)), ValueAndTimestamp.make("3", 500L)),
- KeyValue.pair(new Windowed<>("2", new TimeWindow(50, 150)), ValueAndTimestamp.make("20", 150L)),
- KeyValue.pair(new Windowed<>("2", new TimeWindow(100, 200)), ValueAndTimestamp.make("10+20", 200L)),
- KeyValue.pair(new Windowed<>("2", new TimeWindow(151, 251)), ValueAndTimestamp.make("10", 200L)))));
+ assertEquals(
+ List.of(
+ KeyValue.pair(new Windowed<>("1", new TimeWindow(0, 100)), ValueAndTimestamp.make("1", 100L)),
+ KeyValue.pair(new Windowed<>("1", new TimeWindow(50, 150)), ValueAndTimestamp.make("1+2", 150L)),
+ KeyValue.pair(new Windowed<>("1", new TimeWindow(101, 201)), ValueAndTimestamp.make("2", 150L)),
+ KeyValue.pair(new Windowed<>("1", new TimeWindow(400, 500)), ValueAndTimestamp.make("3", 500L)),
+ KeyValue.pair(new Windowed<>("2", new TimeWindow(50, 150)), ValueAndTimestamp.make("20", 150L)),
+ KeyValue.pair(new Windowed<>("2", new TimeWindow(100, 200)), ValueAndTimestamp.make("10+20", 200L)),
+ KeyValue.pair(new Windowed<>("2", new TimeWindow(151, 251)), ValueAndTimestamp.make("10", 200L))),
+ data);
}
}
}
@@ -308,28 +315,32 @@ public class SlidingWindowedKStreamImplTest {
final WindowStore<String, String> windowStore = driver.getWindowStore("aggregated");
final List<KeyValue<Windowed<String>, String>> data =
StreamsTestUtils.toListAndCloseIterator(windowStore.fetch("1", "2", ofEpochMilli(0), ofEpochMilli(1000L)));
- assertThat(data, equalTo(Arrays.asList(
- KeyValue.pair(new Windowed<>("1", new TimeWindow(0, 100)), "0+1"),
- KeyValue.pair(new Windowed<>("1", new TimeWindow(50, 150)), "0+1+2"),
- KeyValue.pair(new Windowed<>("1", new TimeWindow(101, 201)), "0+2"),
- KeyValue.pair(new Windowed<>("1", new TimeWindow(400, 500)), "0+3"),
- KeyValue.pair(new Windowed<>("2", new TimeWindow(50, 150)), "0+20"),
- KeyValue.pair(new Windowed<>("2", new TimeWindow(100, 200)), "0+10+20"),
- KeyValue.pair(new Windowed<>("2", new TimeWindow(151, 251)), "0+10"))));
+ assertEquals(
+ List.of(
+ KeyValue.pair(new Windowed<>("1", new TimeWindow(0, 100)), "0+1"),
+ KeyValue.pair(new Windowed<>("1", new TimeWindow(50, 150)), "0+1+2"),
+ KeyValue.pair(new Windowed<>("1", new TimeWindow(101, 201)), "0+2"),
+ KeyValue.pair(new Windowed<>("1", new TimeWindow(400, 500)), "0+3"),
+ KeyValue.pair(new Windowed<>("2", new TimeWindow(50, 150)), "0+20"),
+ KeyValue.pair(new Windowed<>("2", new TimeWindow(100, 200)), "0+10+20"),
+ KeyValue.pair(new Windowed<>("2", new TimeWindow(151, 251)), "0+10")),
+ data);
}
{
final WindowStore<String, ValueAndTimestamp<Long>> windowStore =
driver.getTimestampedWindowStore("aggregated");
final List<KeyValue<Windowed<String>, ValueAndTimestamp<Long>>> data =
StreamsTestUtils.toListAndCloseIterator(windowStore.fetch("1", "2", ofEpochMilli(0), ofEpochMilli(1000L)));
- assertThat(data, equalTo(Arrays.asList(
- KeyValue.pair(new Windowed<>("1", new TimeWindow(0, 100)), ValueAndTimestamp.make("0+1", 100L)),
- KeyValue.pair(new Windowed<>("1", new TimeWindow(50, 150)), ValueAndTimestamp.make("0+1+2", 150L)),
- KeyValue.pair(new Windowed<>("1", new TimeWindow(101, 201)), ValueAndTimestamp.make("0+2", 150L)),
- KeyValue.pair(new Windowed<>("1", new TimeWindow(400, 500)), ValueAndTimestamp.make("0+3", 500L)),
- KeyValue.pair(new Windowed<>("2", new TimeWindow(50, 150)), ValueAndTimestamp.make("0+20", 150L)),
- KeyValue.pair(new Windowed<>("2", new TimeWindow(100, 200)), ValueAndTimestamp.make("0+10+20", 200L)),
- KeyValue.pair(new Windowed<>("2", new TimeWindow(151, 251)), ValueAndTimestamp.make("0+10", 200L)))));
+ assertEquals(
+ List.of(
+ KeyValue.pair(new Windowed<>("1", new TimeWindow(0, 100)), ValueAndTimestamp.make("0+1", 100L)),
+ KeyValue.pair(new Windowed<>("1", new TimeWindow(50, 150)), ValueAndTimestamp.make("0+1+2", 150L)),
+ KeyValue.pair(new Windowed<>("1", new TimeWindow(101, 201)), ValueAndTimestamp.make("0+2", 150L)),
+ KeyValue.pair(new Windowed<>("1", new TimeWindow(400, 500)), ValueAndTimestamp.make("0+3", 500L)),
+ KeyValue.pair(new Windowed<>("2", new TimeWindow(50, 150)), ValueAndTimestamp.make("0+20", 150L)),
+ KeyValue.pair(new Windowed<>("2", new TimeWindow(100, 200)), ValueAndTimestamp.make("0+10+20", 200L)),
+ KeyValue.pair(new Windowed<>("2", new TimeWindow(151, 251)), ValueAndTimestamp.make("0+10", 200L))),
+ data);
}
}
}
@@ -453,18 +464,22 @@ public class SlidingWindowedKStreamImplTest {
final WindowStore<String, String> windowStore = driver.getWindowStore("aggregated");
final List<KeyValue<Windowed<String>, String>> data =
StreamsTestUtils.toListAndCloseIterator(windowStore.fetch("1", "1", ofEpochMilli(0), ofEpochMilli(10000L)));
- assertThat(data, equalTo(Arrays.asList(
- KeyValue.pair(new Windowed<>("1", new TimeWindow(900, 1000)), "0+4"),
- KeyValue.pair(new Windowed<>("1", new TimeWindow(1900, 2000)), "0+5"))));
+ assertEquals(
+ List.of(
+ KeyValue.pair(new Windowed<>("1", new TimeWindow(900, 1000)), "0+4"),
+ KeyValue.pair(new Windowed<>("1", new TimeWindow(1900, 2000)), "0+5")),
+ data);
}
{
final WindowStore<String, ValueAndTimestamp<Long>> windowStore =
driver.getTimestampedWindowStore("aggregated");
final List<KeyValue<Windowed<String>, ValueAndTimestamp<Long>>> data =
StreamsTestUtils.toListAndCloseIterator(windowStore.fetch("1", "1", ofEpochMilli(0), ofEpochMilli(2000L)));
- assertThat(data, equalTo(Arrays.asList(
- KeyValue.pair(new Windowed<>("1", new TimeWindow(900, 1000)), ValueAndTimestamp.make("0+4", 1000L)),
- KeyValue.pair(new Windowed<>("1", new TimeWindow(1900, 2000)), ValueAndTimestamp.make("0+5", 2000L)))));
+ assertEquals(
+ List.of(
+ KeyValue.pair(new Windowed<>("1", new TimeWindow(900, 1000)), ValueAndTimestamp.make("0+4", 1000L)),
+ KeyValue.pair(new Windowed<>("1", new TimeWindow(1900, 2000)), ValueAndTimestamp.make("0+5", 2000L))),
+ data);
}
}
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SuppressScenarioTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SuppressScenarioTest.java
index ccd9babe228..f829bbb8f65 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SuppressScenarioTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SuppressScenarioTest.java
@@ -68,8 +68,7 @@ import static org.apache.kafka.streams.kstream.Suppressed.BufferConfig.maxRecord
import static org.apache.kafka.streams.kstream.Suppressed.BufferConfig.unbounded;
import static org.apache.kafka.streams.kstream.Suppressed.untilTimeLimit;
import static org.apache.kafka.streams.kstream.Suppressed.untilWindowCloses;
-import static org.hamcrest.CoreMatchers.equalTo;
-import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertEquals;
@SuppressWarnings("deprecation")
public class SuppressScenarioTest {
@@ -858,7 +857,7 @@ public class SuppressScenarioTest {
for (final TestRecord<K, V> result : results) {
final KeyValueTimestamp<K, V> expected = expectedIterator.next();
try {
- assertThat(result, equalTo(new TestRecord<>(expected.key(), expected.value(), null, expected.timestamp())));
+ assertEquals(new TestRecord<>(expected.key(), expected.value(), null, expected.timestamp()), result);
} catch (final AssertionError e) {
throw new AssertionError(printRecords(results) + " != " + expectedResults, e);
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SuppressTopologyTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SuppressTopologyTest.java
index 2254f8c10af..79332359112 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SuppressTopologyTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SuppressTopologyTest.java
@@ -41,8 +41,7 @@ import static java.time.Duration.ofMillis;
import static org.apache.kafka.streams.kstream.Suppressed.BufferConfig.unbounded;
import static org.apache.kafka.streams.kstream.Suppressed.untilTimeLimit;
import static org.apache.kafka.streams.kstream.Suppressed.untilWindowCloses;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.core.Is.is;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
public class SuppressTopologyTest {
@@ -162,7 +161,7 @@ public class SuppressTopologyTest {
final String anonymousNodeTopology = anonymousNodeBuilder.build().describe().toString();
// without the name, the suppression node increments the topology index
- assertThat(anonymousNodeTopology, is(ANONYMOUS_FINAL_TOPOLOGY));
+ assertEquals(ANONYMOUS_FINAL_TOPOLOGY, anonymousNodeTopology);
}
@Test
@@ -180,7 +179,7 @@ public class SuppressTopologyTest {
final String namedNodeTopology = namedNodeBuilder.build().describe().toString();
// without the name, the suppression node does not increment the topology index
- assertThat(namedNodeTopology, is(NAMED_FINAL_TOPOLOGY));
+ assertEquals(NAMED_FINAL_TOPOLOGY, namedNodeTopology);
}
@Test
@@ -196,7 +195,7 @@ public class SuppressTopologyTest {
final String anonymousNodeTopology = anonymousNodeBuilder.build().describe().toString();
// without the name, the suppression node increments the topology index
- assertThat(anonymousNodeTopology, is(ANONYMOUS_INTERMEDIATE_TOPOLOGY));
+ assertEquals(ANONYMOUS_INTERMEDIATE_TOPOLOGY, anonymousNodeTopology);
}
@Test
@@ -212,7 +211,7 @@ public class SuppressTopologyTest {
final String namedNodeTopology = namedNodeBuilder.build().describe().toString();
// without the name, the suppression node does not increment the topology index
- assertThat(namedNodeTopology, is(NAMED_INTERMEDIATE_TOPOLOGY));
+ assertEquals(NAMED_INTERMEDIATE_TOPOLOGY, namedNodeTopology);
}
@Test
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/TimeWindowedCogroupedKStreamImplTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/TimeWindowedCogroupedKStreamImplTest.java
index 1f133b8a252..7534d0f1ee7 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/TimeWindowedCogroupedKStreamImplTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/TimeWindowedCogroupedKStreamImplTest.java
@@ -51,8 +51,7 @@ import org.junit.jupiter.params.provider.ValueSource;
import java.util.Properties;
import static java.time.Duration.ofMillis;
-import static org.hamcrest.CoreMatchers.equalTo;
-import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
public class TimeWindowedCogroupedKStreamImplTest {
@@ -158,7 +157,7 @@ public class TimeWindowedCogroupedKStreamImplTest {
.windowedBy(TimeWindows.ofSizeWithNoGrace(ofMillis(WINDOW_SIZE)))
.aggregate(MockInitializer.STRING_INIT, Named.as("foo"));
- assertThat(builder.build().describe().toString(), equalTo(
+ assertEquals(
"Topologies:\n" +
" Sub-topology: 0\n" +
" Source: KSTREAM-SOURCE-0000000000 (topics: [topic])\n" +
@@ -168,7 +167,8 @@ public class TimeWindowedCogroupedKStreamImplTest {
" <-- KSTREAM-SOURCE-0000000000\n" +
" Processor: foo-cogroup-merge (stores: [])\n" +
" --> none\n" +
- " <-- foo-cogroup-agg-0\n\n"));
+ " <-- foo-cogroup-agg-0\n\n",
+ builder.build().describe().toString());
}
@ParameterizedTest
@@ -354,6 +354,6 @@ public class TimeWindowedCogroupedKStreamImplTest {
final TestRecord<String, String> nonWindowedRecord = new TestRecord<>(
realRecord.getKey().key(), realRecord.getValue(), null, realRecord.timestamp());
final TestRecord<String, String> testRecord = new TestRecord<>(expectedKey, expectedValue, null, expectedTimestamp);
- assertThat(nonWindowedRecord, equalTo(testRecord));
+ assertEquals(testRecord, nonWindowedRecord);
}
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/TimeWindowedKStreamImplTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/TimeWindowedKStreamImplTest.java
index d734cd21fd5..1f53dd9a752 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/TimeWindowedKStreamImplTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/TimeWindowedKStreamImplTest.java
@@ -50,7 +50,6 @@ import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
import java.util.ArrayList;
-import java.util.Collections;
import java.util.List;
import java.util.Properties;
import java.util.stream.Stream;
@@ -58,8 +57,6 @@ import java.util.stream.Stream;
import static java.time.Duration.ofMillis;
import static java.time.Instant.ofEpochMilli;
import static java.util.Arrays.asList;
-import static org.hamcrest.CoreMatchers.equalTo;
-import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -250,21 +247,24 @@ public class TimeWindowedKStreamImplTest {
if (withCache) {
// with cache returns all records (expired from underneath as well) as part of
// the merge process
- assertThat(data, equalTo(asList(
+ assertEquals(
+ List.of(
KeyValue.pair(new Windowed<>("1", new TimeWindow(0, 500)), 2L),
KeyValue.pair(new Windowed<>("1", new TimeWindow(500, 1000)), 1L),
KeyValue.pair(new Windowed<>("2", new TimeWindow(500, 1000)), 2L),
- KeyValue.pair(new Windowed<>("2", new TimeWindow(1000, 1500)), 1L))));
+ KeyValue.pair(new Windowed<>("2", new TimeWindow(1000, 1500)), 1L)),
+ data);
} else {
// without cache, we get only non-expired record from underlying store.
if (!emitFinal) {
- assertThat(data, equalTo(Collections.singletonList(
- KeyValue.pair(new Windowed<>("2", new TimeWindow(1000, 1500)), 1L))));
+ assertEquals(List.of(KeyValue.pair(new Windowed<>("2", new TimeWindow(1000, 1500)), 1L)), data);
} else {
- assertThat(data, equalTo(asList(
+ assertEquals(
+ List.of(
KeyValue.pair(new Windowed<>("1", new TimeWindow(500, 1000)), 1L),
KeyValue.pair(new Windowed<>("2", new TimeWindow(500, 1000)), 2L),
- KeyValue.pair(new Windowed<>("2", new TimeWindow(1000, 1500)), 1L))));
+ KeyValue.pair(new Windowed<>("2", new TimeWindow(1000, 1500)), 1L)),
+ data);
}
}
}
@@ -276,20 +276,25 @@ public class TimeWindowedKStreamImplTest {
// the same values and logic described above applies here as well.
if (withCache) {
- assertThat(data, equalTo(asList(
+ assertEquals(
+ List.of(
KeyValue.pair(new Windowed<>("1", new TimeWindow(0, 500)), ValueAndTimestamp.make(2L, 15L)),
KeyValue.pair(new Windowed<>("1", new TimeWindow(500, 1000)), ValueAndTimestamp.make(1L, 500L)),
KeyValue.pair(new Windowed<>("2", new TimeWindow(500, 1000)), ValueAndTimestamp.make(2L, 550L)),
- KeyValue.pair(new Windowed<>("2", new TimeWindow(1000, 1500)), ValueAndTimestamp.make(1L, 1000L)))));
+ KeyValue.pair(new Windowed<>("2", new TimeWindow(1000, 1500)), ValueAndTimestamp.make(1L, 1000L))),
+ data);
} else {
if (!emitFinal) {
- assertThat(data, equalTo(Collections.singletonList(
- KeyValue.pair(new Windowed<>("2", new TimeWindow(1000, 1500)), ValueAndTimestamp.make(1L, 1000L)))));
+ assertEquals(
+ List.of(KeyValue.pair(new Windowed<>("2", new TimeWindow(1000, 1500)), ValueAndTimestamp.make(1L, 1000L))),
+ data);
} else {
- assertThat(data, equalTo(asList(
+ assertEquals(
+ List.of(
KeyValue.pair(new Windowed<>("1", new TimeWindow(500, 1000)), ValueAndTimestamp.make(1L, 500L)),
KeyValue.pair(new Windowed<>("2", new TimeWindow(500, 1000)), ValueAndTimestamp.make(2L, 550L)),
- KeyValue.pair(new Windowed<>("2", new TimeWindow(1000, 1500)), ValueAndTimestamp.make(1L, 1000L)))));
+ KeyValue.pair(new Windowed<>("2", new TimeWindow(1000, 1500)), ValueAndTimestamp.make(1L, 1000L))),
+ data);
}
}
}
@@ -316,16 +321,18 @@ public class TimeWindowedKStreamImplTest {
if (withCache) {
// with cache returns all records (expired from underneath as well) as part of
// the merge process
- assertThat(data, equalTo(asList(
+ assertEquals(
+ List.of(
KeyValue.pair(new Windowed<>("1", new TimeWindow(0, 500)), "1+2"),
KeyValue.pair(new Windowed<>("1", new TimeWindow(500, 1000)), "3"),
KeyValue.pair(new Windowed<>("2", new TimeWindow(500, 1000)), "10+20"),
- KeyValue.pair(new Windowed<>("2", new TimeWindow(1000, 1500)), "30"))));
+ KeyValue.pair(new Windowed<>("2", new TimeWindow(1000, 1500)), "30")),
+ data);
} else {
// without cache, we get only non-expired record from underlying store.
// actualFrom = observedStreamTime(1500) - retentionPeriod(1000) + 1 = 501.
// only 1 record is non expired and would be returned.
- assertThat(data, equalTo(Collections.singletonList(KeyValue.pair(new Windowed<>("2", new TimeWindow(1000, 1500)), "30"))));
+ assertEquals(List.of(KeyValue.pair(new Windowed<>("2", new TimeWindow(1000, 1500)), "30")), data);
}
}
{
@@ -335,14 +342,17 @@ public class TimeWindowedKStreamImplTest {
// same logic/data as explained above.
if (withCache) {
- assertThat(data, equalTo(asList(
+ assertEquals(
+ List.of(
KeyValue.pair(new Windowed<>("1", new TimeWindow(0, 500)), ValueAndTimestamp.make("1+2", 15L)),
KeyValue.pair(new Windowed<>("1", new TimeWindow(500, 1000)), ValueAndTimestamp.make("3", 500L)),
KeyValue.pair(new Windowed<>("2", new TimeWindow(500, 1000)), ValueAndTimestamp.make("10+20", 550L)),
- KeyValue.pair(new Windowed<>("2", new TimeWindow(1000, 1500)), ValueAndTimestamp.make("30", 1000L)))));
+ KeyValue.pair(new Windowed<>("2", new TimeWindow(1000, 1500)), ValueAndTimestamp.make("30", 1000L))),
+ data);
} else {
- assertThat(data, equalTo(Collections.singletonList(
- KeyValue.pair(new Windowed<>("2", new TimeWindow(1000, 1500)), ValueAndTimestamp.make("30", 1000L)))));
+ assertEquals(
+ List.of(KeyValue.pair(new Windowed<>("2", new TimeWindow(1000, 1500)), ValueAndTimestamp.make("30", 1000L))),
+ data);
}
}
}
@@ -369,17 +379,18 @@ public class TimeWindowedKStreamImplTest {
if (withCache) {
// with cache returns all records (expired from underneath as well) as part of
// the merge process
- assertThat(data, equalTo(asList(
+ assertEquals(
+ List.of(
KeyValue.pair(new Windowed<>("1", new TimeWindow(0, 500)), "0+1+2"),
KeyValue.pair(new Windowed<>("1", new TimeWindow(500, 1000)), "0+3"),
KeyValue.pair(new Windowed<>("2", new TimeWindow(500, 1000)), "0+10+20"),
- KeyValue.pair(new Windowed<>("2", new TimeWindow(1000, 1500)), "0+30"))));
+ KeyValue.pair(new Windowed<>("2", new TimeWindow(1000, 1500)), "0+30")),
+ data);
} else {
// without cache, we get only non-expired record from underlying store.
// actualFrom = observedStreamTime(1500) - retentionPeriod(1000) + 1 = 501.
// only 1 record is non expired and would be returned.
- assertThat(data, equalTo(Collections
- .singletonList(KeyValue.pair(new Windowed<>("2", new TimeWindow(1000, 1500)), "0+30"))));
+ assertEquals(List.of(KeyValue.pair(new Windowed<>("2", new TimeWindow(1000, 1500)), "0+30")), data);
}
}
{
@@ -387,14 +398,17 @@ public class TimeWindowedKStreamImplTest {
final List<KeyValue<Windowed<String>, ValueAndTimestamp<String>>> data =
StreamsTestUtils.toListAndCloseIterator(windowStore.fetch("1", "2", ofEpochMilli(0), ofEpochMilli(1000L)));
if (withCache) {
- assertThat(data, equalTo(asList(
+ assertEquals(
+ List.of(
KeyValue.pair(new Windowed<>("1", new TimeWindow(0, 500)), ValueAndTimestamp.make("0+1+2", 15L)),
KeyValue.pair(new Windowed<>("1", new TimeWindow(500, 1000)), ValueAndTimestamp.make("0+3", 500L)),
KeyValue.pair(new Windowed<>("2", new TimeWindow(500, 1000)), ValueAndTimestamp.make("0+10+20", 550L)),
- KeyValue.pair(new Windowed<>("2", new TimeWindow(1000, 1500)), ValueAndTimestamp.make("0+30", 1000L)))));
+ KeyValue.pair(new Windowed<>("2", new TimeWindow(1000, 1500)), ValueAndTimestamp.make("0+30", 1000L))),
+ data);
} else {
- assertThat(data, equalTo(Collections.singletonList(
- KeyValue.pair(new Windowed<>("2", new TimeWindow(1000, 1500)), ValueAndTimestamp.make("0+30", 1000L)))));
+ assertEquals(
+ List.of(KeyValue.pair(new Windowed<>("2", new TimeWindow(1000, 1500)), ValueAndTimestamp.make("0+30", 1000L))),
+ data);
}
}
}