You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ca...@apache.org on 2023/07/20 14:01:28 UTC
[kafka] branch trunk updated: KAFKA-14936: Check the versioned table's history retention and compare to grace period (4/N) (#13942)
This is an automated email from the ASF dual-hosted git repository.
cadonna 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 e9fe2a2eeaf KAFKA-14936: Check the versioned table's history retention and compare to grace period (4/N) (#13942)
e9fe2a2eeaf is described below
commit e9fe2a2eeaf449cddd7770a1a7fa07bc75918c69
Author: Walker Carlson <18...@users.noreply.github.com>
AuthorDate: Thu Jul 20 09:01:21 2023 -0500
KAFKA-14936: Check the versioned table's history retention and compare to grace period (4/N) (#13942)
Check the history retention of the ktable of the grace period join.
Reviewers: Reviewers: Victoria Xia <vi...@confluent.io>, Bruno Cadonna <ca...@apache.org>
---
.../streams/kstream/internals/KStreamImpl.java | 5 +-
.../internals/graph/StreamTableJoinNode.java | 17 +++++-
.../internals/InternalTopologyBuilder.java | 9 +++
.../kstream/internals/KStreamKTableJoinTest.java | 70 ++++++++++++++++++++++
4 files changed, 98 insertions(+), 3 deletions(-)
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java
index 1540a9091b8..00d8b179ce2 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java
@@ -1232,7 +1232,7 @@ public class KStreamImpl<K, V> extends AbstractStream<K, V> implements KStream<K
leftJoin);
final ProcessorParameters<K, V, ?, ?> processorParameters = new ProcessorParameters<>(processorSupplier, name);
final StreamTableJoinNode<K, V> streamTableJoinNode =
- new StreamTableJoinNode<>(name, processorParameters, new String[] {}, null);
+ new StreamTableJoinNode<>(name, processorParameters, new String[] {}, null, null);
builder.addGraphNode(graphNode, streamTableJoinNode);
@@ -1286,7 +1286,8 @@ public class KStreamImpl<K, V> extends AbstractStream<K, V> implements KStream<K
name,
processorParameters,
((KTableImpl<K, ?, VO>) table).valueGetterSupplier().storeNames(),
- this.name
+ this.name,
+ joined.gracePeriod()
);
builder.addGraphNode(graphNode, streamTableJoinNode);
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/StreamTableJoinNode.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/StreamTableJoinNode.java
index a4db1ba24cd..d99df941754 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/StreamTableJoinNode.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/StreamTableJoinNode.java
@@ -20,6 +20,7 @@ package org.apache.kafka.streams.kstream.internals.graph;
import org.apache.kafka.streams.processor.api.ProcessorSupplier;
import org.apache.kafka.streams.processor.internals.InternalTopologyBuilder;
+import java.time.Duration;
import java.util.Arrays;
/**
@@ -31,17 +32,21 @@ public class StreamTableJoinNode<K, V> extends GraphNode {
private final String[] storeNames;
private final ProcessorParameters<K, V, ?, ?> processorParameters;
private final String otherJoinSideNodeName;
+ private final Duration gracePeriod;
+
public StreamTableJoinNode(final String nodeName,
final ProcessorParameters<K, V, ?, ?> processorParameters,
final String[] storeNames,
- final String otherJoinSideNodeName) {
+ final String otherJoinSideNodeName,
+ final Duration gracePeriod) {
super(nodeName);
// in the case of Stream-Table join the state stores associated with the KTable
this.storeNames = storeNames;
this.processorParameters = processorParameters;
this.otherJoinSideNodeName = otherJoinSideNodeName;
+ this.gracePeriod = gracePeriod;
}
@Override
@@ -64,6 +69,16 @@ public class StreamTableJoinNode<K, V> extends GraphNode {
// Steam - KTable join only
if (otherJoinSideNodeName != null) {
topologyBuilder.connectProcessorAndStateStores(processorName, storeNames);
+ if (gracePeriod != null) {
+ for (final String storeName : storeNames) {
+ if (!topologyBuilder.isStoreVersioned(storeName)) {
+ throw new IllegalArgumentException("KTable must be versioned to use a grace period in a stream table join.");
+ }
+ if (gracePeriod.toMillis() > topologyBuilder.getHistoryRetention(storeName)) {
+ throw new IllegalArgumentException("History retention must be at least grace period.");
+ }
+ }
+ }
}
}
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java
index d763bb0d9bb..17409aba0c1 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java
@@ -696,6 +696,15 @@ public class InternalTopologyBuilder {
}
}
+ public Long getHistoryRetention(final String storeName) {
+ return stateFactories.get(storeName).historyRetention();
+ }
+
+ public boolean isStoreVersioned(final String storeName) {
+ return stateFactories.get(storeName).isVersionedStore();
+ }
+
+
public final void connectProcessorAndStateStores(final String processorName,
final String... stateStoreNames) {
Objects.requireNonNull(processorName, "processorName can't be null");
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 a62f39b1885..c015f5a4d73 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
@@ -164,6 +164,76 @@ public class KStreamKTableJoinTest {
);
}
+ @Test
+ public void shouldFailIfTableIsNotVersionedButMaterializationIsInherited() {
+ final StreamsBuilder builder = new StreamsBuilder();
+ final Properties props = new Properties();
+ props.put(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG, StreamsConfig.NO_OPTIMIZATION);
+ 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")));
+ final KTable<String, String> tableB = source.filter((k, v) -> true);
+ // the filter operation forces the table materialization to be inherited
+
+ streamA.join(tableB, (value1, value2) -> value1 + value2, Joined.with(Serdes.String(), Serdes.String(), Serdes.String(), "first-join", Duration.ofMillis(6))).to("out-one");
+
+ final IllegalArgumentException exception = assertThrows(IllegalArgumentException.class, builder::build);
+ assertThat(
+ exception.getMessage(),
+ is("KTable must be versioned to use a grace period in a stream table join.")
+ );
+ }
+
+ @Test
+ public void shouldNotFailIfTableIsVersionedButMaterializationIsInherited() {
+ final StreamsBuilder builder = new StreamsBuilder();
+ final Properties props = new Properties();
+ props.put(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG, StreamsConfig.NO_OPTIMIZATION);
+ 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))));
+ final KTable<String, String> tableB = source.filter((k, v) -> true);
+ // the filter operation forces the table materialization to be inherited
+
+ streamA.join(tableB, (value1, value2) -> value1 + value2, Joined.with(Serdes.String(), Serdes.String(), Serdes.String(), "first-join", Duration.ofMillis(6))).to("out-one");
+
+ //should not throw an error
+ builder.build();
+ }
+
+ @Test
+ public void shouldFailIfGracePeriodIsLongerThanHistoryRetention() {
+ final StreamsBuilder builder = new StreamsBuilder();
+ final Properties props = new Properties();
+ props.put(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG, StreamsConfig.NO_OPTIMIZATION);
+ 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))));
+
+ streamA.join(tableB, (value1, value2) -> value1 + value2, Joined.with(Serdes.String(), Serdes.String(), Serdes.String(), "first-join", Duration.ofMinutes(6))).to("out-one");
+
+ final IllegalArgumentException exception = assertThrows(IllegalArgumentException.class, () -> builder.build(props));
+ assertThat(exception.getMessage(), is("History retention must be at least grace period."));
+ }
+
+ @Test
+ public void shouldFailIfGracePeriodIsLongerThanHistoryRetentionAndInheritedStore() {
+ final StreamsBuilder builder = new StreamsBuilder();
+ final Properties props = new Properties();
+ props.put(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG, StreamsConfig.NO_OPTIMIZATION);
+ 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))));
+ final KTable<String, String> tableB = source.filter((k, v) -> true);
+ // the filter operation forces the table materialization to be inherited
+
+ streamA.join(tableB, (value1, value2) -> value1 + value2, Joined.with(Serdes.String(), Serdes.String(), Serdes.String(), "first-join", Duration.ofMillis(6))).to("out-one");
+
+ final IllegalArgumentException exception = assertThrows(IllegalArgumentException.class, () -> builder.build(props));
+ assertThat(exception.getMessage(), is("History retention must be at least grace period."));
+ }
+
+
@Test
public void shouldDelayJoinByGracePeriod() {
makeJoin(Duration.ofMillis(2));