Skip to content

KAFKA-9925: decorate pseudo-topics with app id #8574

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
vvcephei merged 3 commits into from
Apr 29, 2020
Merged

KAFKA-9925: decorate pseudo-topics with app id #8574

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
8e8467d
KAFKA-9925: decorate pseudo-topics with app id
Apr 28, 2020
a863182
checkstyle
Apr 28, 2020
ea21699
checkstyle
Apr 28, 2020
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
24 changes: 19 additions & 5 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 @@ -77,6 +77,7 @@
import java.util.Objects;
import java.util.Set;
import java.util.function.Function;
import java.util.function.Supplier;

import static org.apache.kafka.streams.kstream.internals.graph.GraphGraceSearchUtil.findAndVerifyWindowGrace;

Expand Down Expand Up @@ -972,13 +973,26 @@ private <VR, KO, VO> KTable<K, VR> doJoinOnForeignKey(final KTable<KO, VO> forei
//This occurs whenever the extracted foreignKey changes values.
enableSendingOldValues();

final NamedInternal renamed = new NamedInternal(joinName);

final String subscriptionTopicName = renamed.suffixWithOrElseGet(
"-subscription-registration",
builder,
SUBSCRIPTION_REGISTRATION
) + TOPIC_SUFFIX;

// the decoration can't be performed until we have the configuration available when the app runs,
// so we pass Suppliers into the components, which they can call at run time
Comment on lines +984 to +985

@vvcephei vvcephei Apr 28, 2020

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.

Hopefully, this explains what's going on here.

@abbccdda abbccdda Apr 29, 2020

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

I understood


final Supplier<String> subscriptionPrimaryKeySerdePseudoTopic =
() -> internalTopologyBuilder().decoratePseudoTopic(subscriptionTopicName + "-pk");

final Supplier<String> subscriptionForeignKeySerdePseudoTopic =
() -> internalTopologyBuilder().decoratePseudoTopic(subscriptionTopicName + "-fk");

final Supplier<String> valueHashSerdePseudoTopic =
() -> internalTopologyBuilder().decoratePseudoTopic(subscriptionTopicName + "-vh");

final NamedInternal renamed = new NamedInternal(joinName);
final String subscriptionTopicName = renamed.suffixWithOrElseGet("-subscription-registration", builder, SUBSCRIPTION_REGISTRATION) + TOPIC_SUFFIX;
final String subscriptionPrimaryKeySerdePseudoTopic = subscriptionTopicName + "-pk";
final String subscriptionForeignKeySerdePseudoTopic = subscriptionTopicName + "-fk";
final String valueHashSerdePseudoTopic = subscriptionTopicName + "-vh";
builder.internalTopologyBuilder.addInternalTopic(subscriptionTopicName, InternalTopicProperties.empty());

final Serde<KO> foreignKeySerde = ((KTableImpl<KO, VO, ?>) foreignKeyTable).keySerde;
Expand Down
17 changes: 11 additions & 6 deletions ...ain/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/CombinedKeySchema.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,24 +23,27 @@
import org.apache.kafka.streams.processor.ProcessorContext;

import java.nio.ByteBuffer;
import java.util.function.Supplier;

/**
* Factory for creating CombinedKey serializers / deserializers.
*/
public class CombinedKeySchema<KO, K> {
private final String primaryKeySerdeTopic;
private final String foreignKeySerdeTopic;
private final Supplier<String> undecoratedPrimaryKeySerdeTopicSupplier;
private final Supplier<String> undecoratedForeignKeySerdeTopicSupplier;
private String primaryKeySerdeTopic;
private String foreignKeySerdeTopic;
private Serializer<K> primaryKeySerializer;
private Deserializer<K> primaryKeyDeserializer;
private Serializer<KO> foreignKeySerializer;
private Deserializer<KO> foreignKeyDeserializer;

public CombinedKeySchema(final String foreignKeySerdeTopic,
public CombinedKeySchema(final Supplier<String> foreignKeySerdeTopicSupplier,
final Serde<KO> foreignKeySerde,
final String primaryKeySerdeTopic,
final Supplier<String> primaryKeySerdeTopicSupplier,
final Serde<K> primaryKeySerde) {
this.primaryKeySerdeTopic = primaryKeySerdeTopic;
this.foreignKeySerdeTopic = foreignKeySerdeTopic;
undecoratedPrimaryKeySerdeTopicSupplier = primaryKeySerdeTopicSupplier;
undecoratedForeignKeySerdeTopicSupplier = foreignKeySerdeTopicSupplier;
primaryKeySerializer = primaryKeySerde == null ? null : primaryKeySerde.serializer();
primaryKeyDeserializer = primaryKeySerde == null ? null : primaryKeySerde.deserializer();
foreignKeyDeserializer = foreignKeySerde == null ? null : foreignKeySerde.deserializer();
Expand All @@ -49,6 +52,8 @@ public CombinedKeySchema(final String foreignKeySerdeTopic,

@SuppressWarnings("unchecked")
public void init(final ProcessorContext context) {
primaryKeySerdeTopic = undecoratedPrimaryKeySerdeTopicSupplier.get();
foreignKeySerdeTopic = undecoratedForeignKeySerdeTopicSupplier.get();
primaryKeySerializer = primaryKeySerializer == null ? (Serializer<K>) context.keySerde().serializer() : primaryKeySerializer;
primaryKeyDeserializer = primaryKeyDeserializer == null ? (Deserializer<K>) context.keySerde().deserializer() : primaryKeyDeserializer;
foreignKeySerializer = foreignKeySerializer == null ? (Serializer<KO>) context.keySerde().serializer() : foreignKeySerializer;
Expand Down
17 changes: 11 additions & 6 deletions ...treams/kstream/internals/foreignkeyjoin/ForeignJoinSubscriptionSendProcessorSupplier.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@

import java.util.Arrays;
import java.util.function.Function;
import java.util.function.Supplier;

import static org.apache.kafka.streams.kstream.internals.foreignkeyjoin.SubscriptionWrapper.Instruction.DELETE_KEY_AND_PROPAGATE;
import static org.apache.kafka.streams.kstream.internals.foreignkeyjoin.SubscriptionWrapper.Instruction.DELETE_KEY_NO_PROPAGATE;
Expand All @@ -43,21 +44,21 @@ public class ForeignJoinSubscriptionSendProcessorSupplier<K, KO, V> implements P
private static final Logger LOG = LoggerFactory.getLogger(ForeignJoinSubscriptionSendProcessorSupplier.class);

private final Function<V, KO> foreignKeyExtractor;
private final String foreignKeySerdeTopic;
private final String valueSerdeTopic;
private final Supplier<String> foreignKeySerdeTopicSupplier;
private final Supplier<String> valueSerdeTopicSupplier;
private final boolean leftJoin;
private Serializer<KO> foreignKeySerializer;
private Serializer<V> valueSerializer;

public ForeignJoinSubscriptionSendProcessorSupplier(final Function<V, KO> foreignKeyExtractor,
final String foreignKeySerdeTopic,
final String valueSerdeTopic,
final Supplier<String> foreignKeySerdeTopicSupplier,
final Supplier<String> valueSerdeTopicSupplier,
final Serde<KO> foreignKeySerde,
final Serializer<V> valueSerializer,
final boolean leftJoin) {
this.foreignKeyExtractor = foreignKeyExtractor;
this.foreignKeySerdeTopic = foreignKeySerdeTopic;
this.valueSerdeTopic = valueSerdeTopic;
this.foreignKeySerdeTopicSupplier = foreignKeySerdeTopicSupplier;
this.valueSerdeTopicSupplier = valueSerdeTopicSupplier;
this.valueSerializer = valueSerializer;
this.leftJoin = leftJoin;
foreignKeySerializer = foreignKeySerde == null ? null : foreignKeySerde.serializer();
Expand All @@ -71,11 +72,15 @@ public Processor<K, Change<V>> get() {
private class UnbindChangeProcessor extends AbstractProcessor<K, Change<V>> {

private Sensor droppedRecordsSensor;
private String foreignKeySerdeTopic;
private String valueSerdeTopic;

@SuppressWarnings("unchecked")
@Override
public void init(final ProcessorContext context) {
super.init(context);
foreignKeySerdeTopic = foreignKeySerdeTopicSupplier.get();
valueSerdeTopic = valueSerdeTopicSupplier.get();
// get default key serde if it wasn't supplied directly at construction
if (foreignKeySerializer == null) {
foreignKeySerializer = (Serializer<KO>) context.keySerde().serializer();
Expand Down
10 changes: 7 additions & 3 deletions ...a/streams/kstream/internals/foreignkeyjoin/SubscriptionResolverJoinProcessorSupplier.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,8 @@
import org.apache.kafka.streams.state.ValueAndTimestamp;
import org.apache.kafka.streams.state.internals.Murmur3;

import java.util.function.Supplier;

/**
* Receives {@code SubscriptionResponseWrapper<VO>} events and filters out events which do not match the current hash
* of the primary key. This eliminates race-condition results for rapidly-changing foreign-keys for a given primary key.
Expand All @@ -42,25 +44,26 @@
public class SubscriptionResolverJoinProcessorSupplier<K, V, VO, VR> implements ProcessorSupplier<K, SubscriptionResponseWrapper<VO>> {
private final KTableValueGetterSupplier<K, V> valueGetterSupplier;
private final Serializer<V> constructionTimeValueSerializer;
private final String valueHashSerdePseudoTopic;
private final Supplier<String> valueHashSerdePseudoTopicSupplier;
private final ValueJoiner<V, VO, VR> joiner;
private final boolean leftJoin;

public SubscriptionResolverJoinProcessorSupplier(final KTableValueGetterSupplier<K, V> valueGetterSupplier,
final Serializer<V> valueSerializer,
final String valueHashSerdePseudoTopic,
final Supplier<String> valueHashSerdePseudoTopicSupplier,
final ValueJoiner<V, VO, VR> joiner,
final boolean leftJoin) {
this.valueGetterSupplier = valueGetterSupplier;
constructionTimeValueSerializer = valueSerializer;
this.valueHashSerdePseudoTopic = valueHashSerdePseudoTopic;
this.valueHashSerdePseudoTopicSupplier = valueHashSerdePseudoTopicSupplier;
this.joiner = joiner;
this.leftJoin = leftJoin;
}

@Override
public Processor<K, SubscriptionResponseWrapper<VO>> get() {
return new AbstractProcessor<K, SubscriptionResponseWrapper<VO>>() {
private String valueHashSerdePseudoTopic;
private Serializer<V> runtimeValueSerializer = constructionTimeValueSerializer;

private KTableValueGetter<K, V> valueGetter;
Expand All @@ -69,6 +72,7 @@ public Processor<K, SubscriptionResponseWrapper<VO>> get() {
@Override
public void init(final ProcessorContext context) {
super.init(context);
valueHashSerdePseudoTopic = valueHashSerdePseudoTopicSupplier.get();
valueGetter = valueGetterSupplier.get();
valueGetter.init(context);
if (runtimeValueSerializer == null) {
Expand Down
30 changes: 21 additions & 9 deletions ...a/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionWrapperSerde.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,16 +25,17 @@

import java.nio.ByteBuffer;
import java.util.Objects;
import java.util.function.Supplier;

public class SubscriptionWrapperSerde<K> implements Serde<SubscriptionWrapper<K>> {
private final SubscriptionWrapperSerializer<K> serializer;
private final SubscriptionWrapperDeserializer<K> deserializer;

public SubscriptionWrapperSerde(final String primaryKeySerializationPseudoTopic,
public SubscriptionWrapperSerde(final Supplier<String> primaryKeySerializationPseudoTopicSupplier,
final Serde<K> primaryKeySerde) {
serializer = new SubscriptionWrapperSerializer<>(primaryKeySerializationPseudoTopic,
serializer = new SubscriptionWrapperSerializer<>(primaryKeySerializationPseudoTopicSupplier,
primaryKeySerde == null ? null : primaryKeySerde.serializer());
deserializer = new SubscriptionWrapperDeserializer<>(primaryKeySerializationPseudoTopic,
deserializer = new SubscriptionWrapperDeserializer<>(primaryKeySerializationPseudoTopicSupplier,
primaryKeySerde == null ? null : primaryKeySerde.deserializer());
}

Expand All @@ -51,12 +52,13 @@ public Deserializer<SubscriptionWrapper<K>> deserializer() {
private static class SubscriptionWrapperSerializer<K>
implements Serializer<SubscriptionWrapper<K>>, WrappingNullableSerializer<SubscriptionWrapper<K>, K> {

private final String primaryKeySerializationPseudoTopic;
private final Supplier<String> primaryKeySerializationPseudoTopicSupplier;
private String primaryKeySerializationPseudoTopic = null;
private Serializer<K> primaryKeySerializer;

SubscriptionWrapperSerializer(final String primaryKeySerializationPseudoTopic,
SubscriptionWrapperSerializer(final Supplier<String> primaryKeySerializationPseudoTopicSupplier,
final Serializer<K> primaryKeySerializer) {
this.primaryKeySerializationPseudoTopic = primaryKeySerializationPseudoTopic;
this.primaryKeySerializationPseudoTopicSupplier = primaryKeySerializationPseudoTopicSupplier;
this.primaryKeySerializer = primaryKeySerializer;
}

Expand All @@ -76,6 +78,10 @@ public byte[] serialize(final String ignored, final SubscriptionWrapper<K> data)
throw new UnsupportedVersionException("SubscriptionWrapper version is larger than maximum supported 0x7F");
}

if (primaryKeySerializationPseudoTopic == null) {
primaryKeySerializationPseudoTopic = primaryKeySerializationPseudoTopicSupplier.get();
}
Comment on lines +81 to +83

@vvcephei vvcephei Apr 28, 2020

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 below) is a bit awkward.

Our requirement is not to call the supplier until after the app starts, but we can call it any time after the app starts.

The natural place would be in configure, but unfortunately, that method is basically useless for our internal serdes. The reason is that we previously decided that configure should be called externally to the DSL, but our internal serdes are constructed internal to the DSL. Plus, configure must be called at run time (when the config is available), but by run time, we can no longer tell whether our serde is "internal" or not. So, there's no good place where we can call configure for our internal serdes.

I'm side-stepping the problem here by just invoking the supplier when we first need to use it, which is also at run time.


final byte[] primaryKeySerializedData = primaryKeySerializer.serialize(
primaryKeySerializationPseudoTopic,
data.getPrimaryKey()
Expand Down Expand Up @@ -106,12 +112,13 @@ public byte[] serialize(final String ignored, final SubscriptionWrapper<K> data)
private static class SubscriptionWrapperDeserializer<K>
implements Deserializer<SubscriptionWrapper<K>>, WrappingNullableDeserializer<SubscriptionWrapper<K>, K> {

private final String primaryKeySerializationPseudoTopic;
private final Supplier<String> primaryKeySerializationPseudoTopicSupplier;
private String primaryKeySerializationPseudoTopic = null;
private Deserializer<K> primaryKeyDeserializer;

SubscriptionWrapperDeserializer(final String primaryKeySerializationPseudoTopic,
SubscriptionWrapperDeserializer(final Supplier<String> primaryKeySerializationPseudoTopicSupplier,
final Deserializer<K> primaryKeyDeserializer) {
this.primaryKeySerializationPseudoTopic = primaryKeySerializationPseudoTopic;
this.primaryKeySerializationPseudoTopicSupplier = primaryKeySerializationPseudoTopicSupplier;
this.primaryKeyDeserializer = primaryKeyDeserializer;
}

Expand Down Expand Up @@ -144,6 +151,11 @@ public SubscriptionWrapper<K> deserialize(final String ignored, final byte[] dat

final byte[] primaryKeyRaw = new byte[data.length - lengthSum]; //The remaining data is the serialized pk
buf.get(primaryKeyRaw, 0, primaryKeyRaw.length);

if (primaryKeySerializationPseudoTopic == null) {
primaryKeySerializationPseudoTopic = primaryKeySerializationPseudoTopicSupplier.get();
}

final K primaryKey = primaryKeyDeserializer.deserialize(primaryKeySerializationPseudoTopic,
primaryKeyRaw);

Expand Down
4 changes: 4 additions & 0 deletions ...s/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1224,6 +1224,10 @@ private List<String> maybeDecorateInternalSourceTopics(final Collection<String>
return decoratedTopics;
}

public String decoratePseudoTopic(final String topic) {

@vvcephei vvcephei Apr 28, 2020

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'm adding a new public method for our specific use case here, to document that we should only need to invoke this method publicly for "pseudo" topics.

return decorateTopic(topic);
}

private String decorateTopic(final String topic) {
if (applicationId == null) {
throw new TopologyException("there are internal topics and "
Expand Down
30 changes: 15 additions & 15 deletions ...va/org/apache/kafka/streams/kstream/internals/KTableKTableForeignKeyJoinScenarioTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,6 @@
import org.apache.kafka.streams.state.KeyValueStore;
import org.apache.kafka.streams.utils.UniqueTopicSerdeScope;
import org.apache.kafka.test.TestUtils;
import org.hamcrest.CoreMatchers;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TestName;
Expand Down Expand Up @@ -179,8 +178,9 @@ public void shouldWorkWithDefaultAndProducedSerdes() {

@Test
public void shouldUseExpectedTopicsWithSerde() {
final String applicationId = "ktable-ktable-joinOnForeignKey";
final Properties streamsConfig = mkProperties(mkMap(
mkEntry(StreamsConfig.APPLICATION_ID_CONFIG, "ktable-ktable-joinOnForeignKey"),
mkEntry(StreamsConfig.APPLICATION_ID_CONFIG, applicationId),
mkEntry(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "asdf:0000"),
mkEntry(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getPath())
));
Expand All @@ -191,12 +191,12 @@ public void shouldUseExpectedTopicsWithSerde() {
final KTable<String, String> left = builder.table(
LEFT_TABLE,
Consumed.with(serdeScope.decorateSerde(Serdes.String(), streamsConfig, true),
serdeScope.decorateSerde(Serdes.String(), streamsConfig, false))
serdeScope.decorateSerde(Serdes.String(), streamsConfig, false))
);
final KTable<String, String> right = builder.table(
RIGHT_TABLE,
Consumed.with(serdeScope.decorateSerde(Serdes.String(), streamsConfig, true),
serdeScope.decorateSerde(Serdes.String(), streamsConfig, false))
serdeScope.decorateSerde(Serdes.String(), streamsConfig, false))
);

left.join(
Expand All @@ -218,19 +218,19 @@ public void shouldUseExpectedTopicsWithSerde() {
}
// verifying primarily that no extra pseudo-topics were used, but it's nice to also verify the rest of the
// topics our serdes serialize data for
assertThat(serdeScope.registeredTopics(), CoreMatchers.is(mkSet(
assertThat(serdeScope.registeredTopics(), is(mkSet(
// expected pseudo-topics
"KTABLE-FK-JOIN-SUBSCRIPTION-REGISTRATION-0000000006-topic-fk--key",
"KTABLE-FK-JOIN-SUBSCRIPTION-REGISTRATION-0000000006-topic-pk--key",
"KTABLE-FK-JOIN-SUBSCRIPTION-REGISTRATION-0000000006-topic-vh--value",
applicationId + "-KTABLE-FK-JOIN-SUBSCRIPTION-REGISTRATION-0000000006-topic-fk--key",
applicationId + "-KTABLE-FK-JOIN-SUBSCRIPTION-REGISTRATION-0000000006-topic-pk--key",
applicationId + "-KTABLE-FK-JOIN-SUBSCRIPTION-REGISTRATION-0000000006-topic-vh--value",
Comment on lines 222 to +225

@vvcephei vvcephei Apr 28, 2020

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 verifies the fix: the pseudo topics should also be prefixed. I should have noticed before that they weren't.

// internal topics
"ktable-ktable-joinOnForeignKey-KTABLE-FK-JOIN-SUBSCRIPTION-REGISTRATION-0000000006-topic--key",
"ktable-ktable-joinOnForeignKey-KTABLE-FK-JOIN-SUBSCRIPTION-RESPONSE-0000000014-topic--key",
"ktable-ktable-joinOnForeignKey-KTABLE-FK-JOIN-SUBSCRIPTION-RESPONSE-0000000014-topic--value",
"ktable-ktable-joinOnForeignKey-left_table-STATE-STORE-0000000000-changelog--key",
"ktable-ktable-joinOnForeignKey-left_table-STATE-STORE-0000000000-changelog--value",
"ktable-ktable-joinOnForeignKey-right_table-STATE-STORE-0000000003-changelog--key",
"ktable-ktable-joinOnForeignKey-right_table-STATE-STORE-0000000003-changelog--value",
applicationId + "-KTABLE-FK-JOIN-SUBSCRIPTION-REGISTRATION-0000000006-topic--key",
applicationId + "-KTABLE-FK-JOIN-SUBSCRIPTION-RESPONSE-0000000014-topic--key",
applicationId + "-KTABLE-FK-JOIN-SUBSCRIPTION-RESPONSE-0000000014-topic--value",
applicationId + "-left_table-STATE-STORE-0000000000-changelog--key",
applicationId + "-left_table-STATE-STORE-0000000000-changelog--value",
applicationId + "-right_table-STATE-STORE-0000000003-changelog--key",
applicationId + "-right_table-STATE-STORE-0000000003-changelog--value",
// output topics
"output-topic--key",
"output-topic--value"
Expand Down
Loading