You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by fr...@apache.org on 2026/04/14 13:23:16 UTC
(kafka) branch trunk updated: KAFKA-20329: Test headers dsl.store.format further (4/N) (#21827)
This is an automated email from the ASF dual-hosted git repository.
frankvicky 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 e2b5eb61316 KAFKA-20329: Test headers dsl.store.format further (4/N) (#21827)
e2b5eb61316 is described below
commit e2b5eb61316f02b586eeee3d76de43ec2678edbc
Author: Alieh Saeedi <10...@users.noreply.github.com>
AuthorDate: Tue Apr 14 15:23:08 2026 +0200
KAFKA-20329: Test headers dsl.store.format further (4/N) (#21827)
Parametrized 28 DSL and processor test classes to validate functionality
with both `headers` and `default` store configurations. This ensures
backward compatibility and correctness as Kafka Streams transitions to
supporting record headers in state stores.
Modified Tests: - Materializer Tests - Windowed KStream Tests
- Aggregation Tests - Cogroup Tests - Join Tests - KTable
Tests
Reviewers: Matthias J. Sax <ma...@confluent.io>, TengYao Chi
<fr...@apache.org>
---
.../internals/CogroupedKStreamImplTest.java | 213 ++++++++++++++-------
.../kstream/internals/KStreamKStreamJoinTest.java | 165 +++++++++++-----
.../internals/KStreamKStreamLeftJoinTest.java | 109 +++++++----
.../internals/KStreamKStreamOuterJoinTest.java | 122 ++++++++----
.../internals/KStreamKStreamSelfJoinTest.java | 44 ++++-
.../kstream/internals/KStreamKTableJoinTest.java | 116 +++++++----
.../internals/KStreamKTableLeftJoinTest.java | 54 ++++--
...KStreamSessionWindowAggregateProcessorTest.java | 94 +++++----
.../KStreamSlidingWindowAggregateTest.java | 115 ++++++-----
.../internals/KStreamWindowAggregateTest.java | 86 +++++----
.../kstream/internals/KTableAggregateTest.java | 83 ++++----
.../kstream/internals/KTableFilterTest.java | 98 ++++++----
.../streams/kstream/internals/KTableImplTest.java | 188 ++++++++++++------
.../KTableKTableForeignKeyJoinScenarioTest.java | 70 ++++---
.../internals/KTableKTableInnerJoinTest.java | 69 +++++--
.../internals/KTableKTableLeftJoinTest.java | 45 +++--
.../internals/KTableKTableOuterJoinTest.java | 44 ++++-
.../internals/KTableKTableRightJoinTest.java | 24 ++-
.../kstream/internals/KTableMapKeysTest.java | 25 ++-
.../kstream/internals/KTableMapValuesTest.java | 56 ++++--
.../internals/KTableTransformValuesTest.java | 42 ++--
.../SlidingWindowedCogroupedKStreamImplTest.java | 79 +++++---
.../internals/SlidingWindowedKStreamImplTest.java | 121 ++++++++----
.../TimeWindowedCogroupedKStreamImplTest.java | 97 ++++++----
.../internals/TimeWindowedKStreamImplTest.java | 82 ++++----
25 files changed, 1488 insertions(+), 753 deletions(-)
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 244c1f2c048..dc6b851223d 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
@@ -49,8 +49,8 @@ import org.apache.kafka.test.MockInitializer;
import org.apache.kafka.test.MockValueJoiner;
import org.apache.kafka.test.StreamsTestUtils;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.util.Properties;
@@ -82,8 +82,10 @@ public class CogroupedKStreamImplTest {
private static final Initializer<Integer> SUM_INITIALIZER = () -> 0;
- @BeforeEach
- public void setup() {
+ private void setup(final boolean withHeaders) {
+ if (withHeaders) {
+ props.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
+ }
final StreamsBuilder builder = new StreamsBuilder();
final KStream<String, String> stream = builder.stream(TOPIC, Consumed.with(Serdes.String(), Serdes.String()));
@@ -91,73 +93,101 @@ public class CogroupedKStreamImplTest {
cogroupedStream = groupedStream.cogroup(MockAggregator.TOSTRING_ADDER);
}
- @Test
- public void shouldThrowNPEInCogroupIfKGroupedStreamIsNull() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldThrowNPEInCogroupIfKGroupedStreamIsNull(final boolean withHeaders) {
+ setup(withHeaders);
assertThrows(NullPointerException.class, () -> cogroupedStream.cogroup(null, MockAggregator.TOSTRING_ADDER));
}
- @Test
- public void shouldNotHaveNullAggregatorOnCogroup() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotHaveNullAggregatorOnCogroup(final boolean withHeaders) {
+ setup(withHeaders);
assertThrows(NullPointerException.class, () -> cogroupedStream.cogroup(groupedStream, null));
}
- @Test
- public void shouldNotHaveNullInitializerOnAggregate() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotHaveNullInitializerOnAggregate(final boolean withHeaders) {
+ setup(withHeaders);
assertThrows(NullPointerException.class, () -> cogroupedStream.aggregate(null));
}
- @Test
- public void shouldNotHaveNullInitializerOnAggregateWitNamed() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotHaveNullInitializerOnAggregateWitNamed(final boolean withHeaders) {
+ setup(withHeaders);
assertThrows(NullPointerException.class, () -> cogroupedStream.aggregate(null, Named.as("name")));
}
- @Test
- public void shouldNotHaveNullInitializerOnAggregateWitMaterialized() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotHaveNullInitializerOnAggregateWitMaterialized(final boolean withHeaders) {
+ setup(withHeaders);
assertThrows(NullPointerException.class, () -> cogroupedStream.aggregate(null, Materialized.as("store")));
}
- @Test
- public void shouldNotHaveNullInitializerOnAggregateWitNamedAndMaterialized() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotHaveNullInitializerOnAggregateWitNamedAndMaterialized(final boolean withHeaders) {
+ setup(withHeaders);
assertThrows(NullPointerException.class, () -> cogroupedStream.aggregate(null, Named.as("name"), Materialized.as("store")));
}
- @Test
- public void shouldNotHaveNullNamedOnAggregate() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotHaveNullNamedOnAggregate(final boolean withHeaders) {
+ setup(withHeaders);
assertThrows(NullPointerException.class, () -> cogroupedStream.aggregate(STRING_INITIALIZER, (Named) null));
}
- @Test
- public void shouldNotHaveNullMaterializedOnAggregate() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotHaveNullMaterializedOnAggregate(final boolean withHeaders) {
+ setup(withHeaders);
assertThrows(NullPointerException.class, () -> cogroupedStream.aggregate(STRING_INITIALIZER, (Materialized<String, String, KeyValueStore<Bytes, byte[]>>) null));
}
- @Test
- public void shouldNotHaveNullNamedOnAggregateWithMateriazlied() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotHaveNullNamedOnAggregateWithMateriazlied(final boolean withHeaders) {
+ setup(withHeaders);
assertThrows(NullPointerException.class, () -> cogroupedStream.aggregate(STRING_INITIALIZER, null, Materialized.as("store")));
}
- @Test
- public void shouldNotHaveNullMaterializedOnAggregateWithNames() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotHaveNullMaterializedOnAggregateWithNames(final boolean withHeaders) {
+ setup(withHeaders);
assertThrows(NullPointerException.class, () -> cogroupedStream.aggregate(STRING_INITIALIZER, Named.as("name"), null));
}
- @Test
- public void shouldNotHaveNullWindowOnWindowedByTime() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotHaveNullWindowOnWindowedByTime(final boolean withHeaders) {
+ setup(withHeaders);
assertThrows(NullPointerException.class, () -> cogroupedStream.windowedBy((Windows<? extends Window>) null));
}
- @Test
- public void shouldNotHaveNullWindowOnWindowedBySession() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotHaveNullWindowOnWindowedBySession(final boolean withHeaders) {
+ setup(withHeaders);
assertThrows(NullPointerException.class, () -> cogroupedStream.windowedBy((SessionWindows) null));
}
- @Test
- public void shouldNotHaveNullWindowOnWindowedBySliding() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotHaveNullWindowOnWindowedBySliding(final boolean withHeaders) {
+ setup(withHeaders);
assertThrows(NullPointerException.class, () -> cogroupedStream.windowedBy((SlidingWindows) null));
}
- @Test
- public void shouldNameProcessorsAndStoreBasedOnNamedParameter() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNameProcessorsAndStoreBasedOnNamedParameter(final boolean withHeaders) {
+ setup(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final KStream<String, String> stream1 = builder.stream("one", stringConsumed);
@@ -199,8 +229,10 @@ public class CogroupedKStreamImplTest {
" <-- KTABLE-TOSTREAM-0000000005\n\n"));
}
- @Test
- public void shouldNameRepartitionTopic() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNameRepartitionTopic(final boolean withHeaders) {
+ setup(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final KStream<String, String> stream1 = builder.stream("one", stringConsumed);
@@ -253,8 +285,10 @@ public class CogroupedKStreamImplTest {
" <-- KTABLE-TOSTREAM-0000000010\n\n"));
}
- @Test
- public void shouldInsertRepartitionsTopicForUpstreamKeyModification() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldInsertRepartitionsTopicForUpstreamKeyModification(final boolean withHeaders) {
+ setup(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final KStream<String, String> stream1 = builder.stream("one", stringConsumed);
final KStream<String, String> test2 = builder.stream("two", stringConsumed);
@@ -306,8 +340,10 @@ public class CogroupedKStreamImplTest {
" <-- KTABLE-TOSTREAM-0000000009\n\n"));
}
- @Test
- public void shouldInsertRepartitionsTopicForUpstreamKeyModificationWithGroupedReusedInSameCogroups() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldInsertRepartitionsTopicForUpstreamKeyModificationWithGroupedReusedInSameCogroups(final boolean withHeaders) {
+ setup(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final KStream<String, String> stream1 = builder.stream("one", stringConsumed);
@@ -387,10 +423,11 @@ public class CogroupedKStreamImplTest {
" <-- KTABLE-TOSTREAM-0000000019\n\n"));
}
- @Test
- public void shouldInsertRepartitionsTopicForUpstreamKeyModificationWithGroupedReusedInSameCogroupsWithOptimization() {
- final Properties properties = new Properties();
- properties.setProperty(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG, StreamsConfig.OPTIMIZE);
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldInsertRepartitionsTopicForUpstreamKeyModificationWithGroupedReusedInSameCogroupsWithOptimization(final boolean withHeaders) {
+ setup(withHeaders);
+ props.setProperty(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG, StreamsConfig.OPTIMIZE);
final StreamsBuilder builder = new StreamsBuilder();
final KStream<String, String> stream1 = builder.stream("one", stringConsumed);
@@ -412,7 +449,7 @@ public class CogroupedKStreamImplTest {
cogroupedOne.toStream().to(OUTPUT);
cogroupedTwo.toStream().to("OUTPUT2");
- final String topologyDescription = builder.build(properties).describe().toString();
+ final String topologyDescription = builder.build(props).describe().toString();
assertThat(
topologyDescription,
@@ -463,8 +500,10 @@ public class CogroupedKStreamImplTest {
" <-- KTABLE-TOSTREAM-0000000019\n\n"));
}
- @Test
- public void shouldInsertRepartitionsTopicForUpstreamKeyModificationWithGroupedReusedInDifferentCogroups() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldInsertRepartitionsTopicForUpstreamKeyModificationWithGroupedReusedInDifferentCogroups(final boolean withHeaders) {
+ setup(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final KStream<String, String> stream1 = builder.stream("one", stringConsumed);
@@ -534,8 +573,10 @@ public class CogroupedKStreamImplTest {
" <-- COGROUPKSTREAM-AGGREGATE-0000000008, COGROUPKSTREAM-AGGREGATE-0000000009\n\n"));
}
- @Test
- public void shouldInsertRepartitionsTopicForUpstreamKeyModificationWithGroupedReusedInDifferentCogroupsWithOptimization() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldInsertRepartitionsTopicForUpstreamKeyModificationWithGroupedReusedInDifferentCogroupsWithOptimization(final boolean withHeaders) {
+ setup(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final Properties properties = new Properties();
@@ -600,8 +641,10 @@ public class CogroupedKStreamImplTest {
" <-- COGROUPKSTREAM-AGGREGATE-0000000015, COGROUPKSTREAM-AGGREGATE-0000000016\n\n"));
}
- @Test
- public void shouldInsertRepartitionsTopicForUpstreamKeyModificationWithGroupedReused() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldInsertRepartitionsTopicForUpstreamKeyModificationWithGroupedReused(final boolean withHeaders) {
+ setup(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final KStream<String, String> stream1 = builder.stream("one", stringConsumed);
@@ -659,8 +702,10 @@ public class CogroupedKStreamImplTest {
" <-- KSTREAM-SOURCE-0000000014\n\n"));
}
- @Test
- public void shouldInsertRepartitionsTopicForUpstreamKeyModificationWithGroupedReusedWithOptimization() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldInsertRepartitionsTopicForUpstreamKeyModificationWithGroupedReusedWithOptimization(final boolean withHeaders) {
+ setup(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final Properties properties = new Properties();
@@ -713,8 +758,10 @@ public class CogroupedKStreamImplTest {
" <-- COGROUPKSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition-source\n\n"));
}
- @Test
- public void shouldInsertRepartitionsTopicForUpstreamKeyModificationWithGroupedRemadeWithOptimization() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldInsertRepartitionsTopicForUpstreamKeyModificationWithGroupedRemadeWithOptimization(final boolean withHeaders) {
+ setup(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final Properties properties = new Properties();
@@ -793,8 +840,10 @@ public class CogroupedKStreamImplTest {
" <-- COGROUPKSTREAM-AGGREGATE-0000000016, COGROUPKSTREAM-AGGREGATE-0000000017\n\n"));
}
- @Test
- public void shouldInsertRepartitionsTopicForCogroupsUsedTwice() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldInsertRepartitionsTopicForCogroupsUsedTwice(final boolean withHeaders) {
+ setup(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final Properties properties = new Properties();
@@ -840,8 +889,10 @@ public class CogroupedKStreamImplTest {
" <-- COGROUPKSTREAM-AGGREGATE-0000000012\n\n"));
}
- @Test
- public void shouldCogroupAndAggregateSingleKStreams() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldCogroupAndAggregateSingleKStreams(final boolean withHeaders) {
+ setup(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final KStream<String, String> stream1 = builder.stream("one", stringConsumed);
@@ -870,8 +921,10 @@ public class CogroupedKStreamImplTest {
}
}
- @Test
- public void testCogroupHandleNullValues() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testCogroupHandleNullValues(final boolean withHeaders) {
+ setup(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final KStream<String, String> stream1 = builder.stream("one", stringConsumed);
@@ -899,8 +952,10 @@ public class CogroupedKStreamImplTest {
}
}
- @Test
- public void shouldCogroupAndAggregateTwoKStreamsWithDistinctKeys() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldCogroupAndAggregateTwoKStreamsWithDistinctKeys(final boolean withHeaders) {
+ setup(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final KStream<String, String> stream1 = builder.stream("one", stringConsumed);
final KStream<String, String> stream2 = builder.stream("two", stringConsumed);
@@ -947,8 +1002,10 @@ public class CogroupedKStreamImplTest {
}
}
- @Test
- public void shouldCogroupAndAggregateTwoKStreamsWithSharedKeys() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldCogroupAndAggregateTwoKStreamsWithSharedKeys(final boolean withHeaders) {
+ setup(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final KStream<String, String> stream1 = builder.stream("one", stringConsumed);
final KStream<String, String> stream2 = builder.stream("two", stringConsumed);
@@ -997,8 +1054,10 @@ public class CogroupedKStreamImplTest {
}
}
- @Test
- public void shouldAllowDifferentOutputTypeInCoGroup() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldAllowDifferentOutputTypeInCoGroup(final boolean withHeaders) {
+ setup(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final KStream<String, String> stream1 = builder.stream("one", stringConsumed);
final KStream<String, String> stream2 = builder.stream("two", stringConsumed);
@@ -1050,8 +1109,10 @@ public class CogroupedKStreamImplTest {
}
}
- @Test
- public void shouldCoGroupStreamsWithDifferentInputTypes() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldCoGroupStreamsWithDifferentInputTypes(final boolean withHeaders) {
+ setup(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final Consumed<String, Integer> integerConsumed = Consumed.with(Serdes.String(), Serdes.Integer());
final KStream<String, String> stream1 = builder.stream("one", stringConsumed);
@@ -1102,8 +1163,10 @@ public class CogroupedKStreamImplTest {
}
}
- @Test
- public void testCogroupKeyMixedAggregators() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testCogroupKeyMixedAggregators(final boolean withHeaders) {
+ setup(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final KStream<String, String> stream1 = builder.stream("one", stringConsumed);
final KStream<String, String> stream2 = builder.stream("two", stringConsumed);
@@ -1149,8 +1212,10 @@ public class CogroupedKStreamImplTest {
}
}
- @Test
- public void testCogroupWithThreeGroupedStreams() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testCogroupWithThreeGroupedStreams(final boolean withHeaders) {
+ setup(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final KStream<String, String> stream1 = builder.stream("one", stringConsumed);
final KStream<String, String> stream2 = builder.stream("two", stringConsumed);
@@ -1206,8 +1271,10 @@ public class CogroupedKStreamImplTest {
}
}
- @Test
- public void testCogroupWithKTableKTableInnerJoin() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testCogroupWithKTableKTableInnerJoin(final boolean withHeaders) {
+ setup(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final KGroupedStream<String, String> grouped1 = builder.stream("one", stringConsumed).groupByKey();
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 58596a63850..68091871137 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
@@ -61,7 +61,8 @@ import org.apache.kafka.test.MockInternalProcessorContext;
import org.apache.kafka.test.MockValueJoiner;
import org.apache.kafka.test.StreamsTestUtils;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.Mockito;
import java.time.Duration;
@@ -95,13 +96,15 @@ public class KStreamKStreamJoinTest {
private final String topic1 = "topic1";
private final String topic2 = "topic2";
private final Consumed<Integer, String> consumed = Consumed.with(Serdes.Integer(), Serdes.String());
- private final Properties props = StreamsTestUtils.getStreamsConfig(Serdes.String(), Serdes.String());
+ private Properties props = StreamsTestUtils.getStreamsConfig(Serdes.String(), Serdes.String());
private final JoinWindows joinWindows = JoinWindows.ofTimeDifferenceAndGrace(ofMillis(50), Duration.ofMillis(50));
private final StreamJoined<String, Integer, Integer> streamJoined = StreamJoined.with(Serdes.String(), Serdes.Integer(), Serdes.Integer());
private final String errorMessagePrefix = "Window settings mismatch. WindowBytesStoreSupplier settings";
- @Test
- public void shouldLogAndMeterOnSkippedRecordsWithNullValueWithBuiltInMetricsVersionLatest() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldLogAndMeterOnSkippedRecordsWithNullValueWithBuiltInMetricsVersionLatest(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final KStream<String, Integer> left = builder.stream("left", Consumed.with(Serdes.String(), Serdes.Integer()));
@@ -130,8 +133,10 @@ public class KStreamKStreamJoinTest {
}
}
- @Test
- public void shouldReuseRepartitionTopicWithGeneratedName() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldReuseRepartitionTopicWithGeneratedName(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final Properties props = new Properties();
props.put(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG, StreamsConfig.NO_OPTIMIZATION);
@@ -144,8 +149,10 @@ public class KStreamKStreamJoinTest {
assertEquals(expectedTopologyWithGeneratedRepartitionTopic, builder.build(props).describe().toString());
}
- @Test
- public void shouldCreateRepartitionTopicsWithUserProvidedName() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldCreateRepartitionTopicsWithUserProvidedName(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final Properties props = new Properties();
props.put(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG, StreamsConfig.NO_OPTIMIZATION);
@@ -161,8 +168,10 @@ public class KStreamKStreamJoinTest {
assertEquals(expectedTopologyWithUserNamedRepartitionTopics, topology.describe().toString());
}
- @Test
- public void shouldDisableLoggingOnStreamJoined() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldDisableLoggingOnStreamJoined(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final JoinWindows joinWindows = JoinWindows.ofTimeDifferenceAndGrace(ofMillis(100), Duration.ofMillis(50));
final StreamJoined<String, Integer, Integer> streamJoined = StreamJoined
.with(Serdes.String(), Serdes.Integer(), Serdes.Integer())
@@ -187,8 +196,10 @@ public class KStreamKStreamJoinTest {
assertThat(internalTopologyBuilder.stateStores().get("store-other-join-store").loggingEnabled(), equalTo(false));
}
- @Test
- public void shouldEnableLoggingWithCustomConfigOnStreamJoined() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldEnableLoggingWithCustomConfigOnStreamJoined(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final JoinWindows joinWindows = JoinWindows.ofTimeDifferenceAndGrace(ofMillis(100), Duration.ofMillis(50));
final StreamJoined<String, Integer, Integer> streamJoined = StreamJoined
.with(Serdes.String(), Serdes.Integer(), Serdes.Integer())
@@ -222,8 +233,10 @@ public class KStreamKStreamJoinTest {
}
}
- @Test
- public void shouldThrowExceptionThisStoreSupplierRetentionDoNotMatchWindowsSizeAndGrace() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldThrowExceptionThisStoreSupplierRetentionDoNotMatchWindowsSizeAndGrace(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
// Case where retention of thisJoinStore doesn't match JoinWindows
final WindowBytesStoreSupplier thisStoreSupplier = buildWindowBytesStoreSupplier("in-memory-join-store", 500L, 100L, true);
final WindowBytesStoreSupplier otherStoreSupplier = buildWindowBytesStoreSupplier("in-memory-join-store-other", 150L, 100L, true);
@@ -235,8 +248,10 @@ public class KStreamKStreamJoinTest {
);
}
- @Test
- public void shouldThrowExceptionThisStoreSupplierWindowSizeDoesNotMatchJoinWindowsWindowSize() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldThrowExceptionThisStoreSupplierWindowSizeDoesNotMatchJoinWindowsWindowSize(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
//Case where window size of thisJoinStore doesn't match JoinWindows
final WindowBytesStoreSupplier thisStoreSupplier = buildWindowBytesStoreSupplier("in-memory-join-store", 150L, 150L, true);
final WindowBytesStoreSupplier otherStoreSupplier = buildWindowBytesStoreSupplier("in-memory-join-store-other", 150L, 100L, true);
@@ -248,8 +263,10 @@ public class KStreamKStreamJoinTest {
);
}
- @Test
- public void shouldThrowExceptionWhenThisJoinStoreSetsRetainDuplicatesFalse() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldThrowExceptionWhenThisJoinStoreSetsRetainDuplicatesFalse(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
//Case where thisJoinStore retain duplicates false
final WindowBytesStoreSupplier thisStoreSupplier = buildWindowBytesStoreSupplier("in-memory-join-store", 150L, 100L, false);
final WindowBytesStoreSupplier otherStoreSupplier = buildWindowBytesStoreSupplier("in-memory-join-store-other", 150L, 100L, true);
@@ -261,8 +278,10 @@ public class KStreamKStreamJoinTest {
);
}
- @Test
- public void shouldThrowExceptionOtherStoreSupplierRetentionDoNotMatchWindowsSizeAndGrace() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldThrowExceptionOtherStoreSupplierRetentionDoNotMatchWindowsSizeAndGrace(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
//Case where retention size of otherJoinStore doesn't match JoinWindows
final WindowBytesStoreSupplier thisStoreSupplier = buildWindowBytesStoreSupplier("in-memory-join-store", 150L, 100L, true);
final WindowBytesStoreSupplier otherStoreSupplier = buildWindowBytesStoreSupplier("in-memory-join-store-other", 500L, 100L, true);
@@ -274,8 +293,10 @@ public class KStreamKStreamJoinTest {
);
}
- @Test
- public void shouldThrowExceptionOtherStoreSupplierWindowSizeDoesNotMatchJoinWindowsWindowSize() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldThrowExceptionOtherStoreSupplierWindowSizeDoesNotMatchJoinWindowsWindowSize(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
//Case where window size of otherJoinStore doesn't match JoinWindows
final WindowBytesStoreSupplier thisStoreSupplier = buildWindowBytesStoreSupplier("in-memory-join-store", 150L, 100L, true);
final WindowBytesStoreSupplier otherStoreSupplier = buildWindowBytesStoreSupplier("in-memory-join-store-other", 150L, 150L, true);
@@ -287,8 +308,10 @@ public class KStreamKStreamJoinTest {
);
}
- @Test
- public void shouldThrowExceptionWhenOtherJoinStoreSetsRetainDuplicatesFalse() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldThrowExceptionWhenOtherJoinStoreSetsRetainDuplicatesFalse(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
//Case where otherJoinStore retain duplicates false
final WindowBytesStoreSupplier thisStoreSupplier = buildWindowBytesStoreSupplier("in-memory-join-store", 150L, 100L, true);
final WindowBytesStoreSupplier otherStoreSupplier = buildWindowBytesStoreSupplier("in-memory-join-store-other", 150L, 100L, false);
@@ -300,8 +323,10 @@ public class KStreamKStreamJoinTest {
);
}
- @Test
- public void shouldBuildJoinWithCustomStoresAndCorrectWindowSettings() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldBuildJoinWithCustomStoresAndCorrectWindowSettings(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
//Case where everything matches up
final StreamsBuilder builder = new StreamsBuilder();
final KStream<String, Integer> left = builder.stream("left", Consumed.with(Serdes.String(), Serdes.Integer()));
@@ -316,8 +341,10 @@ public class KStreamKStreamJoinTest {
builder.build();
}
- @Test
- public void shouldExceptionWhenJoinStoresDoNotHaveUniqueNames() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldExceptionWhenJoinStoresDoNotHaveUniqueNames(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final JoinWindows joinWindows = JoinWindows.ofTimeDifferenceAndGrace(ofMillis(100L), Duration.ofMillis(50L));
final StreamJoined<String, Integer, Integer> streamJoined = StreamJoined.with(Serdes.String(), Serdes.Integer(), Serdes.Integer());
final WindowBytesStoreSupplier thisStoreSupplier = buildWindowBytesStoreSupplier("in-memory-join-store", 150L, 100L, true);
@@ -330,8 +357,10 @@ public class KStreamKStreamJoinTest {
);
}
- @Test
- public void shouldJoinWithCustomStoreSuppliers() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldJoinWithCustomStoreSuppliers(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final JoinWindows joinWindows = JoinWindows.ofTimeDifferenceWithNoGrace(ofMillis(100L));
final WindowBytesStoreSupplier thisStoreSupplier = Stores.inMemoryWindowStore(
@@ -371,8 +400,10 @@ public class KStreamKStreamJoinTest {
}
}
- @Test
- public void shouldJoinWithDslStoreSuppliersIfNoStoreSupplied() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldJoinWithDslStoreSuppliersIfNoStoreSupplied(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
TrackingDslStoreSuppliers.NUM_CALLS.set(0);
final JoinWindows joinWindows = JoinWindows.ofTimeDifferenceWithNoGrace(ofMillis(100L));
@@ -407,8 +438,10 @@ public class KStreamKStreamJoinTest {
}
- @Test
- public void shouldJoinWithDslStoreSuppliersFromStreamsConfig() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldJoinWithDslStoreSuppliersFromStreamsConfig(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
TrackingDslStoreSuppliers.NUM_CALLS.set(0);
final JoinWindows joinWindows = JoinWindows.ofTimeDifferenceWithNoGrace(ofMillis(100L));
@@ -433,8 +466,10 @@ public class KStreamKStreamJoinTest {
}
}
- @Test
- public void shouldJoinWithNonTimestampedStore() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldCreateCorrectWindowStoreTypeBasedOnConfiguration(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final JoinWindows joinWindows = JoinWindows.ofTimeDifferenceWithNoGrace(ofMillis(100L));
final CapturingStoreSuppliers storeSuppliers = new CapturingStoreSuppliers();
@@ -443,12 +478,19 @@ public class KStreamKStreamJoinTest {
.withDslStoreSuppliers(storeSuppliers);
runJoin(streamJoined, joinWindows);
- assertThat("Expected stream joined to supply builders that create non-timestamped stores",
+ if (withHeaders) {
+ assertThat("Expected stream joined to supply builders that create headers stores",
+ WrappedStateStore.isHeadersAware(storeSuppliers.capture.get().get()));
+ } else {
+ assertThat("Expected stream joined to supply builders that create non-timestamped stores",
!WrappedStateStore.isTimestamped(storeSuppliers.capture.get().get()));
+ }
}
- @Test
- public void shouldThrottleEmitNonJoinedOuterRecordsEvenWhenClockDrift() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldThrottleEmitNonJoinedOuterRecordsEvenWhenClockDrift(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
/*
* This test is testing something internal to [[KStreamKStreamJoin]], so we had to setup low-level api manually.
*/
@@ -558,8 +600,10 @@ public class KStreamKStreamJoinTest {
}
}
- @Test
- public void testJoin() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testJoin(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final int[] expectedKeys = new int[] {0, 1, 2, 3};
@@ -680,8 +724,10 @@ public class KStreamKStreamJoinTest {
}
}
- @Test
- public void testOuterJoin() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testOuterJoin(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final int[] expectedKeys = new int[] {0, 1, 2, 3};
@@ -802,8 +848,10 @@ public class KStreamKStreamJoinTest {
}
}
- @Test
- public void testWindowing() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testWindowing(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final int[] expectedKeys = new int[] {0, 1, 2, 3};
@@ -1365,8 +1413,10 @@ public class KStreamKStreamJoinTest {
}
}
- @Test
- public void testAsymmetricWindowingAfter() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testAsymmetricWindowingAfter(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final int[] expectedKeys = new int[] {0, 1, 2, 3};
@@ -1633,8 +1683,10 @@ public class KStreamKStreamJoinTest {
}
}
- @Test
- public void testAsymmetricWindowingBefore() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testAsymmetricWindowingBefore(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final int[] expectedKeys = new int[] {0, 1, 2, 3};
@@ -2050,4 +2102,19 @@ public class KStreamKStreamJoinTest {
" <-- KSTREAM-MERGE-0000000011\n" +
" Sink: KSTREAM-SINK-0000000021 (topic: out-to)\n" +
" <-- KSTREAM-MERGE-0000000020\n\n";
+
+ /**
+ * Configures the DSL store format to use headers if enabled.
+ * This is a helper method to reduce boilerplate in parameterized tests that test both
+ * with and without headers mode.
+ *
+ * @param withHeaders Whether to enable headers mode
+ */
+ private void setDslStoreFormat(final boolean withHeaders) {
+ if (withHeaders) {
+ props.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
+ } else {
+ props.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_DEFAULT);
+ }
+ }
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamLeftJoinTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamLeftJoinTest.java
index 74ebaa0e844..ec9dc246170 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamLeftJoinTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamLeftJoinTest.java
@@ -21,6 +21,7 @@ import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.streams.KeyValueTimestamp;
import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.TestInputTopic;
import org.apache.kafka.streams.TopologyTestDriver;
import org.apache.kafka.streams.TopologyWrapper;
@@ -35,7 +36,8 @@ import org.apache.kafka.test.MockApiProcessorSupplier;
import org.apache.kafka.test.MockValueJoiner;
import org.apache.kafka.test.StreamsTestUtils;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.time.Duration;
import java.time.Instant;
@@ -56,8 +58,10 @@ public class KStreamKStreamLeftJoinTest {
private static final Properties PROPS = StreamsTestUtils.getStreamsConfig(Serdes.String(), Serdes.String());
@SuppressWarnings("deprecation") // old join semantics; can be removed when `JoinWindows.of()` is removed
- @Test
- public void testLeftJoinWithSpuriousResultFixDisabledOldApi() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testLeftJoinWithSpuriousResultFixDisabledOldApi(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final int[] expectedKeys = new int[] {0, 1, 2, 3};
@@ -116,8 +120,10 @@ public class KStreamKStreamLeftJoinTest {
}
@SuppressWarnings("deprecation") // old join semantics; can be removed when `JoinWindows.of()` is removed
- @Test
- public void testLeftJoinDuplicatesWithSpuriousResultFixDisabledOldApi() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testLeftJoinDuplicatesWithSpuriousResultFixDisabledOldApi(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final KStream<Integer, String> stream1;
@@ -158,8 +164,10 @@ public class KStreamKStreamLeftJoinTest {
}
}
- @Test
- public void testLeftJoinDuplicates() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testLeftJoinDuplicates(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final KStream<Integer, String> stream1;
@@ -214,8 +222,10 @@ public class KStreamKStreamLeftJoinTest {
}
}
- @Test
- public void testLeftExpiredNonJoinedRecordsAreEmittedByTheLeftProcessor() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testLeftExpiredNonJoinedRecordsAreEmittedByTheLeftProcessor(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final KStream<Integer, String> stream1;
@@ -269,8 +279,10 @@ public class KStreamKStreamLeftJoinTest {
}
}
- @Test
- public void testLeftExpiredNonJoinedRecordsAreEmittedByTheRightProcessor() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testLeftExpiredNonJoinedRecordsAreEmittedByTheRightProcessor(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final KStream<Integer, String> stream1;
@@ -324,8 +336,10 @@ public class KStreamKStreamLeftJoinTest {
}
}
- @Test
- public void testRightNonJoinedRecordsAreNeverEmittedByTheLeftProcessor() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testRightNonJoinedRecordsAreNeverEmittedByTheLeftProcessor(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final KStream<Integer, String> stream1;
@@ -376,8 +390,10 @@ public class KStreamKStreamLeftJoinTest {
}
}
- @Test
- public void testRightNonJoinedRecordsAreNeverEmittedByTheRightProcessor() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testRightNonJoinedRecordsAreNeverEmittedByTheRightProcessor(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final KStream<Integer, String> stream1;
@@ -428,8 +444,10 @@ public class KStreamKStreamLeftJoinTest {
}
}
- @Test
- public void testLeftJoinedRecordsWithZeroAfterAreEmitted() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testLeftJoinedRecordsWithZeroAfterAreEmitted(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final int[] expectedKeys = new int[] {0, 1, 2, 3};
@@ -606,8 +624,9 @@ public class KStreamKStreamLeftJoinTest {
}
}
- @Test
- public void testLeftJoinWithInMemoryCustomSuppliers() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testLeftJoinWithInMemoryCustomSuppliers(final boolean withHeaders) {
final JoinWindows joinWindows = JoinWindows.ofTimeDifferenceAndGrace(ofMillis(100L), ofMillis(0L));
final WindowBytesStoreSupplier thisStoreSupplier = Stores.inMemoryWindowStore("in-memory-join-store",
@@ -620,19 +639,22 @@ public class KStreamKStreamLeftJoinTest {
final StreamJoined<Integer, String, String> streamJoined = StreamJoined.with(Serdes.Integer(), Serdes.String(), Serdes.String());
- runLeftJoin(streamJoined.withThisStoreSupplier(thisStoreSupplier).withOtherStoreSupplier(otherStoreSupplier), joinWindows);
+ runLeftJoin(streamJoined.withThisStoreSupplier(thisStoreSupplier).withOtherStoreSupplier(otherStoreSupplier), joinWindows, withHeaders);
}
- @Test
- public void testLeftJoinWithDefaultSuppliers() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testLeftJoinWithDefaultSuppliers(final boolean withHeaders) {
final JoinWindows joinWindows = JoinWindows.ofTimeDifferenceWithNoGrace(ofMillis(100L));
final StreamJoined<Integer, String, String> streamJoined = StreamJoined.with(Serdes.Integer(), Serdes.String(), Serdes.String());
- runLeftJoin(streamJoined, joinWindows);
+ runLeftJoin(streamJoined, joinWindows, withHeaders);
}
public void runLeftJoin(final StreamJoined<Integer, String, String> streamJoined,
- final JoinWindows joinWindows) {
+ final JoinWindows joinWindows,
+ final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final int[] expectedKeys = new int[] {0, 1, 2, 3};
@@ -744,8 +766,10 @@ public class KStreamKStreamLeftJoinTest {
}
}
- @Test
- public void testOrdering() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testOrdering(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final KStream<Integer, String> stream1;
@@ -795,8 +819,10 @@ public class KStreamKStreamLeftJoinTest {
}
}
- @Test
- public void testGracePeriod() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testGracePeriod(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final int[] expectedKeys = new int[] {0, 1, 2, 3};
@@ -867,8 +893,10 @@ public class KStreamKStreamLeftJoinTest {
}
}
- @Test
- public void testWindowing() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testWindowing(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final int[] expectedKeys = new int[] {0, 1, 2, 3};
@@ -928,8 +956,10 @@ public class KStreamKStreamLeftJoinTest {
}
}
- @Test
- public void shouldNotEmitLeftJoinResultForAsymmetricWindow() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotEmitLeftJoinResultForAsymmetricWindow(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final int[] expectedKeys = new int[] {0, 1, 2, 3};
@@ -1311,4 +1341,19 @@ public class KStreamKStreamLeftJoinTest {
inputTopic1.pipeInput(0, "dummy", time + 300L);
processor.checkAndClearProcessResult(new KeyValueTimestamp<>(0, "dummy+null", 1203L));
}
+
+ /**
+ * Configures the DSL store format to use headers if enabled.
+ * This is a helper method to reduce boilerplate in parameterized tests that test both
+ * with and without headers mode.
+ *
+ * @param withHeaders Whether to enable headers mode
+ */
+ private void setDslStoreFormat(final boolean withHeaders) {
+ if (withHeaders) {
+ PROPS.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
+ } else {
+ PROPS.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_DEFAULT);
+ }
+ }
}
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 5fc4e0a418b..a8eb3c61302 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
@@ -25,6 +25,7 @@ import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.streams.KeyValueTimestamp;
import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.TestInputTopic;
import org.apache.kafka.streams.TopologyTestDriver;
import org.apache.kafka.streams.TopologyWrapper;
@@ -45,7 +46,8 @@ import org.apache.kafka.test.MockApiProcessorSupplier;
import org.apache.kafka.test.MockValueJoiner;
import org.apache.kafka.test.StreamsTestUtils;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.time.Duration;
import java.time.Instant;
@@ -66,9 +68,12 @@ public class KStreamKStreamOuterJoinTest {
private final Consumed<Integer, Long> consumed2 = Consumed.with(Serdes.Integer(), Serdes.Long());
private static final Properties PROPS = StreamsTestUtils.getStreamsConfig(Serdes.String(), Serdes.String());
+
@SuppressWarnings("deprecation") // old join semantics; can be removed when `JoinWindows.of()` is removed
- @Test
- public void testOuterJoinDuplicatesWithFixDisabledOldApi() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testOuterJoinDuplicatesWithFixDisabledOldApi(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final KStream<Integer, String> stream1;
@@ -111,8 +116,10 @@ public class KStreamKStreamOuterJoinTest {
}
}
- @Test
- public void testOuterJoinDuplicates() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testOuterJoinDuplicates(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final KStream<Integer, String> stream1;
@@ -177,8 +184,10 @@ public class KStreamKStreamOuterJoinTest {
}
}
- @Test
- public void testLeftExpiredNonJoinedRecordsAreEmittedByTheLeftProcessor() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testLeftExpiredNonJoinedRecordsAreEmittedByTheLeftProcessor(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final KStream<Integer, String> stream1;
@@ -232,8 +241,10 @@ public class KStreamKStreamOuterJoinTest {
}
}
- @Test
- public void testLeftExpiredNonJoinedRecordsAreEmittedByTheRightProcessor() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testLeftExpiredNonJoinedRecordsAreEmittedByTheRightProcessor(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final KStream<Integer, String> stream1;
@@ -287,8 +298,10 @@ public class KStreamKStreamOuterJoinTest {
}
}
- @Test
- public void testRightExpiredNonJoinedRecordsAreEmittedByTheLeftProcessor() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testRightExpiredNonJoinedRecordsAreEmittedByTheLeftProcessor(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final KStream<Integer, String> stream1;
@@ -342,8 +355,10 @@ public class KStreamKStreamOuterJoinTest {
}
}
- @Test
- public void testRightExpiredNonJoinedRecordsAreEmittedByTheRightProcessor() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testRightExpiredNonJoinedRecordsAreEmittedByTheRightProcessor(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final KStream<Integer, String> stream1;
@@ -397,8 +412,10 @@ public class KStreamKStreamOuterJoinTest {
}
}
- @Test
- public void testOrdering() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testOrdering(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final KStream<Integer, String> stream1;
@@ -447,8 +464,10 @@ public class KStreamKStreamOuterJoinTest {
}
}
- @Test
- public void testGracePeriod() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testGracePeriod(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final KStream<Integer, String> stream1;
@@ -513,8 +532,10 @@ public class KStreamKStreamOuterJoinTest {
}
}
- @Test
- public void testEmitAllNonJoinedResultsForAsymmetricWindow() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testEmitAllNonJoinedResultsForAsymmetricWindow(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final KStream<Integer, String> stream1;
@@ -608,8 +629,9 @@ public class KStreamKStreamOuterJoinTest {
}
}
- @Test
- public void testOuterJoinWithInMemoryCustomSuppliers() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testOuterJoinWithInMemoryCustomSuppliers(final boolean withHeaders) {
final JoinWindows joinWindows = JoinWindows.ofTimeDifferenceWithNoGrace(ofMillis(100L));
final WindowBytesStoreSupplier thisStoreSupplier = Stores.inMemoryWindowStore(
@@ -628,19 +650,22 @@ public class KStreamKStreamOuterJoinTest {
final StreamJoined<Integer, String, Long> streamJoined = StreamJoined.with(Serdes.Integer(), Serdes.String(), Serdes.Long());
- runOuterJoin(streamJoined.withThisStoreSupplier(thisStoreSupplier).withOtherStoreSupplier(otherStoreSupplier), joinWindows);
+ runOuterJoin(streamJoined.withThisStoreSupplier(thisStoreSupplier).withOtherStoreSupplier(otherStoreSupplier), joinWindows, withHeaders);
}
- @Test
- public void testOuterJoinWithDefaultSuppliers() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testOuterJoinWithDefaultSuppliers(final boolean withHeaders) {
final JoinWindows joinWindows = JoinWindows.ofTimeDifferenceWithNoGrace(ofMillis(100L));
final StreamJoined<Integer, String, Long> streamJoined = StreamJoined.with(Serdes.Integer(), Serdes.String(), Serdes.Long());
- runOuterJoin(streamJoined, joinWindows);
+ runOuterJoin(streamJoined, joinWindows, withHeaders);
}
public void runOuterJoin(final StreamJoined<Integer, String, Long> streamJoined,
- final JoinWindows joinWindows) {
+ final JoinWindows joinWindows,
+ final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final int[] expectedKeys = new int[] {0, 1, 2, 3};
@@ -753,8 +778,10 @@ public class KStreamKStreamOuterJoinTest {
}
}
- @Test
- public void testWindowing() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testWindowing(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final int[] expectedKeys = new int[]{0, 1, 2, 3};
@@ -815,8 +842,10 @@ public class KStreamKStreamOuterJoinTest {
}
}
- @Test
- public void testShouldNotEmitLeftJoinResultForAsymmetricBeforeWindow() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testShouldNotEmitLeftJoinResultForAsymmetricBeforeWindow(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final int[] expectedKeys = new int[] {0, 1, 2, 3};
@@ -902,8 +931,10 @@ public class KStreamKStreamOuterJoinTest {
}
}
- @Test
- public void testShouldNotEmitLeftJoinResultForAsymmetricAfterWindow() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testShouldNotEmitLeftJoinResultForAsymmetricAfterWindow(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final int[] expectedKeys = new int[] {0, 1, 2, 3};
@@ -994,8 +1025,10 @@ public class KStreamKStreamOuterJoinTest {
* NOTE: Header forwarding is undefined behavior, but we still want to understand the
* behavior so that we can make decisions about defining it in the future.
*/
- @Test
- public void testShouldForwardCurrentHeaders() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testShouldForwardCurrentHeaders(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final KStream<Integer, String> stream1;
@@ -1419,8 +1452,10 @@ public class KStreamKStreamOuterJoinTest {
}
}
- @Test
- public void testShouldJoinWithNonTimestampedStore() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testShouldJoinWithNonTimestampedStore(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final CapturingStoreSuppliers suppliers = new CapturingStoreSuppliers();
final StreamJoined<Integer, String, String> streamJoined =
StreamJoined.with(Serdes.Integer(), Serdes.String(), Serdes.String())
@@ -1449,4 +1484,19 @@ public class KStreamKStreamOuterJoinTest {
!WrappedStateStore.isTimestamped(suppliers.capture.get().get()));
}
}
+
+ /**
+ * Configures the DSL store format to use headers if enabled.
+ * This is a helper method to reduce boilerplate in parameterized tests that test both
+ * with and without headers mode.
+ *
+ * @param withHeaders Whether to enable headers mode
+ */
+ private void setDslStoreFormat(final boolean withHeaders) {
+ if (withHeaders) {
+ PROPS.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
+ } else {
+ PROPS.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_DEFAULT);
+ }
+ }
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamSelfJoinTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamSelfJoinTest.java
index c3e8341cd2e..f609dd514e8 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamSelfJoinTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamSelfJoinTest.java
@@ -33,7 +33,8 @@ import org.apache.kafka.test.MockApiProcessor;
import org.apache.kafka.test.MockApiProcessorSupplier;
import org.apache.kafka.test.StreamsTestUtils;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.util.List;
import java.util.Properties;
@@ -44,12 +45,14 @@ import static java.time.Duration.ofSeconds;
public class KStreamKStreamSelfJoinTest {
private final String topic1 = "topic1";
private final String topic2 = "topic2";
- private final Properties props = StreamsTestUtils.getStreamsConfig(Serdes.String(), Serdes.String());
+ private Properties props = StreamsTestUtils.getStreamsConfig(Serdes.String(), Serdes.String());
- @Test
- public void shouldMatchInnerJoinWithSelfJoinWithSingleStream() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldMatchInnerJoinWithSelfJoinWithSingleStream(final boolean withHeaders) {
props.setProperty(StreamsConfig.BUILT_IN_METRICS_VERSION_CONFIG, StreamsConfig.METRICS_LATEST);
props.put(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG, StreamsConfig.OPTIMIZE);
+ setDslStoreFormat(withHeaders);
final ValueJoiner<String, String, String> valueJoiner = (v, v2) -> v + v2;
final List<KeyValueTimestamp<String, String>> expected;
final StreamsBuilder streamsBuilder = new StreamsBuilder();
@@ -113,10 +116,12 @@ public class KStreamKStreamSelfJoinTest {
}
- @Test
- public void shouldMatchInnerJoinWithSelfJoinWithTwoStreams() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldMatchInnerJoinWithSelfJoinWithTwoStreams(final boolean withHeaders) {
props.setProperty(StreamsConfig.BUILT_IN_METRICS_VERSION_CONFIG, StreamsConfig.METRICS_LATEST);
props.put(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG, StreamsConfig.OPTIMIZE);
+ setDslStoreFormat(withHeaders);
final ValueJoiner<String, String, String> valueJoiner = (v, v2) -> v + v2;
final List<KeyValueTimestamp<String, String>> expected;
final StreamsBuilder streamsBuilder = new StreamsBuilder();
@@ -183,10 +188,12 @@ public class KStreamKStreamSelfJoinTest {
}
}
- @Test
- public void shouldMatchInnerJoinWithSelfJoinDifferentBeforeAfterWindows() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldMatchInnerJoinWithSelfJoinDifferentBeforeAfterWindows(final boolean withHeaders) {
props.setProperty(StreamsConfig.BUILT_IN_METRICS_VERSION_CONFIG, StreamsConfig.METRICS_LATEST);
props.put(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG, StreamsConfig.OPTIMIZE);
+ setDslStoreFormat(withHeaders);
final ValueJoiner<String, String, String> valueJoiner = (v, v2) -> v + v2;
final List<KeyValueTimestamp<String, String>> expected;
final StreamsBuilder streamsBuilder = new StreamsBuilder();
@@ -258,10 +265,12 @@ public class KStreamKStreamSelfJoinTest {
}
}
- @Test
- public void shouldMatchInnerJoinWithSelfJoinOutOfOrderMessages() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldMatchInnerJoinWithSelfJoinOutOfOrderMessages(final boolean withHeaders) {
props.setProperty(StreamsConfig.BUILT_IN_METRICS_VERSION_CONFIG, StreamsConfig.METRICS_LATEST);
props.put(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG, StreamsConfig.OPTIMIZE);
+ setDslStoreFormat(withHeaders);
final ValueJoiner<String, String, String> valueJoiner = (v, v2) -> v + v2;
final List<KeyValueTimestamp<String, String>> expected;
final StreamsBuilder streamsBuilder = new StreamsBuilder();
@@ -336,4 +345,19 @@ public class KStreamKStreamSelfJoinTest {
processor.checkAndClearProcessResult(expected.toArray(new KeyValueTimestamp[0]));
}
}
+
+ /**
+ * Configures the DSL store format to use headers if enabled.
+ * This is a helper method to reduce boilerplate in parameterized tests that test both
+ * with and without headers mode.
+ *
+ * @param withHeaders Whether to enable headers mode
+ */
+ private void setDslStoreFormat(final boolean withHeaders) {
+ if (withHeaders) {
+ props.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
+ } else {
+ props.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_DEFAULT);
+ }
+ }
}
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 19a8b5e5205..222e3e9914e 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
@@ -41,8 +41,8 @@ import org.apache.kafka.test.MockValueJoiner;
import org.apache.kafka.test.StreamsTestUtils;
import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.time.Duration;
import java.time.Instant;
@@ -73,8 +73,7 @@ public class KStreamKTableJoinTest {
private StreamsBuilder builder;
private final MockApiProcessorSupplier<Integer, String, Void, Void> supplier = new MockApiProcessorSupplier<>();
- @BeforeEach
- public void setUp() {
+ public void setUp(final boolean withHeaders) {
builder = new StreamsBuilder();
final KStream<Integer, String> stream;
@@ -85,6 +84,9 @@ public class KStreamKTableJoinTest {
table = builder.table(tableTopic, consumed);
stream.join(table, MockValueJoiner.TOSTRING_JOINER).process(supplier);
final Properties props = StreamsTestUtils.getStreamsConfig(Serdes.Integer(), Serdes.String());
+ if (withHeaders) {
+ props.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
+ }
driver = new TopologyTestDriver(builder.build(), props);
inputStreamTopic = driver.createInputTopic(streamTopic, new IntegerSerializer(), new StringSerializer(), Instant.ofEpochMilli(0L), Duration.ZERO);
inputTableTopic = driver.createInputTopic(tableTopic, new IntegerSerializer(), new StringSerializer(), Instant.ofEpochMilli(0L), Duration.ZERO);
@@ -94,7 +96,9 @@ public class KStreamKTableJoinTest {
@AfterEach
public void cleanup() {
- driver.close();
+ if (driver != null) {
+ driver.close();
+ }
}
private void pushToStream(final int messageCount, final String valuePrefix) {
@@ -151,11 +155,13 @@ public class KStreamKTableJoinTest {
processor = supplier.theCapturedProcessor();
}
- @Test
- public void shouldFailIfTableIsNotVersioned() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldFailIfTableIsNotVersioned(final boolean withHeaders) {
final StreamsBuilder builder = new StreamsBuilder();
final Properties props = new Properties();
props.put(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG, StreamsConfig.NO_OPTIMIZATION);
+ maybeSetDslStoreFormatHeaders(props, withHeaders);
final KStream<String, String> streamA = builder.stream("topic", Consumed.with(Serdes.String(), Serdes.String()));
final KTable<String, String> tableB = builder.table("topic2", Consumed.with(Serdes.String(), Serdes.String()));
@@ -167,11 +173,13 @@ public class KStreamKTableJoinTest {
);
}
- @Test
- public void shouldFailIfTableIsNotVersionedButMaterializationIsInherited() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldFailIfTableIsNotVersionedButMaterializationIsInherited(final boolean withHeaders) {
final StreamsBuilder builder = new StreamsBuilder();
final Properties props = new Properties();
props.put(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG, StreamsConfig.NO_OPTIMIZATION);
+ maybeSetDslStoreFormatHeaders(props, withHeaders);
final KStream<String, String> streamA = builder.stream("topic", Consumed.with(Serdes.String(), Serdes.String()));
final KTable<String, String> source = builder.table("topic2", Consumed.with(Serdes.String(), Serdes.String()),
Materialized.as(Stores.inMemoryKeyValueStore("tableB")));
@@ -187,11 +195,13 @@ public class KStreamKTableJoinTest {
);
}
- @Test
- public void shouldNotFailIfTableIsVersionedButMaterializationIsInherited() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotFailIfTableIsVersionedButMaterializationIsInherited(final boolean withHeaders) {
final StreamsBuilder builder = new StreamsBuilder();
final Properties props = new Properties();
props.put(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG, StreamsConfig.NO_OPTIMIZATION);
+ maybeSetDslStoreFormatHeaders(props, withHeaders);
final KStream<String, String> streamA = builder.stream("topic", Consumed.with(Serdes.String(), Serdes.String()));
final KTable<String, String> source = builder.table("topic2", Consumed.with(Serdes.String(), Serdes.String()),
Materialized.as(Stores.persistentVersionedKeyValueStore("tableB", Duration.ofMinutes(5))));
@@ -204,11 +214,13 @@ public class KStreamKTableJoinTest {
builder.build();
}
- @Test
- public void shouldFailIfGracePeriodIsLongerThanHistoryRetention() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldFailIfGracePeriodIsLongerThanHistoryRetention(final boolean withHeaders) {
final StreamsBuilder builder = new StreamsBuilder();
final Properties props = new Properties();
props.put(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG, StreamsConfig.NO_OPTIMIZATION);
+ maybeSetDslStoreFormatHeaders(props, withHeaders);
final KStream<String, String> streamA = builder.stream("topic", Consumed.with(Serdes.String(), Serdes.String()));
final KTable<String, String> tableB = builder.table("topic2", Consumed.with(Serdes.String(), Serdes.String()),
Materialized.as(Stores.persistentVersionedKeyValueStore("tableB", Duration.ofMinutes(5))));
@@ -219,11 +231,13 @@ public class KStreamKTableJoinTest {
assertThat(exception.getMessage(), is("History retention must be at least grace period."));
}
- @Test
- public void shouldFailIfGracePeriodIsLongerThanHistoryRetentionAndInheritedStore() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldFailIfGracePeriodIsLongerThanHistoryRetentionAndInheritedStore(final boolean withHeaders) {
final StreamsBuilder builder = new StreamsBuilder();
final Properties props = new Properties();
props.put(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG, StreamsConfig.NO_OPTIMIZATION);
+ maybeSetDslStoreFormatHeaders(props, withHeaders);
final KStream<String, String> streamA = builder.stream("topic", Consumed.with(Serdes.String(), Serdes.String()));
final KTable<String, String> source = builder.table("topic2", Consumed.with(Serdes.String(), Serdes.String()),
Materialized.as(Stores.persistentVersionedKeyValueStore("V-grace", Duration.ofMinutes(0))));
@@ -237,8 +251,10 @@ public class KStreamKTableJoinTest {
}
- @Test
- public void shouldDelayJoinByGracePeriod() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldDelayJoinByGracePeriod(final boolean withHeaders) {
+ setUp(withHeaders);
makeJoin(Duration.ofMillis(2));
// push four items to the table. this should not produce any item.
@@ -275,8 +291,10 @@ public class KStreamKTableJoinTest {
processor.checkAndClearProcessResult(EMPTY);
}
- @Test
- public void shouldHandleLateJoinsWithGracePeriod() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldHandleLateJoinsWithGracePeriod(final boolean withHeaders) {
+ setUp(withHeaders);
makeJoin(Duration.ofMillis(2));
// push four items to the table. this should not produce any item.
@@ -295,11 +313,13 @@ public class KStreamKTableJoinTest {
new KeyValueTimestamp<>(0, "X0+Y0", 0));
}
- @Test
- public void shouldReuseRepartitionTopicWithGeneratedName() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldReuseRepartitionTopicWithGeneratedName(final boolean withHeaders) {
final StreamsBuilder builder = new StreamsBuilder();
final Properties props = new Properties();
props.put(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG, StreamsConfig.NO_OPTIMIZATION);
+ maybeSetDslStoreFormatHeaders(props, withHeaders);
final KStream<String, String> streamA = builder.stream("topic", Consumed.with(Serdes.String(), Serdes.String()));
final KTable<String, String> tableB = builder.table("topic2", Consumed.with(Serdes.String(), Serdes.String()));
final KTable<String, String> tableC = builder.table("topic3", Consumed.with(Serdes.String(), Serdes.String()));
@@ -310,11 +330,13 @@ public class KStreamKTableJoinTest {
assertEquals(expectedTopologyWithGeneratedRepartitionTopicNames, topology.describe().toString());
}
- @Test
- public void shouldCreateRepartitionTopicsWithUserProvidedName() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldCreateRepartitionTopicsWithUserProvidedName(final boolean withHeaders) {
final StreamsBuilder builder = new StreamsBuilder();
final Properties props = new Properties();
props.put(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG, StreamsConfig.NO_OPTIMIZATION);
+ maybeSetDslStoreFormatHeaders(props, withHeaders);
final KStream<String, String> streamA = builder.stream("topic", Consumed.with(Serdes.String(), Serdes.String()));
final KTable<String, String> tableB = builder.table("topic2", Consumed.with(Serdes.String(), Serdes.String()));
final KTable<String, String> tableC = builder.table("topic3", Consumed.with(Serdes.String(), Serdes.String()));
@@ -327,8 +349,10 @@ public class KStreamKTableJoinTest {
assertEquals(expectedTopologyWithUserProvidedRepartitionTopicNames, topology.describe().toString());
}
- @Test
- public void shouldRequireCopartitionedStreams() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldRequireCopartitionedStreams(final boolean withHeaders) {
+ setUp(withHeaders);
final Collection<Set<String>> copartitionGroups =
TopologyWrapper.getInternalTopologyBuilder(builder.build()).copartitionGroups();
@@ -336,15 +360,19 @@ public class KStreamKTableJoinTest {
assertEquals(Set.of(streamTopic, tableTopic), copartitionGroups.iterator().next());
}
- @Test
- public void shouldNotJoinWithEmptyTableOnStreamUpdates() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotJoinWithEmptyTableOnStreamUpdates(final boolean withHeaders) {
+ setUp(withHeaders);
// push two items to the primary stream. the table is empty
pushToStream(2, "X");
processor.checkAndClearProcessResult(EMPTY);
}
- @Test
- public void shouldNotJoinOnTableUpdates() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotJoinOnTableUpdates(final boolean withHeaders) {
+ setUp(withHeaders);
// push two items to the primary stream. the table is empty
pushToStream(2, "X");
processor.checkAndClearProcessResult(EMPTY);
@@ -378,8 +406,10 @@ public class KStreamKTableJoinTest {
processor.checkAndClearProcessResult(EMPTY);
}
- @Test
- public void shouldJoinOnlyIfMatchFoundOnStreamUpdates() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldJoinOnlyIfMatchFoundOnStreamUpdates(final boolean withHeaders) {
+ setUp(withHeaders);
// push two items to the table. this should not produce any item.
pushToTable(2, "Y");
processor.checkAndClearProcessResult(EMPTY);
@@ -392,8 +422,10 @@ public class KStreamKTableJoinTest {
);
}
- @Test
- public void shouldClearTableEntryOnNullValueUpdates() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldClearTableEntryOnNullValueUpdates(final boolean withHeaders) {
+ setUp(withHeaders);
// push all four items to the table. this should not produce any item.
pushToTable(4, "Y");
processor.checkAndClearProcessResult(EMPTY);
@@ -419,8 +451,10 @@ public class KStreamKTableJoinTest {
);
}
- @Test
- public void shouldLogAndMeterWhenSkippingNullLeftKey() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldLogAndMeterWhenSkippingNullLeftKey(final boolean withHeaders) {
+ setUp(withHeaders);
try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister(KStreamKTableJoin.class)) {
final TestInputTopic<Integer, String> inputTopic =
driver.createInputTopic(streamTopic, new IntegerSerializer(), new StringSerializer());
@@ -448,8 +482,10 @@ public class KStreamKTableJoinTest {
);
}
- @Test
- public void shouldLogAndMeterWhenSkippingNullLeftValue() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldLogAndMeterWhenSkippingNullLeftValue(final boolean withHeaders) {
+ setUp(withHeaders);
try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister(KStreamKTableJoin.class)) {
final TestInputTopic<Integer, String> inputTopic =
driver.createInputTopic(streamTopic, new IntegerSerializer(), new StringSerializer());
@@ -564,4 +600,10 @@ public class KStreamKTableJoinTest {
+ " Processor: KTABLE-SOURCE-0000000006 (stores: [topic3-STATE-STORE-0000000004])\n"
+ " --> none\n"
+ " <-- KSTREAM-SOURCE-0000000005\n\n";
+
+ private static void maybeSetDslStoreFormatHeaders(final Properties streamsConfig, final boolean withHeaders) {
+ if (withHeaders) {
+ streamsConfig.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
+ }
+ }
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKTableLeftJoinTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKTableLeftJoinTest.java
index f78432b95e7..a64985a019f 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKTableLeftJoinTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKTableLeftJoinTest.java
@@ -23,6 +23,7 @@ import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.common.utils.LogCaptureAppender;
import org.apache.kafka.streams.KeyValueTimestamp;
import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.TestInputTopic;
import org.apache.kafka.streams.TopologyTestDriver;
import org.apache.kafka.streams.TopologyWrapper;
@@ -35,8 +36,8 @@ import org.apache.kafka.test.MockValueJoiner;
import org.apache.kafka.test.StreamsTestUtils;
import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.time.Duration;
import java.time.Instant;
@@ -63,8 +64,8 @@ public class KStreamKTableLeftJoinTest {
private MockApiProcessor<Integer, String, Void, Void> processor;
private StreamsBuilder builder;
- @BeforeEach
- public void setUp() {
+
+ public void setUp(final boolean withHeaders) {
builder = new StreamsBuilder();
final KStream<Integer, String> stream;
@@ -77,6 +78,9 @@ public class KStreamKTableLeftJoinTest {
stream.leftJoin(table, MockValueJoiner.TOSTRING_JOINER).process(supplier);
final Properties props = StreamsTestUtils.getStreamsConfig(Serdes.Integer(), Serdes.String());
+ if (withHeaders) {
+ props.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
+ }
driver = new TopologyTestDriver(builder.build(), props);
inputStreamTopic = driver.createInputTopic(streamTopic, new IntegerSerializer(), new StringSerializer(), Instant.ofEpochMilli(0L), Duration.ZERO);
inputTableTopic = driver.createInputTopic(tableTopic, new IntegerSerializer(), new StringSerializer(), Instant.ofEpochMilli(0L), Duration.ZERO);
@@ -111,8 +115,10 @@ public class KStreamKTableLeftJoinTest {
}
}
- @Test
- public void shouldRequireCopartitionedStreams() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldRequireCopartitionedStreams(final boolean withHeaders) {
+ setUp(withHeaders);
final Collection<Set<String>> copartitionGroups =
TopologyWrapper.getInternalTopologyBuilder(builder.build()).copartitionGroups();
@@ -120,16 +126,20 @@ public class KStreamKTableLeftJoinTest {
assertEquals(Set.of(streamTopic, tableTopic), copartitionGroups.iterator().next());
}
- @Test
- public void shouldJoinWithEmptyTableOnStreamUpdates() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldJoinWithEmptyTableOnStreamUpdates(final boolean withHeaders) {
+ setUp(withHeaders);
// push two items to the primary stream. the table is empty
pushToStream(2, "X");
processor.checkAndClearProcessResult(new KeyValueTimestamp<>(0, "X0+null", 0),
new KeyValueTimestamp<>(1, "X1+null", 1));
}
- @Test
- public void shouldNotJoinOnTableUpdates() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotJoinOnTableUpdates(final boolean withHeaders) {
+ setUp(withHeaders);
// push two items to the primary stream. the table is empty
pushToStream(2, "X");
processor.checkAndClearProcessResult(new KeyValueTimestamp<>(0, "X0+null", 0),
@@ -162,8 +172,10 @@ public class KStreamKTableLeftJoinTest {
processor.checkAndClearProcessResult(EMPTY);
}
- @Test
- public void shouldJoinRegardlessIfMatchFoundOnStreamUpdates() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldJoinRegardlessIfMatchFoundOnStreamUpdates(final boolean withHeaders) {
+ setUp(withHeaders);
// push two items to the table. this should not produce any item.
pushToTable(2, "Y");
processor.checkAndClearProcessResult(EMPTY);
@@ -177,8 +189,10 @@ public class KStreamKTableLeftJoinTest {
}
- @Test
- public void shouldClearTableEntryOnNullValueUpdates() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldClearTableEntryOnNullValueUpdates(final boolean withHeaders) {
+ setUp(withHeaders);
// push all four items to the table. this should not produce any item.
pushToTable(4, "Y");
processor.checkAndClearProcessResult(EMPTY);
@@ -202,8 +216,10 @@ public class KStreamKTableLeftJoinTest {
new KeyValueTimestamp<>(3, "XX3+Y3", 3));
}
- @Test
- public void shouldNotDropLeftNullKey() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotDropLeftNullKey(final boolean withHeaders) {
+ setUp(withHeaders);
// push all four items to the table. this should not produce any item.
pushToTable(1, "Y");
processor.checkAndClearProcessResult(EMPTY);
@@ -234,8 +250,10 @@ public class KStreamKTableLeftJoinTest {
);
}
- @Test
- public void shouldLogAndMeterWhenSkippingNullLeftValue() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldLogAndMeterWhenSkippingNullLeftValue(final boolean withHeaders) {
+ setUp(withHeaders);
try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister(KStreamKTableJoin.class)) {
final TestInputTopic<Integer, String> inputTopic =
driver.createInputTopic(streamTopic, new IntegerSerializer(), new StringSerializer());
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 a43b5689106..f00f1806161 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
@@ -53,7 +53,8 @@ import org.apache.kafka.test.TestUtils;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.EnumSource;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
import java.util.ArrayList;
import java.util.Arrays;
@@ -61,6 +62,7 @@ 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;
import static org.apache.kafka.common.utils.Utils.mkEntry;
@@ -93,17 +95,29 @@ public class KStreamSessionWindowAggregateProcessorTest {
private KStreamSessionWindowAggregate<String, String, Long> sessionAggregator;
private Processor<String, String, Windowed<String>, Change<Long>> processor;
private SessionStoreWithHeaders<String, Long> sessionStore;
-
+
public EmitStrategy.StrategyType type;
private EmitStrategy emitStrategy;
private boolean emitFinal;
- private void setup(final EmitStrategy.StrategyType inputType, final boolean enableCaching) {
+ public static Stream<Arguments> emitStrategyAndHeadersMatrix() {
+ return Stream.of(
+ Arguments.of(EmitStrategy.StrategyType.ON_WINDOW_UPDATE, true),
+ Arguments.of(EmitStrategy.StrategyType.ON_WINDOW_UPDATE, false),
+ Arguments.of(EmitStrategy.StrategyType.ON_WINDOW_UPDATE, true),
+ Arguments.of(EmitStrategy.StrategyType.ON_WINDOW_UPDATE, false)
+ );
+ }
+
+ private void setup(final EmitStrategy.StrategyType inputType, final boolean enableCaching, final boolean withHeaders) {
type = inputType;
// Always process
final Properties prop = StreamsTestUtils.getStreamsConfig();
prop.put(StreamsConfig.InternalConfig.EMIT_INTERVAL_MS_KSTREAMS_WINDOWED_AGGREGATION, 0);
+ if (withHeaders) {
+ prop.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
+ }
final StreamsConfig config = new StreamsConfig(prop);
mockContext = new InternalMockProcessorContext<>(
@@ -175,9 +189,9 @@ public class KStreamSessionWindowAggregateProcessorTest {
}
@ParameterizedTest
- @EnumSource(EmitStrategy.StrategyType.class)
- public void shouldCreateSingleSessionWhenWithinGap(final EmitStrategy.StrategyType inputType) {
- setup(inputType, true);
+ @MethodSource("emitStrategyAndHeadersMatrix")
+ public void shouldCreateSingleSessionWhenWithinGap(final EmitStrategy.StrategyType inputType, final boolean withHeaders) {
+ setup(inputType, true, withHeaders);
processor.process(new Record<>("john", "first", 0L));
processor.process(new Record<>("john", "second", 500L));
@@ -189,9 +203,9 @@ public class KStreamSessionWindowAggregateProcessorTest {
}
@ParameterizedTest
- @EnumSource(EmitStrategy.StrategyType.class)
- public void shouldMergeSessions(final EmitStrategy.StrategyType inputType) {
- setup(inputType, true);
+ @MethodSource("emitStrategyAndHeadersMatrix")
+ public void shouldMergeSessions(final EmitStrategy.StrategyType inputType, final boolean withHeaders) {
+ setup(inputType, true, withHeaders);
final String sessionId = "mel";
processor.process(new Record<>(sessionId, "first", 0L));
try (final KeyValueIterator<Windowed<String>, AggregationWithHeaders<Long>> iterator = sessionStore.findSessions(sessionId, 0, 0)) {
@@ -220,9 +234,9 @@ public class KStreamSessionWindowAggregateProcessorTest {
}
@ParameterizedTest
- @EnumSource(EmitStrategy.StrategyType.class)
- public void shouldUpdateSessionIfTheSameTime(final EmitStrategy.StrategyType inputType) {
- setup(inputType, true);
+ @MethodSource("emitStrategyAndHeadersMatrix")
+ public void shouldUpdateSessionIfTheSameTime(final EmitStrategy.StrategyType inputType, final boolean withHeaders) {
+ setup(inputType, true, withHeaders);
processor.process(new Record<>("mel", "first", 0L));
processor.process(new Record<>("mel", "second", 0L));
try (final KeyValueIterator<Windowed<String>, AggregationWithHeaders<Long>> iterator =
@@ -233,9 +247,10 @@ public class KStreamSessionWindowAggregateProcessorTest {
}
@ParameterizedTest
- @EnumSource(EmitStrategy.StrategyType.class)
- public void shouldHaveMultipleSessionsForSameIdWhenTimestampApartBySessionGap(final EmitStrategy.StrategyType inputType) {
- setup(inputType, true);
+ @MethodSource("emitStrategyAndHeadersMatrix")
+ public void shouldHaveMultipleSessionsForSameIdWhenTimestampApartBySessionGap(final EmitStrategy.StrategyType inputType, final boolean withHeaders) {
+ setup(inputType, true, withHeaders);
+
final String sessionId = "mel";
long now = 0;
processor.process(new Record<>(sessionId, "first", now));
@@ -285,9 +300,9 @@ public class KStreamSessionWindowAggregateProcessorTest {
}
@ParameterizedTest
- @EnumSource(EmitStrategy.StrategyType.class)
- public void shouldRemoveMergedSessionsFromStateStore(final EmitStrategy.StrategyType inputType) {
- setup(inputType, true);
+ @MethodSource("emitStrategyAndHeadersMatrix")
+ public void shouldRemoveMergedSessionsFromStateStore(final EmitStrategy.StrategyType inputType, final boolean withHeaders) {
+ setup(inputType, true, withHeaders);
processor.process(new Record<>("a", "1", 0L));
// first ensure it is in the store
@@ -312,9 +327,10 @@ public class KStreamSessionWindowAggregateProcessorTest {
}
@ParameterizedTest
- @EnumSource(EmitStrategy.StrategyType.class)
- public void shouldHandleMultipleSessionsAndMerging(final EmitStrategy.StrategyType inputType) {
- setup(inputType, true);
+ @MethodSource("emitStrategyAndHeadersMatrix")
+ public void shouldHandleMultipleSessionsAndMerging(final EmitStrategy.StrategyType inputType, final boolean withHeaders) {
+ setup(inputType, true, withHeaders);
+
processor.process(new Record<>("a", "1", 0L));
processor.process(new Record<>("b", "1", 0L));
processor.process(new Record<>("c", "1", 0L));
@@ -385,9 +401,9 @@ public class KStreamSessionWindowAggregateProcessorTest {
}
@ParameterizedTest
- @EnumSource(EmitStrategy.StrategyType.class)
- public void shouldGetAggregatedValuesFromValueGetter(final EmitStrategy.StrategyType inputType) {
- setup(inputType, true);
+ @MethodSource("emitStrategyAndHeadersMatrix")
+ public void shouldGetAggregatedValuesFromValueGetter(final EmitStrategy.StrategyType inputType, final boolean withHeaders) {
+ setup(inputType, true, withHeaders);
final KTableValueGetter<Windowed<String>, Long> getter = sessionAggregator.view().get();
getter.init(mockContext);
processor.process(new Record<>("a", "1", 0L));
@@ -400,9 +416,9 @@ public class KStreamSessionWindowAggregateProcessorTest {
}
@ParameterizedTest
- @EnumSource(EmitStrategy.StrategyType.class)
- public void shouldImmediatelyForwardNewSessionWhenNonCachedStore(final EmitStrategy.StrategyType inputType) {
- setup(inputType, true);
+ @MethodSource("emitStrategyAndHeadersMatrix")
+ public void shouldImmediatelyForwardNewSessionWhenNonCachedStore(final EmitStrategy.StrategyType inputType, final boolean withHeaders) {
+ setup(inputType, true, withHeaders);
if (emitFinal)
return;
@@ -433,9 +449,9 @@ public class KStreamSessionWindowAggregateProcessorTest {
}
@ParameterizedTest
- @EnumSource(EmitStrategy.StrategyType.class)
- public void shouldImmediatelyForwardRemovedSessionsWhenMerging(final EmitStrategy.StrategyType inputType) {
- setup(inputType, true);
+ @MethodSource("emitStrategyAndHeadersMatrix")
+ public void shouldImmediatelyForwardRemovedSessionsWhenMerging(final EmitStrategy.StrategyType inputType, final boolean withHeaders) {
+ setup(inputType, true, withHeaders);
if (emitFinal)
return;
@@ -464,9 +480,9 @@ public class KStreamSessionWindowAggregateProcessorTest {
}
@ParameterizedTest
- @EnumSource(EmitStrategy.StrategyType.class)
- public void shouldLogAndMeterWhenSkippingNullKeyWithBuiltInMetrics(final EmitStrategy.StrategyType inputType) {
- setup(inputType, false);
+ @MethodSource("emitStrategyAndHeadersMatrix")
+ public void shouldLogAndMeterWhenSkippingNullKeyWithBuiltInMetrics(final EmitStrategy.StrategyType inputType, final boolean withHeaders) {
+ setup(inputType, false, withHeaders);
mockContext.setRecordContext(
new ProcessorRecordContext(-1, -2, -3, "topic", new RecordHeaders())
);
@@ -492,9 +508,9 @@ public class KStreamSessionWindowAggregateProcessorTest {
}
@ParameterizedTest
- @EnumSource(EmitStrategy.StrategyType.class)
- public void shouldLogAndMeterWhenSkippingLateRecordWithZeroGrace(final EmitStrategy.StrategyType inputType) {
- setup(inputType, false);
+ @MethodSource("emitStrategyAndHeadersMatrix")
+ public void shouldLogAndMeterWhenSkippingLateRecordWithZeroGrace(final EmitStrategy.StrategyType inputType, final boolean withHeaders) {
+ setup(inputType, false, withHeaders);
final Processor<String, String, Windowed<String>, Change<Long>> processor = new KStreamSessionWindowAggregate<>(
SessionWindows.ofInactivityGapAndGrace(ofMillis(10L), ofMillis(0L)),
mockStoreFactory(STORE_NAME),
@@ -559,9 +575,9 @@ public class KStreamSessionWindowAggregateProcessorTest {
}
@ParameterizedTest
- @EnumSource(EmitStrategy.StrategyType.class)
- public void shouldLogAndMeterWhenSkippingLateRecordWithNonzeroGrace(final EmitStrategy.StrategyType inputType) {
- setup(inputType, false);
+ @MethodSource("emitStrategyAndHeadersMatrix")
+ public void shouldLogAndMeterWhenSkippingLateRecordWithNonzeroGrace(final EmitStrategy.StrategyType inputType, final boolean withHeaders) {
+ setup(inputType, false, withHeaders);
final Processor<String, String, Windowed<String>, Change<Long>> processor = new KStreamSessionWindowAggregate<>(
SessionWindows.ofInactivityGapAndGrace(ofMillis(10L), ofMillis(1L)),
mockStoreFactory(STORE_NAME),
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 19f8ffeb287..a62a86d9670 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
@@ -90,16 +90,24 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
public class KStreamSlidingWindowAggregateTest {
- public static Stream<Arguments> data() {
+ public static Stream<Arguments> testParameters() {
return Stream.of(
- Arguments.of(StrategyType.ON_WINDOW_UPDATE, true, true),
- Arguments.of(StrategyType.ON_WINDOW_UPDATE, true, false),
- Arguments.of(StrategyType.ON_WINDOW_UPDATE, false, true),
- Arguments.of(StrategyType.ON_WINDOW_UPDATE, false, false),
- Arguments.of(StrategyType.ON_WINDOW_CLOSE, true, true),
- Arguments.of(StrategyType.ON_WINDOW_CLOSE, true, false),
- Arguments.of(StrategyType.ON_WINDOW_CLOSE, false, true),
- Arguments.of(StrategyType.ON_WINDOW_CLOSE, false, false)
+ Arguments.of(StrategyType.ON_WINDOW_UPDATE, true, true, false),
+ Arguments.of(StrategyType.ON_WINDOW_UPDATE, true, true, true),
+ Arguments.of(StrategyType.ON_WINDOW_UPDATE, true, false, false),
+ Arguments.of(StrategyType.ON_WINDOW_UPDATE, true, false, true),
+ Arguments.of(StrategyType.ON_WINDOW_UPDATE, false, true, false),
+ Arguments.of(StrategyType.ON_WINDOW_UPDATE, false, true, true),
+ Arguments.of(StrategyType.ON_WINDOW_UPDATE, false, false, false),
+ Arguments.of(StrategyType.ON_WINDOW_UPDATE, false, false, true),
+ Arguments.of(StrategyType.ON_WINDOW_CLOSE, true, true, false),
+ Arguments.of(StrategyType.ON_WINDOW_CLOSE, true, true, true),
+ Arguments.of(StrategyType.ON_WINDOW_CLOSE, true, false, false),
+ Arguments.of(StrategyType.ON_WINDOW_CLOSE, true, false, true),
+ Arguments.of(StrategyType.ON_WINDOW_CLOSE, false, true, false),
+ Arguments.of(StrategyType.ON_WINDOW_CLOSE, false, true, true),
+ Arguments.of(StrategyType.ON_WINDOW_CLOSE, false, false, false),
+ Arguments.of(StrategyType.ON_WINDOW_CLOSE, false, false, true)
);
}
public StrategyType type;
@@ -112,18 +120,21 @@ public class KStreamSlidingWindowAggregateTest {
private final Properties props = StreamsTestUtils.getStreamsConfig(Serdes.String(), Serdes.String());
private final String threadId = Thread.currentThread().getName();
- public void setup(final StrategyType inputType, final boolean inputInOrderIterator, final boolean inputWithCache) {
+ public void setup(final StrategyType inputType, final boolean inputInOrderIterator, final boolean inputWithCache, final boolean withHeaders) {
type = inputType;
inOrderIterator = inputInOrderIterator;
withCache = inputWithCache;
emitFinal = type.equals(StrategyType.ON_WINDOW_CLOSE);
emitStrategy = StrategyType.forType(type);
+ if (withHeaders) {
+ props.put(org.apache.kafka.streams.StreamsConfig.DSL_STORE_FORMAT_CONFIG, org.apache.kafka.streams.StreamsConfig.DSL_STORE_FORMAT_HEADERS);
+ }
}
@ParameterizedTest
- @MethodSource("data")
- public void testAggregateSmallInput(final StrategyType inputType, final boolean inputInOrderIterator, final boolean inputWithCache) {
- setup(inputType, inputInOrderIterator, inputWithCache);
+ @MethodSource("testParameters")
+ public void testAggregateSmallInput(final StrategyType inputType, final boolean inputInOrderIterator, final boolean inputWithCache, final boolean withHeaders) {
+ setup(inputType, inputInOrderIterator, inputWithCache, withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final String topic = "topic";
@@ -219,9 +230,9 @@ public class KStreamSlidingWindowAggregateTest {
}
@ParameterizedTest
- @MethodSource("data")
- public void testReduceSmallInput(final StrategyType inputType, final boolean inputInOrderIterator, final boolean inputWithCache) {
- setup(inputType, inputInOrderIterator, inputWithCache);
+ @MethodSource("testParameters")
+ public void testReduceSmallInput(final StrategyType inputType, final boolean inputInOrderIterator, final boolean inputWithCache, final boolean withHeaders) {
+ setup(inputType, inputInOrderIterator, inputWithCache, withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final String topic = "topic";
final WindowBytesStoreSupplier storeSupplier = setupWindowBytesStoreSupplier(1);
@@ -286,9 +297,9 @@ public class KStreamSlidingWindowAggregateTest {
}
@ParameterizedTest
- @MethodSource("data")
- public void testAggregateLargeInput(final StrategyType inputType, final boolean inputInOrderIterator, final boolean inputWithCache) {
- setup(inputType, inputInOrderIterator, inputWithCache);
+ @MethodSource("testParameters")
+ public void testAggregateLargeInput(final StrategyType inputType, final boolean inputInOrderIterator, final boolean inputWithCache, final boolean withHeaders) {
+ setup(inputType, inputInOrderIterator, inputWithCache, withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final String topic1 = "topic1";
final long grace = emitFinal ? 10L : 50L;
@@ -507,9 +518,9 @@ public class KStreamSlidingWindowAggregateTest {
}
@ParameterizedTest
- @MethodSource("data")
- public void testJoin(final StrategyType inputType, final boolean inputInOrderIterator, final boolean inputWithCache) {
- setup(inputType, inputInOrderIterator, inputWithCache);
+ @MethodSource("testParameters")
+ public void testJoin(final StrategyType inputType, final boolean inputInOrderIterator, final boolean inputWithCache, final boolean withHeaders) {
+ setup(inputType, inputInOrderIterator, inputWithCache, withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final String topic1 = "topic1";
final String topic2 = "topic2";
@@ -657,9 +668,9 @@ public class KStreamSlidingWindowAggregateTest {
}
@ParameterizedTest
- @MethodSource("data")
- public void testEarlyRecordsSmallInput(final StrategyType inputType, final boolean inputInOrderIterator, final boolean inputWithCache) {
- setup(inputType, inputInOrderIterator, inputWithCache);
+ @MethodSource("testParameters")
+ public void testEarlyRecordsSmallInput(final StrategyType inputType, final boolean inputInOrderIterator, final boolean inputWithCache, final boolean withHeaders) {
+ setup(inputType, inputInOrderIterator, inputWithCache, withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final String topic = "topic";
@@ -781,9 +792,9 @@ public class KStreamSlidingWindowAggregateTest {
}
@ParameterizedTest
- @MethodSource("data")
- public void testEarlyRecordsRepeatedInput(final StrategyType inputType, final boolean inputInOrderIterator, final boolean inputWithCache) {
- setup(inputType, inputInOrderIterator, inputWithCache);
+ @MethodSource("testParameters")
+ public void testEarlyRecordsRepeatedInput(final StrategyType inputType, final boolean inputInOrderIterator, final boolean inputWithCache, final boolean withHeaders) {
+ setup(inputType, inputInOrderIterator, inputWithCache, withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final String topic = "topic";
@@ -865,9 +876,9 @@ public class KStreamSlidingWindowAggregateTest {
}
@ParameterizedTest
- @MethodSource("data")
- public void testEarlyRecordsLargeInput(final StrategyType inputType, final boolean inputInOrderIterator, final boolean inputWithCache) {
- setup(inputType, inputInOrderIterator, inputWithCache);
+ @MethodSource("testParameters")
+ public void testEarlyRecordsLargeInput(final StrategyType inputType, final boolean inputInOrderIterator, final boolean inputWithCache, final boolean withHeaders) {
+ setup(inputType, inputInOrderIterator, inputWithCache, withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final String topic = "topic";
final WindowBytesStoreSupplier storeSupplier = setupWindowBytesStoreSupplier(1);
@@ -1020,9 +1031,9 @@ public class KStreamSlidingWindowAggregateTest {
}
@ParameterizedTest
- @MethodSource("data")
- public void testEarlyNoGracePeriodSmallInput(final StrategyType inputType, final boolean inputInOrderIterator, final boolean inputWithCache) {
- setup(inputType, inputInOrderIterator, inputWithCache);
+ @MethodSource("testParameters")
+ public void testEarlyNoGracePeriodSmallInput(final StrategyType inputType, final boolean inputInOrderIterator, final boolean inputWithCache, final boolean withHeaders) {
+ setup(inputType, inputInOrderIterator, inputWithCache, withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final String topic = "topic";
@@ -1120,9 +1131,9 @@ public class KStreamSlidingWindowAggregateTest {
}
@ParameterizedTest
- @MethodSource("data")
- public void testNoGracePeriodSmallInput(final StrategyType inputType, final boolean inputInOrderIterator, final boolean inputWithCache) {
- setup(inputType, inputInOrderIterator, inputWithCache);
+ @MethodSource("testParameters")
+ public void testNoGracePeriodSmallInput(final StrategyType inputType, final boolean inputInOrderIterator, final boolean inputWithCache, final boolean withHeaders) {
+ setup(inputType, inputInOrderIterator, inputWithCache, withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final String topic = "topic";
@@ -1212,9 +1223,9 @@ public class KStreamSlidingWindowAggregateTest {
}
@ParameterizedTest
- @MethodSource("data")
- public void testEarlyNoGracePeriodLargeInput(final StrategyType inputType, final boolean inputInOrderIterator, final boolean inputWithCache) {
- setup(inputType, inputInOrderIterator, inputWithCache);
+ @MethodSource("testParameters")
+ public void testEarlyNoGracePeriodLargeInput(final StrategyType inputType, final boolean inputInOrderIterator, final boolean inputWithCache, final boolean withHeaders) {
+ setup(inputType, inputInOrderIterator, inputWithCache, withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final String topic = "topic";
final WindowBytesStoreSupplier storeSupplier =
@@ -1333,9 +1344,9 @@ public class KStreamSlidingWindowAggregateTest {
}
@ParameterizedTest
- @MethodSource("data")
- public void testNoGracePeriodLargeInput(final StrategyType inputType, final boolean inputInOrderIterator, final boolean inputWithCache) {
- setup(inputType, inputInOrderIterator, inputWithCache);
+ @MethodSource("testParameters")
+ public void testNoGracePeriodLargeInput(final StrategyType inputType, final boolean inputInOrderIterator, final boolean inputWithCache, final boolean withHeaders) {
+ setup(inputType, inputInOrderIterator, inputWithCache, withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final String topic = "topic";
final WindowBytesStoreSupplier storeSupplier =
@@ -1458,9 +1469,9 @@ public class KStreamSlidingWindowAggregateTest {
}
@ParameterizedTest
- @MethodSource("data")
- public void shouldLogAndMeterWhenSkippingNullKey(final StrategyType inputType, final boolean inputInOrderIterator, final boolean inputWithCache) {
- setup(inputType, inputInOrderIterator, inputWithCache);
+ @MethodSource("testParameters")
+ public void shouldLogAndMeterWhenSkippingNullKey(final StrategyType inputType, final boolean inputInOrderIterator, final boolean inputWithCache, final boolean withHeaders) {
+ setup(inputType, inputInOrderIterator, inputWithCache, withHeaders);
final String builtInMetricsVersion = StreamsConfig.METRICS_LATEST;
final StreamsBuilder builder = new StreamsBuilder();
final String topic = "topic";
@@ -1491,9 +1502,9 @@ public class KStreamSlidingWindowAggregateTest {
}
@ParameterizedTest
- @MethodSource("data")
- public void shouldLogAndMeterWhenSkippingExpiredWindowByGrace(final StrategyType inputType, final boolean inputInOrderIterator, final boolean inputWithCache) {
- setup(inputType, inputInOrderIterator, inputWithCache);
+ @MethodSource("testParameters")
+ public void shouldLogAndMeterWhenSkippingExpiredWindowByGrace(final StrategyType inputType, final boolean inputInOrderIterator, final boolean inputWithCache, final boolean withHeaders) {
+ setup(inputType, inputInOrderIterator, inputWithCache, withHeaders);
final String builtInMetricsVersion = StreamsConfig.METRICS_LATEST;
final StreamsBuilder builder = new StreamsBuilder();
final String topic = "topic";
@@ -1570,9 +1581,9 @@ public class KStreamSlidingWindowAggregateTest {
}
@ParameterizedTest
- @MethodSource("data")
- public void testAggregateRandomInput(final StrategyType inputType, final boolean inputInOrderIterator, final boolean inputWithCache) {
- setup(inputType, inputInOrderIterator, inputWithCache);
+ @MethodSource("testParameters")
+ public void testAggregateRandomInput(final StrategyType inputType, final boolean inputInOrderIterator, final boolean inputWithCache, final boolean withHeaders) {
+ setup(inputType, inputInOrderIterator, inputWithCache, withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final String topic1 = "topic1";
final WindowBytesStoreSupplier storeSupplier =
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 7d4ea159c83..eed3fd998e8 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
@@ -26,6 +26,7 @@ import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.KeyValueTimestamp;
import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.StreamsConfig.InternalConfig;
import org.apache.kafka.streams.TestInputTopic;
import org.apache.kafka.streams.TestOutputTopic;
@@ -90,7 +91,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
public class KStreamWindowAggregateTest {
private static final String WINDOW_STORE_NAME = "dummy-store-name";
- private final Properties props = StreamsTestUtils.getStreamsConfig(Serdes.String(), Serdes.String());
+ private Properties props = StreamsTestUtils.getStreamsConfig(Serdes.String(), Serdes.String());
private final String threadId = Thread.currentThread().getName();
private final StoreFactory storeFactory = mockStoreFactory(WINDOW_STORE_NAME);
@@ -102,26 +103,35 @@ public class KStreamWindowAggregateTest {
private boolean emitFinal;
- public static Stream<Arguments> getEmitStrategy() {
+ public static Stream<Arguments> testParameters() {
return Stream.of(
- Arguments.of(StrategyType.ON_WINDOW_UPDATE, true),
- Arguments.of(StrategyType.ON_WINDOW_UPDATE, false),
- Arguments.of(StrategyType.ON_WINDOW_CLOSE, true),
- Arguments.of(StrategyType.ON_WINDOW_CLOSE, false)
+ Arguments.of(StrategyType.ON_WINDOW_UPDATE, true, false),
+ Arguments.of(StrategyType.ON_WINDOW_UPDATE, true, true),
+ Arguments.of(StrategyType.ON_WINDOW_UPDATE, false, false),
+ Arguments.of(StrategyType.ON_WINDOW_UPDATE, false, true),
+ Arguments.of(StrategyType.ON_WINDOW_CLOSE, true, false),
+ Arguments.of(StrategyType.ON_WINDOW_CLOSE, true, true),
+ Arguments.of(StrategyType.ON_WINDOW_CLOSE, false, false),
+ Arguments.of(StrategyType.ON_WINDOW_CLOSE, false, true)
);
}
-
- public void setup(final StrategyType inputType, final boolean inputWithCache) {
+
+ public void setup(final StrategyType inputType, final boolean inputWithCache, final boolean withHeaders) {
type = inputType;
withCache = inputWithCache;
emitFinal = type.equals(StrategyType.ON_WINDOW_CLOSE);
emitStrategy = StrategyType.forType(type);
+ if (withHeaders) {
+ props.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
+ } else {
+ props.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_DEFAULT);
+ }
}
@ParameterizedTest
- @MethodSource("getEmitStrategy")
- public void testAggBasic(final StrategyType inputType, final boolean inputWithCache) {
- setup(inputType, inputWithCache);
+ @MethodSource("testParameters")
+ public void testAggBasic(final StrategyType inputType, final boolean inputWithCache, final boolean withHeaders) {
+ setup(inputType, inputWithCache, withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final String topic1 = "topic1";
@@ -215,9 +225,9 @@ public class KStreamWindowAggregateTest {
}
@ParameterizedTest
- @MethodSource("getEmitStrategy")
- public void testJoin(final StrategyType inputType, final boolean inputWithCache) {
- setup(inputType, inputWithCache);
+ @MethodSource("testParameters")
+ public void testJoin(final StrategyType inputType, final boolean inputWithCache, final boolean withHeaders) {
+ setup(inputType, inputWithCache, withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final String topic1 = "topic1";
final String topic2 = "topic2";
@@ -465,9 +475,9 @@ public class KStreamWindowAggregateTest {
}
@ParameterizedTest
- @MethodSource("getEmitStrategy")
- public void shouldLogAndMeterWhenSkippingNullKey(final StrategyType inputType, final boolean inputWithCache) {
- setup(inputType, inputWithCache);
+ @MethodSource("testParameters")
+ public void shouldLogAndMeterWhenSkippingNullKey(final StrategyType inputType, final boolean inputWithCache, final boolean withHeaders) {
+ setup(inputType, inputWithCache, withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final String topic = "topic";
@@ -493,9 +503,9 @@ public class KStreamWindowAggregateTest {
}
@ParameterizedTest
- @MethodSource("getEmitStrategy")
- public void shouldLogAndMeterWhenSkippingExpiredWindow(final StrategyType inputType, final boolean inputWithCache) {
- setup(inputType, inputWithCache);
+ @MethodSource("testParameters")
+ public void shouldLogAndMeterWhenSkippingExpiredWindow(final StrategyType inputType, final boolean inputWithCache, final boolean withHeaders) {
+ setup(inputType, inputWithCache, withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final String topic = "topic";
@@ -579,9 +589,9 @@ public class KStreamWindowAggregateTest {
}
@ParameterizedTest
- @MethodSource("getEmitStrategy")
- public void shouldLogAndMeterWhenSkippingExpiredWindowByGrace(final StrategyType inputType, final boolean inputWithCache) {
- setup(inputType, inputWithCache);
+ @MethodSource("testParameters")
+ public void shouldLogAndMeterWhenSkippingExpiredWindowByGrace(final StrategyType inputType, final boolean inputWithCache, final boolean withHeaders) {
+ setup(inputType, inputWithCache, withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final String topic = "topic";
@@ -636,9 +646,9 @@ public class KStreamWindowAggregateTest {
}
@ParameterizedTest
- @MethodSource("getEmitStrategy")
- public void shouldNotEmitFinalIfNotProgressEnough(final StrategyType inputType, final boolean inputWithCache) throws IOException {
- setup(inputType, inputWithCache);
+ @MethodSource("testParameters")
+ public void shouldNotEmitFinalIfNotProgressEnough(final StrategyType inputType, final boolean inputWithCache, final boolean withHeaders) throws IOException {
+ setup(inputType, inputWithCache, withHeaders);
final File stateDir = TestUtils.tempDirectory();
final long windowSize = 10L;
final Windows<TimeWindow> windows = TimeWindows.ofSizeAndGrace(ofMillis(windowSize), ofMillis(5)).advanceBy(ofMillis(5));
@@ -726,9 +736,9 @@ public class KStreamWindowAggregateTest {
}
@ParameterizedTest
- @MethodSource("getEmitStrategy")
- public void shouldEmitWithInterval0(final StrategyType inputType, final boolean inputWithCache) throws IOException {
- setup(inputType, inputWithCache);
+ @MethodSource("testParameters")
+ public void shouldEmitWithInterval0(final StrategyType inputType, final boolean inputWithCache, final boolean withHeaders) throws IOException {
+ setup(inputType, inputWithCache, withHeaders);
final File stateDir = TestUtils.tempDirectory();
final long windowSize = 10L;
final Windows<TimeWindow> windows = TimeWindows.ofSizeAndGrace(ofMillis(windowSize), ofMillis(5)).advanceBy(ofMillis(5));
@@ -795,9 +805,9 @@ public class KStreamWindowAggregateTest {
}
@ParameterizedTest
- @MethodSource("getEmitStrategy")
- public void shouldEmitWithLargeInterval(final StrategyType inputType, final boolean inputWithCache) throws IOException {
- setup(inputType, inputWithCache);
+ @MethodSource("testParameters")
+ public void shouldEmitWithLargeInterval(final StrategyType inputType, final boolean inputWithCache, final boolean withHeaders) throws IOException {
+ setup(inputType, inputWithCache, withHeaders);
final File stateDir = TestUtils.tempDirectory();
final long windowSize = 10L;
final Windows<TimeWindow> windows = TimeWindows.ofSizeAndGrace(ofMillis(windowSize), ofMillis(5)).advanceBy(ofMillis(5));
@@ -896,9 +906,9 @@ public class KStreamWindowAggregateTest {
}
@ParameterizedTest
- @MethodSource("getEmitStrategy")
- public void shouldEmitFromLastEmitTime(final StrategyType inputType, final boolean inputWithCache) throws IOException {
- setup(inputType, inputWithCache);
+ @MethodSource("testParameters")
+ public void shouldEmitFromLastEmitTime(final StrategyType inputType, final boolean inputWithCache, final boolean withHeaders) throws IOException {
+ setup(inputType, inputWithCache, withHeaders);
final File stateDir = TestUtils.tempDirectory();
final long windowSize = 10L;
final Windows<TimeWindow> windows = TimeWindows.ofSizeAndGrace(ofMillis(windowSize), ofMillis(5)).advanceBy(ofMillis(5));
@@ -978,9 +988,9 @@ public class KStreamWindowAggregateTest {
}
@ParameterizedTest
- @MethodSource("getEmitStrategy")
- public void showThrowIfEmitFinalUsedWithUnlimitedWindow(final StrategyType inputType, final boolean inputWithCache) {
- setup(inputType, inputWithCache);
+ @MethodSource("testParameters")
+ public void showThrowIfEmitFinalUsedWithUnlimitedWindow(final StrategyType inputType, final boolean inputWithCache, final boolean withHeaders) {
+ setup(inputType, inputWithCache, withHeaders);
if (emitFinal) {
final IllegalArgumentException e = assertThrows(
IllegalArgumentException.class, () -> new KStreamWindowAggregate<>(
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableAggregateTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableAggregateTest.java
index 250e10169b6..35bdbee4d1d 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableAggregateTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableAggregateTest.java
@@ -53,6 +53,8 @@ import org.apache.kafka.test.StreamsTestUtils;
import org.apache.kafka.test.TestUtils;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
@@ -76,17 +78,21 @@ public class KTableAggregateTest {
private final MockApiProcessorSupplier<String, Object, Void, Void> supplier = new MockApiProcessorSupplier<>();
private static final Properties CONFIG = mkProperties(mkMap(
mkEntry(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory("kafka-test").getAbsolutePath())));
-
- private StreamsBuilder createStreamBuilderInMemory() {
+
+ private StreamsBuilder createStreamBuilderInMemory(final boolean withHeaders) {
final Properties props = StreamsTestUtils.getStreamsConfig(Serdes.String(), Serdes.String());
props.put(StreamsConfig.DSL_STORE_SUPPLIERS_CLASS_CONFIG,
BuiltInDslStoreSuppliers.InMemoryDslStoreSuppliers.class.getName());
+ if (withHeaders) {
+ props.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
+ }
return new StreamsBuilder(new TopologyConfig(new StreamsConfig(props)));
}
- @Test
- public void testAggBasic() {
- final StreamsBuilder builder = createStreamBuilderInMemory();
+ @ParameterizedTest
+ @CsvSource({"true", "false"})
+ public void testAggBasic(final boolean withHeaders) {
+ final StreamsBuilder builder = createStreamBuilderInMemory(withHeaders);
final String topic1 = "topic1";
final KTable<String, String> table1 = builder.table(topic1, consumed);
@@ -134,9 +140,10 @@ public class KTableAggregateTest {
}
}
- @Test
- public void testAggRepartition() {
- final StreamsBuilder builder = createStreamBuilderInMemory();
+ @ParameterizedTest
+ @CsvSource({"true", "false"})
+ public void testAggRepartition(final boolean withHeaders) {
+ final StreamsBuilder builder = createStreamBuilderInMemory(withHeaders);
final String topic1 = "topic1";
final KTable<String, String> table1 = builder.table(topic1, consumed);
@@ -275,9 +282,10 @@ public class KTableAggregateTest {
}
- @Test
- public void testCount() {
- final StreamsBuilder builder = createStreamBuilderInMemory();
+ @ParameterizedTest
+ @CsvSource({"true", "false"})
+ public void testCount(final boolean withHeaders) {
+ final StreamsBuilder builder = createStreamBuilderInMemory(withHeaders);
final String input = "count-test-input";
builder
@@ -290,9 +298,10 @@ public class KTableAggregateTest {
testCountHelper(builder, input, supplier);
}
- @Test
- public void testCountWithInternalStore() {
- final StreamsBuilder builder = createStreamBuilderInMemory();
+ @ParameterizedTest
+ @CsvSource({"true", "false"})
+ public void testCountWithInternalStore(final boolean withHeaders) {
+ final StreamsBuilder builder = createStreamBuilderInMemory(withHeaders);
final String input = "count-test-input";
builder
@@ -344,9 +353,10 @@ public class KTableAggregateTest {
}
}
- @Test
- public void testRemoveOldBeforeAddNew() {
- final StreamsBuilder builder = createStreamBuilderInMemory();
+ @ParameterizedTest
+ @CsvSource({"true", "false"})
+ public void testRemoveOldBeforeAddNew(final boolean withHeaders) {
+ final StreamsBuilder builder = createStreamBuilderInMemory(withHeaders);
final String input = "count-test-input";
final MockApiProcessorSupplier<String, String, Void, Void> supplier = new MockApiProcessorSupplier<>();
@@ -391,8 +401,8 @@ public class KTableAggregateTest {
}
}
- private void testUpgradeFromConfig(final Properties config, final List<KeyValueTimestamp<String, Long>> expected) {
- final StreamsBuilder builder = createStreamBuilderInMemory();
+ private void testUpgradeFromConfig(final Properties config, final List<KeyValueTimestamp<String, Long>> expected, final boolean withHeaders) {
+ final StreamsBuilder builder = createStreamBuilderInMemory(withHeaders);
final String input = "input-topic";
final String output = "output-topic";
final Serde<String> stringSerde = Serdes.String();
@@ -421,8 +431,9 @@ public class KTableAggregateTest {
}
}
- @Test
- public void testShouldSendTransientStateWhenUpgrading() {
+ @ParameterizedTest
+ @CsvSource({"true", "false"})
+ public void testShouldSendTransientStateWhenUpgrading(final boolean withHeaders) {
final Properties upgradingConfig = new Properties();
upgradingConfig.putAll(CONFIG);
upgradingConfig.put(StreamsConfig.UPGRADE_FROM_CONFIG, StreamsConfig.UPGRADE_FROM_33);
@@ -430,15 +441,16 @@ public class KTableAggregateTest {
new KeyValueTimestamp<>("1", 1L, 8),
new KeyValueTimestamp<>("1", 0L, 9), // transient inconsistent state
new KeyValueTimestamp<>("1", 1L, 9)
- ));
+ ), withHeaders);
}
- @Test
- public void testShouldNotSendTransientStateIfNotUpgrading() {
+ @ParameterizedTest
+ @CsvSource({"true", "false"})
+ public void testShouldNotSendTransientStateIfNotUpgrading(final boolean withHeaders) {
testUpgradeFromConfig(CONFIG, asList(
new KeyValueTimestamp<>("1", 1L, 8),
new KeyValueTimestamp<>("1", 1L, 9)
- ));
+ ), withHeaders);
}
private static class NoEqualsImpl {
@@ -474,8 +486,9 @@ public class KTableAggregateTest {
private void testKeyWithNoEquals(
final KeyValueMapper<NoEqualsImpl, NoEqualsImpl, KeyValue<NoEqualsImpl, NoEqualsImpl>> keyValueMapper,
- final List<TestRecord<NoEqualsImpl, Long>> expected) {
- final StreamsBuilder builder = createStreamBuilderInMemory();
+ final List<TestRecord<NoEqualsImpl, Long>> expected,
+ final boolean withHeaders) {
+ final StreamsBuilder builder = createStreamBuilderInMemory(withHeaders);
final String input = "input-topic";
final String output = "output-topic";
final Serde<NoEqualsImpl> noEqualsImplSerde = new NoEqualsImplSerde();
@@ -507,8 +520,9 @@ public class KTableAggregateTest {
}
}
- @Test
- public void testNoEqualsAndNotSameObject() {
+ @ParameterizedTest
+ @CsvSource({"true", "false"})
+ public void testNoEqualsAndNotSameObject(final boolean withHeaders) {
testKeyWithNoEquals(
// key changes, different object reference (deserializer returns a new object reference)
(k, v) -> new KeyValue<>(v, v),
@@ -516,19 +530,22 @@ public class KTableAggregateTest {
new TestRecord<>(new NoEqualsImpl("1"), 1L, Instant.ofEpochMilli(8)),
new TestRecord<>(new NoEqualsImpl("1"), 0L, Instant.ofEpochMilli(9)), // transient inconsistent state
new TestRecord<>(new NoEqualsImpl("1"), 1L, Instant.ofEpochMilli(9))
- )
+ ),
+ withHeaders
);
}
- @Test
- public void testNoEqualsAndSameObject() {
+ @ParameterizedTest
+ @CsvSource({"true", "false"})
+ public void testNoEqualsAndSameObject(final boolean withHeaders) {
testKeyWithNoEquals(
// key does not change, same object reference
KeyValue::new,
asList(
new TestRecord<>(new NoEqualsImpl("1"), 1L, Instant.ofEpochMilli(8)),
new TestRecord<>(new NoEqualsImpl("1"), 1L, Instant.ofEpochMilli(9))
- )
+ ),
+ withHeaders
);
}
}
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 45adf9951f6..8c9b5c40e0b 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
@@ -43,8 +43,8 @@ import org.apache.kafka.test.MockMapper;
import org.apache.kafka.test.MockReducer;
import org.apache.kafka.test.StreamsTestUtils;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.time.Duration;
import java.time.Instant;
@@ -59,12 +59,16 @@ import static org.junit.jupiter.api.Assertions.assertNull;
@SuppressWarnings("unchecked")
public class KTableFilterTest {
private final Consumed<String, Integer> consumed = Consumed.with(Serdes.String(), Serdes.Integer());
- private final Properties props = StreamsTestUtils.getStreamsConfig(Serdes.String(), Serdes.Integer());
+ private Properties props = StreamsTestUtils.getStreamsConfig(Serdes.String(), Serdes.Integer());
- @BeforeEach
- public void setUp() {
+ private void setup(final boolean withHeaders) {
// disable caching at the config level
props.setProperty(StreamsConfig.STATESTORE_CACHE_MAX_BYTES_CONFIG, "0");
+ if (withHeaders) {
+ props.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
+ } else {
+ props.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_DEFAULT);
+ }
}
private final Predicate<String, Integer> predicate = (key, value) -> (value % 2) == 0;
@@ -104,8 +108,10 @@ public class KTableFilterTest {
new KeyValueTimestamp<>("B", null, 15));
}
- @Test
- public void shouldPassThroughWithoutMaterialization() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldPassThroughWithoutMaterialization(final boolean withHeaders) {
+ setup(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final String topic1 = "topic1";
@@ -120,8 +126,10 @@ public class KTableFilterTest {
doTestKTable(builder, table2, table3, topic1);
}
- @Test
- public void shouldPassThroughOnMaterialization() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldPassThroughOnMaterialization(final boolean withHeaders) {
+ setup(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final String topic1 = "topic1";
@@ -206,8 +214,10 @@ public class KTableFilterTest {
}
}
- @Test
- public void shouldGetValuesOnMaterialization() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldGetValuesOnMaterialization(final boolean withHeaders) {
+ setup(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final String topic1 = "topic1";
@@ -277,8 +287,10 @@ public class KTableFilterTest {
}
- @Test
- public void shouldNotSendOldValuesWithoutMaterialization() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotSendOldValuesWithoutMaterialization(final boolean withHeaders) {
+ setup(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final String topic1 = "topic1";
@@ -289,8 +301,10 @@ public class KTableFilterTest {
doTestNotSendingOldValue(builder, table1, table2, topic1);
}
- @Test
- public void shouldNotSendOldValuesOnMaterialization() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotSendOldValuesOnMaterialization(final boolean withHeaders) {
+ setup(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final String topic1 = "topic1";
@@ -302,8 +316,10 @@ public class KTableFilterTest {
doTestNotSendingOldValue(builder, table1, table2, topic1);
}
- @Test
- public void shouldNotEnableSendingOldValuesIfNotAlreadyMaterializedAndNotForcedToMaterialize() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotEnableSendingOldValuesIfNotAlreadyMaterializedAndNotForcedToMaterialize(final boolean withHeaders) {
+ setup(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final String topic1 = "topic1";
@@ -381,8 +397,10 @@ public class KTableFilterTest {
}
}
- @Test
- public void shouldEnableSendOldValuesWhenNotMaterializedAlreadyButForcedToMaterialize() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldEnableSendOldValuesWhenNotMaterializedAlreadyButForcedToMaterialize(final boolean withHeaders) {
+ setup(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final String topic1 = "topic1";
@@ -399,8 +417,10 @@ public class KTableFilterTest {
doTestSendingOldValue(builder, table1, table2, topic1);
}
- @Test
- public void shouldEnableSendOldValuesWhenMaterializedAlreadyAndForcedToMaterialize() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldEnableSendOldValuesWhenMaterializedAlreadyAndForcedToMaterialize(final boolean withHeaders) {
+ setup(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final String topic1 = "topic1";
@@ -417,8 +437,10 @@ public class KTableFilterTest {
doTestSendingOldValue(builder, table1, table2, topic1);
}
- @Test
- public void shouldSendOldValuesWhenEnabledOnUpStreamMaterialization() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldSendOldValuesWhenEnabledOnUpStreamMaterialization(final boolean withHeaders) {
+ setup(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final String topic1 = "topic1";
@@ -468,8 +490,10 @@ public class KTableFilterTest {
}
}
- @Test
- public void shouldSkipNullToRepartitionWithoutMaterialization() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldSkipNullToRepartitionWithoutMaterialization(final boolean withHeaders) {
+ setup(withHeaders);
// Do not explicitly set enableSendingOldValues. Let a further downstream stateful operator trigger it instead.
final StreamsBuilder builder = new StreamsBuilder();
@@ -486,8 +510,10 @@ public class KTableFilterTest {
doTestSkipNullOnMaterialization(builder, table1, table2, topic1, true);
}
- @Test
- public void shouldSkipNullToRepartitionOnMaterialization() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldSkipNullToRepartitionOnMaterialization(final boolean withHeaders) {
+ setup(withHeaders);
// Do not explicitly set enableSendingOldValues. Let a further downstream stateful operator trigger it instead.
final StreamsBuilder builder = new StreamsBuilder();
@@ -504,8 +530,10 @@ public class KTableFilterTest {
doTestSkipNullOnMaterialization(builder, table1, table2, topic1, true);
}
- @Test
- public void shouldNotSkipNullIfVersionedUpstream() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotSkipNullIfVersionedUpstream(final boolean withHeaders) {
+ setup(withHeaders);
// stateful downstream operation enables sendOldValues, but duplicate nulls will still
// be sent because the source table is versioned
final StreamsBuilder builder = new StreamsBuilder();
@@ -525,8 +553,10 @@ public class KTableFilterTest {
doTestSkipNullOnMaterialization(builder, table1, table2, topic1, false);
}
- @Test
- public void shouldSkipNullIfVersionedDownstream() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldSkipNullIfVersionedDownstream(final boolean withHeaders) {
+ setup(withHeaders);
// materializing the result of the filter as a versioned store does not prevent duplicate
// tombstones from being sent, as it's whether the input table is versioned or not that
// determines whether the optimization is enabled
@@ -547,8 +577,10 @@ public class KTableFilterTest {
doTestSkipNullOnMaterialization(builder, table1, table2, topic1, true);
}
- @Test
- public void testTypeVariance() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testTypeVariance(final boolean withHeaders) {
+ setup(withHeaders);
final Predicate<Number, Object> numberKeyPredicate = (key, value) -> false;
new StreamsBuilder()
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 2f6f29fdbda..8825828ea07 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
@@ -23,6 +23,7 @@ import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.KeyValueTimestamp;
import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.TestInputTopic;
import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.TopologyDescription;
@@ -54,8 +55,9 @@ import org.apache.kafka.test.MockValueJoiner;
import org.apache.kafka.test.StreamsTestUtils;
import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.junit.jupiter.MockitoExtension;
import org.mockito.junit.jupiter.MockitoSettings;
import org.mockito.quality.Strictness;
@@ -80,18 +82,28 @@ public class KTableImplTest {
private final Consumed<String, String> stringConsumed = Consumed.with(Serdes.String(), Serdes.String());
private final Consumed<String, String> consumed = Consumed.with(Serdes.String(), Serdes.String());
private final Produced<String, String> produced = Produced.with(Serdes.String(), Serdes.String());
- private final Properties props = StreamsTestUtils.getStreamsConfig(Serdes.String(), Serdes.String());
+ private Properties props = StreamsTestUtils.getStreamsConfig(Serdes.String(), Serdes.String());
private final Serde<String> mySerde = new Serdes.StringSerde();
private KTable<String, String> table;
+ private void setDslStoreFormat(final boolean withHeaders) {
+ if (withHeaders) {
+ props.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
+ } else {
+ props.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_DEFAULT);
+ }
+ }
+
@BeforeEach
public void setUp() {
table = new StreamsBuilder().table("test");
}
- @Test
- public void testKTable() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testKTable(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final String topic1 = "topic1";
@@ -158,8 +170,10 @@ public class KTableImplTest {
processors.get(3).processed());
}
- @Test
- public void testMaterializedKTable() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testMaterializedKTable(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final String topic1 = "topic1";
@@ -223,8 +237,10 @@ public class KTableImplTest {
processors.get(3).processed());
}
- @Test
- public void shouldPreserveSerdesForOperators() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldPreserveSerdesForOperators(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final KTable<String, String> table1 = builder.table("topic-2", stringConsumed);
final ConsumedInternal<String, String> consumedInternal = new ConsumedInternal<>(stringConsumed);
@@ -347,8 +363,10 @@ public class KTableImplTest {
mySerde);
}
- @Test
- public void testStateStoreLazyEval() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testStateStoreLazyEval(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final String topic1 = "topic1";
final String topic2 = "topic2";
@@ -364,8 +382,10 @@ public class KTableImplTest {
}
}
- @Test
- public void testStateStore() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testStateStore(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final String topic1 = "topic1";
final String topic2 = "topic2";
@@ -382,8 +402,10 @@ public class KTableImplTest {
}
}
- @Test
- public void shouldNotEnableSendingOldValuesIfNotMaterializedAlreadyAndNotForcedToMaterialize() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotEnableSendingOldValuesIfNotMaterializedAlreadyAndNotForcedToMaterialize(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final var table = assertInstanceOf(KTableImpl.class, builder.table("topic1", consumed));
@@ -392,8 +414,10 @@ public class KTableImplTest {
assertThat(table.sendingOldValueEnabled(), is(false));
}
- @Test
- public void shouldEnableSendingOldValuesIfNotMaterializedAlreadyButForcedToMaterialize() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldEnableSendingOldValuesIfNotMaterializedAlreadyButForcedToMaterialize(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final var table = assertInstanceOf(KTableImpl.class, builder.table("topic1", consumed));
@@ -414,8 +438,10 @@ public class KTableImplTest {
+ "found in the provided Topology:\n" + topology.describe());
}
- @Test
- public void shouldCreateSourceAndSinkNodesForRepartitioningTopic() throws Exception {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldCreateSourceAndSinkNodesForRepartitioningTopic(final boolean withHeaders) throws Exception {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final String topic1 = "topic1";
final String storeName1 = "storeName1";
@@ -467,124 +493,166 @@ public class KTableImplTest {
}
}
- @Test
- public void shouldNotAllowNullSelectorOnToStream() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotAllowNullSelectorOnToStream(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
assertThrows(NullPointerException.class, () -> table.toStream((KeyValueMapper<String, String, ?>) null));
}
- @Test
- public void shouldNotAllowNullPredicateOnFilter() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotAllowNullPredicateOnFilter(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
assertThrows(NullPointerException.class, () -> table.filter(null));
}
- @Test
- public void shouldNotAllowNullPredicateOnFilterNot() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotAllowNullPredicateOnFilterNot(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
assertThrows(NullPointerException.class, () -> table.filterNot(null));
}
- @Test
- public void shouldNotAllowNullMapperOnMapValues() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotAllowNullMapperOnMapValues(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
assertThrows(NullPointerException.class, () -> table.mapValues((ValueMapper<String, ?>) null));
}
- @Test
- public void shouldNotAllowNullMapperOnMapValueWithKey() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotAllowNullMapperOnMapValueWithKey(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
assertThrows(NullPointerException.class, () -> table.mapValues((ValueMapperWithKey<String, String, ?>) null));
}
- @Test
- public void shouldNotAllowNullSelectorOnGroupBy() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotAllowNullSelectorOnGroupBy(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
assertThrows(NullPointerException.class, () -> table.groupBy(null));
}
- @Test
- public void shouldNotAllowNullOtherTableOnJoin() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotAllowNullOtherTableOnJoin(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
assertThrows(NullPointerException.class, () -> table.join(null, MockValueJoiner.TOSTRING_JOINER));
}
- @Test
- public void shouldAllowNullStoreInJoin() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldAllowNullStoreInJoin(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
table.join(table, MockValueJoiner.TOSTRING_JOINER);
}
- @Test
- public void shouldNotAllowNullJoinerJoin() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotAllowNullJoinerJoin(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
assertThrows(NullPointerException.class, () -> table.join(table, null));
}
- @Test
- public void shouldNotAllowNullOtherTableOnOuterJoin() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotAllowNullOtherTableOnOuterJoin(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
assertThrows(NullPointerException.class, () -> table.outerJoin(null, MockValueJoiner.TOSTRING_JOINER));
}
- @Test
- public void shouldNotAllowNullJoinerOnOuterJoin() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotAllowNullJoinerOnOuterJoin(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
assertThrows(NullPointerException.class, () -> table.outerJoin(table, null));
}
- @Test
- public void shouldNotAllowNullJoinerOnLeftJoin() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotAllowNullJoinerOnLeftJoin(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
assertThrows(NullPointerException.class, () -> table.leftJoin(table, null));
}
- @Test
- public void shouldNotAllowNullOtherTableOnLeftJoin() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotAllowNullOtherTableOnLeftJoin(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
assertThrows(NullPointerException.class, () -> table.leftJoin(null, MockValueJoiner.TOSTRING_JOINER));
}
- @Test
- public void shouldThrowNullPointerOnFilterWhenMaterializedIsNull() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldThrowNullPointerOnFilterWhenMaterializedIsNull(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
assertThrows(
NullPointerException.class,
() -> table.filter((key, value) -> false, (Materialized<String, String, KeyValueStore<Bytes, byte[]>>) null)
);
}
- @Test
- public void shouldThrowNullPointerOnFilterNotWhenMaterializedIsNull() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldThrowNullPointerOnFilterNotWhenMaterializedIsNull(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
assertThrows(
NullPointerException.class,
() -> table.filterNot((key, value) -> false, (Materialized<String, String, KeyValueStore<Bytes, byte[]>>) null)
);
}
- @Test
- public void shouldThrowNullPointerOnJoinWhenMaterializedIsNull() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldThrowNullPointerOnJoinWhenMaterializedIsNull(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
assertThrows(
NullPointerException.class,
() -> table.join(table, MockValueJoiner.TOSTRING_JOINER, (Materialized<String, String, KeyValueStore<Bytes, byte[]>>) null)
);
}
- @Test
- public void shouldThrowNullPointerOnLeftJoinWhenMaterializedIsNull() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldThrowNullPointerOnLeftJoinWhenMaterializedIsNull(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
assertThrows(
NullPointerException.class,
() -> table.leftJoin(table, MockValueJoiner.TOSTRING_JOINER, (Materialized<String, String, KeyValueStore<Bytes, byte[]>>) null)
);
}
- @Test
- public void shouldThrowNullPointerOnOuterJoinWhenMaterializedIsNull() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldThrowNullPointerOnOuterJoinWhenMaterializedIsNull(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
assertThrows(
NullPointerException.class,
() -> table.outerJoin(table, MockValueJoiner.TOSTRING_JOINER, (Materialized<String, String, KeyValueStore<Bytes, byte[]>>) null)
);
}
- @Test
- public void shouldThrowNullPointerOnTransformValuesWithKeyWhenTransformerSupplierIsNull() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldThrowNullPointerOnTransformValuesWithKeyWhenTransformerSupplierIsNull(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
assertThrows(NullPointerException.class, () -> table.transformValues(null));
}
- @Test
- public void shouldThrowNullPointerOnTransformValuesWithKeyWhenMaterializedIsNull() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldThrowNullPointerOnTransformValuesWithKeyWhenMaterializedIsNull(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
assertThrows(NullPointerException.class, () -> table.transformValues(mock(), (Materialized<String, Object, KeyValueStore<Bytes, byte[]>>) null));
}
- @Test
- public void shouldThrowNullPointerOnTransformValuesWithKeyWhenStoreNamesNull() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldThrowNullPointerOnTransformValuesWithKeyWhenStoreNamesNull(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
assertThrows(NullPointerException.class, () -> table.transformValues(mock(), (String[]) null));
}
}
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 5606933733c..2fa32d3cbf4 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
@@ -36,7 +36,8 @@ import org.apache.kafka.streams.state.KeyValueStore;
import org.apache.kafka.streams.utils.UniqueTopicSerdeScope;
import org.apache.kafka.test.TestUtils;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.util.Collections;
import java.util.HashMap;
@@ -56,8 +57,9 @@ public class KTableKTableForeignKeyJoinScenarioTest {
private static final String RIGHT_TABLE = "right_table";
private static final String OUTPUT = "output-topic";
- @Test
- public void shouldWorkWithDefaultSerdes() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldWorkWithDefaultSerdes(final boolean withHeaders) {
final StreamsBuilder builder = new StreamsBuilder();
final KTable<Integer, String> aTable = builder.table("A");
final KTable<Integer, String> bTable = builder.table("B");
@@ -76,11 +78,12 @@ public class KTableKTableForeignKeyJoinScenarioTest {
finalJoinResult.toStream().to("output");
- validateTopologyCanProcessData(builder);
+ validateTopologyCanProcessData(builder, withHeaders);
}
- @Test
- public void shouldWorkWithDefaultAndConsumedSerdes() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldWorkWithDefaultAndConsumedSerdes(final boolean withHeaders) {
final StreamsBuilder builder = new StreamsBuilder();
final KTable<Integer, String> aTable = builder.table("A", Consumed.with(Serdes.Integer(), Serdes.String()));
final KTable<Integer, String> bTable = builder.table("B");
@@ -99,11 +102,12 @@ public class KTableKTableForeignKeyJoinScenarioTest {
finalJoinResult.toStream().to("output");
- validateTopologyCanProcessData(builder);
+ validateTopologyCanProcessData(builder, withHeaders);
}
- @Test
- public void shouldWorkWithDefaultAndJoinResultSerdes() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldWorkWithDefaultAndJoinResultSerdes(final boolean withHeaders) {
final StreamsBuilder builder = new StreamsBuilder();
final KTable<Integer, String> aTable = builder.table("A");
final KTable<Integer, String> bTable = builder.table("B");
@@ -124,11 +128,12 @@ public class KTableKTableForeignKeyJoinScenarioTest {
finalJoinResult.toStream().to("output");
- validateTopologyCanProcessData(builder);
+ validateTopologyCanProcessData(builder, withHeaders);
}
- @Test
- public void shouldWorkWithDefaultAndEquiJoinResultSerdes() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldWorkWithDefaultAndEquiJoinResultSerdes(final boolean withHeaders) {
final StreamsBuilder builder = new StreamsBuilder();
final KTable<Integer, String> aTable = builder.table("A");
final KTable<Integer, String> bTable = builder.table("B");
@@ -148,11 +153,12 @@ public class KTableKTableForeignKeyJoinScenarioTest {
finalJoinResult.toStream().to("output");
- validateTopologyCanProcessData(builder);
+ validateTopologyCanProcessData(builder, withHeaders);
}
- @Test
- public void shouldWorkWithDefaultAndProducedSerdes() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldWorkWithDefaultAndProducedSerdes(final boolean withHeaders) {
final StreamsBuilder builder = new StreamsBuilder();
final KTable<Integer, String> aTable = builder.table("A");
final KTable<Integer, String> bTable = builder.table("B");
@@ -171,16 +177,20 @@ public class KTableKTableForeignKeyJoinScenarioTest {
finalJoinResult.toStream().to("output", Produced.with(Serdes.Integer(), Serdes.String()));
- validateTopologyCanProcessData(builder);
+ validateTopologyCanProcessData(builder, withHeaders);
}
- @Test
- public void shouldUseExpectedTopicsWithSerde() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldUseExpectedTopicsWithSerde(final boolean withHeaders) {
final String applicationId = "ktable-ktable-joinOnForeignKey";
final Properties streamsConfig = mkProperties(mkMap(
mkEntry(StreamsConfig.APPLICATION_ID_CONFIG, applicationId),
mkEntry(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getPath())
));
+ if (withHeaders) {
+ streamsConfig.setProperty(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
+ }
final UniqueTopicSerdeScope serdeScope = new UniqueTopicSerdeScope();
final StreamsBuilder builder = new StreamsBuilder();
@@ -234,8 +244,9 @@ public class KTableKTableForeignKeyJoinScenarioTest {
)));
}
- @Test
- public void shouldWorkWithCompositeKeyAndProducerIdInValue() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldWorkWithCompositeKeyAndProducerIdInValue(final boolean withHeaders) {
final StreamsBuilder builder = new StreamsBuilder();
// Left table keyed by <producer_id, product_id>
@@ -260,7 +271,7 @@ public class KTableKTableForeignKeyJoinScenarioTest {
joined.toStream().to("output");
- try (final TopologyTestDriver driver = createTopologyTestDriver(builder)) {
+ try (final TopologyTestDriver driver = createTopologyTestDriver(builder, withHeaders)) {
final TestInputTopic<String, String> leftInput = driver.createInputTopic(
"left_table",
new StringSerializer(),
@@ -295,8 +306,9 @@ public class KTableKTableForeignKeyJoinScenarioTest {
}
}
- @Test
- public void shouldWorkWithCompositeKeyAndBiFunctionExtractor() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldWorkWithCompositeKeyAndBiFunctionExtractor(final boolean withHeaders) {
final StreamsBuilder builder = new StreamsBuilder();
// Left table keyed by <producer_id, product_id>
@@ -321,7 +333,7 @@ public class KTableKTableForeignKeyJoinScenarioTest {
joined.toStream().to("output");
- try (final TopologyTestDriver driver = createTopologyTestDriver(builder)) {
+ try (final TopologyTestDriver driver = createTopologyTestDriver(builder, withHeaders)) {
final TestInputTopic<String, String> leftInput = driver.createInputTopic(
"left_table",
new StringSerializer(),
@@ -355,21 +367,27 @@ public class KTableKTableForeignKeyJoinScenarioTest {
}
}
- private TopologyTestDriver createTopologyTestDriver(final StreamsBuilder builder) {
+ private TopologyTestDriver createTopologyTestDriver(final StreamsBuilder builder, final boolean withHeaders) {
final Properties config = new Properties();
config.setProperty(StreamsConfig.APPLICATION_ID_CONFIG, "test-app");
config.setProperty(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:1234");
config.setProperty(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getAbsolutePath());
config.setProperty(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
config.setProperty(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
+ if (withHeaders) {
+ config.setProperty(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
+ }
return new TopologyTestDriver(builder.build(), config);
}
- private void validateTopologyCanProcessData(final StreamsBuilder builder) {
+ private void validateTopologyCanProcessData(final StreamsBuilder builder, final boolean withHeaders) {
final Properties config = new Properties();
config.setProperty(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.IntegerSerde.class.getName());
config.setProperty(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.StringSerde.class.getName());
config.setProperty(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getAbsolutePath());
+ if (withHeaders) {
+ config.setProperty(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
+ }
try (final TopologyTestDriver topologyTestDriver = new TopologyTestDriver(builder.build(), config)) {
final TestInputTopic<Integer, String> aTopic = topologyTestDriver.createInputTopic("A", new IntegerSerializer(), new StringSerializer());
final TestInputTopic<Integer, String> bTopic = topologyTestDriver.createInputTopic("B", new IntegerSerializer(), new StringSerializer());
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 9d1e7adec95..d336b69a324 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
@@ -21,6 +21,7 @@ import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.common.utils.LogCaptureAppender;
import org.apache.kafka.streams.KeyValueTimestamp;
import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.TestInputTopic;
import org.apache.kafka.streams.TestOutputTopic;
import org.apache.kafka.streams.TopologyTestDriver;
@@ -38,7 +39,8 @@ import org.apache.kafka.test.MockApiProcessorSupplier;
import org.apache.kafka.test.MockValueJoiner;
import org.apache.kafka.test.StreamsTestUtils;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.time.Duration;
import java.time.Instant;
@@ -62,10 +64,12 @@ public class KTableKTableInnerJoinTest {
private final Consumed<Integer, String> consumed = Consumed.with(Serdes.Integer(), Serdes.String());
private final Materialized<Integer, String, KeyValueStore<Bytes, byte[]>> materialized =
Materialized.with(Serdes.Integer(), Serdes.String());
- private final Properties props = StreamsTestUtils.getStreamsConfig(Serdes.Integer(), Serdes.String());
+ private Properties props = StreamsTestUtils.getStreamsConfig(Serdes.Integer(), Serdes.String());
- @Test
- public void testJoin() {
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testJoin(final boolean withHeaders) {
final StreamsBuilder builder = new StreamsBuilder();
final int[] expectedKeys = new int[] {0, 1, 2, 3};
@@ -78,11 +82,12 @@ public class KTableKTableInnerJoinTest {
joined = table1.join(table2, MockValueJoiner.TOSTRING_JOINER);
joined.toStream().to(output);
- doTestJoin(builder, expectedKeys);
+ doTestJoin(builder, expectedKeys, withHeaders);
}
- @Test
- public void testQueryableJoin() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testQueryableJoin(final boolean withHeaders) {
final StreamsBuilder builder = new StreamsBuilder();
final int[] expectedKeys = new int[] {0, 1, 2, 3};
@@ -95,11 +100,12 @@ public class KTableKTableInnerJoinTest {
table3 = table1.join(table2, MockValueJoiner.TOSTRING_JOINER, materialized);
table3.toStream().to(output);
- doTestJoin(builder, expectedKeys);
+ doTestJoin(builder, expectedKeys, withHeaders);
}
- @Test
- public void testQueryableNotSendingOldValues() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testQueryableNotSendingOldValues(final boolean withHeaders) {
final StreamsBuilder builder = new StreamsBuilder();
final int[] expectedKeys = new int[] {0, 1, 2, 3};
@@ -114,11 +120,12 @@ public class KTableKTableInnerJoinTest {
joined = table1.join(table2, MockValueJoiner.TOSTRING_JOINER, materialized);
builder.build().addProcessor("proc", supplier, ((KTableImpl<?, ?, ?>) joined).name);
- doTestNotSendingOldValues(builder, expectedKeys, table1, table2, supplier, joined);
+ doTestNotSendingOldValues(builder, expectedKeys, table1, table2, supplier, joined, withHeaders);
}
- @Test
- public void testNotSendingOldValues() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testNotSendingOldValues(final boolean withHeaders) {
final StreamsBuilder builder = new StreamsBuilder();
final int[] expectedKeys = new int[] {0, 1, 2, 3};
@@ -133,11 +140,12 @@ public class KTableKTableInnerJoinTest {
joined = table1.join(table2, MockValueJoiner.TOSTRING_JOINER);
builder.build().addProcessor("proc", supplier, ((KTableImpl<?, ?, ?>) joined).name);
- doTestNotSendingOldValues(builder, expectedKeys, table1, table2, supplier, joined);
+ doTestNotSendingOldValues(builder, expectedKeys, table1, table2, supplier, joined, withHeaders);
}
- @Test
- public void testSendingOldValues() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testSendingOldValues(final boolean withHeaders) {
final StreamsBuilder builder = new StreamsBuilder();
final int[] expectedKeys = new int[] {0, 1, 2, 3};
@@ -155,6 +163,7 @@ public class KTableKTableInnerJoinTest {
builder.build().addProcessor("proc", supplier, ((KTableImpl<?, ?, ?>) joined).name);
+ setDslStoreFormat(withHeaders);
try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {
final TestInputTopic<Integer, String> inputTopic1 =
driver.createInputTopic(topic1, Serdes.Integer().serializer(), Serdes.String().serializer(), Instant.ofEpochMilli(0L), Duration.ZERO);
@@ -245,8 +254,9 @@ public class KTableKTableInnerJoinTest {
}
}
- @Test
- public void shouldLogAndMeterSkippedRecordsDueToNullLeftKey() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldLogAndMeterSkippedRecordsDueToNullLeftKey(final boolean withHeaders) {
final StreamsBuilder builder = new StreamsBuilder();
@SuppressWarnings("unchecked")
@@ -256,6 +266,7 @@ public class KTableKTableInnerJoinTest {
null
).get();
+ setDslStoreFormat(withHeaders);
final MockProcessorContext<String, Change<Object>> context = new MockProcessorContext<>(props);
context.setRecordMetadata("left", -1, -2);
join.init(context);
@@ -275,8 +286,10 @@ public class KTableKTableInnerJoinTest {
final KTable<Integer, String> table1,
final KTable<Integer, String> table2,
final MockApiProcessorSupplier<Integer, String, Void, Void> supplier,
- final KTable<Integer, String> joined) {
+ final KTable<Integer, String> joined,
+ final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {
final TestInputTopic<Integer, String> inputTopic1 =
driver.createInputTopic(topic1, Serdes.Integer().serializer(), Serdes.String().serializer(), Instant.ofEpochMilli(0L), Duration.ZERO);
@@ -367,13 +380,14 @@ public class KTableKTableInnerJoinTest {
}
}
- private void doTestJoin(final StreamsBuilder builder, final int[] expectedKeys) {
+ private void doTestJoin(final StreamsBuilder builder, final int[] expectedKeys, final boolean withHeaders) {
final Collection<Set<String>> copartitionGroups =
TopologyWrapper.getInternalTopologyBuilder(builder.build()).copartitionGroups();
assertEquals(1, copartitionGroups.size());
assertEquals(Set.of(topic1, topic2), copartitionGroups.iterator().next());
+ setDslStoreFormat(withHeaders);
try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {
final TestInputTopic<Integer, String> inputTopic1 =
driver.createInputTopic(topic1, Serdes.Integer().serializer(), Serdes.String().serializer(), Instant.ofEpochMilli(0L), Duration.ZERO);
@@ -477,4 +491,19 @@ public class KTableKTableInnerJoinTest {
assertThat(outputTopic.readRecord(), equalTo(new TestRecord<>(expectedKey, expectedValue, null, expectedTimestamp)));
}
+ /**
+ * Configures the DSL store format to use headers if enabled.
+ * This is a helper method to reduce boilerplate in parameterized tests that test both
+ * with and without headers mode.
+ *
+ * @param withHeaders Whether to enable headers mode
+ */
+ private void setDslStoreFormat(final boolean withHeaders) {
+ if (withHeaders) {
+ props.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
+ } else {
+ props.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_DEFAULT);
+ }
+ }
+
}
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 87a38cc7a44..de56b921787 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
@@ -46,7 +46,8 @@ import org.apache.kafka.test.MockReducer;
import org.apache.kafka.test.MockValueJoiner;
import org.apache.kafka.test.StreamsTestUtils;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.time.Duration;
import java.time.Instant;
@@ -69,16 +70,20 @@ public class KTableKTableLeftJoinTest {
private final String topic2 = "topic2";
private final String output = "output";
private final Consumed<Integer, String> consumed = Consumed.with(Serdes.Integer(), Serdes.String());
- private final Properties props = StreamsTestUtils.getStreamsConfig(Serdes.Integer(), Serdes.String());
+ private Properties props = StreamsTestUtils.getStreamsConfig(Serdes.Integer(), Serdes.String());
- private StreamsBuilder createStreamBuilderInMemory() {
+ private StreamsBuilder createStreamBuilderInMemory(final boolean withHeaders) {
props.put(StreamsConfig.DSL_STORE_SUPPLIERS_CLASS_CONFIG, BuiltInDslStoreSuppliers.InMemoryDslStoreSuppliers.class.getName());
+ if (withHeaders) {
+ props.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
+ }
return new StreamsBuilder(new TopologyConfig(new StreamsConfig(props)));
}
- @Test
- public void testJoin() {
- final StreamsBuilder builder = createStreamBuilderInMemory();
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testJoin(final boolean withHeaders) {
+ final StreamsBuilder builder = createStreamBuilderInMemory(withHeaders);
final int[] expectedKeys = new int[] {0, 1, 2, 3};
@@ -197,9 +202,10 @@ public class KTableKTableLeftJoinTest {
}
}
- @Test
- public void testNotSendingOldValue() {
- final StreamsBuilder builder = createStreamBuilderInMemory();
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testNotSendingOldValue(final boolean withHeaders) {
+ final StreamsBuilder builder = createStreamBuilderInMemory(withHeaders);
final int[] expectedKeys = new int[] {0, 1, 2, 3};
@@ -313,9 +319,10 @@ public class KTableKTableLeftJoinTest {
}
}
- @Test
- public void testSendingOldValue() {
- final StreamsBuilder builder = createStreamBuilderInMemory();
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testSendingOldValue(final boolean withHeaders) {
+ final StreamsBuilder builder = createStreamBuilderInMemory(withHeaders);
final int[] expectedKeys = new int[] {0, 1, 2, 3};
@@ -438,8 +445,9 @@ public class KTableKTableLeftJoinTest {
* It is based on a fairly complicated join used by the developer that reported the bug.
* Before the fix this would trigger an IllegalStateException.
*/
- @Test
- public void shouldNotThrowIllegalStateExceptionWhenMultiCacheEvictions() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotThrowIllegalStateExceptionWhenMultiCacheEvictions(final boolean withHeaders) {
final String agg = "agg";
final String tableOne = "tableOne";
final String tableTwo = "tableTwo";
@@ -449,7 +457,7 @@ public class KTableKTableLeftJoinTest {
final String tableSix = "tableSix";
final String[] inputs = {agg, tableOne, tableTwo, tableThree, tableFour, tableFive, tableSix};
- final StreamsBuilder builder = createStreamBuilderInMemory();
+ final StreamsBuilder builder = createStreamBuilderInMemory(withHeaders);
final Consumed<Long, String> consumed = Consumed.with(Serdes.Long(), Serdes.String());
final KTable<Long, String> aggTable = builder
.table(agg, consumed, Materialized.as(Stores.inMemoryKeyValueStore("agg-base-store")))
@@ -501,9 +509,10 @@ public class KTableKTableLeftJoinTest {
}
}
- @Test
- public void shouldLogAndMeterSkippedRecordsDueToNullLeftKey() {
- final StreamsBuilder builder = createStreamBuilderInMemory();
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldLogAndMeterSkippedRecordsDueToNullLeftKey(final boolean withHeaders) {
+ final StreamsBuilder builder = createStreamBuilderInMemory(withHeaders);
@SuppressWarnings("unchecked")
final Processor<String, Change<String>, String, Change<Object>> join = new KTableKTableLeftJoin<>(
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 080cdbd82d1..886e6db4f9e 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
@@ -20,6 +20,7 @@ import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.LogCaptureAppender;
import org.apache.kafka.streams.KeyValueTimestamp;
import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.TestInputTopic;
import org.apache.kafka.streams.TestOutputTopic;
import org.apache.kafka.streams.Topology;
@@ -36,7 +37,8 @@ import org.apache.kafka.test.MockApiProcessorSupplier;
import org.apache.kafka.test.MockValueJoiner;
import org.apache.kafka.test.StreamsTestUtils;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.time.Duration;
import java.time.Instant;
@@ -58,8 +60,10 @@ public class KTableKTableOuterJoinTest {
private final Consumed<Integer, String> consumed = Consumed.with(Serdes.Integer(), Serdes.String());
private final Properties props = StreamsTestUtils.getStreamsConfig(Serdes.Integer(), Serdes.String());
- @Test
- public void testJoin() {
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testJoin(final boolean withHeaders) {
final StreamsBuilder builder = new StreamsBuilder();
final int[] expectedKeys = new int[]{0, 1, 2, 3};
@@ -79,6 +83,7 @@ public class KTableKTableOuterJoinTest {
assertEquals(1, copartitionGroups.size());
assertEquals(Set.of(topic1, topic2), copartitionGroups.iterator().next());
+ setDslStoreFormat(withHeaders);
try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {
final TestInputTopic<Integer, String> inputTopic1 =
driver.createInputTopic(topic1, Serdes.Integer().serializer(), Serdes.String().serializer(), Instant.ofEpochMilli(0L), Duration.ZERO);
@@ -182,8 +187,9 @@ public class KTableKTableOuterJoinTest {
}
}
- @Test
- public void testNotSendingOldValue() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testNotSendingOldValue(final boolean withHeaders) {
final StreamsBuilder builder = new StreamsBuilder();
final int[] expectedKeys = new int[]{0, 1, 2, 3};
@@ -199,6 +205,7 @@ public class KTableKTableOuterJoinTest {
final Topology topology = builder.build().addProcessor("proc", supplier, ((KTableImpl<?, ?, ?>) joined).name);
+ setDslStoreFormat(withHeaders);
try (final TopologyTestDriver driver = new TopologyTestDriver(topology, props)) {
final TestInputTopic<Integer, String> inputTopic1 =
driver.createInputTopic(topic1, Serdes.Integer().serializer(), Serdes.String().serializer(), Instant.ofEpochMilli(0L), Duration.ZERO);
@@ -291,8 +298,9 @@ public class KTableKTableOuterJoinTest {
}
}
- @Test
- public void testSendingOldValue() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testSendingOldValue(final boolean withHeaders) {
final StreamsBuilder builder = new StreamsBuilder();
final int[] expectedKeys = new int[]{0, 1, 2, 3};
@@ -310,6 +318,7 @@ public class KTableKTableOuterJoinTest {
final Topology topology = builder.build().addProcessor("proc", supplier, ((KTableImpl<?, ?, ?>) joined).name);
+ setDslStoreFormat(withHeaders);
try (final TopologyTestDriver driver = new TopologyTestDriver(topology, props)) {
final TestInputTopic<Integer, String> inputTopic1 =
driver.createInputTopic(topic1, Serdes.Integer().serializer(), Serdes.String().serializer(), Instant.ofEpochMilli(0L), Duration.ZERO);
@@ -402,8 +411,9 @@ public class KTableKTableOuterJoinTest {
}
}
- @Test
- public void shouldLogAndMeterSkippedRecordsDueToNullLeftKey() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldLogAndMeterSkippedRecordsDueToNullLeftKey(final boolean withHeaders) {
final StreamsBuilder builder = new StreamsBuilder();
@SuppressWarnings("unchecked")
@@ -413,6 +423,7 @@ public class KTableKTableOuterJoinTest {
null
).get();
+ setDslStoreFormat(withHeaders);
final MockProcessorContext<String, Change<Object>> context = new MockProcessorContext<>(props);
context.setRecordMetadata("left", -1, -2);
join.init(context);
@@ -434,4 +445,19 @@ public class KTableKTableOuterJoinTest {
assertThat(outputTopic.readRecord(), equalTo(new TestRecord<>(expectedKey, expectedValue, null, expectedTimestamp)));
}
+ /**
+ * Configures the DSL store format to use headers if enabled.
+ * This is a helper method to reduce boilerplate in parameterized tests that test both
+ * with and without headers mode.
+ *
+ * @param withHeaders Whether to enable headers mode
+ */
+ private void setDslStoreFormat(final boolean withHeaders) {
+ if (withHeaders) {
+ props.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
+ } else {
+ props.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_DEFAULT);
+ }
+ }
+
}
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 61be038e562..416968bf8a4 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
@@ -27,7 +27,8 @@ import org.apache.kafka.streams.processor.api.Processor;
import org.apache.kafka.streams.processor.api.Record;
import org.apache.kafka.test.StreamsTestUtils;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.util.Properties;
import java.util.stream.Collectors;
@@ -39,8 +40,9 @@ public class KTableKTableRightJoinTest {
private final Properties props = StreamsTestUtils.getStreamsConfig(Serdes.String(), Serdes.String());
- @Test
- public void shouldLogAndMeterSkippedRecordsDueToNullLeftKeyWithBuiltInMetricsVersionLatest() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldLogAndMeterSkippedRecordsDueToNullLeftKeyWithBuiltInMetricsVersionLatest(final boolean withHeaders) {
final StreamsBuilder builder = new StreamsBuilder();
@SuppressWarnings("unchecked")
@@ -50,6 +52,7 @@ public class KTableKTableRightJoinTest {
null
).get();
+ setDslStoreFormat(withHeaders);
props.setProperty(StreamsConfig.BUILT_IN_METRICS_VERSION_CONFIG, StreamsConfig.METRICS_LATEST);
final MockProcessorContext<String, Change<Object>> context = new MockProcessorContext<>(props);
context.setRecordMetadata("left", -1, -2);
@@ -67,4 +70,19 @@ public class KTableKTableRightJoinTest {
);
}
}
+
+ /**
+ * Configures the DSL store format to use headers if enabled.
+ * This is a helper method to reduce boilerplate in parameterized tests that test both
+ * with and without headers mode.
+ *
+ * @param withHeaders Whether to enable headers mode
+ */
+ private void setDslStoreFormat(final boolean withHeaders) {
+ if (withHeaders) {
+ props.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
+ } else {
+ props.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_DEFAULT);
+ }
+ }
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableMapKeysTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableMapKeysTest.java
index 0ecf7857fd3..ab7829b5aa0 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableMapKeysTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableMapKeysTest.java
@@ -21,6 +21,7 @@ import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.streams.KeyValueTimestamp;
import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.TestInputTopic;
import org.apache.kafka.streams.TopologyTestDriver;
import org.apache.kafka.streams.kstream.Consumed;
@@ -29,7 +30,8 @@ import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.test.MockApiProcessorSupplier;
import org.apache.kafka.test.StreamsTestUtils;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.util.HashMap;
import java.util.Map;
@@ -40,8 +42,10 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
public class KTableMapKeysTest {
private final Properties props = StreamsTestUtils.getStreamsConfig(Serdes.Integer(), Serdes.String());
- @Test
- public void testMapKeysConvertingToStream() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testMapKeysConvertingToStream(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final String topic1 = "topic_map_keys";
@@ -76,4 +80,19 @@ public class KTableMapKeysTest {
assertEquals(expected[i], supplier.theCapturedProcessor().processed().get(i));
}
}
+
+ /**
+ * Configures the DSL store format to use headers if enabled.
+ * This is a helper method to reduce boilerplate in parameterized tests that test both
+ * with and without headers mode.
+ *
+ * @param withHeaders Whether to enable headers mode
+ */
+ private void setDslStoreFormat(final boolean withHeaders) {
+ if (withHeaders) {
+ props.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
+ } else {
+ props.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_DEFAULT);
+ }
+ }
}
\ No newline at end of file
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 6ca6a8320d7..fc1ece7f1db 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
@@ -22,6 +22,7 @@ import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.KeyValueTimestamp;
import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.TestInputTopic;
import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.TopologyTestDriver;
@@ -37,7 +38,8 @@ import org.apache.kafka.test.MockApiProcessor;
import org.apache.kafka.test.MockApiProcessorSupplier;
import org.apache.kafka.test.StreamsTestUtils;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.time.Duration;
import java.time.Instant;
@@ -55,6 +57,7 @@ public class KTableMapValuesTest {
private final Consumed<String, String> consumed = Consumed.with(Serdes.String(), Serdes.String());
private final Properties props = StreamsTestUtils.getStreamsConfig(Serdes.String(), Serdes.String());
+
private void doTestKTable(final StreamsBuilder builder,
final String topic1,
final MockApiProcessorSupplier<String, Integer, Void, Void> supplier) {
@@ -72,8 +75,10 @@ public class KTableMapValuesTest {
}
}
- @Test
- public void testKTable() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testKTable(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final String topic1 = "topic1";
@@ -86,8 +91,10 @@ public class KTableMapValuesTest {
doTestKTable(builder, topic1, supplier);
}
- @Test
- public void testQueryableKTable() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testQueryableKTable(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final String topic1 = "topic1";
@@ -172,8 +179,10 @@ public class KTableMapValuesTest {
}
}
- @Test
- public void testQueryableValueGetter() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testQueryableValueGetter(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final String topic1 = "topic1";
final String storeName2 = "store2";
@@ -201,8 +210,10 @@ public class KTableMapValuesTest {
doTestValueGetter(builder, topic1, table2, table3);
}
- @Test
- public void testNotSendingOldValue() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testNotSendingOldValue(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final String topic1 = "topic1";
@@ -242,8 +253,10 @@ public class KTableMapValuesTest {
}
}
- @Test
- public void shouldEnableSendingOldValuesOnParentIfMapValuesNotMaterialized() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldEnableSendingOldValuesOnParentIfMapValuesNotMaterialized(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final String topic1 = "topic1";
@@ -260,8 +273,10 @@ public class KTableMapValuesTest {
testSendingOldValues(builder, topic1, table2);
}
- @Test
- public void shouldNotEnableSendingOldValuesOnParentIfMapValuesMaterialized() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotEnableSendingOldValuesOnParentIfMapValuesMaterialized(final boolean withHeaders) {
+ setDslStoreFormat(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final String topic1 = "topic1";
@@ -321,4 +336,19 @@ public class KTableMapValuesTest {
);
}
}
+
+ /**
+ * Configures the DSL store format to use headers if enabled.
+ * This is a helper method to reduce boilerplate in parameterized tests that test both
+ * with and without headers mode.
+ *
+ * @param withHeaders Whether to enable headers mode
+ */
+ private void setDslStoreFormat(final boolean withHeaders) {
+ if (withHeaders) {
+ props.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
+ } else {
+ props.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_DEFAULT);
+ }
+ }
}
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 625556d7454..7c84d4437fe 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
@@ -54,6 +54,8 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import org.mockito.junit.jupiter.MockitoSettings;
@@ -353,8 +355,9 @@ public class KTableTransformValuesTest {
getter.close();
}
- @Test
- public void shouldTransformValuesWithKey() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldTransformValuesWithKey(final boolean withHeaders) {
builder
.addStateStore(storeBuilder(STORE_NAME))
.addStateStore(storeBuilder(OTHER_STORE_NAME))
@@ -365,7 +368,7 @@ public class KTableTransformValuesTest {
.toStream()
.process(capture);
- driver = new TopologyTestDriver(builder.build(), props());
+ driver = new TopologyTestDriver(builder.build(), props(withHeaders));
final TestInputTopic<String, String> inputTopic =
driver.createInputTopic(INPUT_TOPIC, new StringSerializer(), new StringSerializer());
@@ -381,8 +384,9 @@ public class KTableTransformValuesTest {
assertNull(driver.getKeyValueStore(QUERYABLE_NAME), "Store should not be materialized");
}
- @Test
- public void shouldTransformValuesWithKeyAndMaterialize() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldTransformValuesWithKeyAndMaterialize(final boolean withHeaders) {
builder
.addStateStore(storeBuilder(STORE_NAME))
.table(INPUT_TOPIC, CONSUMED)
@@ -395,7 +399,7 @@ public class KTableTransformValuesTest {
.toStream()
.process(capture);
- driver = new TopologyTestDriver(builder.build(), props());
+ driver = new TopologyTestDriver(builder.build(), props(withHeaders));
final TestInputTopic<String, String> inputTopic =
driver.createInputTopic(INPUT_TOPIC, new StringSerializer(), new StringSerializer());
inputTopic.pipeInput("A", "a", 5L);
@@ -420,8 +424,9 @@ public class KTableTransformValuesTest {
}
}
- @Test
- public void shouldCalculateCorrectOldValuesIfMaterializedEvenIfStateful() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldCalculateCorrectOldValuesIfMaterializedEvenIfStateful(final boolean withHeaders) {
builder
.table(INPUT_TOPIC, CONSUMED)
.transformValues(
@@ -435,7 +440,7 @@ public class KTableTransformValuesTest {
.toStream()
.process(capture);
- driver = new TopologyTestDriver(builder.build(), props());
+ driver = new TopologyTestDriver(builder.build(), props(withHeaders));
final TestInputTopic<String, String> inputTopic =
driver.createInputTopic(INPUT_TOPIC, new StringSerializer(), new StringSerializer());
@@ -453,8 +458,9 @@ public class KTableTransformValuesTest {
equalTo(Set.of(QUERYABLE_NAME, "KTABLE-AGGREGATE-STATE-STORE-0000000005")));
}
- @Test
- public void shouldCalculateCorrectOldValuesIfNotStatefulEvenIfNotMaterialized() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldCalculateCorrectOldValuesIfNotStatefulEvenIfNotMaterialized(final boolean withHeaders) {
builder
.table(INPUT_TOPIC, CONSUMED)
.transformValues(new StatelessTransformerSupplier())
@@ -464,7 +470,7 @@ public class KTableTransformValuesTest {
.toStream()
.process(capture);
- driver = new TopologyTestDriver(builder.build(), props());
+ driver = new TopologyTestDriver(builder.build(), props(withHeaders));
final TestInputTopic<String, String> inputTopic =
driver.createInputTopic(INPUT_TOPIC, new StringSerializer(), new StringSerializer());
@@ -479,8 +485,9 @@ public class KTableTransformValuesTest {
equalTo(Set.of("inputTopic-STATE-STORE-0000000000", "KTABLE-AGGREGATE-STATE-STORE-0000000005")));
}
- @Test
- public void shouldCalculateCorrectOldValuesIfNotStatefulEvenNotMaterializedNoQueryableName() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldCalculateCorrectOldValuesIfNotStatefulEvenNotMaterializedNoQueryableName(final boolean withHeaders) {
builder
.table(INPUT_TOPIC, CONSUMED)
.transformValues(new StatelessTransformerSupplier(),
@@ -492,7 +499,7 @@ public class KTableTransformValuesTest {
.toStream()
.process(capture);
- driver = new TopologyTestDriver(builder.build(), props());
+ driver = new TopologyTestDriver(builder.build(), props(withHeaders));
final TestInputTopic<String, String> inputTopic =
driver.createInputTopic(INPUT_TOPIC, new StringSerializer(), new StringSerializer());
@@ -523,11 +530,14 @@ public class KTableTransformValuesTest {
return Stores.keyValueStoreBuilder(Stores.persistentKeyValueStore(storeName), Serdes.Long(), Serdes.Long());
}
- public static Properties props() {
+ public static Properties props(final boolean withHeaders) {
final Properties props = new Properties();
props.setProperty(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getAbsolutePath());
props.setProperty(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.Integer().getClass().getName());
props.setProperty(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.Integer().getClass().getName());
+ if (withHeaders) {
+ props.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
+ }
return props;
}
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 b0fdf879c86..24072ba7107 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
@@ -22,6 +22,7 @@ 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.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.TestInputTopic;
import org.apache.kafka.streams.TestOutputTopic;
import org.apache.kafka.streams.TopologyTestDriver;
@@ -43,8 +44,8 @@ import org.apache.kafka.test.MockAggregator;
import org.apache.kafka.test.MockInitializer;
import org.apache.kafka.test.StreamsTestUtils;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.util.LinkedList;
import java.util.List;
@@ -70,8 +71,12 @@ public class SlidingWindowedCogroupedKStreamImplTest {
private final Properties props = StreamsTestUtils.getStreamsConfig(Serdes.String(), Serdes.String());
- @BeforeEach
- public void setup() {
+ public void setup(final boolean withHeaders) {
+ if (withHeaders) {
+ props.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
+ } else {
+ props.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_DEFAULT);
+ }
final KStream<String, String> stream = builder.stream(TOPIC, Consumed
.with(Serdes.String(), Serdes.String()));
final KStream<String, String> stream2 = builder.stream(TOPIC2, Consumed
@@ -85,48 +90,66 @@ public class SlidingWindowedCogroupedKStreamImplTest {
WINDOW_SIZE_MS), ofMillis(2000L)));
}
- @Test
- public void shouldNotHaveNullInitializerOnAggregate() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotHaveNullInitializerOnAggregate(final boolean withHeaders) {
+ setup(withHeaders);
assertThrows(NullPointerException.class, () -> windowedCogroupedStream.aggregate(null));
}
- @Test
- public void shouldNotHaveNullMaterializedOnTwoOptionAggregate() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotHaveNullMaterializedOnTwoOptionAggregate(final boolean withHeaders) {
+ setup(withHeaders);
assertThrows(NullPointerException.class, () -> windowedCogroupedStream.aggregate(MockInitializer.STRING_INIT, (Materialized<String, String, WindowStore<Bytes, byte[]>>) null));
}
- @Test
- public void shouldNotHaveNullNamedTwoOptionOnAggregate() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotHaveNullNamedTwoOptionOnAggregate(final boolean withHeaders) {
+ setup(withHeaders);
assertThrows(NullPointerException.class, () -> windowedCogroupedStream.aggregate(MockInitializer.STRING_INIT, (Named) null));
}
- @Test
- public void shouldNotHaveNullInitializerTwoOptionNamedOnAggregate() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotHaveNullInitializerTwoOptionNamedOnAggregate(final boolean withHeaders) {
+ setup(withHeaders);
assertThrows(NullPointerException.class, () -> windowedCogroupedStream.aggregate(null, Named.as("test")));
}
- @Test
- public void shouldNotHaveNullInitializerTwoOptionMaterializedOnAggregate() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotHaveNullInitializerTwoOptionMaterializedOnAggregate(final boolean withHeaders) {
+ setup(withHeaders);
assertThrows(NullPointerException.class, () -> windowedCogroupedStream.aggregate(null, Materialized.as("test")));
}
- @Test
- public void shouldNotHaveNullInitializerThreeOptionOnAggregate() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotHaveNullInitializerThreeOptionOnAggregate(final boolean withHeaders) {
+ setup(withHeaders);
assertThrows(NullPointerException.class, () -> windowedCogroupedStream.aggregate(null, Named.as("test"), Materialized.as("test")));
}
- @Test
- public void shouldNotHaveNullMaterializedOnAggregate() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotHaveNullMaterializedOnAggregate(final boolean withHeaders) {
+ setup(withHeaders);
assertThrows(NullPointerException.class, () -> windowedCogroupedStream.aggregate(MockInitializer.STRING_INIT, Named.as("Test"), null));
}
- @Test
- public void shouldNotHaveNullNamedOnAggregate() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotHaveNullNamedOnAggregate(final boolean withHeaders) {
+ setup(withHeaders);
assertThrows(NullPointerException.class, () -> windowedCogroupedStream.aggregate(MockInitializer.STRING_INIT, null, Materialized.as("test")));
}
- @Test
- public void namedParamShouldSetName() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void namedParamShouldSetName(final boolean withHeaders) {
+ setup(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final KStream<String, String> stream = builder.stream(TOPIC, Consumed
.with(Serdes.String(), Serdes.String()));
@@ -148,8 +171,10 @@ public class SlidingWindowedCogroupedKStreamImplTest {
" <-- foo-cogroup-agg-0\n\n"));
}
- @Test
- public void slidingWindowAggregateStreamsTest() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void slidingWindowAggregateStreamsTest(final boolean withHeaders) {
+ setup(withHeaders);
final KTable<Windowed<String>, String> customers = windowedCogroupedStream.aggregate(
MockInitializer.STRING_INIT, Materialized.with(Serdes.String(), Serdes.String()));
customers.toStream().to(OUTPUT);
@@ -205,8 +230,10 @@ public class SlidingWindowedCogroupedKStreamImplTest {
}
}
- @Test
- public void slidingWindowAggregateOverlappingWindowsTest() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void slidingWindowAggregateOverlappingWindowsTest(final boolean withHeaders) {
+ setup(withHeaders);
final KTable<Windowed<String>, String> customers = groupedStream.cogroup(MockAggregator.TOSTRING_ADDER)
.windowedBy(SlidingWindows.ofTimeDifferenceAndGrace(ofMillis(WINDOW_SIZE_MS), ofMillis(2000L))).aggregate(
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 e9c52c33831..b7929d5f1be 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
@@ -22,6 +22,7 @@ import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.TestInputTopic;
import org.apache.kafka.streams.TopologyTestDriver;
import org.apache.kafka.streams.kstream.Consumed;
@@ -42,8 +43,8 @@ import org.apache.kafka.test.MockInitializer;
import org.apache.kafka.test.MockReducer;
import org.apache.kafka.test.StreamsTestUtils;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.util.Arrays;
import java.util.List;
@@ -62,16 +63,22 @@ public class SlidingWindowedKStreamImplTest {
private final Properties props = StreamsTestUtils.getStreamsConfig(Serdes.String(), Serdes.String());
private TimeWindowedKStream<String, String> windowedStream;
- @BeforeEach
- public void before() {
+ public void before(final boolean withHeaders) {
+ if (withHeaders) {
+ props.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
+ } else {
+ props.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_DEFAULT);
+ }
final KStream<String, String> stream = builder.stream(TOPIC, Consumed.with(Serdes.String(), Serdes.String()));
windowedStream = stream.
groupByKey(Grouped.with(Serdes.String(), Serdes.String()))
.windowedBy(SlidingWindows.ofTimeDifferenceAndGrace(ofMillis(100L), ofMillis(1000L)));
}
- @Test
- public void shouldCountSlidingWindows() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldCountSlidingWindows(final boolean withHeaders) {
+ before(withHeaders);
final MockApiProcessorSupplier<Windowed<String>, Long, Void, Void> supplier = new MockApiProcessorSupplier<>();
windowedStream
.count()
@@ -111,8 +118,10 @@ public class SlidingWindowedKStreamImplTest {
equalTo(ValueAndTimestamp.make(1L, 200L)));
}
- @Test
- public void shouldReduceSlidingWindows() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldReduceSlidingWindows(final boolean withHeaders) {
+ before(withHeaders);
final MockApiProcessorSupplier<Windowed<String>, String, Void, Void> supplier = new MockApiProcessorSupplier<>();
windowedStream
.reduce(MockReducer.STRING_ADDER)
@@ -152,8 +161,10 @@ public class SlidingWindowedKStreamImplTest {
equalTo(ValueAndTimestamp.make("10", 200L)));
}
- @Test
- public void shouldAggregateSlidingWindows() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldAggregateSlidingWindows(final boolean withHeaders) {
+ before(withHeaders);
final MockApiProcessorSupplier<Windowed<String>, String, Void, Void> supplier = new MockApiProcessorSupplier<>();
windowedStream
.aggregate(
@@ -196,8 +207,10 @@ public class SlidingWindowedKStreamImplTest {
equalTo(ValueAndTimestamp.make("0+10", 200L)));
}
- @Test
- public void shouldMaterializeCount() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldMaterializeCount(final boolean withHeaders) {
+ before(withHeaders);
windowedStream.count(
Materialized.<String, Long, WindowStore<Bytes, byte[]>>as("count-store")
.withKeySerde(Serdes.String())
@@ -235,8 +248,10 @@ public class SlidingWindowedKStreamImplTest {
}
}
- @Test
- public void shouldMaterializeReduced() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldMaterializeReduced(final boolean withHeaders) {
+ before(withHeaders);
windowedStream.reduce(
MockReducer.STRING_ADDER,
Materialized.<String, String, WindowStore<Bytes, byte[]>>as("reduced")
@@ -275,8 +290,10 @@ public class SlidingWindowedKStreamImplTest {
}
}
- @Test
- public void shouldMaterializeAggregated() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldMaterializeAggregated(final boolean withHeaders) {
+ before(withHeaders);
windowedStream.aggregate(
MockInitializer.STRING_INIT,
MockAggregator.TOSTRING_ADDER,
@@ -316,28 +333,38 @@ public class SlidingWindowedKStreamImplTest {
}
}
- @Test
- public void shouldThrowNullPointerOnAggregateIfInitializerIsNull() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldThrowNullPointerOnAggregateIfInitializerIsNull(final boolean withHeaders) {
+ before(withHeaders);
assertThrows(NullPointerException.class, () -> windowedStream.aggregate(null, MockAggregator.TOSTRING_ADDER));
}
- @Test
- public void shouldThrowNullPointerOnAggregateIfAggregatorIsNull() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldThrowNullPointerOnAggregateIfAggregatorIsNull(final boolean withHeaders) {
+ before(withHeaders);
assertThrows(NullPointerException.class, () -> windowedStream.aggregate(MockInitializer.STRING_INIT, null));
}
- @Test
- public void shouldThrowNullPointerOnReduceIfReducerIsNull() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldThrowNullPointerOnReduceIfReducerIsNull(final boolean withHeaders) {
+ before(withHeaders);
assertThrows(NullPointerException.class, () -> windowedStream.reduce(null));
}
- @Test
- public void shouldThrowNullPointerOnMaterializedAggregateIfInitializerIsNull() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldThrowNullPointerOnMaterializedAggregateIfInitializerIsNull(final boolean withHeaders) {
+ before(withHeaders);
assertThrows(NullPointerException.class, () -> windowedStream.aggregate(null, MockAggregator.TOSTRING_ADDER, Materialized.as("store")));
}
- @Test
- public void shouldThrowNullPointerOnMaterializedAggregateIfAggregatorIsNull() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldThrowNullPointerOnMaterializedAggregateIfAggregatorIsNull(final boolean withHeaders) {
+ before(withHeaders);
assertThrows(NullPointerException.class, () -> windowedStream.aggregate(
MockInitializer.STRING_INIT,
null,
@@ -345,34 +372,46 @@ public class SlidingWindowedKStreamImplTest {
}
@SuppressWarnings("unchecked")
- @Test
- public void shouldThrowNullPointerOnMaterializedAggregateIfMaterializedIsNull() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldThrowNullPointerOnMaterializedAggregateIfMaterializedIsNull(final boolean withHeaders) {
+ before(withHeaders);
assertThrows(NullPointerException.class, () -> windowedStream.aggregate(MockInitializer.STRING_INIT, MockAggregator.TOSTRING_ADDER, (Materialized) null));
}
- @Test
- public void shouldThrowNullPointerOnMaterializedReduceIfReducerIsNull() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldThrowNullPointerOnMaterializedReduceIfReducerIsNull(final boolean withHeaders) {
+ before(withHeaders);
assertThrows(NullPointerException.class, () -> windowedStream.reduce(null, Materialized.as("store")));
}
- @Test
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
@SuppressWarnings("unchecked")
- public void shouldThrowNullPointerOnMaterializedReduceIfMaterializedIsNull() {
+ public void shouldThrowNullPointerOnMaterializedReduceIfMaterializedIsNull(final boolean withHeaders) {
+ before(withHeaders);
assertThrows(NullPointerException.class, () -> windowedStream.reduce(MockReducer.STRING_ADDER, (Materialized) null));
}
- @Test
- public void shouldThrowNullPointerOnMaterializedReduceIfNamedIsNull() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldThrowNullPointerOnMaterializedReduceIfNamedIsNull(final boolean withHeaders) {
+ before(withHeaders);
assertThrows(NullPointerException.class, () -> windowedStream.reduce(MockReducer.STRING_ADDER, (Named) null));
}
- @Test
- public void shouldThrowNullPointerOnCountIfMaterializedIsNull() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldThrowNullPointerOnCountIfMaterializedIsNull(final boolean withHeaders) {
+ before(withHeaders);
assertThrows(NullPointerException.class, () -> windowedStream.count((Materialized<String, Long, WindowStore<Bytes, byte[]>>) null));
}
- @Test
- public void shouldThrowIllegalArgumentWhenRetentionIsTooSmall() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldThrowIllegalArgumentWhenRetentionIsTooSmall(final boolean withHeaders) {
+ before(withHeaders);
assertThrows(IllegalArgumentException.class, () -> windowedStream
.aggregate(
MockInitializer.STRING_INIT,
@@ -386,8 +425,10 @@ public class SlidingWindowedKStreamImplTest {
);
}
- @Test
- public void shouldDropWindowsOutsideOfRetention() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldDropWindowsOutsideOfRetention(final boolean withHeaders) {
+ before(withHeaders);
final WindowBytesStoreSupplier storeSupplier = Stores.inMemoryWindowStore("aggregated", ofMillis(1200L), ofMillis(100L), false);
windowedStream.aggregate(
MockInitializer.STRING_INIT,
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 de157ce3fdc..c86315862d1 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
@@ -22,6 +22,7 @@ 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.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.TestInputTopic;
import org.apache.kafka.streams.TestOutputTopic;
import org.apache.kafka.streams.TopologyTestDriver;
@@ -43,8 +44,8 @@ import org.apache.kafka.test.MockAggregator;
import org.apache.kafka.test.MockInitializer;
import org.apache.kafka.test.StreamsTestUtils;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.util.Properties;
@@ -69,8 +70,12 @@ public class TimeWindowedCogroupedKStreamImplTest {
private final Properties props = StreamsTestUtils.getStreamsConfig(Serdes.String(), Serdes.String());
- @BeforeEach
- public void setup() {
+ public void setup(final boolean withHeaders) {
+ if (withHeaders) {
+ props.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
+ } else {
+ props.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_DEFAULT);
+ }
final KStream<String, String> stream = builder.stream(TOPIC, Consumed
.with(Serdes.String(), Serdes.String()));
final KStream<String, String> stream2 = builder.stream(TOPIC2, Consumed
@@ -83,49 +88,67 @@ public class TimeWindowedCogroupedKStreamImplTest {
windowedCogroupedStream = cogroupedStream.windowedBy(TimeWindows.ofSizeWithNoGrace(ofMillis(WINDOW_SIZE)));
}
- @Test
- public void shouldNotHaveNullInitializerOnAggregate() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotHaveNullInitializerOnAggregate(final boolean withHeaders) {
+ setup(withHeaders);
assertThrows(NullPointerException.class, () -> windowedCogroupedStream.aggregate(null));
}
- @Test
- public void shouldNotHaveNullMaterializedOnTwoOptionAggregate() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotHaveNullMaterializedOnTwoOptionAggregate(final boolean withHeaders) {
+ setup(withHeaders);
assertThrows(NullPointerException.class, () -> windowedCogroupedStream.aggregate(MockInitializer.STRING_INIT,
(Materialized<String, String, WindowStore<Bytes, byte[]>>) null));
}
- @Test
- public void shouldNotHaveNullNamedTwoOptionOnAggregate() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotHaveNullNamedTwoOptionOnAggregate(final boolean withHeaders) {
+ setup(withHeaders);
assertThrows(NullPointerException.class, () -> windowedCogroupedStream.aggregate(MockInitializer.STRING_INIT, (Named) null));
}
- @Test
- public void shouldNotHaveNullInitializerTwoOptionNamedOnAggregate() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotHaveNullInitializerTwoOptionNamedOnAggregate(final boolean withHeaders) {
+ setup(withHeaders);
assertThrows(NullPointerException.class, () -> windowedCogroupedStream.aggregate(null, Named.as("test")));
}
- @Test
- public void shouldNotHaveNullInitializerTwoOptionMaterializedOnAggregate() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotHaveNullInitializerTwoOptionMaterializedOnAggregate(final boolean withHeaders) {
+ setup(withHeaders);
assertThrows(NullPointerException.class, () -> windowedCogroupedStream.aggregate(null, Materialized.as("test")));
}
- @Test
- public void shouldNotHaveNullInitializerThreeOptionOnAggregate() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotHaveNullInitializerThreeOptionOnAggregate(final boolean withHeaders) {
+ setup(withHeaders);
assertThrows(NullPointerException.class, () -> windowedCogroupedStream.aggregate(null, Named.as("test"), Materialized.as("test")));
}
- @Test
- public void shouldNotHaveNullMaterializedOnAggregate() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotHaveNullMaterializedOnAggregate(final boolean withHeaders) {
+ setup(withHeaders);
assertThrows(NullPointerException.class, () -> windowedCogroupedStream.aggregate(MockInitializer.STRING_INIT, Named.as("Test"), null));
}
- @Test
- public void shouldNotHaveNullNamedOnAggregate() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void shouldNotHaveNullNamedOnAggregate(final boolean withHeaders) {
+ setup(withHeaders);
assertThrows(NullPointerException.class, () -> windowedCogroupedStream.aggregate(MockInitializer.STRING_INIT, null, Materialized.as("test")));
}
- @Test
- public void namedParamShouldSetName() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void namedParamShouldSetName(final boolean withHeaders) {
+ setup(withHeaders);
final StreamsBuilder builder = new StreamsBuilder();
final KStream<String, String> stream = builder.stream(TOPIC, Consumed
.with(Serdes.String(), Serdes.String()));
@@ -147,8 +170,10 @@ public class TimeWindowedCogroupedKStreamImplTest {
" <-- foo-cogroup-agg-0\n\n"));
}
- @Test
- public void timeWindowAggregateTestStreamsTest() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void timeWindowAggregateTestStreamsTest(final boolean withHeaders) {
+ setup(withHeaders);
final KTable<Windowed<String>, String> customers = windowedCogroupedStream.aggregate(
MockInitializer.STRING_INIT, Materialized.with(Serdes.String(), Serdes.String()));
@@ -181,8 +206,10 @@ public class TimeWindowedCogroupedKStreamImplTest {
}
- @Test
- public void timeWindowMixAggregatorsTest() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void timeWindowMixAggregatorsTest(final boolean withHeaders) {
+ setup(withHeaders);
final KTable<Windowed<String>, String> customers = windowedCogroupedStream.aggregate(
MockInitializer.STRING_INIT, Materialized.with(Serdes.String(), Serdes.String()));
@@ -217,8 +244,10 @@ public class TimeWindowedCogroupedKStreamImplTest {
}
- @Test
- public void timeWindowAggregateManyWindowsTest() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void timeWindowAggregateManyWindowsTest(final boolean withHeaders) {
+ setup(withHeaders);
final KTable<Windowed<String>, String> customers = groupedStream.cogroup(MockAggregator.TOSTRING_ADDER)
.windowedBy(TimeWindows.ofSizeWithNoGrace(ofMillis(500L))).aggregate(
@@ -243,8 +272,10 @@ public class TimeWindowedCogroupedKStreamImplTest {
}
}
- @Test
- public void timeWindowAggregateOverlappingWindowsTest() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void timeWindowAggregateOverlappingWindowsTest(final boolean withHeaders) {
+ setup(withHeaders);
final KTable<Windowed<String>, String> customers = groupedStream.cogroup(MockAggregator.TOSTRING_ADDER)
.windowedBy(TimeWindows.ofSizeWithNoGrace(ofMillis(500L)).advanceBy(ofMillis(200L))).aggregate(
@@ -277,8 +308,10 @@ public class TimeWindowedCogroupedKStreamImplTest {
}
}
- @Test
- public void timeWindowMixAggregatorsManyWindowsTest() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void timeWindowMixAggregatorsManyWindowsTest(final boolean withHeaders) {
+ setup(withHeaders);
final KTable<Windowed<String>, String> customers = windowedCogroupedStream.aggregate(
MockInitializer.STRING_INIT, Materialized.with(Serdes.String(), Serdes.String()));
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 c526b09f9c6..4fcdf72ae75 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
@@ -23,6 +23,7 @@ import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.KeyValueTimestamp;
import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.TestInputTopic;
import org.apache.kafka.streams.TopologyTestDriver;
import org.apache.kafka.streams.kstream.Consumed;
@@ -82,18 +83,25 @@ public class TimeWindowedKStreamImplTest {
public static Stream<Arguments> data() {
return Stream.of(
- Arguments.of(StrategyType.ON_WINDOW_UPDATE, true),
- Arguments.of(StrategyType.ON_WINDOW_UPDATE, false),
- Arguments.of(StrategyType.ON_WINDOW_CLOSE, true),
- Arguments.of(StrategyType.ON_WINDOW_CLOSE, false)
+ Arguments.of(StrategyType.ON_WINDOW_UPDATE, true, false),
+ Arguments.of(StrategyType.ON_WINDOW_UPDATE, true, true),
+ Arguments.of(StrategyType.ON_WINDOW_UPDATE, false, false),
+ Arguments.of(StrategyType.ON_WINDOW_UPDATE, false, true),
+ Arguments.of(StrategyType.ON_WINDOW_CLOSE, true, false),
+ Arguments.of(StrategyType.ON_WINDOW_CLOSE, true, true),
+ Arguments.of(StrategyType.ON_WINDOW_CLOSE, false, false),
+ Arguments.of(StrategyType.ON_WINDOW_CLOSE, false, true)
);
}
- public void setup(final StrategyType inputType, final boolean inputWithCache) {
+ public void setup(final StrategyType inputType, final boolean inputWithCache, final boolean withHeaders) {
type = inputType;
withCache = inputWithCache;
emitFinal = type.equals(StrategyType.ON_WINDOW_CLOSE);
emitStrategy = StrategyType.forType(type);
+ if (withHeaders) {
+ props.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
+ }
final KStream<String, String> stream = builder.stream(TOPIC, Consumed.with(Serdes.String(), Serdes.String()));
windowedStream = stream.groupByKey(Grouped.with(Serdes.String(), Serdes.String()))
.windowedBy(TimeWindows.ofSizeWithNoGrace(ofMillis(500L)));
@@ -101,8 +109,8 @@ public class TimeWindowedKStreamImplTest {
@ParameterizedTest
@MethodSource("data")
- public void shouldCountWindowed(final StrategyType inputType, final boolean inputWithCache) {
- setup(inputType, inputWithCache);
+ public void shouldCountWindowed(final StrategyType inputType, final boolean inputWithCache, final boolean withHeaders) {
+ setup(inputType, inputWithCache, withHeaders);
final MockApiProcessorSupplier<Windowed<String>, Long, Void, Void> supplier = new MockApiProcessorSupplier<>();
windowedStream
.emitStrategy(emitStrategy)
@@ -141,8 +149,8 @@ public class TimeWindowedKStreamImplTest {
@ParameterizedTest
@MethodSource("data")
- public void shouldReduceWindowed(final StrategyType inputType, final boolean inputWithCache) {
- setup(inputType, inputWithCache);
+ public void shouldReduceWindowed(final StrategyType inputType, final boolean inputWithCache, final boolean withHeaders) {
+ setup(inputType, inputWithCache, withHeaders);
final MockApiProcessorSupplier<Windowed<String>, String, Void, Void> supplier = new MockApiProcessorSupplier<>();
windowedStream
.emitStrategy(emitStrategy)
@@ -181,8 +189,8 @@ public class TimeWindowedKStreamImplTest {
@ParameterizedTest
@MethodSource("data")
- public void shouldAggregateWindowed(final StrategyType inputType, final boolean inputWithCache) {
- setup(inputType, inputWithCache);
+ public void shouldAggregateWindowed(final StrategyType inputType, final boolean inputWithCache, final boolean withHeaders) {
+ setup(inputType, inputWithCache, withHeaders);
final MockApiProcessorSupplier<Windowed<String>, String, Void, Void> supplier = new MockApiProcessorSupplier<>();
windowedStream
.emitStrategy(emitStrategy)
@@ -225,8 +233,8 @@ public class TimeWindowedKStreamImplTest {
@ParameterizedTest
@MethodSource("data")
- public void shouldMaterializeCount(final StrategyType inputType, final boolean inputWithCache) {
- setup(inputType, inputWithCache);
+ public void shouldMaterializeCount(final StrategyType inputType, final boolean inputWithCache, final boolean withHeaders) {
+ setup(inputType, inputWithCache, withHeaders);
windowedStream
.emitStrategy(emitStrategy)
.count(
@@ -292,8 +300,8 @@ public class TimeWindowedKStreamImplTest {
@ParameterizedTest
@MethodSource("data")
- public void shouldMaterializeReduced(final StrategyType inputType, final boolean inputWithCache) {
- setup(inputType, inputWithCache);
+ public void shouldMaterializeReduced(final StrategyType inputType, final boolean inputWithCache, final boolean withHeaders) {
+ setup(inputType, inputWithCache, withHeaders);
windowedStream.reduce(
MockReducer.STRING_ADDER,
setMaterializedCache(Materialized.<String, String, WindowStore<Bytes, byte[]>>as("reduced")
@@ -344,8 +352,8 @@ public class TimeWindowedKStreamImplTest {
@ParameterizedTest
@MethodSource("data")
- public void shouldMaterializeAggregated(final StrategyType inputType, final boolean inputWithCache) {
- setup(inputType, inputWithCache);
+ public void shouldMaterializeAggregated(final StrategyType inputType, final boolean inputWithCache, final boolean withHeaders) {
+ setup(inputType, inputWithCache, withHeaders);
windowedStream.aggregate(
MockInitializer.STRING_INIT,
MockAggregator.TOSTRING_ADDER,
@@ -396,29 +404,29 @@ public class TimeWindowedKStreamImplTest {
@ParameterizedTest
@MethodSource("data")
- public void shouldThrowNullPointerOnAggregateIfInitializerIsNull(final StrategyType inputType, final boolean inputWithCache) {
- setup(inputType, inputWithCache);
+ public void shouldThrowNullPointerOnAggregateIfInitializerIsNull(final StrategyType inputType, final boolean inputWithCache, final boolean withHeaders) {
+ setup(inputType, inputWithCache, withHeaders);
assertThrows(NullPointerException.class, () -> windowedStream.aggregate(null, MockAggregator.TOSTRING_ADDER));
}
@ParameterizedTest
@MethodSource("data")
- public void shouldThrowNullPointerOnAggregateIfAggregatorIsNull(final StrategyType inputType, final boolean inputWithCache) {
- setup(inputType, inputWithCache);
+ public void shouldThrowNullPointerOnAggregateIfAggregatorIsNull(final StrategyType inputType, final boolean inputWithCache, final boolean withHeaders) {
+ setup(inputType, inputWithCache, withHeaders);
assertThrows(NullPointerException.class, () -> windowedStream.aggregate(MockInitializer.STRING_INIT, null));
}
@ParameterizedTest
@MethodSource("data")
- public void shouldThrowNullPointerOnReduceIfReducerIsNull(final StrategyType inputType, final boolean inputWithCache) {
- setup(inputType, inputWithCache);
+ public void shouldThrowNullPointerOnReduceIfReducerIsNull(final StrategyType inputType, final boolean inputWithCache, final boolean withHeaders) {
+ setup(inputType, inputWithCache, withHeaders);
assertThrows(NullPointerException.class, () -> windowedStream.reduce(null));
}
@ParameterizedTest
@MethodSource("data")
- public void shouldThrowNullPointerOnMaterializedAggregateIfInitializerIsNull(final StrategyType inputType, final boolean inputWithCache) {
- setup(inputType, inputWithCache);
+ public void shouldThrowNullPointerOnMaterializedAggregateIfInitializerIsNull(final StrategyType inputType, final boolean inputWithCache, final boolean withHeaders) {
+ setup(inputType, inputWithCache, withHeaders);
assertThrows(NullPointerException.class, () -> windowedStream.aggregate(
null,
MockAggregator.TOSTRING_ADDER,
@@ -427,8 +435,8 @@ public class TimeWindowedKStreamImplTest {
@ParameterizedTest
@MethodSource("data")
- public void shouldThrowNullPointerOnMaterializedAggregateIfAggregatorIsNull(final StrategyType inputType, final boolean inputWithCache) {
- setup(inputType, inputWithCache);
+ public void shouldThrowNullPointerOnMaterializedAggregateIfAggregatorIsNull(final StrategyType inputType, final boolean inputWithCache, final boolean withHeaders) {
+ setup(inputType, inputWithCache, withHeaders);
assertThrows(NullPointerException.class, () -> windowedStream.aggregate(
MockInitializer.STRING_INIT,
null,
@@ -438,8 +446,8 @@ public class TimeWindowedKStreamImplTest {
@SuppressWarnings("unchecked")
@ParameterizedTest
@MethodSource("data")
- public void shouldThrowNullPointerOnMaterializedAggregateIfMaterializedIsNull(final StrategyType inputType, final boolean inputWithCache) {
- setup(inputType, inputWithCache);
+ public void shouldThrowNullPointerOnMaterializedAggregateIfMaterializedIsNull(final StrategyType inputType, final boolean inputWithCache, final boolean withHeaders) {
+ setup(inputType, inputWithCache, withHeaders);
assertThrows(NullPointerException.class, () -> windowedStream.aggregate(
MockInitializer.STRING_INIT,
MockAggregator.TOSTRING_ADDER,
@@ -448,8 +456,8 @@ public class TimeWindowedKStreamImplTest {
@ParameterizedTest
@MethodSource("data")
- public void shouldThrowNullPointerOnMaterializedReduceIfReducerIsNull(final StrategyType inputType, final boolean inputWithCache) {
- setup(inputType, inputWithCache);
+ public void shouldThrowNullPointerOnMaterializedReduceIfReducerIsNull(final StrategyType inputType, final boolean inputWithCache, final boolean withHeaders) {
+ setup(inputType, inputWithCache, withHeaders);
assertThrows(NullPointerException.class, () -> windowedStream.reduce(
null,
setMaterializedCache(Materialized.as("store"))));
@@ -458,8 +466,8 @@ public class TimeWindowedKStreamImplTest {
@ParameterizedTest
@MethodSource("data")
@SuppressWarnings("unchecked")
- public void shouldThrowNullPointerOnMaterializedReduceIfMaterializedIsNull(final StrategyType inputType, final boolean inputWithCache) {
- setup(inputType, inputWithCache);
+ public void shouldThrowNullPointerOnMaterializedReduceIfMaterializedIsNull(final StrategyType inputType, final boolean inputWithCache, final boolean withHeaders) {
+ setup(inputType, inputWithCache, withHeaders);
assertThrows(NullPointerException.class, () -> windowedStream.reduce(
MockReducer.STRING_ADDER,
(Materialized) null));
@@ -467,8 +475,8 @@ public class TimeWindowedKStreamImplTest {
@ParameterizedTest
@MethodSource("data")
- public void shouldThrowNullPointerOnMaterializedReduceIfNamedIsNull(final StrategyType inputType, final boolean inputWithCache) {
- setup(inputType, inputWithCache);
+ public void shouldThrowNullPointerOnMaterializedReduceIfNamedIsNull(final StrategyType inputType, final boolean inputWithCache, final boolean withHeaders) {
+ setup(inputType, inputWithCache, withHeaders);
assertThrows(NullPointerException.class, () -> windowedStream.reduce(
MockReducer.STRING_ADDER,
(Named) null));
@@ -476,8 +484,8 @@ public class TimeWindowedKStreamImplTest {
@ParameterizedTest
@MethodSource("data")
- public void shouldThrowNullPointerOnCountIfMaterializedIsNull(final StrategyType inputType, final boolean inputWithCache) {
- setup(inputType, inputWithCache);
+ public void shouldThrowNullPointerOnCountIfMaterializedIsNull(final StrategyType inputType, final boolean inputWithCache, final boolean withHeaders) {
+ setup(inputType, inputWithCache, withHeaders);
assertThrows(NullPointerException.class, () -> windowedStream.count((Materialized<String, Long, WindowStore<Bytes, byte[]>>) null));
}