You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "wcarlson5 (via GitHub)" <gi...@apache.org> on 2023/07/01 03:26:25 UTC
[GitHub] [kafka] wcarlson5 opened a new pull request, #13942: KAFKA-14936: Check the versioned table's history retention and compare to grace period (4/N)
wcarlson5 opened a new pull request, #13942:
URL: https://github.com/apache/kafka/pull/13942
Check the history retention of the ktable of the grace period join.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] wcarlson5 commented on a diff in pull request #13942: KAFKA-14936: Check the versioned table's history retention and compare to grace period (4/N)
Posted by "wcarlson5 (via GitHub)" <gi...@apache.org>.
wcarlson5 commented on code in PR #13942:
URL: https://github.com/apache/kafka/pull/13942#discussion_r1265518582
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBytesStoreTest.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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.state.internals;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.header.internals.RecordHeaders;
+import org.apache.kafka.common.metrics.Metrics;
+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.common.utils.LogContext;
+import org.apache.kafka.streams.processor.StateStoreContext;
+import org.apache.kafka.streams.processor.internals.MockStreamsMetrics;
+import org.apache.kafka.streams.processor.internals.ProcessorRecordContext;
+import org.apache.kafka.test.InternalMockProcessorContext;
+import org.apache.kafka.test.MockRecordCollector;
+import org.apache.kafka.test.TestUtils;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.rocksdb.WriteBatch;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+
+class RocksDBTimeOrderedKeyValueBytesStoreTest {
+
+ private InternalMockProcessorContext context;
+ private RocksDBTimeOrderedKeyValueBytesStore bytesStore;
+ private File stateDir;
+ final String storeName = "bytes-store";
+ private final static String METRICS_SCOPE = "metrics-scope";
+ private final String topic = "changelog";
+
+
+ @BeforeEach
+ public void before() {
+ bytesStore = new RocksDBTimeOrderedKeyValueBytesStore(storeName, METRICS_SCOPE);
+
+ stateDir = TestUtils.tempDirectory();
+ context = new InternalMockProcessorContext<>(
+ stateDir,
+ Serdes.String(),
+ Serdes.Long(),
+ new MockRecordCollector(),
+ new ThreadCache(new LogContext("testCache "), 0, new MockStreamsMetrics(new Metrics()))
+ );
+ bytesStore.init((StateStoreContext) context, bytesStore);
+ }
+
+ @AfterEach
+ public void close() {
+ bytesStore.close();
+ }
+
+ @Test
+ public void shouldCreateWriteBatches() {
Review Comment:
Gonna move this file to a different PR
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] wcarlson5 commented on a diff in pull request #13942: KAFKA-14936: Check the versioned table's history retention and compare to grace period (4/N)
Posted by "wcarlson5 (via GitHub)" <gi...@apache.org>.
wcarlson5 commented on code in PR #13942:
URL: https://github.com/apache/kafka/pull/13942#discussion_r1265517147
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBytesStore.java:
##########
@@ -42,7 +46,29 @@ protected KeyValue<Bytes, byte[]> getIndexKeyValue(final Bytes baseKey, final by
@Override
Map<KeyValueSegment, WriteBatch> getWriteBatches(final Collection<ConsumerRecord<byte[], byte[]>> records) {
- throw new UnsupportedOperationException("Do not use for TimeOrderedKeyValueStore");
+ final Map<KeyValueSegment, WriteBatch> writeBatchMap = new HashMap<>();
Review Comment:
Yeah that should be in a different PR, not sure how that snuck in
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cadonna commented on a diff in pull request #13942: KAFKA-14936: Check the versioned table's history retention and compare to grace period (4/N)
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #13942:
URL: https://github.com/apache/kafka/pull/13942#discussion_r1262545184
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/StreamTableJoinNode.java:
##########
@@ -31,17 +32,22 @@ 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;
+
Review Comment:
nit:
```suggestion
```
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBytesStoreTest.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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.state.internals;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.header.internals.RecordHeaders;
+import org.apache.kafka.common.metrics.Metrics;
+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.common.utils.LogContext;
+import org.apache.kafka.streams.processor.StateStoreContext;
+import org.apache.kafka.streams.processor.internals.MockStreamsMetrics;
+import org.apache.kafka.streams.processor.internals.ProcessorRecordContext;
+import org.apache.kafka.test.InternalMockProcessorContext;
+import org.apache.kafka.test.MockRecordCollector;
+import org.apache.kafka.test.TestUtils;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.rocksdb.WriteBatch;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+
+class RocksDBTimeOrderedKeyValueBytesStoreTest {
+
+ private InternalMockProcessorContext context;
+ private RocksDBTimeOrderedKeyValueBytesStore bytesStore;
+ private File stateDir;
+ final String storeName = "bytes-store";
+ private final static String METRICS_SCOPE = "metrics-scope";
+ private final String topic = "changelog";
+
+
+ @BeforeEach
+ public void before() {
+ bytesStore = new RocksDBTimeOrderedKeyValueBytesStore(storeName, METRICS_SCOPE);
+
+ stateDir = TestUtils.tempDirectory();
+ context = new InternalMockProcessorContext<>(
+ stateDir,
+ Serdes.String(),
+ Serdes.Long(),
+ new MockRecordCollector(),
+ new ThreadCache(new LogContext("testCache "), 0, new MockStreamsMetrics(new Metrics()))
+ );
+ bytesStore.init((StateStoreContext) context, bytesStore);
+ }
+
+ @AfterEach
+ public void close() {
+ bytesStore.close();
+ }
+
+ @Test
+ public void shouldCreateWriteBatches() {
+ final String key = "a";
+ final Collection<ConsumerRecord<byte[], byte[]>> records = new ArrayList<>();
+ records.add(new ConsumerRecord<>("", 0, 0L, serializeKey(key, 0, 0L).get(), serializeValue(50L)));
+ records.add(new ConsumerRecord<>("", 0, 0L, serializeKey(key, 1, 1L).get(), serializeValue(100L)));
+ final Map<KeyValueSegment, WriteBatch> writeBatchMap = bytesStore.getWriteBatches(records);
+ assertEquals(1, writeBatchMap.size());
+
+ for (final WriteBatch batch : writeBatchMap.values()) {
+ // 2 includes base and index record
Review Comment:
This comment lies. There is no index record here. Please remove.
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBytesStoreTest.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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.state.internals;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.header.internals.RecordHeaders;
+import org.apache.kafka.common.metrics.Metrics;
+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.common.utils.LogContext;
+import org.apache.kafka.streams.processor.StateStoreContext;
+import org.apache.kafka.streams.processor.internals.MockStreamsMetrics;
+import org.apache.kafka.streams.processor.internals.ProcessorRecordContext;
+import org.apache.kafka.test.InternalMockProcessorContext;
+import org.apache.kafka.test.MockRecordCollector;
+import org.apache.kafka.test.TestUtils;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.rocksdb.WriteBatch;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+
+class RocksDBTimeOrderedKeyValueBytesStoreTest {
+
+ private InternalMockProcessorContext context;
+ private RocksDBTimeOrderedKeyValueBytesStore bytesStore;
+ private File stateDir;
+ final String storeName = "bytes-store";
+ private final static String METRICS_SCOPE = "metrics-scope";
+ private final String topic = "changelog";
+
+
+ @BeforeEach
+ public void before() {
+ bytesStore = new RocksDBTimeOrderedKeyValueBytesStore(storeName, METRICS_SCOPE);
+
+ stateDir = TestUtils.tempDirectory();
+ context = new InternalMockProcessorContext<>(
+ stateDir,
+ Serdes.String(),
+ Serdes.Long(),
+ new MockRecordCollector(),
+ new ThreadCache(new LogContext("testCache "), 0, new MockStreamsMetrics(new Metrics()))
+ );
+ bytesStore.init((StateStoreContext) context, bytesStore);
+ }
+
+ @AfterEach
+ public void close() {
+ bytesStore.close();
+ }
+
+ @Test
+ public void shouldCreateWriteBatches() {
Review Comment:
Could you please add a test for the edge case when the the collection of records is empty?
##########
streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinTest.java:
##########
@@ -164,6 +164,75 @@ public void shouldFailIfTableIsNotVersioned() {
);
}
+ @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));
Review Comment:
Could you please structure your tests in such a way that it is clearer what the setup, the actual call under test and the verification is. I usually separate these three parts by an empty line. A clear separation is not always possible, though.
##########
streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinTest.java:
##########
@@ -164,6 +164,75 @@ public void shouldFailIfTableIsNotVersioned() {
);
}
+ @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));
Review Comment:
I was wondering if it would not be better testing this exceptions in `StreamTableJoinNodeTest` (does not yet exist). In such a way, the tests might become simpler, because we could verify the exceptions just for the case that a versioned state store is connected to the node in `StreamTableJoinNodeTest`. Here we would merely test if the materializations are correctly inherited (maybe there are already such tests somewhere).
You do not need to do that for this PR.
But we should try to avoid writing more tests than necessary and at the same time we should cover all cases.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] wcarlson5 commented on a diff in pull request #13942: KAFKA-14936: Check the versioned table's history retention and compare to grace period (4/N)
Posted by "wcarlson5 (via GitHub)" <gi...@apache.org>.
wcarlson5 commented on code in PR #13942:
URL: https://github.com/apache/kafka/pull/13942#discussion_r1265518582
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBytesStoreTest.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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.state.internals;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.header.internals.RecordHeaders;
+import org.apache.kafka.common.metrics.Metrics;
+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.common.utils.LogContext;
+import org.apache.kafka.streams.processor.StateStoreContext;
+import org.apache.kafka.streams.processor.internals.MockStreamsMetrics;
+import org.apache.kafka.streams.processor.internals.ProcessorRecordContext;
+import org.apache.kafka.test.InternalMockProcessorContext;
+import org.apache.kafka.test.MockRecordCollector;
+import org.apache.kafka.test.TestUtils;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.rocksdb.WriteBatch;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+
+class RocksDBTimeOrderedKeyValueBytesStoreTest {
+
+ private InternalMockProcessorContext context;
+ private RocksDBTimeOrderedKeyValueBytesStore bytesStore;
+ private File stateDir;
+ final String storeName = "bytes-store";
+ private final static String METRICS_SCOPE = "metrics-scope";
+ private final String topic = "changelog";
+
+
+ @BeforeEach
+ public void before() {
+ bytesStore = new RocksDBTimeOrderedKeyValueBytesStore(storeName, METRICS_SCOPE);
+
+ stateDir = TestUtils.tempDirectory();
+ context = new InternalMockProcessorContext<>(
+ stateDir,
+ Serdes.String(),
+ Serdes.Long(),
+ new MockRecordCollector(),
+ new ThreadCache(new LogContext("testCache "), 0, new MockStreamsMetrics(new Metrics()))
+ );
+ bytesStore.init((StateStoreContext) context, bytesStore);
+ }
+
+ @AfterEach
+ public void close() {
+ bytesStore.close();
+ }
+
+ @Test
+ public void shouldCreateWriteBatches() {
Review Comment:
Gonna move this file to a different PR, but also yes
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] wcarlson5 commented on a diff in pull request #13942: KAFKA-14936: Check the versioned table's history retention and compare to grace period (4/N)
Posted by "wcarlson5 (via GitHub)" <gi...@apache.org>.
wcarlson5 commented on code in PR #13942:
URL: https://github.com/apache/kafka/pull/13942#discussion_r1259980518
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/StreamTableJoinNode.java:
##########
@@ -64,6 +70,16 @@ public void writeToTopology(final InternalTopologyBuilder topologyBuilder) {
// 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 history retention must be at least grace period, but it should be larger.");
Review Comment:
This change is fine, just equal is enough, if not recommended
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] wcarlson5 commented on a diff in pull request #13942: KAFKA-14936: Check the versioned table's history retention and compare to grace period (4/N)
Posted by "wcarlson5 (via GitHub)" <gi...@apache.org>.
wcarlson5 commented on code in PR #13942:
URL: https://github.com/apache/kafka/pull/13942#discussion_r1265535458
##########
streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinTest.java:
##########
@@ -164,6 +164,75 @@ public void shouldFailIfTableIsNotVersioned() {
);
}
+ @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));
Review Comment:
I added an empty line I hope that helps
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cadonna commented on pull request #13942: KAFKA-14936: Check the versioned table's history retention and compare to grace period (4/N)
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on PR #13942:
URL: https://github.com/apache/kafka/pull/13942#issuecomment-1643982125
Build failures are unrelated:
```
Build / JDK 20 and Scala 2.13 / org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationExactlyOnceTest.testOffsetTranslationBehindReplicationFlow()
Build / JDK 20 and Scala 2.13 / org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationExactlyOnceTest.testOffsetTranslationBehindReplicationFlow()
Build / JDK 20 and Scala 2.13 / org.apache.kafka.connect.mirror.integration.MirrorConnectorsWithCustomForwardingAdminIntegrationTest.testReplication()
Build / JDK 20 and Scala 2.13 / org.apache.kafka.controller.QuorumControllerTest.testBalancePartitionLeaders()
Build / JDK 17 and Scala 2.13 / org.apache.kafka.connect.mirror.integration.IdentityReplicationIntegrationTest.testReplicateSourceDefault()
Build / JDK 17 and Scala 2.13 / integration.kafka.server.FetchFromFollowerIntegrationTest.testRackAwareRangeAssignor()
Build / JDK 17 and Scala 2.13 / org.apache.kafka.controller.QuorumControllerTest.testBalancePartitionLeaders()
Build / JDK 11 and Scala 2.13 / org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationExactlyOnceTest.testOffsetTranslationBehindReplicationFlow()
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cadonna commented on a diff in pull request #13942: KAFKA-14936: Check the versioned table's history retention and compare to grace period (4/N)
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #13942:
URL: https://github.com/apache/kafka/pull/13942#discussion_r1251652218
##########
streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinTest.java:
##########
@@ -164,6 +165,53 @@ public void shouldFailIfTableIsNotVersioned() {
);
}
+ @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);
+ 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() {
Review Comment:
Ah, I see! Thanks for the clarification!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] wcarlson5 commented on a diff in pull request #13942: KAFKA-14936: Check the versioned table's history retention and compare to grace period (4/N)
Posted by "wcarlson5 (via GitHub)" <gi...@apache.org>.
wcarlson5 commented on code in PR #13942:
URL: https://github.com/apache/kafka/pull/13942#discussion_r1259987822
##########
streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinTest.java:
##########
@@ -164,6 +165,53 @@ public void shouldFailIfTableIsNotVersioned() {
);
}
+ @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);
+ 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);
+ 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() {
Review Comment:
Sure, easy enough
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] vcrfxia commented on a diff in pull request #13942: KAFKA-14936: Check the versioned table's history retention and compare to grace period (4/N)
Posted by "vcrfxia (via GitHub)" <gi...@apache.org>.
vcrfxia commented on code in PR #13942:
URL: https://github.com/apache/kafka/pull/13942#discussion_r1262600782
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBytesStore.java:
##########
@@ -42,7 +46,29 @@ protected KeyValue<Bytes, byte[]> getIndexKeyValue(final Bytes baseKey, final by
@Override
Map<KeyValueSegment, WriteBatch> getWriteBatches(final Collection<ConsumerRecord<byte[], byte[]>> records) {
- throw new UnsupportedOperationException("Do not use for TimeOrderedKeyValueStore");
+ final Map<KeyValueSegment, WriteBatch> writeBatchMap = new HashMap<>();
Review Comment:
Looks like this PR is now also adding restoration logic, in addition to comparing table history retention and join grace period. Was that intentional? Whenever we add the restoration logic, we'll want to add integration tests for that -- maybe it'd be easier to have it be a separate PR. WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] wcarlson5 commented on a diff in pull request #13942: KAFKA-14936: Check the versioned table's history retention and compare to grace period (4/N)
Posted by "wcarlson5 (via GitHub)" <gi...@apache.org>.
wcarlson5 commented on code in PR #13942:
URL: https://github.com/apache/kafka/pull/13942#discussion_r1265539055
##########
streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinTest.java:
##########
@@ -164,6 +164,75 @@ public void shouldFailIfTableIsNotVersioned() {
);
}
+ @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));
Review Comment:
I think there is. For some reason it seems that streams has some aversion to making tests like `StreamTableJoinNodeTest` as not are set up like this
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cadonna merged pull request #13942: KAFKA-14936: Check the versioned table's history retention and compare to grace period (4/N)
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna merged PR #13942:
URL: https://github.com/apache/kafka/pull/13942
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cadonna commented on a diff in pull request #13942: KAFKA-14936: Check the versioned table's history retention and compare to grace period (4/N)
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #13942:
URL: https://github.com/apache/kafka/pull/13942#discussion_r1250620690
##########
streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinTest.java:
##########
@@ -164,6 +165,53 @@ public void shouldFailIfTableIsNotVersioned() {
);
}
+ @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);
+ 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() {
Review Comment:
Do we also need a test where the materialization is not inherited?
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/StreamTableJoinNode.java:
##########
@@ -64,6 +70,16 @@ public void writeToTopology(final InternalTopologyBuilder topologyBuilder) {
// 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 history retention must be at least grace period, but it should be larger.");
Review Comment:
```suggestion
throw new IllegalArgumentException("History retention must be at least grace period.");
```
Must history retention be larger than or is it enough to be equal to the grace period? That is not clear from your original error message.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] wcarlson5 commented on a diff in pull request #13942: KAFKA-14936: Check the versioned table's history retention and compare to grace period (4/N)
Posted by "wcarlson5 (via GitHub)" <gi...@apache.org>.
wcarlson5 commented on code in PR #13942:
URL: https://github.com/apache/kafka/pull/13942#discussion_r1259992272
##########
streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinTest.java:
##########
@@ -164,6 +165,53 @@ public void shouldFailIfTableIsNotVersioned() {
);
}
+ @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);
+ 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() {
Review Comment:
good idea
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] vcrfxia commented on a diff in pull request #13942: KAFKA-14936: Check the versioned table's history retention and compare to grace period (4/N)
Posted by "vcrfxia (via GitHub)" <gi...@apache.org>.
vcrfxia commented on code in PR #13942:
URL: https://github.com/apache/kafka/pull/13942#discussion_r1251360190
##########
streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinTest.java:
##########
@@ -164,6 +165,53 @@ public void shouldFailIfTableIsNotVersioned() {
);
}
+ @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);
+ 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() {
Review Comment:
I believe those already exist in this file -- there's `shouldFailIfTableIsNotVersioned()` to check the negative case and `shouldDelayJoinByGracePeriod()` for the positive case -- so I think we should be good?
A minor suggestion: a bunch of these new tests look similar even though they test different setups and to an unfamiliar reader it's difficult to pick out which lines of the test are important. You could add a comment to the `filter()` line in the topology to call out that the materialization as a versioned store is inherited through this node; that's the crux of this test that distinguishes it from the other test which already exists.
##########
streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinTest.java:
##########
@@ -164,6 +165,53 @@ public void shouldFailIfTableIsNotVersioned() {
);
}
+ @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);
+ 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);
+ 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() {
Review Comment:
Can you also add the equivalent of this test, for the case where materialization as a versioned store is inherited?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org