You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by mi...@apache.org on 2023/04/14 09:44:03 UTC
[kafka] branch trunk updated: KAFKA-14834: [9/N] Disable versioned-stores for unsupported operations (#13565)
This is an automated email from the ASF dual-hosted git repository.
mimaison 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 b1830e4aa2d KAFKA-14834: [9/N] Disable versioned-stores for unsupported operations (#13565)
b1830e4aa2d is described below
commit b1830e4aa2d83d2f213faf5f7432a119742df80c
Author: Matthias J. Sax <ma...@confluent.io>
AuthorDate: Fri Apr 14 02:43:54 2023 -0700
KAFKA-14834: [9/N] Disable versioned-stores for unsupported operations (#13565)
Using versioned-stores for global-KTables is not allowed, because a
global-table is bootstrapped on startup, and a stream-globalTable join
does not support temporal semantics.
Furthermore, `suppress()` does not support temporal semantics and thus
cannot be applied to an versioned-KTable.
This PR disallows both use-cases explicitely.
Part of KIP-914.
Reviewers: Bill Bejeck <bb...@gmail.com>, Victoria Xia <vi...@confluent.io>
---
.../org/apache/kafka/streams/StreamsBuilder.java | 8 ++++
.../org/apache/kafka/streams/kstream/KTable.java | 4 +-
.../kstream/internals/InternalStreamsBuilder.java | 16 ++++++++
.../streams/kstream/internals/KTableImpl.java | 12 +++++-
.../kstream/internals/graph/TableSuppressNode.java | 27 +++++++++++++
.../apache/kafka/streams/StreamsBuilderTest.java | 15 +++++++
.../VersionedKeyValueStoreIntegrationTest.java | 47 ----------------------
.../internals/InternalStreamsBuilderTest.java | 20 +++++++++
.../internals/KStreamGlobalKTableJoinTest.java | 26 ------------
.../internals/KStreamGlobalKTableLeftJoinTest.java | 29 -------------
.../kstream/internals/SuppressTopologyTest.java | 34 ++++++++++++++++
11 files changed, 134 insertions(+), 104 deletions(-)
diff --git a/streams/src/main/java/org/apache/kafka/streams/StreamsBuilder.java b/streams/src/main/java/org/apache/kafka/streams/StreamsBuilder.java
index 39ae3821450..bed61271e9c 100644
--- a/streams/src/main/java/org/apache/kafka/streams/StreamsBuilder.java
+++ b/streams/src/main/java/org/apache/kafka/streams/StreamsBuilder.java
@@ -369,6 +369,8 @@ public class StreamsBuilder {
* <p>
* Note that {@link GlobalKTable} always applies {@code "auto.offset.reset"} strategy {@code "earliest"}
* regardless of the specified value in {@link StreamsConfig} or {@link Consumed}.
+ * Furthermore, {@link GlobalKTable} cannot be a {@link org.apache.kafka.streams.state.VersionedBytesStoreSupplier
+ * versioned state store}.
*
* @param topic the topic name; cannot be {@code null}
* @param consumed the instance of {@link Consumed} used to define optional parameters
@@ -400,6 +402,8 @@ public class StreamsBuilder {
* <p>
* Note that {@link GlobalKTable} always applies {@code "auto.offset.reset"} strategy {@code "earliest"}
* regardless of the specified value in {@link StreamsConfig}.
+ * Furthermore, {@link GlobalKTable} cannot be a {@link org.apache.kafka.streams.state.VersionedBytesStoreSupplier
+ * versioned state store}.
*
* @param topic the topic name; cannot be {@code null}
* @return a {@link GlobalKTable} for the specified topic
@@ -434,6 +438,8 @@ public class StreamsBuilder {
* }</pre>
* Note that {@link GlobalKTable} always applies {@code "auto.offset.reset"} strategy {@code "earliest"}
* regardless of the specified value in {@link StreamsConfig} or {@link Consumed}.
+ * Furthermore, {@link GlobalKTable} cannot be a {@link org.apache.kafka.streams.state.VersionedBytesStoreSupplier
+ * versioned state store}.
*
* @param topic the topic name; cannot be {@code null}
* @param consumed the instance of {@link Consumed} used to define optional parameters; can't be {@code null}
@@ -476,6 +482,8 @@ public class StreamsBuilder {
* }</pre>
* Note that {@link GlobalKTable} always applies {@code "auto.offset.reset"} strategy {@code "earliest"}
* regardless of the specified value in {@link StreamsConfig}.
+ * Furthermore, {@link GlobalKTable} cannot be a {@link org.apache.kafka.streams.state.VersionedBytesStoreSupplier
+ * versioned state store}.
*
* @param topic the topic name; cannot be {@code null}
* @param materialized the instance of {@link Materialized} used to materialize a state store; cannot be {@code null}
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/KTable.java b/streams/src/main/java/org/apache/kafka/streams/kstream/KTable.java
index 28404eb50bc..cf7af720abe 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/KTable.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/KTable.java
@@ -715,8 +715,10 @@ public interface KTable<K, V> {
/**
* Suppress some updates from this changelog stream, determined by the supplied {@link Suppressed} configuration.
- *
* This controls what updates downstream table and stream operations will receive.
+ * <p>
+ * Note that {@code suppress()} cannot be applied to
+ * {@link org.apache.kafka.streams.state.VersionedBytesStoreSupplier versioned KTables}.
*
* @param suppressed Configuration object determining what, if any, updates to suppress
* @return A new KTable with the desired suppression characteristics.
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/InternalStreamsBuilder.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/InternalStreamsBuilder.java
index 6f17a5d4f1d..f9517afe425 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/InternalStreamsBuilder.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/InternalStreamsBuilder.java
@@ -35,6 +35,7 @@ import org.apache.kafka.streams.kstream.internals.graph.StateStoreNode;
import org.apache.kafka.streams.kstream.internals.graph.StreamSourceNode;
import org.apache.kafka.streams.kstream.internals.graph.GraphNode;
import org.apache.kafka.streams.kstream.internals.graph.StreamStreamJoinNode;
+import org.apache.kafka.streams.kstream.internals.graph.TableSuppressNode;
import org.apache.kafka.streams.kstream.internals.graph.TableSourceNode;
import org.apache.kafka.streams.kstream.internals.graph.VersionedSemanticsGraphNode;
import org.apache.kafka.streams.kstream.internals.graph.WindowedStreamProcessorNode;
@@ -74,6 +75,7 @@ public class InternalStreamsBuilder implements InternalNameProvider {
private final LinkedHashSet<GraphNode> mergeNodes = new LinkedHashSet<>();
private final LinkedHashSet<GraphNode> tableSourceNodes = new LinkedHashSet<>();
private final LinkedHashSet<GraphNode> versionedSemanticsNodes = new LinkedHashSet<>();
+ private final LinkedHashSet<TableSuppressNode> tableSuppressNodesNodes = new LinkedHashSet<>();
private static final String TOPOLOGY_ROOT = "root";
private static final Logger LOG = LoggerFactory.getLogger(InternalStreamsBuilder.class);
@@ -164,6 +166,11 @@ public class InternalStreamsBuilder implements InternalNameProvider {
final MaterializedInternal<K, V, KeyValueStore<Bytes, byte[]>> materialized) {
Objects.requireNonNull(consumed, "consumed can't be null");
Objects.requireNonNull(materialized, "materialized can't be null");
+
+ if (materialized.storeSupplier() instanceof VersionedBytesStoreSupplier) {
+ throw new TopologyException("GlobalTables cannot be versioned.");
+ }
+
// explicitly disable logging for global stores
materialized.withLoggingDisabled();
@@ -282,6 +289,9 @@ public class InternalStreamsBuilder implements InternalNameProvider {
if (node instanceof VersionedSemanticsGraphNode) {
versionedSemanticsNodes.add(node);
}
+ if (node instanceof TableSuppressNode) {
+ tableSuppressNodesNodes.add((TableSuppressNode<?, ?>) node);
+ }
}
// use this method for testing only
@@ -611,6 +621,12 @@ public class InternalStreamsBuilder implements InternalNameProvider {
private void enableVersionedSemantics() {
versionedSemanticsNodes.forEach(node -> ((VersionedSemanticsGraphNode) node).enableVersionedSemantics(isVersionedUpstream(node)));
+ tableSuppressNodesNodes.forEach(node -> {
+ if (isVersionedUpstream(node)) {
+ throw new TopologyException("suppress() is only supported for non-versioned KTables " +
+ "(note that version semantics might be inherited from upstream)");
+ }
+ });
}
/**
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableImpl.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableImpl.java
index 1da24985a92..75942ce7608 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableImpl.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableImpl.java
@@ -20,6 +20,7 @@ import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.errors.TopologyException;
import org.apache.kafka.streams.kstream.Consumed;
import org.apache.kafka.streams.kstream.Grouped;
import org.apache.kafka.streams.kstream.KGroupedTable;
@@ -54,6 +55,7 @@ import org.apache.kafka.streams.kstream.internals.graph.StatefulProcessorNode;
import org.apache.kafka.streams.kstream.internals.graph.StreamSinkNode;
import org.apache.kafka.streams.kstream.internals.graph.StreamSourceNode;
import org.apache.kafka.streams.kstream.internals.graph.GraphNode;
+import org.apache.kafka.streams.kstream.internals.graph.TableSuppressNode;
import org.apache.kafka.streams.kstream.internals.graph.TableFilterNode;
import org.apache.kafka.streams.kstream.internals.graph.TableProcessorNode;
import org.apache.kafka.streams.kstream.internals.graph.TableRepartitionMapNode;
@@ -539,6 +541,14 @@ public class KTableImpl<K, S, V> extends AbstractStream<K, V> implements KTable<
@Override
public KTable<K, V> suppress(final Suppressed<? super K> suppressed) {
+ // this is an eager, but insufficient check
+ // the check only works if the direct parent is materialized
+ // the actual check for "version inheritance" can only be done in the build-phase later
+ // we keep this check to get a better stack trace if possible
+ if (graphNode.isOutputVersioned().isPresent() && graphNode.isOutputVersioned().get()) {
+ throw new TopologyException("suppress() is only supported for non-versioned KTables");
+ }
+
final String name;
if (suppressed instanceof NamedSuppressed) {
final String givenName = ((NamedSuppressed<?>) suppressed).name();
@@ -575,7 +585,7 @@ public class KTableImpl<K, S, V> extends AbstractStream<K, V> implements KTable<
.withLoggingDisabled();
}
- final ProcessorGraphNode<K, Change<V>> node = new StatefulProcessorNode<>(
+ final ProcessorGraphNode<K, Change<V>> node = new TableSuppressNode<>(
name,
new ProcessorParameters<>(suppressionSupplier, name),
storeBuilder
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/TableSuppressNode.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/TableSuppressNode.java
new file mode 100644
index 00000000000..1ac6241cf4b
--- /dev/null
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/TableSuppressNode.java
@@ -0,0 +1,27 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.kstream.internals.graph;
+
+import org.apache.kafka.streams.state.StoreBuilder;
+
+public class TableSuppressNode<K, V> extends StatefulProcessorNode<K, V> {
+ public TableSuppressNode(final String nodeName,
+ final ProcessorParameters<K, V, ?, ?> processorParameters,
+ final StoreBuilder<?> materializedKTableStoreBuilder) {
+ super(nodeName, processorParameters, materializedKTableStoreBuilder);
+ }
+}
diff --git a/streams/src/test/java/org/apache/kafka/streams/StreamsBuilderTest.java b/streams/src/test/java/org/apache/kafka/streams/StreamsBuilderTest.java
index d7553c85b69..630c68db425 100644
--- a/streams/src/test/java/org/apache/kafka/streams/StreamsBuilderTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/StreamsBuilderTest.java
@@ -433,6 +433,21 @@ public class StreamsBuilderTest {
}
}
+ @Test
+ public void shouldThrowOnVersionedStoreSupplierForGlobalTable() {
+ final String topic = "topic";
+ assertThrows(
+ TopologyException.class,
+ () -> builder.globalTable(
+ topic,
+ Materialized.<Long, String>as(Stores.persistentVersionedKeyValueStore("store", Duration.ZERO))
+ .withKeySerde(Serdes.Long())
+ .withValueSerde(Serdes.String()
+ )
+ )
+ );
+ }
+
@Test
public void shouldNotMaterializeStoresIfNotRequired() {
final String topic = "topic";
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/VersionedKeyValueStoreIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/VersionedKeyValueStoreIntegrationTest.java
index 29a070f577b..94d2644905b 100644
--- a/streams/src/test/java/org/apache/kafka/streams/integration/VersionedKeyValueStoreIntegrationTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/integration/VersionedKeyValueStoreIntegrationTest.java
@@ -45,7 +45,6 @@ import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster;
import org.apache.kafka.streams.integration.utils.IntegrationTestUtils;
import org.apache.kafka.streams.kstream.Consumed;
-import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.Produced;
import org.apache.kafka.streams.processor.StateStore;
import org.apache.kafka.streams.processor.StateStoreContext;
@@ -320,52 +319,6 @@ public class VersionedKeyValueStoreIntegrationTest {
assertThat(result.getOnlyPartitionResult().getResult(), equalTo("success"));
}
- @Test
- public void shouldCreateGlobalTable() throws Exception {
- // produce data to global store topic and track in-memory for processor to verify
- final DataTracker data = new DataTracker();
- produceDataToTopic(globalTableTopic, data, baseTimestamp, KeyValue.pair(1, "a0"), KeyValue.pair(2, "b0"), KeyValue.pair(3, null));
- produceDataToTopic(globalTableTopic, data, baseTimestamp + 5, KeyValue.pair(1, "a5"), KeyValue.pair(2, null), KeyValue.pair(3, "c5"));
- produceDataToTopic(globalTableTopic, data, baseTimestamp + 2, KeyValue.pair(1, "a2"), KeyValue.pair(2, "b2"), KeyValue.pair(3, null)); // out-of-order data
-
- // build topology and start app
- final StreamsBuilder streamsBuilder = new StreamsBuilder();
-
- streamsBuilder
- .globalTable(
- globalTableTopic,
- Consumed.with(Serdes.Integer(), Serdes.String()),
- Materialized
- .<Integer, String>as(Stores.persistentVersionedKeyValueStore(STORE_NAME, Duration.ofMillis(HISTORY_RETENTION)))
- .withKeySerde(Serdes.Integer())
- .withValueSerde(Serdes.String()));
- streamsBuilder
- .stream(inputStream, Consumed.with(Serdes.Integer(), Serdes.String()))
- .process(() -> new VersionedStoreContentCheckerProcessor(false, data))
- .to(outputStream, Produced.with(Serdes.Integer(), Serdes.Integer()));
-
- final Properties props = props();
- kafkaStreams = new KafkaStreams(streamsBuilder.build(), props);
- kafkaStreams.start();
-
- // produce source data to trigger store verifications in processor
- final int numRecordsProduced = produceDataToTopic(inputStream, baseTimestamp + 8, KeyValue.pair(1, "a8"), KeyValue.pair(2, "b8"), KeyValue.pair(3, "c8"));
-
- // wait for output and verify
- final List<KeyValue<Integer, Integer>> receivedRecords = IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(
- TestUtils.consumerConfig(
- CLUSTER.bootstrapServers(),
- IntegerDeserializer.class,
- IntegerDeserializer.class),
- outputStream,
- numRecordsProduced);
-
- for (final KeyValue<Integer, Integer> receivedRecord : receivedRecords) {
- // verify zero failed checks for each record
- assertThat(receivedRecord.value, equalTo(0));
- }
- }
-
@Test
public void shouldManualUpgradeFromNonVersionedTimestampedToVersioned() throws Exception {
// build non-versioned (timestamped) topology
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/InternalStreamsBuilderTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/InternalStreamsBuilderTest.java
index 420e4afbd9a..7449f8d3cde 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/InternalStreamsBuilderTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/InternalStreamsBuilderTest.java
@@ -25,6 +25,7 @@ import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.errors.TopologyException;
import org.apache.kafka.streams.kstream.Consumed;
import org.apache.kafka.streams.kstream.GlobalKTable;
import org.apache.kafka.streams.kstream.JoinWindows;
@@ -66,6 +67,7 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThrows;
import static org.junit.Assert.assertTrue;
public class InternalStreamsBuilderTest {
@@ -177,6 +179,24 @@ public class InternalStreamsBuilderTest {
assertEquals("globalTable", stateStores.get(0).name());
}
+ @Test
+ public void shouldThrowOnVersionedStoreSupplierForGlobalTable() {
+ final MaterializedInternal<String, String, KeyValueStore<Bytes, byte[]>> materializedInternal =
+ new MaterializedInternal<>(
+ Materialized.as(Stores.persistentVersionedKeyValueStore("store", Duration.ZERO)),
+ builder,
+ storePrefix
+ );
+
+ assertThrows(
+ TopologyException.class,
+ () -> builder.globalTable(
+ "table",
+ consumed,
+ materializedInternal)
+ );
+ }
+
private void doBuildGlobalTopologyWithAllGlobalTables() {
final ProcessorTopology topology = builder.internalTopologyBuilder
.rewriteTopology(new StreamsConfig(StreamsTestUtils.getStreamsConfig(APP_ID)))
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoinTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoinTest.java
index 49068f99c52..7160381c690 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoinTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoinTest.java
@@ -262,30 +262,4 @@ public class KStreamGlobalKTableJoinTest {
processor.checkAndClearProcessResult(new KeyValueTimestamp<>(null, "X0,FKey0+Y0", 0),
new KeyValueTimestamp<>(1, "X1,FKey1+Y1", 1));
}
-
- @Test
- public void shouldPerformTimestampedGet() {
- initWithVersionedStore(1000);
-
- // do not auto-advance stream timestamps for this test
- inputStreamTopic = driver.createInputTopic(streamTopic, new IntegerSerializer(), new StringSerializer());
-
- // produce out-of-order records including nulls to table
- inputTableTopic.pipeInput("FKey1", "ValueT10", 10);
- inputTableTopic.pipeInput("FKey1", "ValueT5", 5);
- inputTableTopic.pipeInput("FKey1", null, 7);
- inputTableTopic.pipeInput("FKey1", "ValueT12", 12);
-
- // produce records to stream side
- inputStreamTopic.pipeInput(1, "ValueS8,FKey1", 8);
- inputStreamTopic.pipeInput(2, "ValueS12,FKey1", 12);
- inputStreamTopic.pipeInput(3, "ValueS6,FKey1", 6);
- inputStreamTopic.pipeInput(4, "ValueS10,FKey1", 10);
- inputStreamTopic.pipeInput(5, "ValueS2,FKey1", 2);
-
- processor.checkAndClearProcessResult(
- new KeyValueTimestamp<>(2, "ValueS12,FKey1+ValueT12", 12),
- new KeyValueTimestamp<>(3, "ValueS6,FKey1+ValueT5", 6),
- new KeyValueTimestamp<>(4, "ValueS10,FKey1+ValueT10", 10));
- }
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableLeftJoinTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableLeftJoinTest.java
index 08fcb338e54..f1fedab4732 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableLeftJoinTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableLeftJoinTest.java
@@ -272,33 +272,4 @@ public class KStreamGlobalKTableLeftJoinTest {
new KeyValueTimestamp<>(2, "X2,FKey2+Y2", 2),
new KeyValueTimestamp<>(3, "X3,FKey3+Y3", 3));
}
-
- @Test
- public void shouldPerformTimestampedGet() {
- initWithVersionedStore(1000);
-
- // do not auto-advance timestamps for this test
- inputStreamTopic = driver.createInputTopic(streamTopic, new IntegerSerializer(), new StringSerializer());
- inputTableTopic = driver.createInputTopic(globalTableTopic, new StringSerializer(), new StringSerializer());
-
- // produce out-of-order records including nulls to table
- inputTableTopic.pipeInput("FKey1", "ValueT10", 10);
- inputTableTopic.pipeInput("FKey1", "ValueT5", 5);
- inputTableTopic.pipeInput("FKey1", null, 7);
- inputTableTopic.pipeInput("FKey1", "ValueT12", 12);
-
- // produce records to stream side
- inputStreamTopic.pipeInput(1, "ValueS8,FKey1", 8);
- inputStreamTopic.pipeInput(2, "ValueS12,FKey1", 12);
- inputStreamTopic.pipeInput(3, "ValueS6,FKey1", 6);
- inputStreamTopic.pipeInput(4, "ValueS10,FKey1", 10);
- inputStreamTopic.pipeInput(5, "ValueS2,FKey1", 2);
-
- processor.checkAndClearProcessResult(
- new KeyValueTimestamp<>(1, "ValueS8,FKey1+null", 8),
- new KeyValueTimestamp<>(2, "ValueS12,FKey1+ValueT12", 12),
- new KeyValueTimestamp<>(3, "ValueS6,FKey1+ValueT5", 6),
- new KeyValueTimestamp<>(4, "ValueS10,FKey1+ValueT10", 10),
- new KeyValueTimestamp<>(5, "ValueS2,FKey1+null", 2));
- }
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SuppressTopologyTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SuppressTopologyTest.java
index d775796d9aa..7493cfd3faf 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SuppressTopologyTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SuppressTopologyTest.java
@@ -21,13 +21,17 @@ import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.errors.TopologyException;
import org.apache.kafka.streams.kstream.Consumed;
import org.apache.kafka.streams.kstream.Grouped;
+import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.Produced;
import org.apache.kafka.streams.kstream.SessionWindows;
+import org.apache.kafka.streams.kstream.Suppressed;
import org.apache.kafka.streams.kstream.Windowed;
import org.apache.kafka.streams.state.SessionStore;
+import org.apache.kafka.streams.state.Stores;
import org.junit.Test;
import java.time.Duration;
@@ -38,6 +42,7 @@ import static org.apache.kafka.streams.kstream.Suppressed.untilTimeLimit;
import static org.apache.kafka.streams.kstream.Suppressed.untilWindowCloses;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.core.Is.is;
+import static org.junit.Assert.assertThrows;
@SuppressWarnings("deprecation")
public class SuppressTopologyTest {
@@ -209,4 +214,33 @@ public class SuppressTopologyTest {
// without the name, the suppression node does not increment the topology index
assertThat(namedNodeTopology, is(NAMED_INTERMEDIATE_TOPOLOGY));
}
+
+ @Test
+ public void shouldThrowOnSuppressForMaterializedVersionedTable() {
+ final StreamsBuilder builder = new StreamsBuilder();
+ final KTable<Object, Object> versionedTable = builder.table(
+ "input",
+ Materialized.as(Stores.persistentVersionedKeyValueStore("store", Duration.ZERO))
+ );
+
+ assertThrows(
+ TopologyException.class,
+ () -> versionedTable.suppress(Suppressed.untilTimeLimit(Duration.ZERO, Suppressed.BufferConfig.unbounded()))
+ );
+ }
+
+ @Test
+ public void shouldThrowOnSuppressForNonMaterializedVersionedTable() {
+ final StreamsBuilder builder = new StreamsBuilder();
+ builder.table(
+ "input",
+ Materialized.as(Stores.persistentVersionedKeyValueStore("store", Duration.ZERO))
+ ).filter((k, v) -> true)
+ .suppress(Suppressed.untilTimeLimit(Duration.ZERO, Suppressed.BufferConfig.unbounded()));
+
+ assertThrows(
+ TopologyException.class,
+ builder::build
+ );
+ }
}