Skip to content

KAFKA-14834: [5/N] Drop out-of-order records from FK join with versioned tables #13522

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 3 commits into from
Apr 13, 2023
Merged

KAFKA-14834: [5/N] Drop out-of-order records from FK join with versioned tables #13522

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
24dbd3a
update fk join to drop out-of-order records when versioned
vcrfxia Mar 30, 2023
8decdda
refactor to address cyclomatic complexity
vcrfxia Apr 6, 2023
5126d1c
unit test fix
vcrfxia Apr 12, 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
6 changes: 3 additions & 3 deletions checkstyle/suppressions.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -230,19 +230,19 @@
files=".*[/\\]streams[/\\].*test[/\\].*.java"/>

<suppress checks="CyclomaticComplexity"
files="(EosV2UpgradeIntegrationTest|KStreamKStreamJoinTest|KTableKTableForeignKeyJoinIntegrationTest|RocksDBGenericOptionsToDbOptionsColumnFamilyOptionsAdapterTest|RelationalSmokeTest|MockProcessorContextStateStoreTest).java"/>
files="(EosV2UpgradeIntegrationTest|KStreamKStreamJoinTest|KTableKTableForeignKeyJoinIntegrationTest|KTableKTableForeignKeyVersionedJoinIntegrationTest|RocksDBGenericOptionsToDbOptionsColumnFamilyOptionsAdapterTest|RelationalSmokeTest|MockProcessorContextStateStoreTest).java"/>

<suppress checks="JavaNCSS"
files="(EosV2UpgradeIntegrationTest|KStreamKStreamJoinTest|StreamThreadTest|TaskManagerTest).java"/>

<suppress checks="NPathComplexity"
files="(EosV2UpgradeIntegrationTest|EosTestDriver|KStreamKStreamJoinTest|KTableKTableForeignKeyJoinIntegrationTest|RelationalSmokeTest|MockProcessorContextStateStoreTest|TopologyTestDriverTest).java"/>
files="(EosV2UpgradeIntegrationTest|EosTestDriver|KStreamKStreamJoinTest|KTableKTableForeignKeyJoinIntegrationTest|KTableKTableForeignKeyVersionedJoinIntegrationTest|RelationalSmokeTest|MockProcessorContextStateStoreTest|TopologyTestDriverTest).java"/>

<suppress checks="(FinalLocalVariable|WhitespaceAround|LocalVariableName|ImportControl|AvoidStarImport)"
files="Murmur3Test.java"/>

<suppress checks="MethodLength"
files="(KStreamSlidingWindowAggregateTest|KStreamKStreamLeftJoinTest|KStreamKStreamOuterJoinTest).java"/>
files="(KStreamSlidingWindowAggregateTest|KStreamKStreamLeftJoinTest|KStreamKStreamOuterJoinTest|KTableKTableForeignKeyVersionedJoinIntegrationTest).java"/>

<suppress checks="ClassFanOutComplexity"
files="StreamTaskTest.java"/>
Expand Down
22 changes: 13 additions & 9 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 @@ -1116,18 +1116,22 @@ private <VR, KO, VO> KTable<K, VR> doJoinOnForeignKey(final KTable<KO, VO> forei
keySerde
);

final ProcessorGraphNode<K, Change<V>> subscriptionNode = new ProcessorGraphNode<>(
final KTableValueGetterSupplier<K, V> primaryKeyValueGetter = valueGetterSupplier();
final StatefulProcessorNode<K, Change<V>> subscriptionNode = new StatefulProcessorNode<>(
new ProcessorParameters<>(
new ForeignJoinSubscriptionSendProcessorSupplier<>(
foreignKeyExtractor,
subscriptionForeignKeySerdePseudoTopic,
valueHashSerdePseudoTopic,
foreignKeySerde,
valueSerde == null ? null : valueSerde.serializer(),
leftJoin
leftJoin,
primaryKeyValueGetter
),
renamed.suffixWithOrElseGet("-subscription-registration-processor", builder, SUBSCRIPTION_REGISTRATION)
)
),
Collections.emptySet(),
Collections.singleton(primaryKeyValueGetter)
);
builder.addGraphNode(graphNode, subscriptionNode);

Expand Down Expand Up @@ -1179,26 +1183,27 @@ private <VR, KO, VO> KTable<K, VR> doJoinOnForeignKey(final KTable<KO, VO> forei
);
builder.addGraphNode(subscriptionSource, subscriptionReceiveNode);

final KTableValueGetterSupplier<KO, VO> foreignKeyValueGetter = ((KTableImpl<KO, VO, VO>) foreignKeyTable).valueGetterSupplier();
final StatefulProcessorNode<CombinedKey<KO, K>, Change<ValueAndTimestamp<SubscriptionWrapper<K>>>> subscriptionJoinForeignNode =
new StatefulProcessorNode<>(
new ProcessorParameters<>(
new SubscriptionJoinForeignProcessorSupplier<>(
((KTableImpl<KO, VO, VO>) foreignKeyTable).valueGetterSupplier()
foreignKeyValueGetter
),
renamed.suffixWithOrElseGet("-subscription-join-foreign", builder, SUBSCRIPTION_PROCESSOR)
),
Collections.emptySet(),
Collections.singleton(((KTableImpl<KO, VO, VO>) foreignKeyTable).valueGetterSupplier())
Collections.singleton(foreignKeyValueGetter)
);
builder.addGraphNode(subscriptionReceiveNode, subscriptionJoinForeignNode);

final StatefulProcessorNode<KO, Change<Object>> foreignJoinSubscriptionNode = new StatefulProcessorNode<>(
final StatefulProcessorNode<KO, Change<VO>> foreignJoinSubscriptionNode = new StatefulProcessorNode<>(
new ProcessorParameters<>(
new ForeignJoinSubscriptionProcessorSupplier<>(subscriptionStore, combinedKeySchema),
new ForeignJoinSubscriptionProcessorSupplier<>(subscriptionStore, combinedKeySchema, foreignKeyValueGetter),
renamed.suffixWithOrElseGet("-foreign-join-subscription", builder, SUBSCRIPTION_PROCESSOR)
),
Collections.singleton(subscriptionStore),
Collections.emptySet()
Collections.singleton(foreignKeyValueGetter)
);
builder.addGraphNode(((KTableImpl<KO, VO, ?>) foreignKeyTable).graphNode, foreignJoinSubscriptionNode);

Expand Down Expand Up @@ -1232,7 +1237,6 @@ 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 SubscriptionResolverJoinProcessorSupplier<K, V, VO, VR> resolverProcessorSupplier = new SubscriptionResolverJoinProcessorSupplier<>(
primaryKeyValueGetter,
valueSerde == null ? null : valueSerde.serializer(),
Expand Down
36 changes: 31 additions & 5 deletions ...ka/streams/kstream/internals/foreignkeyjoin/ForeignJoinSubscriptionProcessorSupplier.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,8 @@
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;
Expand All @@ -43,24 +45,32 @@ public class ForeignJoinSubscriptionProcessorSupplier<K, KO, VO> implements
private static final Logger LOG = LoggerFactory.getLogger(ForeignJoinSubscriptionProcessorSupplier.class);
private final StoreBuilder<TimestampedKeyValueStore<Bytes, SubscriptionWrapper<K>>> storeBuilder;
private final CombinedKeySchema<KO, K> keySchema;
private final KTableValueGetterSupplier<KO, VO> foreignKeyValueGetterSupplier;

public ForeignJoinSubscriptionProcessorSupplier(
final StoreBuilder<TimestampedKeyValueStore<Bytes, SubscriptionWrapper<K>>> storeBuilder,
final CombinedKeySchema<KO, K> keySchema) {
final CombinedKeySchema<KO, K> keySchema,
final KTableValueGetterSupplier<KO, VO> foreignKeyValueGetterSupplier) {

this.storeBuilder = storeBuilder;
this.keySchema = keySchema;
this.foreignKeyValueGetterSupplier = foreignKeyValueGetterSupplier;
}

@Override
public Processor<KO, Change<VO>, K, SubscriptionResponseWrapper<VO>> get() {
return new KTableKTableJoinProcessor();
return new KTableKTableJoinProcessor(foreignKeyValueGetterSupplier.get());
}


private final class KTableKTableJoinProcessor extends ContextualProcessor<KO, Change<VO>, K, SubscriptionResponseWrapper<VO>> {
private Sensor droppedRecordsSensor;
private TimestampedKeyValueStore<Bytes, SubscriptionWrapper<K>> store;
private TimestampedKeyValueStore<Bytes, SubscriptionWrapper<K>> subscriptionStore;
private final KTableValueGetter<KO, VO> foreignKeyValueGetter;

private KTableKTableJoinProcessor(final KTableValueGetter<KO, VO> foreignKeyValueGetter) {
this.foreignKeyValueGetter = foreignKeyValueGetter;
}

@Override
public void init(final ProcessorContext<K, SubscriptionResponseWrapper<VO>> context) {
Expand All @@ -71,7 +81,8 @@ public void init(final ProcessorContext<K, SubscriptionResponseWrapper<VO>> cont
internalProcessorContext.taskId().toString(),
internalProcessorContext.metrics()
);
store = internalProcessorContext.getStateStore(storeBuilder);
subscriptionStore = internalProcessorContext.getStateStore(storeBuilder);
foreignKeyValueGetter.init(context);
}

@Override
Expand All @@ -95,11 +106,21 @@ public void process(final Record<KO, Change<VO>> record) {
return;
}

// drop out-of-order records from versioned tables (cf. KIP-914)
if (foreignKeyValueGetter.isVersioned()) {
final ValueAndTimestamp<VO> 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;
}
}

final Bytes prefixBytes = keySchema.prefixBytes(record.key());

//Perform the prefixScan and propagate the results
try (final KeyValueIterator<Bytes, ValueAndTimestamp<SubscriptionWrapper<K>>> prefixScanResults =
store.range(prefixBytes, Bytes.increment(prefixBytes))) {
subscriptionStore.range(prefixBytes, Bytes.increment(prefixBytes))) {

while (prefixScanResults.hasNext()) {
final KeyValue<Bytes, ValueAndTimestamp<SubscriptionWrapper<K>>> next = prefixScanResults.next();
Expand All @@ -118,6 +139,11 @@ public void process(final Record<KO, Change<VO>> 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);
Expand Down
93 changes: 52 additions & 41 deletions ...treams/kstream/internals/foreignkeyjoin/ForeignJoinSubscriptionSendProcessorSupplier.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,8 @@
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;
Expand All @@ -29,6 +31,7 @@
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;
Expand All @@ -49,6 +52,7 @@ public class ForeignJoinSubscriptionSendProcessorSupplier<K, KO, V> implements P
private final Supplier<String> foreignKeySerdeTopicSupplier;
private final Supplier<String> valueSerdeTopicSupplier;
private final boolean leftJoin;
private final KTableValueGetterSupplier<K, V> primaryKeyValueGetterSupplier;
private Serializer<KO> foreignKeySerializer;
private Serializer<V> valueSerializer;

Expand All @@ -57,25 +61,32 @@ public ForeignJoinSubscriptionSendProcessorSupplier(final Function<V, KO> foreig
final Supplier<String> valueSerdeTopicSupplier,
final Serde<KO> foreignKeySerde,
final Serializer<V> valueSerializer,
final boolean leftJoin) {
final boolean leftJoin,
final KTableValueGetterSupplier<K, V> primaryKeyValueGetterSupplier) {
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<K, Change<V>, KO, SubscriptionWrapper<K>> get() {
return new UnbindChangeProcessor();
return new UnbindChangeProcessor(primaryKeyValueGetterSupplier.get());
}

private class UnbindChangeProcessor extends ContextualProcessor<K, Change<V>, KO, SubscriptionWrapper<K>> {

private Sensor droppedRecordsSensor;
private String foreignKeySerdeTopic;
private String valueSerdeTopic;
private final KTableValueGetter<K, V> primaryKeyValueGetter;

private UnbindChangeProcessor(final KTableValueGetter<K, V> primaryKeyValueGetter) {
this.primaryKeyValueGetter = primaryKeyValueGetter;
}

@SuppressWarnings("unchecked")
@Override
Expand All @@ -95,10 +106,25 @@ public void init(final ProcessorContext<KO, SubscriptionWrapper<K>> context) {
context.taskId().toString(),
(StreamsMetricsImpl) context.metrics()
);
primaryKeyValueGetter.init(context);
}

@Override
public void process(final Record<K, Change<V>> 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<V> 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;
}
}
}

final long[] currentHash = record.value().newValue == null ?
null :
Murmur3.hash128(valueSerializer.serialize(valueSerdeTopic, record.value().newValue));
Expand All @@ -107,37 +133,13 @@ public void process(final Record<K, Change<V>> record) {
if (record.value().oldValue != null) {
final KO oldForeignKey = foreignKeyExtractor.apply(record.value().oldValue);
if (oldForeignKey == null) {
if (context().recordMetadata().isPresent()) {
final RecordMetadata recordMetadata = context().recordMetadata().get();
LOG.warn(
"Skipping record due to null foreign key. "
+ "topic=[{}] partition=[{}] offset=[{}]",
recordMetadata.topic(), recordMetadata.partition(), recordMetadata.offset()
);
} else {
LOG.warn(
"Skipping record due to null foreign key. Topic, partition, and offset not known."
);
}
droppedRecordsSensor.record();
logSkippedRecordDueToNullForeignKey();
return;
}
if (record.value().newValue != null) {
final KO newForeignKey = foreignKeyExtractor.apply(record.value().newValue);
if (newForeignKey == null) {
if (context().recordMetadata().isPresent()) {
final RecordMetadata recordMetadata = context().recordMetadata().get();
LOG.warn(
"Skipping record due to null foreign key. "
+ "topic=[{}] partition=[{}] offset=[{}]",
recordMetadata.topic(), recordMetadata.partition(), recordMetadata.offset()
);
} else {
LOG.warn(
"Skipping record due to null foreign key. Topic, partition, and offset not known."
);
}
droppedRecordsSensor.record();
logSkippedRecordDueToNullForeignKey();
return;
}

Expand Down Expand Up @@ -193,19 +195,7 @@ public void process(final Record<K, Change<V>> record) {
}
final KO newForeignKey = foreignKeyExtractor.apply(record.value().newValue);
if (newForeignKey == null) {
if (context().recordMetadata().isPresent()) {
final RecordMetadata recordMetadata = context().recordMetadata().get();
LOG.warn(
"Skipping record due to null foreign key. "
+ "topic=[{}] partition=[{}] offset=[{}]",
recordMetadata.topic(), recordMetadata.partition(), recordMetadata.offset()
);
} else {
LOG.warn(
"Skipping record due to null foreign key. Topic, partition, and offset not known."
);
}
droppedRecordsSensor.record();
logSkippedRecordDueToNullForeignKey();
} else {
context().forward(
record.withKey(newForeignKey)
Expand All @@ -217,5 +207,26 @@ public void process(final Record<K, Change<V>> record) {
}
}
}

@Override
public void close() {
primaryKeyValueGetter.close();
}

private void logSkippedRecordDueToNullForeignKey() {

@vcrfxia vcrfxia Apr 6, 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.

Extracting the common code from process() into this helper method is purely a refactor, unrelated to the changes in this PR. I had to make the refactor in order to avoid a CyclomaticComplexity checkstyle failure upon adding a clause into process() for dropping out-of-order records.

if (context().recordMetadata().isPresent()) {
final RecordMetadata recordMetadata = context().recordMetadata().get();
LOG.warn(
"Skipping record due to null foreign key. "
+ "topic=[{}] partition=[{}] offset=[{}]",
recordMetadata.topic(), recordMetadata.partition(), recordMetadata.offset()
);
} else {
LOG.warn(
"Skipping record due to null foreign key. Topic, partition, and offset not known."
);
}
droppedRecordsSensor.record();
}
}
}
Loading