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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
