Skip to content

KAFKA-8254: Pass Changelog as Topic in Suppress Serdes #6602

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
guozhangwang merged 8 commits into from
Apr 26, 2019
Merged

KAFKA-8254: Pass Changelog as Topic in Suppress Serdes #6602

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
f8ad73d
KAFKA-8254: Pass Changelog as Topic in Suppress Serdes
vvcephei Apr 18, 2019
e1642ba
fix test
vvcephei Apr 24, 2019
19127de
fix serde handling
vvcephei Apr 24, 2019
4c2a457
suppress unchecked
vvcephei Apr 24, 2019
a9aa4fa
unused imports
vvcephei Apr 24, 2019
c26718a
style
vvcephei Apr 24, 2019
a51c4fc
cr feedback
vvcephei Apr 25, 2019
7414ed0
qualify serdes
vvcephei Apr 26, 2019
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
8 changes: 4 additions & 4 deletions streams/src/main/java/org/apache/kafka/streams/kstream/internals/FullChangeSerde.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,21 +25,21 @@

import static java.util.Objects.requireNonNull;

public class FullChangeSerde<T> implements Serde<Change<T>> {
public final class FullChangeSerde<T> implements Serde<Change<T>> {
private final Serde<T> inner;

@SuppressWarnings("unchecked")
public static <T> FullChangeSerde<T> castOrWrap(final Serde<?> serde) {
public static <T> FullChangeSerde<T> castOrWrap(final Serde<T> serde) {
if (serde == null) {
return null;
} else if (serde instanceof FullChangeSerde) {
return (FullChangeSerde<T>) serde;
} else {
return new FullChangeSerde<T>((Serde<T>) serde);
return new FullChangeSerde<>(serde);
}
}

public FullChangeSerde(final Serde<T> inner) {
private FullChangeSerde(final Serde<T> inner) {

@vvcephei vvcephei Apr 24, 2019

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.

Forcing all callers to use castOrWrap, which propagates null (aka, no serde is defined). See the usage in KTableImpl.

this.inner = requireNonNull(inner);
}

Expand Down
36 changes: 29 additions & 7 deletions streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -384,18 +384,26 @@ public KStream<K, V> peek(final ForeachAction<? super K, ? super V> action) {

@Override
public KStream<K, V> through(final String topic) {
return through(topic, Produced.with(null, null, null));
return through(topic, Produced.with(keySerde, valSerde, null));
Comment thread
vvcephei marked this conversation as resolved.
}

@Override
public KStream<K, V> through(final String topic, final Produced<K, V> produced) {
Objects.requireNonNull(topic, "topic can't be null");
Objects.requireNonNull(produced, "Produced can't be null");
final ProducedInternal<K, V> producedInternal = new ProducedInternal<>(produced);
if (producedInternal.keySerde() == null) {
producedInternal.withKeySerde(keySerde);
}
if (producedInternal.valueSerde() == null) {
producedInternal.withValueSerde(valSerde);
}
to(topic, producedInternal);
return builder.stream(
Collections.singleton(topic),
new ConsumedInternal<>(
producedInternal.keySerde() != null ? producedInternal.keySerde() : keySerde,
producedInternal.valueSerde() != null ? producedInternal.valueSerde() : valSerde,
producedInternal.keySerde(),
producedInternal.valueSerde(),

@mjsax mjsax Apr 26, 2019

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.

Why this rewrite? (Just curious.)

@vvcephei vvcephei Apr 26, 2019

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.

it's just no longer necessary, since we're already setting the key/value serdes if they were null in producedInternal on lines 395-400.

@mjsax mjsax Apr 26, 2019

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.

Sure. The question was about both changes in combination. What is it better to split out compare to just do a one liner?

new FailOnInvalidTimestamp(),
null
)
Expand All @@ -404,26 +412,40 @@ public KStream<K, V> through(final String topic, final Produced<K, V> produced)

@Override
public void to(final String topic) {
to(topic, Produced.with(null, null, null));
to(topic, Produced.with(keySerde, valSerde, null));
}

@Override
public void to(final String topic, final Produced<K, V> produced) {
Objects.requireNonNull(topic, "topic can't be null");
Objects.requireNonNull(produced, "Produced can't be null");
to(new StaticTopicNameExtractor<>(topic), new ProducedInternal<>(produced));
final ProducedInternal<K, V> producedInternal = new ProducedInternal<>(produced);
if (producedInternal.keySerde() == null) {
producedInternal.withKeySerde(keySerde);
}
if (producedInternal.valueSerde() == null) {
producedInternal.withValueSerde(valSerde);
}
to(new StaticTopicNameExtractor<>(topic), producedInternal);
}

@Override
public void to(final TopicNameExtractor<K, V> topicExtractor) {
to(topicExtractor, Produced.with(null, null, null));
to(topicExtractor, Produced.with(keySerde, valSerde, null));
}

@Override
public void to(final TopicNameExtractor<K, V> topicExtractor, final Produced<K, V> produced) {
Objects.requireNonNull(topicExtractor, "topic extractor can't be null");
Objects.requireNonNull(produced, "Produced can't be null");
to(topicExtractor, new ProducedInternal<>(produced));
final ProducedInternal<K, V> producedInternal = new ProducedInternal<>(produced);
if (producedInternal.keySerde() == null) {
producedInternal.withKeySerde(keySerde);
}
if (producedInternal.valueSerde() == null) {
producedInternal.withValueSerde(valSerde);
}
to(topicExtractor, producedInternal);
}

private void to(final TopicNameExtractor<K, V> topicExtractor, final ProducedInternal<K, V> produced) {
Expand Down
9 changes: 2 additions & 7 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 @@ -402,18 +402,13 @@ public KTable<K, V> suppress(final Suppressed<? super K> suppressed) {
suppressedInternal.name() != null ? suppressedInternal.name() + "-store" : builder.newStoreName(SUPPRESS_NAME);

final ProcessorSupplier<K, Change<V>> suppressionSupplier =
() -> new KTableSuppressProcessor<>(
suppressedInternal,
storeName,
keySerde,
valSerde == null ? null : new FullChangeSerde<>(valSerde)
);
() -> new KTableSuppressProcessor<>(suppressedInternal, storeName);


final ProcessorGraphNode<K, Change<V>> node = new StatefulProcessorNode<>(
name,
new ProcessorParameters<>(suppressionSupplier, name),
new InMemoryTimeOrderedKeyValueBuffer.Builder(storeName)
new InMemoryTimeOrderedKeyValueBuffer.Builder<>(storeName, keySerde, FullChangeSerde.castOrWrap(valSerde))
);

builder.addGraphNode(streamsGraphNode, node);
Expand Down
42 changes: 14 additions & 28 deletions ...ain/java/org/apache/kafka/streams/kstream/internals/suppress/KTableSuppressProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,8 +19,6 @@
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.metrics.Sensor;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.errors.StreamsException;
import org.apache.kafka.streams.kstream.internals.Change;
import org.apache.kafka.streams.kstream.internals.FullChangeSerde;
Expand All @@ -30,7 +28,6 @@
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.internals.InternalProcessorContext;
import org.apache.kafka.streams.processor.internals.ProcessorRecordContext;
import org.apache.kafka.streams.state.internals.ContextualRecord;
import org.apache.kafka.streams.state.internals.TimeOrderedKeyValueBuffer;

import static java.util.Objects.requireNonNull;
Expand All @@ -44,22 +41,14 @@ public class KTableSuppressProcessor<K, V> implements Processor<K, Change<V>> {
private final boolean safeToDropTombstones;
private final String storeName;

private TimeOrderedKeyValueBuffer buffer;
private TimeOrderedKeyValueBuffer<K, Change<V>> buffer;
private InternalProcessorContext internalProcessorContext;
private Sensor suppressionEmitSensor;
private Serde<K> keySerde;
private FullChangeSerde<V> valueSerde;

private long observedStreamTime = ConsumerRecord.NO_TIMESTAMP;

public KTableSuppressProcessor(final SuppressedInternal<K> suppress,
final String storeName,
final Serde<K> keySerde,
final FullChangeSerde<V> valueSerde) {
public KTableSuppressProcessor(final SuppressedInternal<K> suppress, final String storeName) {
this.storeName = storeName;
requireNonNull(suppress);
this.keySerde = keySerde;
this.valueSerde = valueSerde;
maxRecords = suppress.bufferConfig().maxRecords();
maxBytes = suppress.bufferConfig().maxBytes();
suppressDurationMillis = suppress.timeToWaitForMoreEvents().toMillis();
Expand All @@ -74,9 +63,8 @@ public void init(final ProcessorContext context) {
internalProcessorContext = (InternalProcessorContext) context;
suppressionEmitSensor = Sensors.suppressionEmitSensor(internalProcessorContext);

keySerde = keySerde == null ? (Serde<K>) context.keySerde() : keySerde;
valueSerde = valueSerde == null ? FullChangeSerde.castOrWrap(context.valueSerde()) : valueSerde;
buffer = requireNonNull((TimeOrderedKeyValueBuffer) context.getStateStore(storeName));
buffer = requireNonNull((TimeOrderedKeyValueBuffer<K, Change<V>>) context.getStateStore(storeName));
buffer.setSerdesIfNull((Serde<K>) context.keySerde(), FullChangeSerde.castOrWrap((Serde<V>) context.valueSerde()));
}

@Override
Expand All @@ -88,12 +76,7 @@ public void process(final K key, final Change<V> value) {

private void buffer(final K key, final Change<V> value) {
final long bufferTime = bufferTimeDefinition.time(internalProcessorContext, key);
final ProcessorRecordContext recordContext = internalProcessorContext.recordContext();

final Bytes serializedKey = Bytes.wrap(keySerde.serializer().serialize(null, key));
final byte[] serializedValue = valueSerde.serializer().serialize(null, value);

buffer.put(bufferTime, serializedKey, new ContextualRecord(serializedValue, recordContext));
buffer.put(bufferTime, key, value, internalProcessorContext.recordContext());
}

private void enforceConstraints() {
Expand All @@ -114,6 +97,11 @@ private void enforceConstraints() {
buffer.numRecords(), maxRecords,
buffer.bufferSize(), maxBytes
));
default:

@vvcephei vvcephei Apr 18, 2019

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 is to fix a warning about the switch being incomplete.

throw new UnsupportedOperationException(
"The bufferFullStrategy [" + bufferFullStrategy +
"] is not implemented. This is a bug in Kafka Streams."
);
}
}
}
Expand All @@ -122,14 +110,12 @@ private boolean overCapacity() {
return buffer.numRecords() > maxRecords || buffer.bufferSize() > maxBytes;
}

private void emit(final KeyValue<Bytes, ContextualRecord> toEmit) {
final Change<V> value = valueSerde.deserializer().deserialize(null, toEmit.value.value());
if (shouldForward(value)) {
private void emit(final TimeOrderedKeyValueBuffer.Eviction<K, Change<V>> toEmit) {
if (shouldForward(toEmit.value())) {
final ProcessorRecordContext prevRecordContext = internalProcessorContext.recordContext();
internalProcessorContext.setRecordContext(toEmit.value.recordContext());
internalProcessorContext.setRecordContext(toEmit.recordContext());
try {
final K key = keySerde.deserializer().deserialize(null, toEmit.key.get());
internalProcessorContext.forward(key, value);
internalProcessorContext.forward(toEmit.key(), toEmit.value());
suppressionEmitSensor.record();
} finally {
internalProcessorContext.setRecordContext(prevRecordContext);
Expand Down
4 changes: 2 additions & 2 deletions streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -160,7 +160,7 @@ public <K, V> void forward(final K key,
final V value,
final To to) {
final ProcessorNode previousNode = currentNode();
final long currentTimestamp = recordContext.timestamp;
final long currentTimestamp = recordContext.timestamp();

try {
toInternal.update(to);
Expand All @@ -183,7 +183,7 @@ public <K, V> void forward(final K key,
forward(child, key, value);
}
} finally {
recordContext.timestamp = currentTimestamp;
recordContext.setTimestamp(currentTimestamp);
setCurrentNode(previousNode);
}
}
Expand Down
23 changes: 10 additions & 13 deletions ...ms/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorRecordContext.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,11 +29,11 @@

public class ProcessorRecordContext implements RecordContext {

long timestamp;
final long offset;
final String topic;
final int partition;
final Headers headers;
private long timestamp;
private final long offset;
private final String topic;
private final int partition;
private final Headers headers;

@vvcephei vvcephei Apr 24, 2019

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.

some on-the-side cleanup of ProcessorRecordContext I encountered while writing my test.


public ProcessorRecordContext(final long timestamp,
final long offset,
Expand All @@ -48,13 +48,6 @@ public ProcessorRecordContext(final long timestamp,
this.headers = headers;
}

public ProcessorRecordContext(final long timestamp,
final long offset,
final int partition,
final String topic) {
this(timestamp, offset, partition, topic, null);
}

public void setTimestamp(final long timestamp) {
this.timestamp = timestamp;
}
Expand Down Expand Up @@ -225,9 +218,13 @@ public boolean equals(final Object o) {
Objects.equals(headers, that.headers);
}

/**
* Equality is implemented in support of tests, *not* for use in Hash collections, since this class is mutable.
*/
@Deprecated

@mjsax mjsax Apr 26, 2019

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.

No need to deprecate internal method. If we want to remove it, we should just do it.

@vvcephei vvcephei Apr 26, 2019 •
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.

We can't remove hashCode, since it's defined by Object. Marking it as deprecated isn't necessary, since any attempt to use it results in an exception, but it's just extra documentation that hashCode is not to be used on ProcessorRecordContext.

I implemented it because implementing equals without hashCode is a bug, and a very subtle one. But I wanted equals in support of the test validation. And we also can't provide a correct implementation of hashCode, since there is mutable state in this object. This all basically amounts to a flaw in Java, that you can't define equality without implying that an object is stably hashable. Ideally, we could plug in a different notion of equality, but all the testing frameworks have built-in dependency on equals. Throwing the exception and adding a deprecation is essentially finding a middle ground, we wind up with safe code that can also be used in tests.

@mjsax mjsax Apr 26, 2019

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.

What I meant by "remove" was, to remove the overwrite :)

Thanks to clarify why you want to overwrite and throw. But this does not mean that the method is deprecated... Using the annotation implies that we want to remove this overwrite in the future was is miss leading. I would prefer to remove the annotation. \cc @guozhangwang @bbejeck

@guozhangwang guozhangwang Apr 26, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

I think that throwing the exception in its override implementation is sufficient for future bug-proof, as regarding to documentation I feel neutral either way.

@mjsax mjsax Apr 26, 2019

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.

We should still add JavaDocs and explain that we throw and why. But deprecation != documentation IMHO.

@Override
public int hashCode() {
return Objects.hash(timestamp, offset, topic, partition, headers);
throw new UnsupportedOperationException("ProcessorRecordContext is unsafe for use in Hash collections");
}

@Override
Expand Down
Loading