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 f9517afe42586..d928f5b92d362 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 @@ -620,11 +620,17 @@ private GroupedInternal getRepartitionSerdes(final Collection ((VersionedSemanticsGraphNode) node).enableVersionedSemantics(isVersionedUpstream(node))); + versionedSemanticsNodes.forEach(node -> { + for (final GraphNode parentNode : node.parentNodes()) { + if (isVersionedOrVersionedUpstream(parentNode)) { + ((VersionedSemanticsGraphNode) node).enableVersionedSemantics(true, parentNode.nodeName()); + } + } + }); 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)"); + "(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 494969460f005..0527c69c55e67 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 @@ -48,6 +48,8 @@ import org.apache.kafka.streams.kstream.internals.foreignkeyjoin.SubscriptionReceiveProcessorSupplier; import org.apache.kafka.streams.kstream.internals.foreignkeyjoin.SubscriptionWrapper; import org.apache.kafka.streams.kstream.internals.foreignkeyjoin.SubscriptionWrapperSerde; +import org.apache.kafka.streams.kstream.internals.graph.ForeignTableJoinNode; +import org.apache.kafka.streams.kstream.internals.graph.ForeignJoinSubscriptionSendNode; import org.apache.kafka.streams.kstream.internals.graph.KTableKTableJoinNode; import org.apache.kafka.streams.kstream.internals.graph.ProcessorGraphNode; import org.apache.kafka.streams.kstream.internals.graph.ProcessorParameters; @@ -804,6 +806,7 @@ private KTable doJoin(final KTable other, kTableKTableJoinNode.setOutputVersioned(isOutputVersioned); builder.addGraphNode(this.graphNode, kTableKTableJoinNode); + builder.addGraphNode(((KTableImpl) other).graphNode, kTableKTableJoinNode); // we can inherit parent key serde if user do not provide specific overrides return new KTableImpl, VR>( @@ -1098,7 +1101,7 @@ private KTable doJoinOnForeignKey(final KTable forei //not be done needlessly. ((KTableImpl) foreignKeyTable).enableSendingOldValues(true); - //Old values must be sent such that the ForeignJoinSubscriptionSendProcessorSupplier can propagate deletions to the correct node. + //Old values must be sent such that the SubscriptionSendProcessorSupplier can propagate deletions to the correct node. //This occurs whenever the extracted foreignKey changes values. enableSendingOldValues(true); @@ -1138,8 +1141,7 @@ private KTable doJoinOnForeignKey(final KTable forei keySerde ); - final KTableValueGetterSupplier primaryKeyValueGetter = valueGetterSupplier(); - final StatefulProcessorNode> subscriptionSendNode = new StatefulProcessorNode<>( + final ProcessorGraphNode> subscriptionSendNode = new ForeignJoinSubscriptionSendNode<>( new ProcessorParameters<>( new SubscriptionSendProcessorSupplier<>( foreignKeyExtractor, @@ -1147,13 +1149,10 @@ private KTable doJoinOnForeignKey(final KTable forei valueHashSerdePseudoTopic, foreignKeySerde, valueSerde == null ? null : valueSerde.serializer(), - leftJoin, - primaryKeyValueGetter + leftJoin ), renamed.suffixWithOrElseGet("-subscription-registration-processor", builder, SUBSCRIPTION_REGISTRATION) - ), - Collections.emptySet(), - Collections.singleton(primaryKeyValueGetter) + ) ); builder.addGraphNode(graphNode, subscriptionSendNode); @@ -1219,13 +1218,13 @@ private KTable doJoinOnForeignKey(final KTable forei ); builder.addGraphNode(subscriptionReceiveNode, subscriptionJoinNode); - final StatefulProcessorNode> foreignTableJoinNode = new StatefulProcessorNode<>( + final StatefulProcessorNode> foreignTableJoinNode = new ForeignTableJoinNode<>( new ProcessorParameters<>( - new ForeignTableJoinProcessorSupplier<>(subscriptionStore, combinedKeySchema, foreignKeyValueGetter), + new ForeignTableJoinProcessorSupplier<>(subscriptionStore, combinedKeySchema), renamed.suffixWithOrElseGet("-foreign-join-subscription", builder, SUBSCRIPTION_PROCESSOR) ), Collections.singleton(subscriptionStore), - Collections.singleton(foreignKeyValueGetter) + Collections.emptySet() ); builder.addGraphNode(((KTableImpl) foreignKeyTable).graphNode, foreignTableJoinNode); @@ -1259,6 +1258,7 @@ private KTable doJoinOnForeignKey(final KTable forei resultSourceNodes.add(foreignResponseSource.nodeName()); builder.internalTopologyBuilder.copartitionSources(resultSourceNodes); + final KTableValueGetterSupplier primaryKeyValueGetter = valueGetterSupplier(); final StatefulProcessorNode> responseJoinNode = new StatefulProcessorNode<>( new ProcessorParameters<>( new ResponseJoinProcessorSupplier<>( diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableAbstractJoin.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableAbstractJoin.java index 665d839f2f436..21339c0b64966 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableAbstractJoin.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableAbstractJoin.java @@ -18,17 +18,16 @@ import org.apache.kafka.streams.kstream.ValueJoiner; -abstract class KTableKTableAbstractJoin implements +public abstract class KTableKTableAbstractJoin implements KTableProcessorSupplier { private final KTableImpl table1; private final KTableImpl table2; final KTableValueGetterSupplier valueGetterSupplier1; final KTableValueGetterSupplier valueGetterSupplier2; - final KTableValueGetter valueGetter1; - final KTableValueGetter valueGetter2; final ValueJoiner joiner; + boolean useVersionedSemantics = false; boolean sendOldValues = false; KTableKTableAbstractJoin(final KTableImpl table1, @@ -38,8 +37,6 @@ abstract class KTableKTableAbstractJoin implements this.table2 = table2; this.valueGetterSupplier1 = table1.valueGetterSupplier(); this.valueGetterSupplier2 = table2.valueGetterSupplier(); - this.valueGetter1 = valueGetterSupplier1.get(); - this.valueGetter2 = valueGetterSupplier2.get(); this.joiner = joiner; } @@ -51,4 +48,17 @@ public final boolean enableSendingOldValues(final boolean forceMaterialization) sendOldValues = true; return true; } + + public void setUseVersionedSemantics(final boolean useVersionedSemantics) { + this.useVersionedSemantics = useVersionedSemantics; + } + + // VisibleForTesting + public boolean isUseVersionedSemantics() { + return useVersionedSemantics; + } + + public String joinThisParentNodeName() { + return table1.graphNode.nodeName(); + } } diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableInnerJoin.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableInnerJoin.java index 98587b16fe5c0..517a741aaf675 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableInnerJoin.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableInnerJoin.java @@ -42,7 +42,7 @@ class KTableKTableInnerJoin extends KTableKTableAbstractJoin, K, Change> get() { - return new KTableKTableJoinProcessor(valueGetter1, valueGetter2); + return new KTableKTableJoinProcessor(valueGetterSupplier2.get()); } @Override @@ -64,14 +64,11 @@ public KTableValueGetter get() { private class KTableKTableJoinProcessor extends ContextualProcessor, K, Change> { - private final KTableValueGetter thisValueGetter; - private final KTableValueGetter otherValueGetter; + private final KTableValueGetter valueGetter; private Sensor droppedRecordsSensor; - KTableKTableJoinProcessor(final KTableValueGetter thisValueGetter, - final KTableValueGetter otherValueGetter) { - this.thisValueGetter = thisValueGetter; - this.otherValueGetter = otherValueGetter; + KTableKTableJoinProcessor(final KTableValueGetter valueGetter) { + this.valueGetter = valueGetter; } @Override @@ -82,8 +79,7 @@ public void init(final ProcessorContext> context) { context.taskId().toString(), (StreamsMetricsImpl) context.metrics() ); - thisValueGetter.init(context); - otherValueGetter.init(context); + valueGetter.init(context); } @Override @@ -107,20 +103,17 @@ public void process(final Record> record) { } // drop out-of-order records from versioned tables (cf. KIP-914) - if (thisValueGetter.isVersioned()) { - final ValueAndTimestamp valueAndTimestampLeft = thisValueGetter.get(record.key()); - if (valueAndTimestampLeft != null && valueAndTimestampLeft.timestamp() > record.timestamp()) { - LOG.info("Skipping out-of-order record from versioned table while performing table-table join."); - droppedRecordsSensor.record(); - return; - } + if (useVersionedSemantics && !record.value().isLatest) { + LOG.info("Skipping out-of-order record from versioned table while performing table-table join."); + droppedRecordsSensor.record(); + return; } VOut newValue = null; final long resultTimestamp; VOut oldValue = null; - final ValueAndTimestamp valueAndTimestampRight = otherValueGetter.get(record.key()); + final ValueAndTimestamp valueAndTimestampRight = valueGetter.get(record.key()); final V2 valueRight = getValueOrNull(valueAndTimestampRight); if (valueRight == null) { return; @@ -141,8 +134,7 @@ public void process(final Record> record) { @Override public void close() { - thisValueGetter.close(); - otherValueGetter.close(); + valueGetter.close(); } } diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableLeftJoin.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableLeftJoin.java index 46bcbda4e5eb8..7484c8efc1a32 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableLeftJoin.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableLeftJoin.java @@ -43,7 +43,7 @@ class KTableKTableLeftJoin extends KTableKTableAbstractJoin, K, Change> get() { - return new KTableKTableLeftJoinProcessor(valueGetter1, valueGetter2); + return new KTableKTableLeftJoinProcessor(valueGetterSupplier2.get()); } @Override @@ -66,14 +66,11 @@ public KTableValueGetter get() { private class KTableKTableLeftJoinProcessor extends ContextualProcessor, K, Change> { - private final KTableValueGetter thisValueGetter; - private final KTableValueGetter otherValueGetter; + private final KTableValueGetter valueGetter; private Sensor droppedRecordsSensor; - KTableKTableLeftJoinProcessor(final KTableValueGetter thisValueGetter, - final KTableValueGetter otherValueGetter) { - this.thisValueGetter = thisValueGetter; - this.otherValueGetter = otherValueGetter; + KTableKTableLeftJoinProcessor(final KTableValueGetter valueGetter) { + this.valueGetter = valueGetter; } @Override @@ -84,8 +81,7 @@ public void init(final ProcessorContext> context) { context.taskId().toString(), (StreamsMetricsImpl) context.metrics() ); - thisValueGetter.init(context); - otherValueGetter.init(context); + valueGetter.init(context); } @Override @@ -109,20 +105,17 @@ public void process(final Record> record) { } // drop out-of-order records from versioned tables (cf. KIP-914) - if (thisValueGetter.isVersioned()) { - final ValueAndTimestamp valueAndTimestampLeft = thisValueGetter.get(record.key()); - if (valueAndTimestampLeft != null && valueAndTimestampLeft.timestamp() > record.timestamp()) { - LOG.info("Skipping out-of-order record from versioned table while performing table-table join."); - droppedRecordsSensor.record(); - return; - } + if (useVersionedSemantics && !record.value().isLatest) { + LOG.info("Skipping out-of-order record from versioned table while performing table-table join."); + droppedRecordsSensor.record(); + return; } VOut newValue = null; final long resultTimestamp; VOut oldValue = null; - final ValueAndTimestamp valueAndTimestampRight = otherValueGetter.get(record.key()); + final ValueAndTimestamp valueAndTimestampRight = valueGetter.get(record.key()); final V2 value2 = getValueOrNull(valueAndTimestampRight); final long timestampRight; @@ -150,8 +143,7 @@ public void process(final Record> record) { @Override public void close() { - thisValueGetter.close(); - otherValueGetter.close(); + valueGetter.close(); } } diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableOuterJoin.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableOuterJoin.java index a06a7cc447983..e8f74075d45b6 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableOuterJoin.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableOuterJoin.java @@ -43,7 +43,7 @@ class KTableKTableOuterJoin extends KTableKTableAbstractJoin, K, Change> get() { - return new KTableKTableOuterJoinProcessor(valueGetter1, valueGetter2); + return new KTableKTableOuterJoinProcessor(valueGetterSupplier2.get()); } @Override @@ -65,14 +65,11 @@ public KTableValueGetter get() { private class KTableKTableOuterJoinProcessor extends ContextualProcessor, K, Change> { - private final KTableValueGetter otherValueGetter; - private final KTableValueGetter thisValueGetter; + private final KTableValueGetter valueGetter; private Sensor droppedRecordsSensor; - KTableKTableOuterJoinProcessor(final KTableValueGetter thisValueGetter, - final KTableValueGetter otherValueGetter) { - this.thisValueGetter = thisValueGetter; - this.otherValueGetter = otherValueGetter; + KTableKTableOuterJoinProcessor(final KTableValueGetter valueGetter) { + this.valueGetter = valueGetter; } @Override @@ -83,8 +80,7 @@ public void init(final ProcessorContext> context) { context.taskId().toString(), (StreamsMetricsImpl) context.metrics() ); - thisValueGetter.init(context); - otherValueGetter.init(context); + valueGetter.init(context); } @Override @@ -108,20 +104,17 @@ public void process(final Record> record) { } // drop out-of-order records from versioned tables (cf. KIP-914) - if (thisValueGetter.isVersioned()) { - final ValueAndTimestamp valueAndTimestamp1 = thisValueGetter.get(record.key()); - if (valueAndTimestamp1 != null && valueAndTimestamp1.timestamp() > record.timestamp()) { - LOG.info("Skipping out-of-order record from versioned table while performing table-table join."); - droppedRecordsSensor.record(); - return; - } + if (useVersionedSemantics && !record.value().isLatest) { + LOG.info("Skipping out-of-order record from versioned table while performing table-table join."); + droppedRecordsSensor.record(); + return; } VOut newValue = null; final long resultTimestamp; VOut oldValue = null; - final ValueAndTimestamp valueAndTimestamp2 = otherValueGetter.get(record.key()); + final ValueAndTimestamp valueAndTimestamp2 = valueGetter.get(record.key()); final V2 value2 = getValueOrNull(valueAndTimestamp2); if (value2 == null) { if (record.value().newValue == null && record.value().oldValue == null) { @@ -145,8 +138,7 @@ public void process(final Record> record) { @Override public void close() { - thisValueGetter.close(); - otherValueGetter.close(); + valueGetter.close(); } } diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableRightJoin.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableRightJoin.java index 08d7241fd70de..082a0fdae0e0d 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableRightJoin.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableRightJoin.java @@ -42,7 +42,7 @@ class KTableKTableRightJoin extends KTableKTableAbstractJoin, K, Change> get() { - return new KTableKTableRightJoinProcessor(valueGetter1, valueGetter2); + return new KTableKTableRightJoinProcessor(valueGetterSupplier2.get()); } @Override @@ -64,14 +64,11 @@ public KTableValueGetter get() { private class KTableKTableRightJoinProcessor extends ContextualProcessor, K, Change> { - private final KTableValueGetter thisValueGetter; - private final KTableValueGetter otherValueGetter; + private final KTableValueGetter valueGetter; private Sensor droppedRecordsSensor; - KTableKTableRightJoinProcessor(final KTableValueGetter thisValueGetter, - final KTableValueGetter otherValueGetter) { - this.thisValueGetter = thisValueGetter; - this.otherValueGetter = otherValueGetter; + KTableKTableRightJoinProcessor(final KTableValueGetter valueGetter) { + this.valueGetter = valueGetter; } @Override @@ -82,8 +79,7 @@ public void init(final ProcessorContext> context) { context.taskId().toString(), (StreamsMetricsImpl) context.metrics() ); - thisValueGetter.init(context); - otherValueGetter.init(context); + valueGetter.init(context); } @Override @@ -107,20 +103,17 @@ public void process(final Record> record) { } // drop out-of-order records from versioned tables (cf. KIP-914) - if (thisValueGetter.isVersioned()) { - final ValueAndTimestamp valueAndTimestampRight = thisValueGetter.get(record.key()); - if (valueAndTimestampRight != null && valueAndTimestampRight.timestamp() > record.timestamp()) { - LOG.info("Skipping out-of-order record from versioned table while performing table-table join."); - droppedRecordsSensor.record(); - return; - } + if (useVersionedSemantics && !record.value().isLatest) { + LOG.info("Skipping out-of-order record from versioned table while performing table-table join."); + droppedRecordsSensor.record(); + return; } final VOut newValue; final long resultTimestamp; VOut oldValue = null; - final ValueAndTimestamp valueAndTimestampLeft = otherValueGetter.get(record.key()); + final ValueAndTimestamp valueAndTimestampLeft = valueGetter.get(record.key()); final V2 valueLeft = getValueOrNull(valueAndTimestampLeft); if (valueLeft == null) { return; @@ -141,8 +134,7 @@ public void process(final Record> record) { @Override public void close() { - thisValueGetter.close(); - otherValueGetter.close(); + valueGetter.close(); } } diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/ForeignTableJoinProcessorSupplier.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/ForeignTableJoinProcessorSupplier.java index a9cae20337813..314ed7cdfd3d8 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/ForeignTableJoinProcessorSupplier.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/ForeignTableJoinProcessorSupplier.java @@ -21,8 +21,6 @@ import org.apache.kafka.common.utils.Bytes; import org.apache.kafka.streams.KeyValue; import org.apache.kafka.streams.kstream.internals.Change; -import org.apache.kafka.streams.kstream.internals.KTableValueGetter; -import org.apache.kafka.streams.kstream.internals.KTableValueGetterSupplier; import org.apache.kafka.streams.processor.api.ContextualProcessor; import org.apache.kafka.streams.processor.api.Processor; import org.apache.kafka.streams.processor.api.ProcessorContext; @@ -45,32 +43,33 @@ public class ForeignTableJoinProcessorSupplier implements private static final Logger LOG = LoggerFactory.getLogger(ForeignTableJoinProcessorSupplier.class); private final StoreBuilder>> storeBuilder; private final CombinedKeySchema keySchema; - private final KTableValueGetterSupplier foreignKeyValueGetterSupplier; + private boolean useVersionedSemantics = false; public ForeignTableJoinProcessorSupplier( final StoreBuilder>> storeBuilder, - final CombinedKeySchema keySchema, - final KTableValueGetterSupplier foreignKeyValueGetterSupplier) { + final CombinedKeySchema keySchema) { this.storeBuilder = storeBuilder; this.keySchema = keySchema; - this.foreignKeyValueGetterSupplier = foreignKeyValueGetterSupplier; } @Override public Processor, K, SubscriptionResponseWrapper> get() { - return new KTableKTableJoinProcessor(foreignKeyValueGetterSupplier.get()); + return new KTableKTableJoinProcessor(); } + public void setUseVersionedSemantics(final boolean useVersionedSemantics) { + this.useVersionedSemantics = useVersionedSemantics; + } + + // VisibleForTesting + public boolean isUseVersionedSemantics() { + return useVersionedSemantics; + } private final class KTableKTableJoinProcessor extends ContextualProcessor, K, SubscriptionResponseWrapper> { private Sensor droppedRecordsSensor; private TimestampedKeyValueStore> subscriptionStore; - private final KTableValueGetter foreignKeyValueGetter; - - private KTableKTableJoinProcessor(final KTableValueGetter foreignKeyValueGetter) { - this.foreignKeyValueGetter = foreignKeyValueGetter; - } @Override public void init(final ProcessorContext> context) { @@ -82,7 +81,6 @@ public void init(final ProcessorContext> cont internalProcessorContext.metrics() ); subscriptionStore = internalProcessorContext.getStateStore(storeBuilder); - foreignKeyValueGetter.init(context); } @Override @@ -107,13 +105,10 @@ public void process(final Record> record) { } // drop out-of-order records from versioned tables (cf. KIP-914) - if (foreignKeyValueGetter.isVersioned()) { - final ValueAndTimestamp latestValueAndTimestamp = foreignKeyValueGetter.get(record.key()); - if (latestValueAndTimestamp != null && latestValueAndTimestamp.timestamp() > record.timestamp()) { - LOG.info("Skipping out-of-order record from versioned table while performing table-table join."); - droppedRecordsSensor.record(); - return; - } + if (useVersionedSemantics && !record.value().isLatest) { + LOG.info("Skipping out-of-order record from versioned table while performing table-table join."); + droppedRecordsSensor.record(); + return; } final Bytes prefixBytes = keySchema.prefixBytes(record.key()); @@ -139,11 +134,6 @@ public void process(final Record> record) { } } - @Override - public void close() { - foreignKeyValueGetter.close(); - } - private boolean prefixEquals(final byte[] x, final byte[] y) { final int min = Math.min(x.length, y.length); final ByteBuffer xSlice = ByteBuffer.wrap(x, 0, min); diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionJoinProcessorSupplier.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionJoinProcessorSupplier.java index 490f70ffc8d38..7d31ef4422345 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionJoinProcessorSupplier.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionJoinProcessorSupplier.java @@ -33,7 +33,7 @@ /** * Receives {@code SubscriptionWrapper} events and processes them according to their Instruction. * Depending on the results, {@code SubscriptionResponseWrapper}s are created, which will be propagated to - * the {@code SubscriptionResolverJoinProcessorSupplier} instance. + * the {@code ResponseJoinProcessorSupplier} instance. * * @param Type of primary keys * @param Type of foreign key diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionReceiveProcessorSupplier.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionReceiveProcessorSupplier.java index 5c386cc735bb5..cf88aec6f9a09 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionReceiveProcessorSupplier.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionReceiveProcessorSupplier.java @@ -103,7 +103,7 @@ public void process(final Record> record) { final ValueAndTimestamp> newValue = ValueAndTimestamp.make(record.value(), record.timestamp()); final ValueAndTimestamp> oldValue = store.get(subscriptionKey); - //This store is used by the prefix scanner in ForeignJoinSubscriptionProcessorSupplier + //This store is used by the prefix scanner in ForeignTableJoinProcessorSupplier if (record.value().getInstruction().equals(SubscriptionWrapper.Instruction.DELETE_KEY_AND_PROPAGATE) || record.value().getInstruction().equals(SubscriptionWrapper.Instruction.DELETE_KEY_NO_PROPAGATE)) { store.delete(subscriptionKey); diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionSendProcessorSupplier.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionSendProcessorSupplier.java index feb06b02e13ea..8a6298c28de45 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionSendProcessorSupplier.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionSendProcessorSupplier.java @@ -21,8 +21,6 @@ import org.apache.kafka.common.serialization.Serde; import org.apache.kafka.common.serialization.Serializer; import org.apache.kafka.streams.kstream.internals.Change; -import org.apache.kafka.streams.kstream.internals.KTableValueGetter; -import org.apache.kafka.streams.kstream.internals.KTableValueGetterSupplier; import org.apache.kafka.streams.processor.api.ContextualProcessor; import org.apache.kafka.streams.processor.api.Processor; import org.apache.kafka.streams.processor.api.ProcessorContext; @@ -31,7 +29,6 @@ import org.apache.kafka.streams.processor.api.RecordMetadata; import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl; import org.apache.kafka.streams.processor.internals.metrics.TaskMetrics; -import org.apache.kafka.streams.state.ValueAndTimestamp; import org.apache.kafka.streams.state.internals.Murmur3; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -52,29 +49,36 @@ public class SubscriptionSendProcessorSupplier implements ProcessorSup private final Supplier foreignKeySerdeTopicSupplier; private final Supplier valueSerdeTopicSupplier; private final boolean leftJoin; - private final KTableValueGetterSupplier primaryKeyValueGetterSupplier; private Serializer foreignKeySerializer; private Serializer valueSerializer; + private boolean useVersionedSemantics; public SubscriptionSendProcessorSupplier(final Function foreignKeyExtractor, final Supplier foreignKeySerdeTopicSupplier, final Supplier valueSerdeTopicSupplier, final Serde foreignKeySerde, final Serializer valueSerializer, - final boolean leftJoin, - final KTableValueGetterSupplier primaryKeyValueGetterSupplier) { + final boolean leftJoin) { this.foreignKeyExtractor = foreignKeyExtractor; this.foreignKeySerdeTopicSupplier = foreignKeySerdeTopicSupplier; this.valueSerdeTopicSupplier = valueSerdeTopicSupplier; this.valueSerializer = valueSerializer; this.leftJoin = leftJoin; - this.primaryKeyValueGetterSupplier = primaryKeyValueGetterSupplier; foreignKeySerializer = foreignKeySerde == null ? null : foreignKeySerde.serializer(); } @Override public Processor, KO, SubscriptionWrapper> get() { - return new UnbindChangeProcessor(primaryKeyValueGetterSupplier.get()); + return new UnbindChangeProcessor(); + } + + public void setUseVersionedSemantics(final boolean useVersionedSemantics) { + this.useVersionedSemantics = useVersionedSemantics; + } + + // VisibleForTesting + public boolean isUseVersionedSemantics() { + return useVersionedSemantics; } private class UnbindChangeProcessor extends ContextualProcessor, KO, SubscriptionWrapper> { @@ -82,11 +86,6 @@ private class UnbindChangeProcessor extends ContextualProcessor, KO private Sensor droppedRecordsSensor; private String foreignKeySerdeTopic; private String valueSerdeTopic; - private final KTableValueGetter primaryKeyValueGetter; - - private UnbindChangeProcessor(final KTableValueGetter primaryKeyValueGetter) { - this.primaryKeyValueGetter = primaryKeyValueGetter; - } @SuppressWarnings("unchecked") @Override @@ -106,23 +105,15 @@ public void init(final ProcessorContext> context) { context.taskId().toString(), (StreamsMetricsImpl) context.metrics() ); - primaryKeyValueGetter.init(context); } @Override public void process(final Record> record) { // drop out-of-order records from versioned tables (cf. KIP-914) - if (primaryKeyValueGetter.isVersioned()) { - // key-value stores do not contain data for null keys, so skip the check - // if the key is null - if (record.key() != null) { - final ValueAndTimestamp latestValueAndTimestamp = primaryKeyValueGetter.get(record.key()); - if (latestValueAndTimestamp != null && latestValueAndTimestamp.timestamp() > record.timestamp()) { - LOG.info("Skipping out-of-order record from versioned table while performing table-table join."); - droppedRecordsSensor.record(); - return; - } - } + if (useVersionedSemantics && !record.value().isLatest) { + LOG.info("Skipping out-of-order record from versioned table while performing table-table join."); + droppedRecordsSensor.record(); + return; } final long[] currentHash = record.value().newValue == null ? @@ -208,11 +199,6 @@ public void process(final Record> record) { } } - @Override - public void close() { - primaryKeyValueGetter.close(); - } - private void logSkippedRecordDueToNullForeignKey() { if (context().recordMetadata().isPresent()) { final RecordMetadata recordMetadata = context().recordMetadata().get(); diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/BaseJoinProcessorNode.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/BaseJoinProcessorNode.java index 121db53794330..0c0dcb3bec9df 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/BaseJoinProcessorNode.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/BaseJoinProcessorNode.java @@ -51,11 +51,13 @@ abstract class BaseJoinProcessorNode extends GraphNode { this.otherJoinSideNodeName = otherJoinSideNodeName; } - ProcessorParameters thisProcessorParameters() { + // VisibleForTesting + public ProcessorParameters thisProcessorParameters() { return joinThisProcessorParameters; } - ProcessorParameters otherProcessorParameters() { + // VisibleForTesting + public ProcessorParameters otherProcessorParameters() { return joinOtherProcessorParameters; } diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/ForeignJoinSubscriptionSendNode.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/ForeignJoinSubscriptionSendNode.java new file mode 100644 index 0000000000000..4efbd9b29f1c1 --- /dev/null +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/ForeignJoinSubscriptionSendNode.java @@ -0,0 +1,40 @@ +/* + * 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.kstream.internals.foreignkeyjoin.SubscriptionSendProcessorSupplier; +import org.apache.kafka.streams.processor.api.ProcessorSupplier; + +public class ForeignJoinSubscriptionSendNode extends ProcessorGraphNode implements VersionedSemanticsGraphNode { + + public ForeignJoinSubscriptionSendNode(final ProcessorParameters processorParameters) { + super(processorParameters); + } + + @SuppressWarnings("unchecked") + @Override + public void enableVersionedSemantics(final boolean useVersionedSemantics, final String parentNodeName) { + final ProcessorSupplier processorSupplier = processorParameters().processorSupplier(); + if (!(processorSupplier instanceof SubscriptionSendProcessorSupplier)) { + throw new IllegalStateException("Unexpected processor type for foreign-key table-table join subscription send processor: " + processorSupplier.getClass().getName()); + } + + final SubscriptionSendProcessorSupplier subscriptionSendProcessor + = (SubscriptionSendProcessorSupplier) processorSupplier; + subscriptionSendProcessor.setUseVersionedSemantics(useVersionedSemantics); + } +} diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/ForeignTableJoinNode.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/ForeignTableJoinNode.java new file mode 100644 index 0000000000000..82d9d6486ad8a --- /dev/null +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/ForeignTableJoinNode.java @@ -0,0 +1,45 @@ +/* + * 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 java.util.Set; +import org.apache.kafka.streams.kstream.internals.KTableValueGetterSupplier; +import org.apache.kafka.streams.kstream.internals.foreignkeyjoin.ForeignTableJoinProcessorSupplier; +import org.apache.kafka.streams.processor.api.ProcessorSupplier; +import org.apache.kafka.streams.state.StoreBuilder; + +public class ForeignTableJoinNode extends StatefulProcessorNode implements VersionedSemanticsGraphNode { + + public ForeignTableJoinNode(final ProcessorParameters processorParameters, + final Set> preRegisteredStores, + final Set> valueGetterSuppliers) { + super(processorParameters, preRegisteredStores, valueGetterSuppliers); + } + + @SuppressWarnings("unchecked") + @Override + public void enableVersionedSemantics(final boolean useVersionedSemantics, final String parentNodeName) { + final ProcessorSupplier processorSupplier = processorParameters().processorSupplier(); + if (!(processorSupplier instanceof ForeignTableJoinProcessorSupplier)) { + throw new IllegalStateException("Unexpected processor type for foreign-key table-table join subscription processor: " + processorSupplier.getClass().getName()); + } + + final ForeignTableJoinProcessorSupplier subscriptionProcessor + = (ForeignTableJoinProcessorSupplier) processorSupplier; + subscriptionProcessor.setUseVersionedSemantics(useVersionedSemantics); + } +} diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/KTableKTableJoinNode.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/KTableKTableJoinNode.java index 8c2f945a64ee0..ea07b4ba2f895 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/KTableKTableJoinNode.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/KTableKTableJoinNode.java @@ -17,11 +17,9 @@ package org.apache.kafka.streams.kstream.internals.graph; -import java.util.Collections; -import java.util.HashSet; -import java.util.Set; import org.apache.kafka.common.serialization.Serde; import org.apache.kafka.streams.kstream.internals.Change; +import org.apache.kafka.streams.kstream.internals.KTableKTableAbstractJoin; import org.apache.kafka.streams.kstream.internals.KTableKTableJoinMerger; import org.apache.kafka.streams.kstream.internals.KTableProcessorSupplier; import org.apache.kafka.streams.processor.api.ProcessorSupplier; @@ -33,7 +31,7 @@ /** * Too much specific information to generalize so the KTable-KTable join requires a specific node. */ -public class KTableKTableJoinNode extends BaseJoinProcessorNode, Change, Change> { +public class KTableKTableJoinNode extends BaseJoinProcessorNode, Change, Change> implements VersionedSemanticsGraphNode { private final Serde keySerde; private final Serde valueSerde; @@ -97,6 +95,27 @@ public KTableKTableJoinMerger joinMerger() { return (KTableKTableJoinMerger) kChangeProcessorSupplier; } + @Override + public void enableVersionedSemantics(final boolean useVersionedSemantics, final String parentNodeName) { + enableVersionedSemantics(thisProcessorParameters(), useVersionedSemantics, parentNodeName); + enableVersionedSemantics(otherProcessorParameters(), useVersionedSemantics, parentNodeName); + } + + @SuppressWarnings("unchecked") + private void enableVersionedSemantics(final ProcessorParameters processorParameters, + final boolean useVersionedSemantics, + final String parentNodeName) { + final ProcessorSupplier processorSupplier = processorParameters.processorSupplier(); + if (!(processorSupplier instanceof KTableKTableAbstractJoin)) { + throw new IllegalStateException("Unexpected processor type for table-table join: " + processorSupplier.getClass().getName()); + } + final KTableKTableAbstractJoin tableJoin = (KTableKTableAbstractJoin) processorSupplier; + + if (parentNodeName.equals(tableJoin.joinThisParentNodeName())) { + tableJoin.setUseVersionedSemantics(useVersionedSemantics); + } + } + @Override public void writeToTopology(final InternalTopologyBuilder topologyBuilder) { final String thisProcessorName = thisProcessorParameters().processorName(); @@ -119,14 +138,8 @@ public void writeToTopology(final InternalTopologyBuilder topologyBuilder) { thisProcessorName, otherProcessorName); - // join processors for both sides of the join should have access to both stores - final Set stores = new HashSet<>(joinThisStoreNames.length + joinOtherStoreNames.length); - Collections.addAll(stores, joinThisStoreNames); - Collections.addAll(stores, joinOtherStoreNames); - final String[] mergedStoreNames = stores.toArray(new String[0]); - - topologyBuilder.connectProcessorAndStateStores(thisProcessorName, mergedStoreNames); - topologyBuilder.connectProcessorAndStateStores(otherProcessorName, mergedStoreNames); + topologyBuilder.connectProcessorAndStateStores(thisProcessorName, joinOtherStoreNames); + topologyBuilder.connectProcessorAndStateStores(otherProcessorName, joinThisStoreNames); if (storeBuilder != null) { topologyBuilder.addStateStore(storeBuilder, mergeProcessorName); diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/TableFilterNode.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/TableFilterNode.java index e53d5b43a0d85..e13adbc9c390f 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/TableFilterNode.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/TableFilterNode.java @@ -31,7 +31,7 @@ public TableFilterNode(final String nodeName, @SuppressWarnings("unchecked") @Override - public void enableVersionedSemantics(final boolean useVersionedSemantics) { + public void enableVersionedSemantics(final boolean useVersionedSemantics, final String parentNodeName) { final ProcessorSupplier processorSupplier = processorParameters().processorSupplier(); if (!(processorSupplier instanceof KTableFilter)) { throw new IllegalStateException("Unexpected processor type for table filter: " + processorSupplier.getClass().getName()); diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/TableRepartitionMapNode.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/TableRepartitionMapNode.java index 4d5e283aa62f2..6f56e052b69e6 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/TableRepartitionMapNode.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/TableRepartitionMapNode.java @@ -29,7 +29,7 @@ public TableRepartitionMapNode(final String nodeName, @SuppressWarnings("unchecked") @Override - public void enableVersionedSemantics(final boolean useVersionedSemantics) { + public void enableVersionedSemantics(final boolean useVersionedSemantics, final String parentNodeName) { final ProcessorSupplier processorSupplier = processorParameters().processorSupplier(); if (!(processorSupplier instanceof KTableRepartitionMap)) { throw new IllegalStateException("Unexpected processor type for table repartition map: " + processorSupplier.getClass().getName()); diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/VersionedSemanticsGraphNode.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/VersionedSemanticsGraphNode.java index 9b7e578bfe09c..c4cc9c3934dad 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/VersionedSemanticsGraphNode.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/VersionedSemanticsGraphNode.java @@ -28,6 +28,6 @@ public interface VersionedSemanticsGraphNode { /** * @param useVersionedSemantics whether versioned semantics should be enabled */ - void enableVersionedSemantics(final boolean useVersionedSemantics); + void enableVersionedSemantics(boolean useVersionedSemantics, String parentNodeName); } diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/AbstractJoinIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/AbstractJoinIntegrationTest.java index 5805ecfeff607..ff4d2e69e1a81 100644 --- a/streams/src/test/java/org/apache/kafka/streams/integration/AbstractJoinIntegrationTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/integration/AbstractJoinIntegrationTest.java @@ -16,6 +16,7 @@ */ package org.apache.kafka.streams.integration; +import java.util.Collections; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.common.serialization.LongDeserializer; import org.apache.kafka.common.serialization.LongSerializer; @@ -87,7 +88,37 @@ public static Collection data() { StreamsBuilder builder; - private final List> input = Arrays.asList( + protected final List> input = Arrays.asList( + new Input<>(INPUT_TOPIC_LEFT, null, 1), + new Input<>(INPUT_TOPIC_RIGHT, null, 2), + new Input<>(INPUT_TOPIC_LEFT, "A", 3), + new Input<>(INPUT_TOPIC_RIGHT, "a", 4), + new Input<>(INPUT_TOPIC_LEFT, "B", 5), + new Input<>(INPUT_TOPIC_RIGHT, "b", 6), + new Input<>(INPUT_TOPIC_LEFT, null, 7), + new Input<>(INPUT_TOPIC_RIGHT, null, 8), + new Input<>(INPUT_TOPIC_LEFT, "C", 9), + new Input<>(INPUT_TOPIC_RIGHT, "c", 10), + new Input<>(INPUT_TOPIC_RIGHT, null, 11), + new Input<>(INPUT_TOPIC_LEFT, null, 12), + new Input<>(INPUT_TOPIC_RIGHT, null, 13), + new Input<>(INPUT_TOPIC_RIGHT, "d", 7), // out-of-order data with null as latest + new Input<>(INPUT_TOPIC_LEFT, "D", 6), + new Input<>(INPUT_TOPIC_LEFT, null, 2), + new Input<>(INPUT_TOPIC_RIGHT, null, 3), + new Input<>(INPUT_TOPIC_RIGHT, "e", 14), + new Input<>(INPUT_TOPIC_LEFT, "E", 15), + new Input<>(INPUT_TOPIC_LEFT, null, 10), // out-of-order data with non-null as latest + new Input<>(INPUT_TOPIC_RIGHT, null, 9), + new Input<>(INPUT_TOPIC_LEFT, "F", 4), + new Input<>(INPUT_TOPIC_RIGHT, "f", 3) + ); + + // used for stream-stream join tests where out-of-order data does not meaningfully affect + // the result, and the main `input` list results in too many result records/test noise. + // also used for table-table multi-join tests, since out-of-order data with table-table + // joins is already tested in non-multi-join settings. + protected final List> inputWithoutOutOfOrderData = Arrays.asList( new Input<>(INPUT_TOPIC_LEFT, null, 1), new Input<>(INPUT_TOPIC_RIGHT, null, 2), new Input<>(INPUT_TOPIC_LEFT, "A", 3), @@ -102,13 +133,10 @@ public static Collection data() { new Input<>(INPUT_TOPIC_LEFT, null, 12), new Input<>(INPUT_TOPIC_RIGHT, null, 13), new Input<>(INPUT_TOPIC_RIGHT, "d", 14), - new Input<>(INPUT_TOPIC_LEFT, "D", 15), - new Input<>(INPUT_TOPIC_LEFT, "E", 4), // out-of-order data - new Input<>(INPUT_TOPIC_RIGHT, "e", 3), - new Input<>(INPUT_TOPIC_RIGHT, "f", 7), - new Input<>(INPUT_TOPIC_LEFT, "F", 8) + new Input<>(INPUT_TOPIC_LEFT, "D", 15) ); + // used for stream-stream self joins where only one input topic is needed private final List> leftInput = Arrays.asList( new Input<>(INPUT_TOPIC_LEFT, null, 1), new Input<>(INPUT_TOPIC_LEFT, "A", 2), @@ -144,11 +172,11 @@ void prepareEnvironment() throws InterruptedException { STREAMS_CONFIG.put(StreamsConfig.STATE_DIR_CONFIG, testFolder.getRoot().getPath()); } - void runTestWithDriver(final List>> expectedResult) { - runTestWithDriver(expectedResult, null); + void runTestWithDriver(final List> input, final List>> expectedResult) { + runTestWithDriver(input, expectedResult, null); } - void runTestWithDriver(final List>> expectedResult, final String storeName) { + void runTestWithDriver(final List> input, final List>> expectedResult, final String storeName) { try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(STREAMS_CONFIG), STREAMS_CONFIG)) { final TestInputTopic right = driver.createInputTopic(INPUT_TOPIC_RIGHT, new LongSerializer(), new StringSerializer()); final TestInputTopic left = driver.createInputTopic(INPUT_TOPIC_LEFT, new LongSerializer(), new StringSerializer()); @@ -175,6 +203,9 @@ void runTestWithDriver(final List>> expectedResult final List> output = outputTopic.readRecordsToList(); assertThat(output, equalTo(updatedExpected)); expectedFinalResult = updatedExpected.get(expected.size() - 1); + } else { + final List> output = outputTopic.readRecordsToList(); + assertThat(output, equalTo(Collections.emptyList())); } } @@ -184,7 +215,7 @@ void runTestWithDriver(final List>> expectedResult } } - void runTestWithDriver(final TestRecord expectedFinalResult, final String storeName) throws InterruptedException { + void runTestWithDriver(final List> input, final TestRecord expectedFinalResult, final String storeName) throws InterruptedException { try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(STREAMS_CONFIG), STREAMS_CONFIG)) { final TestInputTopic right = driver.createInputTopic(INPUT_TOPIC_RIGHT, new LongSerializer(), new StringSerializer()); final TestInputTopic left = driver.createInputTopic(INPUT_TOPIC_LEFT, new LongSerializer(), new StringSerializer()); @@ -258,7 +289,7 @@ private void checkQueryableStore(final String queryableName, final TestRecord { + protected static final class Input { String topic; KeyValue record; long timestamp; diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/KTableKTableForeignKeyVersionedJoinIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/KTableKTableForeignKeyVersionedJoinIntegrationTest.java index 10f632ffdea0f..3703c910388d6 100644 --- a/streams/src/test/java/org/apache/kafka/streams/integration/KTableKTableForeignKeyVersionedJoinIntegrationTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/integration/KTableKTableForeignKeyVersionedJoinIntegrationTest.java @@ -85,7 +85,6 @@ public void shouldIgnoreOutOfOrderRecordsIffVersioned() { // LHS records with match to existing RHS record left.pipeInput("lhs1", "lhsValue1|rhs1", baseTimestamp + 3); left.pipeInput("lhs2", "lhsValue2|rhs1", baseTimestamp + 5); - { final Map expected = mkMap( mkEntry("lhs1", "(lhsValue1|rhs1,rhsValue1)"), @@ -103,8 +102,28 @@ public void shouldIgnoreOutOfOrderRecordsIffVersioned() { } } + // replace with tombstone, to validate behavior when latest record is null + left.pipeInput("lhs2", null, baseTimestamp + 6); + { + assertThat( + outputTopic.readKeyValuesToMap(), + is(mkMap( + mkEntry("lhs2", null) + )) + ); + if (materialized) { + assertThat( + asMap(store), + is(mkMap( + mkEntry("lhs1", "(lhsValue1|rhs1,rhsValue1)") + )) + ); + } + } + // out-of-order LHS record (for existing key) does not produce a new result iff LHS is versioned left.pipeInput("lhs1", "lhsValue1_ooo|rhs1", baseTimestamp + 2); + left.pipeInput("lhs2", "lhsValue2_ooo|rhs1", baseTimestamp + 2); if (leftVersioned) { assertThat( outputTopic.readKeyValuesToMap(), @@ -114,25 +133,23 @@ public void shouldIgnoreOutOfOrderRecordsIffVersioned() { assertThat( asMap(store), is(mkMap( - mkEntry("lhs1", "(lhsValue1|rhs1,rhsValue1)"), - mkEntry("lhs2", "(lhsValue2|rhs1,rhsValue1)") + mkEntry("lhs1", "(lhsValue1|rhs1,rhsValue1)") )) ); } } else { + final Map expected = mkMap( + mkEntry("lhs1", "(lhsValue1_ooo|rhs1,rhsValue1)"), + mkEntry("lhs2", "(lhsValue2_ooo|rhs1,rhsValue1)") + ); assertThat( outputTopic.readKeyValuesToMap(), - is(mkMap( - mkEntry("lhs1", "(lhsValue1_ooo|rhs1,rhsValue1)") - )) + is(expected) ); if (materialized) { assertThat( asMap(store), - is(mkMap( - mkEntry("lhs1", "(lhsValue1_ooo|rhs1,rhsValue1)"), - mkEntry("lhs2", "(lhsValue2|rhs1,rhsValue1)") - )) + is(expected) ); } } @@ -148,8 +165,7 @@ public void shouldIgnoreOutOfOrderRecordsIffVersioned() { assertThat( asMap(store), is(mkMap( - mkEntry("lhs1", "(lhsValue1|rhs1,rhsValue1)"), - mkEntry("lhs2", "(lhsValue2|rhs1,rhsValue1)") + mkEntry("lhs1", "(lhsValue1|rhs1,rhsValue1)") )) ); } @@ -164,7 +180,7 @@ public void shouldIgnoreOutOfOrderRecordsIffVersioned() { assertThat( asMap(store), is(mkMap( - mkEntry("lhs2", "(lhsValue2|rhs1,rhsValue1)") + mkEntry("lhs2", "(lhsValue2_ooo|rhs1,rhsValue1)") )) ); } @@ -172,20 +188,20 @@ public void shouldIgnoreOutOfOrderRecordsIffVersioned() { // LHS record with larger timestamp always produces a new result left.pipeInput("lhs1", "lhsValue1_new|rhs1", baseTimestamp + 8); + left.pipeInput("lhs2", "lhsValue2_new|rhs1", baseTimestamp + 8); { + final Map expected = mkMap( + mkEntry("lhs1", "(lhsValue1_new|rhs1,rhsValue1)"), + mkEntry("lhs2", "(lhsValue2_new|rhs1,rhsValue1)") + ); assertThat( outputTopic.readKeyValuesToMap(), - is(mkMap( - mkEntry("lhs1", "(lhsValue1_new|rhs1,rhsValue1)") - )) + is(expected) ); if (materialized) { assertThat( asMap(store), - is(mkMap( - mkEntry("lhs1", "(lhsValue1_new|rhs1,rhsValue1)"), - mkEntry("lhs2", "(lhsValue2|rhs1,rhsValue1)") - )) + is(expected) ); } } @@ -202,7 +218,7 @@ public void shouldIgnoreOutOfOrderRecordsIffVersioned() { asMap(store), is(mkMap( mkEntry("lhs1", "(lhsValue1_new|rhs1,rhsValue1)"), - mkEntry("lhs2", "(lhsValue2|rhs1,rhsValue1)") + mkEntry("lhs2", "(lhsValue2_new|rhs1,rhsValue1)") )) ); } @@ -211,7 +227,7 @@ public void shouldIgnoreOutOfOrderRecordsIffVersioned() { outputTopic.readKeyValuesToMap(), is(mkMap( mkEntry("lhs1", "(lhsValue1_new|rhs1,rhsValue1_ooo)"), - mkEntry("lhs2", "(lhsValue2|rhs1,rhsValue1_ooo)") + mkEntry("lhs2", "(lhsValue2_new|rhs1,rhsValue1_ooo)") )) ); if (materialized) { @@ -219,7 +235,7 @@ public void shouldIgnoreOutOfOrderRecordsIffVersioned() { asMap(store), is(mkMap( mkEntry("lhs1", "(lhsValue1_new|rhs1,rhsValue1_ooo)"), - mkEntry("lhs2", "(lhsValue2|rhs1,rhsValue1_ooo)") + mkEntry("lhs2", "(lhsValue2_new|rhs1,rhsValue1_ooo)") )) ); } @@ -237,7 +253,7 @@ public void shouldIgnoreOutOfOrderRecordsIffVersioned() { asMap(store), is(mkMap( mkEntry("lhs1", "(lhsValue1_new|rhs1,rhsValue1)"), - mkEntry("lhs2", "(lhsValue2|rhs1,rhsValue1)") + mkEntry("lhs2", "(lhsValue2_new|rhs1,rhsValue1)") )) ); } @@ -247,7 +263,7 @@ public void shouldIgnoreOutOfOrderRecordsIffVersioned() { outputTopic.readKeyValuesToMap(), is(mkMap( mkEntry("lhs1", "(lhsValue1_new|rhs1,null)"), - mkEntry("lhs2", "(lhsValue2|rhs1,null)") + mkEntry("lhs2", "(lhsValue2_new|rhs1,null)") )) ); if (materialized) { @@ -255,7 +271,7 @@ public void shouldIgnoreOutOfOrderRecordsIffVersioned() { asMap(store), is(mkMap( mkEntry("lhs1", "(lhsValue1_new|rhs1,null)"), - mkEntry("lhs2", "(lhsValue2|rhs1,null)") + mkEntry("lhs2", "(lhsValue2_new|rhs1,null)") )) ); } @@ -283,7 +299,7 @@ public void shouldIgnoreOutOfOrderRecordsIffVersioned() { outputTopic.readKeyValuesToMap(), is(mkMap( mkEntry("lhs1", "(lhsValue1_new|rhs1,rhsValue1_new)"), - mkEntry("lhs2", "(lhsValue2|rhs1,rhsValue1_new)") + mkEntry("lhs2", "(lhsValue2_new|rhs1,rhsValue1_new)") )) ); if (materialized) { @@ -291,7 +307,7 @@ public void shouldIgnoreOutOfOrderRecordsIffVersioned() { asMap(store), is(mkMap( mkEntry("lhs1", "(lhsValue1_new|rhs1,rhsValue1_new)"), - mkEntry("lhs2", "(lhsValue2|rhs1,rhsValue1_new)") + mkEntry("lhs2", "(lhsValue2_new|rhs1,rhsValue1_new)") )) ); } diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/StreamStreamJoinIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/StreamStreamJoinIntegrationTest.java index 725ed7e3f6630..554c5692a96ea 100644 --- a/streams/src/test/java/org/apache/kafka/streams/integration/StreamStreamJoinIntegrationTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/integration/StreamStreamJoinIntegrationTest.java @@ -137,31 +137,7 @@ public void testInner() { new TestRecord<>(ANY_UNIQUE_KEY, "D-a", null, 15L), new TestRecord<>(ANY_UNIQUE_KEY, "D-b", null, 15L), new TestRecord<>(ANY_UNIQUE_KEY, "D-c", null, 15L), - new TestRecord<>(ANY_UNIQUE_KEY, "D-d", null, 15L)), - Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "E-a", null, 4L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-b", null, 6L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-c", null, 10L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-d", null, 14L)), - Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "A-e", null, 3L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-e", null, 4L), - new TestRecord<>(ANY_UNIQUE_KEY, "B-e", null, 5L), - new TestRecord<>(ANY_UNIQUE_KEY, "C-e", null, 9L), - new TestRecord<>(ANY_UNIQUE_KEY, "D-e", null, 15L)), - Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "A-f", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-f", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "B-f", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "C-f", null, 9L), - new TestRecord<>(ANY_UNIQUE_KEY, "D-f", null, 15L)), - Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "F-e", null, 8L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-a", null, 8L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-b", null, 8L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-f", null, 8L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-c", null, 10L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-d", null, 14L)) + new TestRecord<>(ANY_UNIQUE_KEY, "D-d", null, 15L)) ); leftStream.join( @@ -170,7 +146,7 @@ public void testInner() { JoinWindows.ofTimeDifferenceAndGrace(ofSeconds(10), ofHours(24)) ).to(OUTPUT_TOPIC); - runTestWithDriver(expectedResult); + runTestWithDriver(inputWithoutOutOfOrderData, expectedResult); } @Test @@ -206,31 +182,7 @@ public void testInnerRepartitioned() { new TestRecord<>(ANY_UNIQUE_KEY, "D-a", null, 15L), new TestRecord<>(ANY_UNIQUE_KEY, "D-b", null, 15L), new TestRecord<>(ANY_UNIQUE_KEY, "D-c", null, 15L), - new TestRecord<>(ANY_UNIQUE_KEY, "D-d", null, 15L)), - Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "E-a", null, 4L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-b", null, 6L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-c", null, 10L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-d", null, 14L)), - Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "A-e", null, 3L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-e", null, 4L), - new TestRecord<>(ANY_UNIQUE_KEY, "B-e", null, 5L), - new TestRecord<>(ANY_UNIQUE_KEY, "C-e", null, 9L), - new TestRecord<>(ANY_UNIQUE_KEY, "D-e", null, 15L)), - Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "A-f", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-f", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "B-f", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "C-f", null, 9L), - new TestRecord<>(ANY_UNIQUE_KEY, "D-f", null, 15L)), - Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "F-e", null, 8L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-a", null, 8L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-b", null, 8L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-f", null, 8L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-c", null, 10L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-d", null, 14L)) + new TestRecord<>(ANY_UNIQUE_KEY, "D-d", null, 15L)) ); leftStream.map(MockMapper.noOpKeyValueMapper()) @@ -241,7 +193,7 @@ public void testInnerRepartitioned() { JoinWindows.ofTimeDifferenceAndGrace(ofSeconds(10), ofHours(24)) ).to(OUTPUT_TOPIC); - runTestWithDriver(expectedResult); + runTestWithDriver(inputWithoutOutOfOrderData, expectedResult); } @Test @@ -277,31 +229,7 @@ public void testLeft() { new TestRecord<>(ANY_UNIQUE_KEY, "D-a", null, 15L), new TestRecord<>(ANY_UNIQUE_KEY, "D-b", null, 15L), new TestRecord<>(ANY_UNIQUE_KEY, "D-c", null, 15L), - new TestRecord<>(ANY_UNIQUE_KEY, "D-d", null, 15L)), - Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "E-a", null, 4L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-b", null, 6L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-c", null, 10L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-d", null, 14L)), - Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "A-e", null, 3L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-e", null, 4L), - new TestRecord<>(ANY_UNIQUE_KEY, "B-e", null, 5L), - new TestRecord<>(ANY_UNIQUE_KEY, "C-e", null, 9L), - new TestRecord<>(ANY_UNIQUE_KEY, "D-e", null, 15L)), - Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "A-f", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-f", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "B-f", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "C-f", null, 9L), - new TestRecord<>(ANY_UNIQUE_KEY, "D-f", null, 15L)), - Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "F-e", null, 8L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-a", null, 8L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-b", null, 8L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-f", null, 8L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-c", null, 10L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-d", null, 14L)) + new TestRecord<>(ANY_UNIQUE_KEY, "D-d", null, 15L)) ); leftStream.leftJoin( @@ -310,7 +238,7 @@ public void testLeft() { JoinWindows.ofTimeDifferenceAndGrace(ofSeconds(10), ofHours(24)) ).to(OUTPUT_TOPIC); - runTestWithDriver(expectedResult); + runTestWithDriver(inputWithoutOutOfOrderData, expectedResult); } @Test @@ -346,31 +274,7 @@ public void testLeftRepartitioned() { new TestRecord<>(ANY_UNIQUE_KEY, "D-a", null, 15L), new TestRecord<>(ANY_UNIQUE_KEY, "D-b", null, 15L), new TestRecord<>(ANY_UNIQUE_KEY, "D-c", null, 15L), - new TestRecord<>(ANY_UNIQUE_KEY, "D-d", null, 15L)), - Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "E-a", null, 4L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-b", null, 6L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-c", null, 10L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-d", null, 14L)), - Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "A-e", null, 3L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-e", null, 4L), - new TestRecord<>(ANY_UNIQUE_KEY, "B-e", null, 5L), - new TestRecord<>(ANY_UNIQUE_KEY, "C-e", null, 9L), - new TestRecord<>(ANY_UNIQUE_KEY, "D-e", null, 15L)), - Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "A-f", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-f", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "B-f", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "C-f", null, 9L), - new TestRecord<>(ANY_UNIQUE_KEY, "D-f", null, 15L)), - Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "F-e", null, 8L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-a", null, 8L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-b", null, 8L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-f", null, 8L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-c", null, 10L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-d", null, 14L)) + new TestRecord<>(ANY_UNIQUE_KEY, "D-d", null, 15L)) ); leftStream.map(MockMapper.noOpKeyValueMapper()) @@ -381,7 +285,7 @@ public void testLeftRepartitioned() { JoinWindows.ofTimeDifferenceAndGrace(ofSeconds(10), ofHours(24)) ).to(OUTPUT_TOPIC); - runTestWithDriver(expectedResult); + runTestWithDriver(inputWithoutOutOfOrderData, expectedResult); } @Test @@ -417,31 +321,7 @@ public void testOuter() { new TestRecord<>(ANY_UNIQUE_KEY, "D-a", null, 15L), new TestRecord<>(ANY_UNIQUE_KEY, "D-b", null, 15L), new TestRecord<>(ANY_UNIQUE_KEY, "D-c", null, 15L), - new TestRecord<>(ANY_UNIQUE_KEY, "D-d", null, 15L)), - Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "E-a", null, 4L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-b", null, 6L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-c", null, 10L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-d", null, 14L)), - Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "A-e", null, 3L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-e", null, 4L), - new TestRecord<>(ANY_UNIQUE_KEY, "B-e", null, 5L), - new TestRecord<>(ANY_UNIQUE_KEY, "C-e", null, 9L), - new TestRecord<>(ANY_UNIQUE_KEY, "D-e", null, 15L)), - Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "A-f", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-f", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "B-f", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "C-f", null, 9L), - new TestRecord<>(ANY_UNIQUE_KEY, "D-f", null, 15L)), - Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "F-e", null, 8L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-a", null, 8L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-b", null, 8L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-f", null, 8L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-c", null, 10L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-d", null, 14L)) + new TestRecord<>(ANY_UNIQUE_KEY, "D-d", null, 15L)) ); leftStream.outerJoin( @@ -450,7 +330,7 @@ public void testOuter() { JoinWindows.ofTimeDifferenceAndGrace(ofSeconds(10), ofHours(24)) ).to(OUTPUT_TOPIC); - runTestWithDriver(expectedResult); + runTestWithDriver(inputWithoutOutOfOrderData, expectedResult); } @Test @@ -486,31 +366,7 @@ public void testOuterRepartitioned() { new TestRecord<>(ANY_UNIQUE_KEY, "D-a", null, 15L), new TestRecord<>(ANY_UNIQUE_KEY, "D-b", null, 15L), new TestRecord<>(ANY_UNIQUE_KEY, "D-c", null, 15L), - new TestRecord<>(ANY_UNIQUE_KEY, "D-d", null, 15L)), - Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "E-a", null, 4L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-b", null, 6L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-c", null, 10L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-d", null, 14L)), - Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "A-e", null, 3L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-e", null, 4L), - new TestRecord<>(ANY_UNIQUE_KEY, "B-e", null, 5L), - new TestRecord<>(ANY_UNIQUE_KEY, "C-e", null, 9L), - new TestRecord<>(ANY_UNIQUE_KEY, "D-e", null, 15L)), - Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "A-f", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-f", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "B-f", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "C-f", null, 9L), - new TestRecord<>(ANY_UNIQUE_KEY, "D-f", null, 15L)), - Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "F-e", null, 8L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-a", null, 8L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-b", null, 8L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-f", null, 8L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-c", null, 10L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-d", null, 14L)) + new TestRecord<>(ANY_UNIQUE_KEY, "D-d", null, 15L)) ); leftStream.map(MockMapper.noOpKeyValueMapper()) @@ -521,7 +377,7 @@ public void testOuterRepartitioned() { JoinWindows.ofTimeDifferenceAndGrace(ofSeconds(10), ofHours(24)) ).to(OUTPUT_TOPIC); - runTestWithDriver(expectedResult); + runTestWithDriver(inputWithoutOutOfOrderData, expectedResult); } @Test @@ -548,7 +404,7 @@ public void testMultiInner() { new TestRecord<>(ANY_UNIQUE_KEY, "C-a-b", null, 9L), new TestRecord<>(ANY_UNIQUE_KEY, "C-b-a", null, 9L), new TestRecord<>(ANY_UNIQUE_KEY, "C-b-b", null, 9L)), - Arrays.>asList( + Arrays.asList( new TestRecord<>(ANY_UNIQUE_KEY, "A-c-a", null, 10L), new TestRecord<>(ANY_UNIQUE_KEY, "A-c-b", null, 10L), new TestRecord<>(ANY_UNIQUE_KEY, "B-c-a", null, 10L), @@ -567,7 +423,7 @@ public void testMultiInner() { null, null, null, - Arrays.>asList( + Arrays.asList( new TestRecord<>(ANY_UNIQUE_KEY, "A-d-a", null, 14L), new TestRecord<>(ANY_UNIQUE_KEY, "A-d-b", null, 14L), new TestRecord<>(ANY_UNIQUE_KEY, "A-d-c", null, 14L), @@ -589,7 +445,7 @@ public void testMultiInner() { new TestRecord<>(ANY_UNIQUE_KEY, "A-d-d", null, 14L), new TestRecord<>(ANY_UNIQUE_KEY, "B-d-d", null, 14L), new TestRecord<>(ANY_UNIQUE_KEY, "C-d-d", null, 14L)), - Arrays.>asList( + Arrays.asList( new TestRecord<>(ANY_UNIQUE_KEY, "D-a-a", null, 15L), new TestRecord<>(ANY_UNIQUE_KEY, "D-a-b", null, 15L), new TestRecord<>(ANY_UNIQUE_KEY, "D-a-c", null, 15L), @@ -605,163 +461,7 @@ public void testMultiInner() { new TestRecord<>(ANY_UNIQUE_KEY, "D-d-a", null, 15L), new TestRecord<>(ANY_UNIQUE_KEY, "D-d-b", null, 15L), new TestRecord<>(ANY_UNIQUE_KEY, "D-d-c", null, 15L), - new TestRecord<>(ANY_UNIQUE_KEY, "D-d-d", null, 15L)), - Arrays.>asList( - new TestRecord<>(ANY_UNIQUE_KEY, "E-a-a", null, 4L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-a-b", null, 6L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-a-c", null, 10L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-a-d", null, 14L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-b-a", null, 6L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-b-b", null, 6L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-b-c", null, 10L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-b-d", null, 14L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-c-a", null, 10L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-c-b", null, 10L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-c-c", null, 10L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-c-d", null, 14L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-d-a", null, 14L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-d-b", null, 14L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-d-c", null, 14L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-d-d", null, 14L)), - Arrays.>asList( - new TestRecord<>(ANY_UNIQUE_KEY, "A-e-a", null, 4L), - new TestRecord<>(ANY_UNIQUE_KEY, "A-e-b", null, 6L), - new TestRecord<>(ANY_UNIQUE_KEY, "A-e-c", null, 10L), - new TestRecord<>(ANY_UNIQUE_KEY, "A-e-d", null, 14L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-e-a", null, 4L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-e-b", null, 6L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-e-c", null, 10L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-e-d", null, 14L), - new TestRecord<>(ANY_UNIQUE_KEY, "B-e-a", null, 5L), - new TestRecord<>(ANY_UNIQUE_KEY, "B-e-b", null, 6L), - new TestRecord<>(ANY_UNIQUE_KEY, "B-e-c", null, 10L), - new TestRecord<>(ANY_UNIQUE_KEY, "B-e-d", null, 14L), - new TestRecord<>(ANY_UNIQUE_KEY, "C-e-a", null, 9L), - new TestRecord<>(ANY_UNIQUE_KEY, "C-e-b", null, 9L), - new TestRecord<>(ANY_UNIQUE_KEY, "C-e-c", null, 10L), - new TestRecord<>(ANY_UNIQUE_KEY, "C-e-d", null, 14L), - new TestRecord<>(ANY_UNIQUE_KEY, "D-e-a", null, 15L), - new TestRecord<>(ANY_UNIQUE_KEY, "D-e-b", null, 15L), - new TestRecord<>(ANY_UNIQUE_KEY, "D-e-c", null, 15L), - new TestRecord<>(ANY_UNIQUE_KEY, "D-e-d", null, 15L), - new TestRecord<>(ANY_UNIQUE_KEY, "A-e-e", null, 3L), - new TestRecord<>(ANY_UNIQUE_KEY, "A-a-e", null, 4L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-a-e", null, 4L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-e-e", null, 4L), - new TestRecord<>(ANY_UNIQUE_KEY, "B-a-e", null, 5L), - new TestRecord<>(ANY_UNIQUE_KEY, "B-e-e", null, 5L), - new TestRecord<>(ANY_UNIQUE_KEY, "A-b-e", null, 6L), - new TestRecord<>(ANY_UNIQUE_KEY, "B-b-e", null, 6L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-b-e", null, 6L), - new TestRecord<>(ANY_UNIQUE_KEY, "C-a-e", null, 9L), - new TestRecord<>(ANY_UNIQUE_KEY, "C-b-e", null, 9L), - new TestRecord<>(ANY_UNIQUE_KEY, "C-e-e", null, 9L), - new TestRecord<>(ANY_UNIQUE_KEY, "A-c-e", null, 10L), - new TestRecord<>(ANY_UNIQUE_KEY, "B-c-e", null, 10L), - new TestRecord<>(ANY_UNIQUE_KEY, "C-c-e", null, 10L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-c-e", null, 10L), - new TestRecord<>(ANY_UNIQUE_KEY, "A-d-e", null, 14L), - new TestRecord<>(ANY_UNIQUE_KEY, "B-d-e", null, 14L), - new TestRecord<>(ANY_UNIQUE_KEY, "C-d-e", null, 14L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-d-e", null, 14L), - new TestRecord<>(ANY_UNIQUE_KEY, "D-a-e", null, 15L), - new TestRecord<>(ANY_UNIQUE_KEY, "D-b-e", null, 15L), - new TestRecord<>(ANY_UNIQUE_KEY, "D-c-e", null, 15L), - new TestRecord<>(ANY_UNIQUE_KEY, "D-d-e", null, 15L), - new TestRecord<>(ANY_UNIQUE_KEY, "D-e-e", null, 15L)), - Arrays.>asList( - new TestRecord<>(ANY_UNIQUE_KEY, "A-f-e", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "A-f-a", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "A-f-b", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "A-f-c", null, 10L), - new TestRecord<>(ANY_UNIQUE_KEY, "A-f-d", null, 14L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-f-e", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-f-a", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-f-b", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-f-c", null, 10L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-f-d", null, 14L), - new TestRecord<>(ANY_UNIQUE_KEY, "B-f-e", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "B-f-a", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "B-f-b", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "B-f-c", null, 10L), - new TestRecord<>(ANY_UNIQUE_KEY, "B-f-d", null, 14L), - new TestRecord<>(ANY_UNIQUE_KEY, "C-f-e", null, 9L), - new TestRecord<>(ANY_UNIQUE_KEY, "C-f-a", null, 9L), - new TestRecord<>(ANY_UNIQUE_KEY, "C-f-b", null, 9L), - new TestRecord<>(ANY_UNIQUE_KEY, "C-f-c", null, 10L), - new TestRecord<>(ANY_UNIQUE_KEY, "C-f-d", null, 14L), - new TestRecord<>(ANY_UNIQUE_KEY, "D-f-e", null, 15L), - new TestRecord<>(ANY_UNIQUE_KEY, "D-f-a", null, 15L), - new TestRecord<>(ANY_UNIQUE_KEY, "D-f-b", null, 15L), - new TestRecord<>(ANY_UNIQUE_KEY, "D-f-c", null, 15L), - new TestRecord<>(ANY_UNIQUE_KEY, "D-f-d", null, 15L), - new TestRecord<>(ANY_UNIQUE_KEY, "A-e-f", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "A-a-f", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-a-f", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-e-f", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "B-a-f", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "B-e-f", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "A-b-f", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "B-b-f", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-b-f", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "A-f-f", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-f-f", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "B-f-f", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "C-a-f", null, 9L), - new TestRecord<>(ANY_UNIQUE_KEY, "C-b-f", null, 9L), - new TestRecord<>(ANY_UNIQUE_KEY, "C-e-f", null, 9L), - new TestRecord<>(ANY_UNIQUE_KEY, "C-f-f", null, 9L), - new TestRecord<>(ANY_UNIQUE_KEY, "A-c-f", null, 10L), - new TestRecord<>(ANY_UNIQUE_KEY, "B-c-f", null, 10L), - new TestRecord<>(ANY_UNIQUE_KEY, "C-c-f", null, 10L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-c-f", null, 10L), - new TestRecord<>(ANY_UNIQUE_KEY, "A-d-f", null, 14L), - new TestRecord<>(ANY_UNIQUE_KEY, "B-d-f", null, 14L), - new TestRecord<>(ANY_UNIQUE_KEY, "C-d-f", null, 14L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-d-f", null, 14L), - new TestRecord<>(ANY_UNIQUE_KEY, "D-a-f", null, 15L), - new TestRecord<>(ANY_UNIQUE_KEY, "D-b-f", null, 15L), - new TestRecord<>(ANY_UNIQUE_KEY, "D-c-f", null, 15L), - new TestRecord<>(ANY_UNIQUE_KEY, "D-d-f", null, 15L), - new TestRecord<>(ANY_UNIQUE_KEY, "D-e-f", null, 15L), - new TestRecord<>(ANY_UNIQUE_KEY, "D-f-f", null, 15L)), - Arrays.>asList( - new TestRecord<>(ANY_UNIQUE_KEY, "F-e-e", null, 8L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-e-a", null, 8L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-e-b", null, 8L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-e-f", null, 8L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-e-c", null, 10L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-e-d", null, 14L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-a-e", null, 8L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-a-a", null, 8L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-a-b", null, 8L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-a-f", null, 8L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-a-c", null, 10L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-a-d", null, 14L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-b-e", null, 8L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-b-a", null, 8L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-b-b", null, 8L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-b-f", null, 8L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-b-c", null, 10L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-b-d", null, 14L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-f-e", null, 8L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-f-a", null, 8L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-f-b", null, 8L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-f-f", null, 8L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-f-c", null, 10L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-f-d", null, 14L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-c-e", null, 10L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-c-a", null, 10L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-c-b", null, 10L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-c-f", null, 10L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-c-c", null, 10L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-c-d", null, 14L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-d-e", null, 14L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-d-a", null, 14L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-d-b", null, 14L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-d-f", null, 14L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-d-c", null, 14L), - new TestRecord<>(ANY_UNIQUE_KEY, "F-d-d", null, 14L)) + new TestRecord<>(ANY_UNIQUE_KEY, "D-d-d", null, 15L)) ); leftStream.join( @@ -774,6 +474,6 @@ public void testMultiInner() { JoinWindows.ofTimeDifferenceAndGrace(ofSeconds(10), ofHours(24)) ).to(OUTPUT_TOPIC); - runTestWithDriver(expectedResult); + runTestWithDriver(inputWithoutOutOfOrderData, expectedResult); } } diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/StreamTableJoinIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/StreamTableJoinIntegrationTest.java index bd95f57b15a54..e4e46ec69a2ea 100644 --- a/streams/src/test/java/org/apache/kafka/streams/integration/StreamTableJoinIntegrationTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/integration/StreamTableJoinIntegrationTest.java @@ -87,14 +87,18 @@ public void testInner() { null, null, null, - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-d", null, 15L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-d", null, 4L)), + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-d", null, 6L)), null, null, - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "F-f", null, 8L)) + null, + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-e", null, 15L)), + null, + null, + null, + null ); - runTestWithDriver(expectedResult); + runTestWithDriver(input, expectedResult); } @Test @@ -120,14 +124,18 @@ public void testLeft() { null, null, null, - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-d", null, 15L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-d", null, 4L)), + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-d", null, 6L)), + null, + null, + null, + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-e", null, 15L)), null, null, - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "F-f", null, 8L)) + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "F-null", null, 4L)), + null ); - runTestWithDriver(expectedResult); + runTestWithDriver(input, expectedResult); } @Test @@ -154,14 +162,18 @@ public void testInnerWithVersionedStore() { null, null, null, - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-d", null, 15L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-a", null, 4L)), + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-b", null, 6L)), + null, + null, + null, + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-e", null, 15L)), null, null, + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "F-a", null, 4L)), null ); - runTestWithDriver(expectedResult); + runTestWithDriver(input, expectedResult); } @Test @@ -188,13 +200,17 @@ public void testLeftWithVersionedStore() { null, null, null, - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-d", null, 15L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-a", null, 4L)), + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-b", null, 6L)), null, null, - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "F-null", null, 8L)) + null, + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-e", null, 15L)), + null, + null, + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "F-a", null, 4L)), + null ); - runTestWithDriver(expectedResult); + runTestWithDriver(input, expectedResult); } } diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/TableTableJoinIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/TableTableJoinIntegrationTest.java index 261ac165cb52c..31bcc9e0e09da 100644 --- a/streams/src/test/java/org/apache/kafka/streams/integration/TableTableJoinIntegrationTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/integration/TableTableJoinIntegrationTest.java @@ -63,11 +63,10 @@ public void prepareTopology() throws InterruptedException { builder = new StreamsBuilder(); } - private final TestRecord expectedFinalJoinResultUnversioned = new TestRecord<>(ANY_UNIQUE_KEY, "F-f", null, 8L); - private final TestRecord expectedFinalJoinResultVersioned = new TestRecord<>(ANY_UNIQUE_KEY, "D-d", null, 15L); - private final TestRecord expectedFinalJoinResultLeftVersionedOnly = new TestRecord<>(ANY_UNIQUE_KEY, "D-f", null, 15L); - private final TestRecord expectedFinalJoinResultRightVersionedOnly = new TestRecord<>(ANY_UNIQUE_KEY, "F-d", null, 14L); - private final TestRecord expectedFinalMultiJoinResult = new TestRecord<>(ANY_UNIQUE_KEY, "F-f-f", null, 8L); + private final TestRecord expectedFinalJoinResultUnversioned = new TestRecord<>(ANY_UNIQUE_KEY, "F-f", null, 4L); + private final TestRecord expectedFinalJoinResultLeftVersionedOnly = new TestRecord<>(ANY_UNIQUE_KEY, "E-f", null, 15L); + private final TestRecord expectedFinalJoinResultRightVersionedOnly = new TestRecord<>(ANY_UNIQUE_KEY, "F-e", null, 14L); + private final TestRecord expectedFinalMultiJoinResult = new TestRecord<>(ANY_UNIQUE_KEY, "F-f-f", null, 4L); private final String storeName = appID + "-store"; private final Materialized> materialized = Materialized.>as(storeName) @@ -85,7 +84,7 @@ public void testInner() throws Exception { leftTable.join(rightTable, valueJoiner, materialized).toStream().to(OUTPUT_TOPIC); if (cacheEnabled) { - runTestWithDriver(expectedFinalJoinResultUnversioned, storeName); + runTestWithDriver(input, expectedFinalJoinResultUnversioned, storeName); } else { final List>> expectedResult = Arrays.asList( null, @@ -102,14 +101,18 @@ public void testInner() throws Exception { null, null, null, - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-d", null, 15L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-d", null, 14L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-e", null, 4L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-f", null, 7L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "F-f", null, 8L)) + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-d", null, 7L)), + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, null, null, 7L)), + null, + null, + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-e", null, 15L)), + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, null, null, 14L)), + null, + null, + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "F-f", null, 4L)) ); - runTestWithDriver(expectedResult, storeName); + runTestWithDriver(input, expectedResult, storeName); } } @@ -122,7 +125,7 @@ public void testLeft() throws Exception { leftTable.leftJoin(rightTable, valueJoiner, materialized).toStream().to(OUTPUT_TOPIC); if (cacheEnabled) { - runTestWithDriver(expectedFinalJoinResultUnversioned, storeName); + runTestWithDriver(input, expectedFinalJoinResultUnversioned, storeName); } else { final List>> expectedResult = Arrays.asList( null, @@ -139,14 +142,18 @@ public void testLeft() throws Exception { Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, null, null, 12L)), null, null, - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-d", null, 15L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-d", null, 14L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-e", null, 4L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-f", null, 7L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "F-f", null, 8L)) + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-d", null, 7L)), + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, null, null, 7L)), + null, + null, + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-e", null, 15L)), + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, null, null, 14L)), + null, + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "F-null", null, 4L)), + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "F-f", null, 4L)) ); - runTestWithDriver(expectedResult, storeName); + runTestWithDriver(input, expectedResult, storeName); } } @@ -159,7 +166,7 @@ public void testOuter() throws Exception { leftTable.outerJoin(rightTable, valueJoiner, materialized).toStream().to(OUTPUT_TOPIC); if (cacheEnabled) { - runTestWithDriver(expectedFinalJoinResultUnversioned, storeName); + runTestWithDriver(input, expectedFinalJoinResultUnversioned, storeName); } else { final List>> expectedResult = Arrays.asList( null, @@ -175,15 +182,19 @@ public void testOuter() throws Exception { Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "C-null", null, 11L)), Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, null, null, 12L)), null, - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "null-d", null, 14L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-d", null, 15L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-d", null, 14L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-e", null, 4L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-f", null, 7L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "F-f", null, 8L)) + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "null-d", null, 7L)), + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-d", null, 7L)), + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "null-d", null, 7L)), + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, null, null, 3L)), + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "null-e", null, 14L)), + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-e", null, 15L)), + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "null-e", null, 14L)), + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, null, null, 9L)), + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "F-null", null, 4L)), + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "F-f", null, 4L)) ); - runTestWithDriver(expectedResult, storeName); + runTestWithDriver(input, expectedResult, storeName); } } @@ -215,14 +226,18 @@ public void testInnerWithVersionedStores() { null, null, null, - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-d", null, 15L)), + null, + null, + null, + null, + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-e", null, 15L)), null, null, null, null ); - runTestWithDriver(expectedResult, storeName); + runTestWithDriver(input, expectedResult, storeName); } @Test @@ -253,14 +268,18 @@ public void testLeftWithVersionedStores() { Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, null, null, 12L)), null, null, - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-d", null, 15L)), + null, + null, + null, + null, + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-e", null, 15L)), null, null, null, null ); - runTestWithDriver(expectedResult, storeName); + runTestWithDriver(input, expectedResult, storeName); } @Test @@ -290,15 +309,19 @@ public void testOuterWithVersionedStores() { Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "C-null", null, 11L)), Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, null, null, 12L)), null, - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "null-d", null, 14L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-d", null, 15L)), + null, + null, + null, + null, + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "null-e", null, 14L)), + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-e", null, 15L)), null, null, null, null ); - runTestWithDriver(expectedResult, storeName); + runTestWithDriver(input, expectedResult, storeName); } @Test @@ -312,7 +335,7 @@ public void testInnerWithLeftVersionedOnly() throws Exception { leftTable.join(rightTable, valueJoiner, materialized).toStream().to(OUTPUT_TOPIC); if (cacheEnabled) { - runTestWithDriver(expectedFinalJoinResultLeftVersionedOnly, storeName); + runTestWithDriver(input, expectedFinalJoinResultLeftVersionedOnly, storeName); } else { final List>> expectedResult = Arrays.asList( null, @@ -329,14 +352,18 @@ public void testInnerWithLeftVersionedOnly() throws Exception { null, null, null, - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-d", null, 15L)), null, - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-e", null, 15L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-f", null, 15L)), - null + null, + null, + null, + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-e", null, 15L)), + null, + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, null, null, 15L)), + null, + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-f", null, 15L)) ); - runTestWithDriver(expectedResult, storeName); + runTestWithDriver(input, expectedResult, storeName); } } @@ -351,7 +378,7 @@ public void testLeftWithLeftVersionedOnly() throws Exception { leftTable.leftJoin(rightTable, valueJoiner, materialized).toStream().to(OUTPUT_TOPIC); if (cacheEnabled) { - runTestWithDriver(expectedFinalJoinResultLeftVersionedOnly, storeName); + runTestWithDriver(input, expectedFinalJoinResultLeftVersionedOnly, storeName); } else { final List>> expectedResult = Arrays.asList( null, @@ -368,14 +395,18 @@ public void testLeftWithLeftVersionedOnly() throws Exception { Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, null, null, 12L)), null, null, - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-d", null, 15L)), null, - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-e", null, 15L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-f", null, 15L)), - null + null, + null, + null, + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-e", null, 15L)), + null, + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-null", null, 15L)), + null, + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-f", null, 15L)) ); - runTestWithDriver(expectedResult, storeName); + runTestWithDriver(input, expectedResult, storeName); } } @@ -390,31 +421,35 @@ public void testOuterWithLeftVersionedOnly() throws Exception { leftTable.outerJoin(rightTable, valueJoiner, materialized).toStream().to(OUTPUT_TOPIC); if (cacheEnabled) { - runTestWithDriver(expectedFinalJoinResultLeftVersionedOnly, storeName); + runTestWithDriver(input, expectedFinalJoinResultLeftVersionedOnly, storeName); } else { final List>> expectedResult = Arrays.asList( null, null, - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "A-null", null, 3L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "A-a", null, 4L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "B-a", null, 5L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "B-b", null, 6L)), + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "A-null", null, 3L)), + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "A-a", null, 4L)), + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "B-a", null, 5L)), + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "B-b", null, 6L)), Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "null-b", null, 7L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, null, null, 8L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "C-null", null, 9L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "C-c", null, 10L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "C-null", null, 11L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, null, null, 12L)), + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, null, null, 8L)), + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "C-null", null, 9L)), + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "C-c", null, 10L)), + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "C-null", null, 11L)), + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, null, null, 12L)), null, - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "null-d", null, 14L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-d", null, 15L)), + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "null-d", null, 7L)), null, - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-e", null, 15L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-f", null, 15L)), - null + null, + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, null, null, 3L)), + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "null-e", null, 14L)), + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-e", null, 15L)), + null, + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-null", null, 15L)), + null, + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-f", null, 15L)) ); - runTestWithDriver(expectedResult, storeName); + runTestWithDriver(input, expectedResult, storeName); } } @@ -429,7 +464,7 @@ public void testInnerWithRightVersionedOnly() throws Exception { leftTable.join(rightTable, valueJoiner, materialized).toStream().to(OUTPUT_TOPIC); if (cacheEnabled) { - runTestWithDriver(expectedFinalJoinResultRightVersionedOnly, storeName); + runTestWithDriver(input, expectedFinalJoinResultRightVersionedOnly, storeName); } else { final List>> expectedResult = Arrays.asList( null, @@ -446,14 +481,18 @@ public void testInnerWithRightVersionedOnly() throws Exception { null, null, null, - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-d", null, 15L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-d", null, 14L)), null, null, - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "F-d", null, 14L)) + null, + null, + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-e", null, 15L)), + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, null, null, 14L)), + null, + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "F-e", null, 14L)), + null ); - runTestWithDriver(expectedResult, storeName); + runTestWithDriver(input, expectedResult, storeName); } } @@ -468,7 +507,7 @@ public void testLeftWithRightVersionedOnly() throws Exception { leftTable.leftJoin(rightTable, valueJoiner, materialized).toStream().to(OUTPUT_TOPIC); if (cacheEnabled) { - runTestWithDriver(expectedFinalJoinResultRightVersionedOnly, storeName); + runTestWithDriver(input, expectedFinalJoinResultRightVersionedOnly, storeName); } else { final List>> expectedResult = Arrays.asList( null, @@ -485,14 +524,18 @@ public void testLeftWithRightVersionedOnly() throws Exception { Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, null, null, 12L)), null, null, - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-d", null, 15L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-d", null, 14L)), + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-null", null, 6L)), + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, null, null, 2L)), + null, null, + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-e", null, 15L)), + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, null, null, 14L)), null, - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "F-d", null, 14L)) + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "F-e", null, 14L)), + null ); - runTestWithDriver(expectedResult, storeName); + runTestWithDriver(input, expectedResult, storeName); } } @@ -507,7 +550,7 @@ public void testOuterWithRightVersionedOnly() throws Exception { leftTable.outerJoin(rightTable, valueJoiner, materialized).toStream().to(OUTPUT_TOPIC); if (cacheEnabled) { - runTestWithDriver(expectedFinalJoinResultRightVersionedOnly, storeName); + runTestWithDriver(input, expectedFinalJoinResultRightVersionedOnly, storeName); } else { final List>> expectedResult = Arrays.asList( null, @@ -523,15 +566,19 @@ public void testOuterWithRightVersionedOnly() throws Exception { Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "C-null", null, 11L)), Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, null, null, 12L)), null, - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "null-d", null, 14L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-d", null, 15L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-d", null, 14L)), null, + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-null", null, 6L)), + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, null, null, 2L)), + null, + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "null-e", null, 14L)), + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-e", null, 15L)), + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "null-e", null, 14L)), null, - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "F-d", null, 14L)) + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "F-e", null, 14L)), + null ); - runTestWithDriver(expectedResult, storeName); + runTestWithDriver(input, expectedResult, storeName); } } @@ -547,7 +594,7 @@ public void testInnerInner() throws Exception { .to(OUTPUT_TOPIC); if (cacheEnabled) { - runTestWithDriver(expectedFinalMultiJoinResult, storeName); + runTestWithDriver(input, expectedFinalMultiJoinResult, storeName); } else { // TODO K6443: the duplicate below for all the multi-joins are due to // KAFKA-6443, should be updated once it is fixed. @@ -573,18 +620,10 @@ public void testInnerInner() throws Exception { null, null, null, - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-d-d", null, 15L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-d-d", null, 14L)), - Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "E-e-e", null, 4L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-e-e", null, 4L)), - Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "E-f-f", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-f-f", null, 7L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "F-f-f", null, 8L)) + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-d-d", null, 15L)) ); - runTestWithDriver(expectedResult, storeName); + runTestWithDriver(inputWithoutOutOfOrderData, expectedResult, storeName); } } @@ -600,7 +639,7 @@ public void testInnerLeft() throws Exception { .to(OUTPUT_TOPIC); if (cacheEnabled) { - runTestWithDriver(expectedFinalMultiJoinResult, storeName); + runTestWithDriver(input, expectedFinalMultiJoinResult, storeName); } else { final List>> expectedResult = Arrays.asList( null, @@ -623,18 +662,10 @@ public void testInnerLeft() throws Exception { null, null, null, - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-d-d", null, 15L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-d-d", null, 14L)), - Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "E-e-e", null, 4L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-e-e", null, 4L)), - Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "E-f-f", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-f-f", null, 7L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "F-f-f", null, 8L)) + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-d-d", null, 15L)) ); - runTestWithDriver(expectedResult, storeName); + runTestWithDriver(inputWithoutOutOfOrderData, expectedResult, storeName); } } @@ -650,7 +681,7 @@ public void testInnerOuter() throws Exception { .to(OUTPUT_TOPIC); if (cacheEnabled) { - runTestWithDriver(expectedFinalMultiJoinResult, storeName); + runTestWithDriver(input, expectedFinalMultiJoinResult, storeName); } else { final List>> expectedResult = Arrays.asList( null, @@ -674,22 +705,12 @@ public void testInnerOuter() throws Exception { new TestRecord<>(ANY_UNIQUE_KEY, null, null, 11L)), null, null, - null, - Arrays.asList( - // incorrect result `null-d` is caused by self-join of `rightTable` - new TestRecord<>(ANY_UNIQUE_KEY, "null-d", null, 14L), - new TestRecord<>(ANY_UNIQUE_KEY, "D-d-d", null, 15L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-d-d", null, 14L)), - Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "E-e-e", null, 4L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-e-e", null, 4L)), - Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "E-f-f", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-f-f", null, 7L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "F-f-f", null, 8L)) + // incorrect result `null-d` is caused by self-join of `rightTable` + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "null-d", null, 14L)), + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-d-d", null, 15L)) ); - runTestWithDriver(expectedResult, storeName); + runTestWithDriver(inputWithoutOutOfOrderData, expectedResult, storeName); } } @@ -705,7 +726,7 @@ public void testLeftInner() throws Exception { .to(OUTPUT_TOPIC); if (cacheEnabled) { - runTestWithDriver(expectedFinalMultiJoinResult, storeName); + runTestWithDriver(input, expectedFinalMultiJoinResult, storeName); } else { final List>> expectedResult = Arrays.asList( null, @@ -728,18 +749,10 @@ public void testLeftInner() throws Exception { null, null, null, - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-d-d", null, 15L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-d-d", null, 14L)), - Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "E-e-e", null, 4L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-e-e", null, 4L)), - Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "E-f-f", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-f-f", null, 7L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "F-f-f", null, 8L)) + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-d-d", null, 15L)) ); - runTestWithDriver(expectedResult, storeName); + runTestWithDriver(inputWithoutOutOfOrderData, expectedResult, storeName); } } @@ -755,14 +768,13 @@ public void testLeftLeft() throws Exception { .to(OUTPUT_TOPIC); if (cacheEnabled) { - runTestWithDriver(expectedFinalMultiJoinResult, storeName); + runTestWithDriver(input, expectedFinalMultiJoinResult, storeName); } else { final List>> expectedResult = Arrays.asList( null, null, - null, + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "A-null-null", null, 3L)), Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "A-null-null", null, 3L), new TestRecord<>(ANY_UNIQUE_KEY, "A-a-a", null, 4L), new TestRecord<>(ANY_UNIQUE_KEY, "A-a-a", null, 4L)), Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "B-a-a", null, 5L)), @@ -771,9 +783,8 @@ public void testLeftLeft() throws Exception { new TestRecord<>(ANY_UNIQUE_KEY, "B-b-b", null, 6L)), Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, null, null, 7L)), null, - null, + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "C-null-null", null, 9L)), Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "C-null-null", null, 9L), new TestRecord<>(ANY_UNIQUE_KEY, "C-c-c", null, 10L), new TestRecord<>(ANY_UNIQUE_KEY, "C-c-c", null, 10L)), Arrays.asList( @@ -782,18 +793,10 @@ public void testLeftLeft() throws Exception { Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, null, null, 12L)), null, null, - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-d-d", null, 15L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-d-d", null, 14L)), - Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "E-e-e", null, 4L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-e-e", null, 4L)), - Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "E-f-f", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-f-f", null, 7L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "F-f-f", null, 8L)) + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-d-d", null, 15L)) ); - runTestWithDriver(expectedResult, storeName); + runTestWithDriver(inputWithoutOutOfOrderData, expectedResult, storeName); } } @@ -809,14 +812,13 @@ public void testLeftOuter() throws Exception { .to(OUTPUT_TOPIC); if (cacheEnabled) { - runTestWithDriver(expectedFinalMultiJoinResult, storeName); + runTestWithDriver(input, expectedFinalMultiJoinResult, storeName); } else { final List>> expectedResult = Arrays.asList( null, null, - null, + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "A-null-null", null, 3L)), Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "A-null-null", null, 3L), new TestRecord<>(ANY_UNIQUE_KEY, "A-a-a", null, 4L), new TestRecord<>(ANY_UNIQUE_KEY, "A-a-a", null, 4L)), Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "B-a-a", null, 5L)), @@ -825,9 +827,8 @@ public void testLeftOuter() throws Exception { new TestRecord<>(ANY_UNIQUE_KEY, "B-b-b", null, 6L)), Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "null-b", null, 7L)), Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, null, null, 8L)), - null, + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "C-null-null", null, 9L)), Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "C-null-null", null, 9L), new TestRecord<>(ANY_UNIQUE_KEY, "C-c-c", null, 10L), new TestRecord<>(ANY_UNIQUE_KEY, "C-c-c", null, 10L)), Arrays.asList( @@ -835,21 +836,11 @@ public void testLeftOuter() throws Exception { new TestRecord<>(ANY_UNIQUE_KEY, "C-null-null", null, 11L)), Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, null, null, 12L)), null, - null, - Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "null-d", null, 14L), - new TestRecord<>(ANY_UNIQUE_KEY, "D-d-d", null, 15L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-d-d", null, 14L)), - Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "E-e-e", null, 4L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-e-e", null, 4L)), - Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "E-f-f", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-f-f", null, 7L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "F-f-f", null, 8L)) + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "null-d", null, 14L)), + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-d-d", null, 15L)) ); - runTestWithDriver(expectedResult, storeName); + runTestWithDriver(inputWithoutOutOfOrderData, expectedResult, storeName); } } @@ -865,7 +856,7 @@ public void testOuterInner() throws Exception { .to(OUTPUT_TOPIC); if (cacheEnabled) { - runTestWithDriver(expectedFinalMultiJoinResult, storeName); + runTestWithDriver(input, expectedFinalMultiJoinResult, storeName); } else { final List>> expectedResult = Arrays.asList( null, @@ -890,18 +881,10 @@ public void testOuterInner() throws Exception { Arrays.asList( new TestRecord<>(ANY_UNIQUE_KEY, "null-d-d", null, 14L), new TestRecord<>(ANY_UNIQUE_KEY, "null-d-d", null, 14L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-d-d", null, 15L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-d-d", null, 14L)), - Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "E-e-e", null, 4L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-e-e", null, 4L)), - Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "E-f-f", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-f-f", null, 7L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "F-f-f", null, 8L)) + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-d-d", null, 15L)) ); - runTestWithDriver(expectedResult, storeName); + runTestWithDriver(inputWithoutOutOfOrderData, expectedResult, storeName); } } @@ -917,14 +900,13 @@ public void testOuterLeft() throws Exception { .to(OUTPUT_TOPIC); if (cacheEnabled) { - runTestWithDriver(expectedFinalMultiJoinResult, storeName); + runTestWithDriver(input, expectedFinalMultiJoinResult, storeName); } else { final List>> expectedResult = Arrays.asList( null, null, - null, + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "A-null-null", null, 3L)), Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "A-null-null", null, 3L), new TestRecord<>(ANY_UNIQUE_KEY, "A-a-a", null, 4L), new TestRecord<>(ANY_UNIQUE_KEY, "A-a-a", null, 4L)), Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "B-a-a", null, 5L)), @@ -933,9 +915,8 @@ public void testOuterLeft() throws Exception { new TestRecord<>(ANY_UNIQUE_KEY, "B-b-b", null, 6L)), Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "null-b-b", null, 7L)), Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, null, null, 8L)), - null, + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "C-null-null", null, 9L)), Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "C-null-null", null, 9L), new TestRecord<>(ANY_UNIQUE_KEY, "C-c-c", null, 10L), new TestRecord<>(ANY_UNIQUE_KEY, "C-c-c", null, 10L)), Arrays.asList( @@ -946,18 +927,10 @@ public void testOuterLeft() throws Exception { Arrays.asList( new TestRecord<>(ANY_UNIQUE_KEY, "null-d-d", null, 14L), new TestRecord<>(ANY_UNIQUE_KEY, "null-d-d", null, 14L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-d-d", null, 15L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-d-d", null, 14L)), - Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "E-e-e", null, 4L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-e-e", null, 4L)), - Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "E-f-f", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-f-f", null, 7L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "F-f-f", null, 8L)) + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-d-d", null, 15L)) ); - runTestWithDriver(expectedResult, storeName); + runTestWithDriver(inputWithoutOutOfOrderData, expectedResult, storeName); } } @@ -973,14 +946,13 @@ public void testOuterOuter() throws Exception { .to(OUTPUT_TOPIC); if (cacheEnabled) { - runTestWithDriver(expectedFinalMultiJoinResult, storeName); + runTestWithDriver(input, expectedFinalMultiJoinResult, storeName); } else { final List>> expectedResult = Arrays.asList( null, null, - null, + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "A-null-null", null, 3L)), Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "A-null-null", null, 3L), new TestRecord<>(ANY_UNIQUE_KEY, "A-a-a", null, 4L), new TestRecord<>(ANY_UNIQUE_KEY, "A-a-a", null, 4L)), Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "B-a-a", null, 5L)), @@ -1000,21 +972,12 @@ public void testOuterOuter() throws Exception { new TestRecord<>(ANY_UNIQUE_KEY, "C-null-null", null, 11L)), Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, null, null, 12L)), null, - null, Arrays.asList( new TestRecord<>(ANY_UNIQUE_KEY, "null-d-d", null, 14L), - new TestRecord<>(ANY_UNIQUE_KEY, "null-d-d", null, 14L), - new TestRecord<>(ANY_UNIQUE_KEY, "D-d-d", null, 15L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "E-d-d", null, 14L)), - Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "E-e-e", null, 4L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-e-e", null, 4L)), - Arrays.asList( - new TestRecord<>(ANY_UNIQUE_KEY, "E-f-f", null, 7L), - new TestRecord<>(ANY_UNIQUE_KEY, "E-f-f", null, 7L)), - Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "F-f-f", null, 8L)) + new TestRecord<>(ANY_UNIQUE_KEY, "null-d-d", null, 14L)), + Collections.singletonList(new TestRecord<>(ANY_UNIQUE_KEY, "D-d-d", null, 15L)) ); - runTestWithDriver(expectedResult, storeName); + runTestWithDriver(inputWithoutOutOfOrderData, expectedResult, storeName); } } } 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 7449f8d3cdef5..1106e914cf362 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 @@ -33,7 +33,12 @@ import org.apache.kafka.streams.kstream.KTable; import org.apache.kafka.streams.kstream.KeyValueMapper; import org.apache.kafka.streams.kstream.Materialized; +import org.apache.kafka.streams.kstream.internals.foreignkeyjoin.ForeignTableJoinProcessorSupplier; +import org.apache.kafka.streams.kstream.internals.foreignkeyjoin.SubscriptionSendProcessorSupplier; +import org.apache.kafka.streams.kstream.internals.graph.ForeignTableJoinNode; +import org.apache.kafka.streams.kstream.internals.graph.ForeignJoinSubscriptionSendNode; import org.apache.kafka.streams.kstream.internals.graph.GraphNode; +import org.apache.kafka.streams.kstream.internals.graph.KTableKTableJoinNode; import org.apache.kafka.streams.kstream.internals.graph.StreamStreamJoinNode; import org.apache.kafka.streams.kstream.internals.graph.TableFilterNode; import org.apache.kafka.streams.kstream.internals.graph.TableRepartitionMapNode; @@ -1009,6 +1014,176 @@ public void shouldSetUseVersionedSemanticsOnTableRepartitionMapWithIntermediateN verifyVersionedSemantics((TableRepartitionMapNode) repartitionMap, true); } + @Test + public void shouldSetUseVersionedSemanticsOnTableJoin() { + // Given: + final MaterializedInternal> versionedMaterialize = + new MaterializedInternal<>(Materialized.as(Stores.persistentVersionedKeyValueStore("versioned", Duration.ofMinutes(5))), builder, storePrefix); + final MaterializedInternal> versionedMaterialize2 = + new MaterializedInternal<>(Materialized.as(Stores.persistentVersionedKeyValueStore("versioned2", Duration.ofMinutes(5))), builder, storePrefix); + final KTable table1 = builder.table("t1", consumed, versionedMaterialize); + final KTable table2 = builder.table("t2", consumed, versionedMaterialize2); + table1.join(table2, (v1, v2) -> v1 + v2); + + // When: + builder.buildAndOptimizeTopology(); + + // Then: + final GraphNode join = getNodeByType(builder.root, KTableKTableJoinNode.class, new HashSet<>()); + assertNotNull(join); + verifyVersionedSemantics((KTableKTableJoinNode) join, true, true); + } + + @Test + public void shouldSetUseVersionedSemanticsOnTableJoinLeftOnly() { + // Given: + final MaterializedInternal> versionedMaterialize = + new MaterializedInternal<>(Materialized.as(Stores.persistentVersionedKeyValueStore("versioned", Duration.ofMinutes(5))), builder, storePrefix); + final MaterializedInternal> unversionedMaterialize = + new MaterializedInternal<>(Materialized.as("unversioned"), builder, storePrefix); + final KTable table1 = builder.table("t1", consumed, versionedMaterialize); + final KTable table2 = builder.table("t2", consumed, unversionedMaterialize); + table1.join(table2, (v1, v2) -> v1 + v2); + + // When: + builder.buildAndOptimizeTopology(); + + // Then: + final GraphNode join = getNodeByType(builder.root, KTableKTableJoinNode.class, new HashSet<>()); + assertNotNull(join); + verifyVersionedSemantics((KTableKTableJoinNode) join, true, false); + } + + @Test + public void shouldSetUseVersionedSemanticsOnTableJoinRightOnly() { + // Given: + final MaterializedInternal> versionedMaterialize = + new MaterializedInternal<>(Materialized.as(Stores.persistentVersionedKeyValueStore("versioned", Duration.ofMinutes(5))), builder, storePrefix); + final MaterializedInternal> unversionedMaterialize = + new MaterializedInternal<>(Materialized.as("unversioned"), builder, storePrefix); + final KTable table1 = builder.table("t1", consumed, unversionedMaterialize); + final KTable table2 = builder.table("t2", consumed, versionedMaterialize); + table1.join(table2, (v1, v2) -> v1 + v2); + + // When: + builder.buildAndOptimizeTopology(); + + // Then: + final GraphNode join = getNodeByType(builder.root, KTableKTableJoinNode.class, new HashSet<>()); + assertNotNull(join); + verifyVersionedSemantics((KTableKTableJoinNode) join, false, true); + } + + @Test + public void shouldSetUseVersionedSemanticsOnTableSelfJoin() { + // Given: + final MaterializedInternal> versionedMaterialize = + new MaterializedInternal<>(Materialized.as(Stores.persistentVersionedKeyValueStore("versioned", Duration.ofMinutes(5))), builder, storePrefix); + final KTable table1 = builder.table("t1", consumed, versionedMaterialize); + table1.join(table1, (v1, v2) -> v1 + v2); + + // When: + builder.buildAndOptimizeTopology(); + + // Then: + final GraphNode join = getNodeByType(builder.root, KTableKTableJoinNode.class, new HashSet<>()); + assertNotNull(join); + verifyVersionedSemantics((KTableKTableJoinNode) join, true, true); + } + + @Test + public void shouldSetUseVersionedSemanticsOnTableForeignJoin() { + // Given: + final MaterializedInternal> versionedMaterialize = + new MaterializedInternal<>(Materialized.as(Stores.persistentVersionedKeyValueStore("versioned", Duration.ofMinutes(5))), builder, storePrefix); + final MaterializedInternal> versionedMaterialize2 = + new MaterializedInternal<>(Materialized.as(Stores.persistentVersionedKeyValueStore("versioned2", Duration.ofMinutes(5))), builder, storePrefix); + final KTable table1 = builder.table("t1", consumed, versionedMaterialize); + final KTable table2 = builder.table("t2", consumed, versionedMaterialize2); + table1.join(table2, v -> v, (v1, v2) -> v1 + v2); + + // When: + builder.buildAndOptimizeTopology(); + + // Then: + final GraphNode joinThis = getNodeByType(builder.root, ForeignJoinSubscriptionSendNode.class, new HashSet<>()); + assertNotNull(joinThis); + verifyVersionedSemantics((ForeignJoinSubscriptionSendNode) joinThis, true); + + final GraphNode joinOther = getNodeByType(builder.root, ForeignTableJoinNode.class, new HashSet<>()); + assertNotNull(joinOther); + verifyVersionedSemantics((ForeignTableJoinNode) joinOther, true); + } + + @Test + public void shouldSetUseVersionedSemanticsOnTableForeignJoinLeftOnly() { + // Given: + final MaterializedInternal> versionedMaterialize = + new MaterializedInternal<>(Materialized.as(Stores.persistentVersionedKeyValueStore("versioned", Duration.ofMinutes(5))), builder, storePrefix); + final MaterializedInternal> unversionedMaterialize = + new MaterializedInternal<>(Materialized.as("unversioned"), builder, storePrefix); + final KTable table1 = builder.table("t1", consumed, versionedMaterialize); + final KTable table2 = builder.table("t2", consumed, unversionedMaterialize); + table1.join(table2, v -> v, (v1, v2) -> v1 + v2); + + // When: + builder.buildAndOptimizeTopology(); + + // Then: + final GraphNode joinThis = getNodeByType(builder.root, ForeignJoinSubscriptionSendNode.class, new HashSet<>()); + assertNotNull(joinThis); + verifyVersionedSemantics((ForeignJoinSubscriptionSendNode) joinThis, true); + + final GraphNode joinOther = getNodeByType(builder.root, ForeignTableJoinNode.class, new HashSet<>()); + assertNotNull(joinOther); + verifyVersionedSemantics((ForeignTableJoinNode) joinOther, false); + } + + @Test + public void shouldSetUseVersionedSemanticsOnTableForeignJoinRightOnly() { + // Given: + final MaterializedInternal> versionedMaterialize = + new MaterializedInternal<>(Materialized.as(Stores.persistentVersionedKeyValueStore("versioned", Duration.ofMinutes(5))), builder, storePrefix); + final MaterializedInternal> unversionedMaterialize = + new MaterializedInternal<>(Materialized.as("unversioned"), builder, storePrefix); + final KTable table1 = builder.table("t1", consumed, unversionedMaterialize); + final KTable table2 = builder.table("t2", consumed, versionedMaterialize); + table1.join(table2, v -> v, (v1, v2) -> v1 + v2); + + // When: + builder.buildAndOptimizeTopology(); + + // Then: + final GraphNode joinThis = getNodeByType(builder.root, ForeignJoinSubscriptionSendNode.class, new HashSet<>()); + assertNotNull(joinThis); + verifyVersionedSemantics((ForeignJoinSubscriptionSendNode) joinThis, false); + + final GraphNode joinOther = getNodeByType(builder.root, ForeignTableJoinNode.class, new HashSet<>()); + assertNotNull(joinOther); + verifyVersionedSemantics((ForeignTableJoinNode) joinOther, true); + } + + @Test + public void shouldSetUseVersionedSemanticsOnTableForeignSelfJoin() { + // Given: + final MaterializedInternal> versionedMaterialize = + new MaterializedInternal<>(Materialized.as(Stores.persistentVersionedKeyValueStore("versioned", Duration.ofMinutes(5))), builder, storePrefix); + final KTable table1 = builder.table("t1", consumed, versionedMaterialize); + table1.join(table1, v -> v, (v1, v2) -> v1 + v2); + + // When: + builder.buildAndOptimizeTopology(); + + // Then: + final GraphNode joinThis = getNodeByType(builder.root, ForeignJoinSubscriptionSendNode.class, new HashSet<>()); + assertNotNull(joinThis); + verifyVersionedSemantics((ForeignJoinSubscriptionSendNode) joinThis, true); + + final GraphNode joinOther = getNodeByType(builder.root, ForeignTableJoinNode.class, new HashSet<>()); + assertNotNull(joinOther); + verifyVersionedSemantics((ForeignTableJoinNode) joinOther, true); + } + private void verifyVersionedSemantics(final TableFilterNode filterNode, final boolean expectedValue) { final ProcessorSupplier processorSupplier = filterNode.processorParameters().processorSupplier(); assertTrue(processorSupplier instanceof KTableFilter); @@ -1023,6 +1198,32 @@ private void verifyVersionedSemantics(final TableRepartitionMapNode repart assertEquals(expectedValue, repartitionMap.isUseVersionedSemantics()); } + private void verifyVersionedSemantics(final KTableKTableJoinNode joinNode, final boolean expectedValueLeft, final boolean expectedValueRight) { + final ProcessorSupplier thisProcessorSupplier = joinNode.thisProcessorParameters().processorSupplier(); + assertTrue(thisProcessorSupplier instanceof KTableKTableAbstractJoin); + final KTableKTableAbstractJoin thisJoin = (KTableKTableAbstractJoin) thisProcessorSupplier; + assertEquals(expectedValueLeft, thisJoin.isUseVersionedSemantics()); + + final ProcessorSupplier otherProcessorSupplier = joinNode.otherProcessorParameters().processorSupplier(); + assertTrue(otherProcessorSupplier instanceof KTableKTableAbstractJoin); + final KTableKTableAbstractJoin otherJoin = (KTableKTableAbstractJoin) otherProcessorSupplier; + assertEquals(expectedValueRight, otherJoin.isUseVersionedSemantics()); + } + + private void verifyVersionedSemantics(final ForeignJoinSubscriptionSendNode joinThisNode, final boolean expectedValue) { + final ProcessorSupplier thisProcessorSupplier = joinThisNode.processorParameters().processorSupplier(); + assertTrue(thisProcessorSupplier instanceof SubscriptionSendProcessorSupplier); + final SubscriptionSendProcessorSupplier joinThis = (SubscriptionSendProcessorSupplier) thisProcessorSupplier; + assertEquals(expectedValue, joinThis.isUseVersionedSemantics()); + } + + private void verifyVersionedSemantics(final ForeignTableJoinNode joinOtherNode, final boolean expectedValue) { + final ProcessorSupplier otherProcessorSupplier = joinOtherNode.processorParameters().processorSupplier(); + assertTrue(otherProcessorSupplier instanceof ForeignTableJoinProcessorSupplier); + final ForeignTableJoinProcessorSupplier joinThis = (ForeignTableJoinProcessorSupplier) otherProcessorSupplier; + assertEquals(expectedValue, joinThis.isUseVersionedSemantics()); + } + private GraphNode getNodeByType( final GraphNode currentNode, final Class clazz, diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamImplTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamImplTest.java index 54bb14aa0ecd8..81a2f1daf5bfb 100644 --- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamImplTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamImplTest.java @@ -3149,7 +3149,7 @@ public void shouldSupportForeignKeyTableTableJoinWithKTableFromKStream() { " Processor: KTABLE-FK-JOIN-OUTPUT-0000000018 (stores: [])\n" + " --> KTABLE-TOSTREAM-0000000020\n" + " <-- KTABLE-FK-JOIN-SUBSCRIPTION-RESPONSE-RESOLVER-PROCESSOR-0000000017\n" + - " Processor: KTABLE-FK-JOIN-SUBSCRIPTION-REGISTRATION-0000000007 (stores: [KSTREAM-TOTABLE-STATE-STORE-0000000002])\n" + + " Processor: KTABLE-FK-JOIN-SUBSCRIPTION-REGISTRATION-0000000007 (stores: [])\n" + " --> KTABLE-SINK-0000000008\n" + " <-- KSTREAM-TOTABLE-0000000001\n" + " Processor: KTABLE-TOSTREAM-0000000020 (stores: [])\n" + @@ -3174,7 +3174,7 @@ public void shouldSupportForeignKeyTableTableJoinWithKTableFromKStream() { " Processor: KTABLE-FK-JOIN-SUBSCRIPTION-PROCESSOR-0000000012 (stores: [KSTREAM-TOTABLE-STATE-STORE-0000000005])\n" + " --> KTABLE-SINK-0000000015\n" + " <-- KTABLE-FK-JOIN-SUBSCRIPTION-PROCESSOR-0000000011\n" + - " Processor: KTABLE-FK-JOIN-SUBSCRIPTION-PROCESSOR-0000000013 (stores: [KSTREAM-TOTABLE-STATE-STORE-0000000005, KTABLE-FK-JOIN-SUBSCRIPTION-STATE-STORE-0000000010])\n" + + " Processor: KTABLE-FK-JOIN-SUBSCRIPTION-PROCESSOR-0000000013 (stores: [KTABLE-FK-JOIN-SUBSCRIPTION-STATE-STORE-0000000010])\n" + " --> KTABLE-SINK-0000000015\n" + " <-- KSTREAM-TOTABLE-0000000004\n" + " Sink: KTABLE-SINK-0000000015 (topic: KTABLE-FK-JOIN-SUBSCRIPTION-RESPONSE-0000000014-topic)\n" + @@ -3254,10 +3254,10 @@ public void shouldSupportTableTableJoinWithKStreamToKTable() { " Processor: KSTREAM-TOTABLE-0000000004 (stores: [KSTREAM-TOTABLE-STATE-STORE-0000000005])\n" + " --> KTABLE-JOINOTHER-0000000008\n" + " <-- KSTREAM-SOURCE-0000000003\n" + - " Processor: KTABLE-JOINOTHER-0000000008 (stores: [KSTREAM-TOTABLE-STATE-STORE-0000000005, KSTREAM-TOTABLE-STATE-STORE-0000000002])\n" + + " Processor: KTABLE-JOINOTHER-0000000008 (stores: [KSTREAM-TOTABLE-STATE-STORE-0000000002])\n" + " --> KTABLE-MERGE-0000000006\n" + " <-- KSTREAM-TOTABLE-0000000004\n" + - " Processor: KTABLE-JOINTHIS-0000000007 (stores: [KSTREAM-TOTABLE-STATE-STORE-0000000005, KSTREAM-TOTABLE-STATE-STORE-0000000002])\n" + + " Processor: KTABLE-JOINTHIS-0000000007 (stores: [KSTREAM-TOTABLE-STATE-STORE-0000000005])\n" + " --> KTABLE-MERGE-0000000006\n" + " <-- KSTREAM-TOTABLE-0000000001\n" + " Processor: KTABLE-MERGE-0000000006 (stores: [])\n" +