Skip to content

KAFKA-14834: [11/N] Update table joins to identify out-of-order records with isLatest #13609

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mjsax merged 4 commits into from
Apr 19, 2023
Merged

KAFKA-14834: [11/N] Update table joins to identify out-of-order records with isLatest #13609

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
8ac5565
update table joins to use isLatest instead of value getter
vcrfxia Apr 14, 2023
9314f2f
rename to be consistent with https://github.com/apache/kafka/pull/13589
vcrfxia Apr 18, 2023
18a6c01
review feedback
vcrfxia Apr 19, 2023
4d021d4
fix check to be stricter
vcrfxia Apr 19, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 8 additions & 2 deletions streams/src/main/java/org/apache/kafka/streams/kstream/internals/InternalStreamsBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -620,11 +620,17 @@ private <K, V> GroupedInternal<K, V> getRepartitionSerdes(final Collection<Optim
}

private void enableVersionedSemantics() {
versionedSemanticsNodes.forEach(node -> ((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)");
}
});
}
Expand Down
22 changes: 11 additions & 11 deletions streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -804,6 +806,7 @@ private <VO, VR> KTable<K, VR> doJoin(final KTable<K, VO> other,
kTableKTableJoinNode.setOutputVersioned(isOutputVersioned);

builder.addGraphNode(this.graphNode, kTableKTableJoinNode);
builder.addGraphNode(((KTableImpl<?, ?, ?>) other).graphNode, kTableKTableJoinNode);

@vcrfxia vcrfxia Apr 18, 2023

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't know why it's currently the case that primary-key table-table join nodes only have one parent, instead of two. Seems more correct to have two, and the GraphNode mechanism for determining whether the joining table is versioned or not will not work without this parent connection.

I have verified that there is no change to the built topology, so AFAICT this addition is internal-only.

@mjsax mjsax Apr 18, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah. Seems to be incorrect, but did apparently not surface as a bug. Nice fix!


// we can inherit parent key serde if user do not provide specific overrides
return new KTableImpl<K, Change<VR>, VR>(
Expand Down Expand Up @@ -1098,7 +1101,7 @@ private <VR, KO, VO> KTable<K, VR> doJoinOnForeignKey(final KTable<KO, VO> 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.

@vcrfxia vcrfxia Apr 18, 2023

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This, and a few other similar renames in comments, are unrelated to this PR but included as cleanup from #13589.

@mjsax mjsax Apr 18, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thank you! -- Could we do a small follow up for 3.4 branch to get it backported?

@vcrfxia vcrfxia Apr 19, 2023 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sure thing: #13610

//This occurs whenever the extracted foreignKey changes values.
enableSendingOldValues(true);

Expand Down Expand Up @@ -1138,22 +1141,18 @@ private <VR, KO, VO> KTable<K, VR> doJoinOnForeignKey(final KTable<KO, VO> forei
keySerde
);

final KTableValueGetterSupplier<K, V> primaryKeyValueGetter = valueGetterSupplier();
final StatefulProcessorNode<K, Change<V>> subscriptionSendNode = new StatefulProcessorNode<>(
final ProcessorGraphNode<K, Change<V>> subscriptionSendNode = new ForeignJoinSubscriptionSendNode<>(
new ProcessorParameters<>(
new SubscriptionSendProcessorSupplier<>(
foreignKeyExtractor,
subscriptionForeignKeySerdePseudoTopic,
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);

Expand Down Expand Up @@ -1219,13 +1218,13 @@ private <VR, KO, VO> KTable<K, VR> doJoinOnForeignKey(final KTable<KO, VO> forei
);
builder.addGraphNode(subscriptionReceiveNode, subscriptionJoinNode);

final StatefulProcessorNode<KO, Change<VO>> foreignTableJoinNode = new StatefulProcessorNode<>(
final StatefulProcessorNode<KO, Change<VO>> 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<KO, VO, ?>) foreignKeyTable).graphNode, foreignTableJoinNode);

Expand Down Expand Up @@ -1259,6 +1258,7 @@ private <VR, KO, VO> KTable<K, VR> doJoinOnForeignKey(final KTable<KO, VO> forei
resultSourceNodes.add(foreignResponseSource.nodeName());
builder.internalTopologyBuilder.copartitionSources(resultSourceNodes);

final KTableValueGetterSupplier<K, V> primaryKeyValueGetter = valueGetterSupplier();
final StatefulProcessorNode<K, SubscriptionResponseWrapper<VO>> responseJoinNode = new StatefulProcessorNode<>(
new ProcessorParameters<>(
new ResponseJoinProcessorSupplier<>(
Expand Down
20 changes: 15 additions & 5 deletions ...ms/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableAbstractJoin.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,17 +18,16 @@

import org.apache.kafka.streams.kstream.ValueJoiner;

abstract class KTableKTableAbstractJoin<K, V1, V2, VOut> implements
public abstract class KTableKTableAbstractJoin<K, V1, V2, VOut> implements
KTableProcessorSupplier<K, V1, K, VOut> {

private final KTableImpl<K, ?, V1> table1;
private final KTableImpl<K, ?, V2> table2;
final KTableValueGetterSupplier<K, V1> valueGetterSupplier1;
final KTableValueGetterSupplier<K, V2> valueGetterSupplier2;
final KTableValueGetter<K, V1> valueGetter1;
final KTableValueGetter<K, V2> valueGetter2;
final ValueJoiner<? super V1, ? super V2, ? extends VOut> joiner;

boolean useVersionedSemantics = false;
boolean sendOldValues = false;

KTableKTableAbstractJoin(final KTableImpl<K, ?, V1> table1,
Expand All @@ -38,8 +37,6 @@ abstract class KTableKTableAbstractJoin<K, V1, V2, VOut> implements
this.table2 = table2;
this.valueGetterSupplier1 = table1.valueGetterSupplier();
this.valueGetterSupplier2 = table2.valueGetterSupplier();
this.valueGetter1 = valueGetterSupplier1.get();
this.valueGetter2 = valueGetterSupplier2.get();
this.joiner = joiner;
}

Expand All @@ -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();
}
}
30 changes: 11 additions & 19 deletions streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableInnerJoin.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@ class KTableKTableInnerJoin<K, V1, V2, VOut> extends KTableKTableAbstractJoin<K,

@Override
public Processor<K, Change<V1>, K, Change<VOut>> get() {
return new KTableKTableJoinProcessor(valueGetter1, valueGetter2);
return new KTableKTableJoinProcessor(valueGetterSupplier2.get());
}

@Override
Expand All @@ -64,14 +64,11 @@ public KTableValueGetter<K, VOut> get() {

private class KTableKTableJoinProcessor extends ContextualProcessor<K, Change<V1>, K, Change<VOut>> {

private final KTableValueGetter<K, V1> thisValueGetter;
private final KTableValueGetter<K, V2> otherValueGetter;
private final KTableValueGetter<K, V2> valueGetter;
private Sensor droppedRecordsSensor;

KTableKTableJoinProcessor(final KTableValueGetter<K, V1> thisValueGetter,
final KTableValueGetter<K, V2> otherValueGetter) {
this.thisValueGetter = thisValueGetter;
this.otherValueGetter = otherValueGetter;
KTableKTableJoinProcessor(final KTableValueGetter<K, V2> valueGetter) {
this.valueGetter = valueGetter;
}

@Override
Expand All @@ -82,8 +79,7 @@ public void init(final ProcessorContext<K, Change<VOut>> context) {
context.taskId().toString(),
(StreamsMetricsImpl) context.metrics()
);
thisValueGetter.init(context);
otherValueGetter.init(context);
valueGetter.init(context);
}

@Override
Expand All @@ -107,20 +103,17 @@ public void process(final Record<K, Change<V1>> record) {
}

// drop out-of-order records from versioned tables (cf. KIP-914)
if (thisValueGetter.isVersioned()) {
final ValueAndTimestamp<V1> 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<V2> valueAndTimestampRight = otherValueGetter.get(record.key());
final ValueAndTimestamp<V2> valueAndTimestampRight = valueGetter.get(record.key());
final V2 valueRight = getValueOrNull(valueAndTimestampRight);
if (valueRight == null) {
return;
Expand All @@ -141,8 +134,7 @@ public void process(final Record<K, Change<V1>> record) {

@Override
public void close() {
thisValueGetter.close();
otherValueGetter.close();
valueGetter.close();
}
}

Expand Down
30 changes: 11 additions & 19 deletions streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableLeftJoin.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,7 @@ class KTableKTableLeftJoin<K, V1, V2, VOut> extends KTableKTableAbstractJoin<K,

@Override
public Processor<K, Change<V1>, K, Change<VOut>> get() {
return new KTableKTableLeftJoinProcessor(valueGetter1, valueGetter2);
return new KTableKTableLeftJoinProcessor(valueGetterSupplier2.get());
}

@Override
Expand All @@ -66,14 +66,11 @@ public KTableValueGetter<K, VOut> get() {

private class KTableKTableLeftJoinProcessor extends ContextualProcessor<K, Change<V1>, K, Change<VOut>> {

private final KTableValueGetter<K, V1> thisValueGetter;
private final KTableValueGetter<K, V2> otherValueGetter;
private final KTableValueGetter<K, V2> valueGetter;
private Sensor droppedRecordsSensor;

KTableKTableLeftJoinProcessor(final KTableValueGetter<K, V1> thisValueGetter,
final KTableValueGetter<K, V2> otherValueGetter) {
this.thisValueGetter = thisValueGetter;
this.otherValueGetter = otherValueGetter;
KTableKTableLeftJoinProcessor(final KTableValueGetter<K, V2> valueGetter) {
this.valueGetter = valueGetter;
}

@Override
Expand All @@ -84,8 +81,7 @@ public void init(final ProcessorContext<K, Change<VOut>> context) {
context.taskId().toString(),
(StreamsMetricsImpl) context.metrics()
);
thisValueGetter.init(context);
otherValueGetter.init(context);
valueGetter.init(context);
}

@Override
Expand All @@ -109,20 +105,17 @@ public void process(final Record<K, Change<V1>> record) {
}

// drop out-of-order records from versioned tables (cf. KIP-914)
if (thisValueGetter.isVersioned()) {
final ValueAndTimestamp<V1> 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<V2> valueAndTimestampRight = otherValueGetter.get(record.key());
final ValueAndTimestamp<V2> valueAndTimestampRight = valueGetter.get(record.key());
final V2 value2 = getValueOrNull(valueAndTimestampRight);
final long timestampRight;

Expand Down Expand Up @@ -150,8 +143,7 @@ public void process(final Record<K, Change<V1>> record) {

@Override
public void close() {
thisValueGetter.close();
otherValueGetter.close();
valueGetter.close();
}
}

Expand Down
30 changes: 11 additions & 19 deletions streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableOuterJoin.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,7 @@ class KTableKTableOuterJoin<K, V1, V2, VOut> extends KTableKTableAbstractJoin<K,

@Override
public Processor<K, Change<V1>, K, Change<VOut>> get() {
return new KTableKTableOuterJoinProcessor(valueGetter1, valueGetter2);
return new KTableKTableOuterJoinProcessor(valueGetterSupplier2.get());
}

@Override
Expand All @@ -65,14 +65,11 @@ public KTableValueGetter<K, VOut> get() {

private class KTableKTableOuterJoinProcessor extends ContextualProcessor<K, Change<V1>, K, Change<VOut>> {

private final KTableValueGetter<K, V2> otherValueGetter;
private final KTableValueGetter<K, V1> thisValueGetter;
private final KTableValueGetter<K, V2> valueGetter;
private Sensor droppedRecordsSensor;

KTableKTableOuterJoinProcessor(final KTableValueGetter<K, V1> thisValueGetter,
final KTableValueGetter<K, V2> otherValueGetter) {
this.thisValueGetter = thisValueGetter;
this.otherValueGetter = otherValueGetter;
KTableKTableOuterJoinProcessor(final KTableValueGetter<K, V2> valueGetter) {
this.valueGetter = valueGetter;
}

@Override
Expand All @@ -83,8 +80,7 @@ public void init(final ProcessorContext<K, Change<VOut>> context) {
context.taskId().toString(),
(StreamsMetricsImpl) context.metrics()
);
thisValueGetter.init(context);
otherValueGetter.init(context);
valueGetter.init(context);
}

@Override
Expand All @@ -108,20 +104,17 @@ public void process(final Record<K, Change<V1>> record) {
}

// drop out-of-order records from versioned tables (cf. KIP-914)
if (thisValueGetter.isVersioned()) {
final ValueAndTimestamp<V1> 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<V2> valueAndTimestamp2 = otherValueGetter.get(record.key());
final ValueAndTimestamp<V2> valueAndTimestamp2 = valueGetter.get(record.key());
final V2 value2 = getValueOrNull(valueAndTimestamp2);
if (value2 == null) {
if (record.value().newValue == null && record.value().oldValue == null) {
Expand All @@ -145,8 +138,7 @@ public void process(final Record<K, Change<V1>> record) {

@Override
public void close() {
thisValueGetter.close();
otherValueGetter.close();
valueGetter.close();
}
}

Expand Down
Loading