KIP-557: Add Emit On Change Support #8254
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -16,9 +16,12 @@ | |
| */ | ||
| package org.apache.kafka.streams.state.internals; | ||
|
|
||
| import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.maybeMeasureLatency; | ||
|
|
||
| import org.apache.kafka.common.serialization.Serde; | ||
| import org.apache.kafka.common.utils.Bytes; | ||
| import org.apache.kafka.common.utils.Time; | ||
| import org.apache.kafka.streams.errors.ProcessorStateException; | ||
| import org.apache.kafka.streams.processor.ProcessorContext; | ||
| import org.apache.kafka.streams.processor.internals.ProcessorStateManager; | ||
| import org.apache.kafka.streams.state.KeyValueStore; | ||
|
|
@@ -35,7 +38,7 @@ | |
| * @param <V> | ||
| */ | ||
| public class MeteredTimestampedKeyValueStore<K, V> | ||
| extends MeteredKeyValueStore<K, ValueAndTimestamp<V>> | ||
| implements TimestampedKeyValueStore<K, V> { | ||
|
|
||
| MeteredTimestampedKeyValueStore(final KeyValueStore<Bytes, byte[]> inner, | ||
|
|
@@ -53,4 +56,48 @@ void initStoreSerde(final ProcessorContext context) { | |
| keySerde == null ? (Serde<K>) context.keySerde() : keySerde, | ||
| valueSerde == null ? new ValueAndTimestampSerde<>((Serde<V>) context.valueSerde()) : valueSerde); | ||
| } | ||
|
|
||
| public RawAndDeserializedValue<V> getWithBinary(final K key) { | ||
|
Member
There was a problem hiding this comment. req: Please add unit tests for this method.
Contributor
Author
There was a problem hiding this comment. Done. |
||
| try { | ||
| return maybeMeasureLatency(() -> { | ||
| final byte[] serializedValue = wrapped().get(keyBytes(key)); | ||
| return new RawAndDeserializedValue<V>(serializedValue, outerValue(serializedValue)); | ||
| }, time, getSensor); | ||
| } catch (final ProcessorStateException e) { | ||
| final String message = String.format(e.getMessage(), key); | ||
| throw new ProcessorStateException(message, e); | ||
| } | ||
| } | ||
|
|
||
| public boolean putIfDifferentValues(final K key, | ||
|
Member
There was a problem hiding this comment. req: Please add unit tests for this method.
Contributor
Author
There was a problem hiding this comment. Done. |
||
| final ValueAndTimestamp<V> newValue, | ||
| final byte[] oldSerializedValue) { | ||
| try { | ||
| return maybeMeasureLatency( | ||
| () -> { | ||
| final byte[] newSerializedValue = serdes.rawValue(newValue); | ||
| if (ValueAndTimestampSerializer.compareValuesAndCheckForIncreasingTimestamp(oldSerializedValue, newSerializedValue)) { | ||
| return false; | ||
| } else { | ||
| wrapped().put(keyBytes(key), newSerializedValue); | ||
|
Contributor
There was a problem hiding this comment. @ConcurrencyPractitioner @vvcephei I'm trying to understand this to debug some broken tests in ksql. Couple questions: When the timestamp of the newer value is lower (ignoring the value), why do we want to put the new value into the store? Surely the store should have the value with the newer timestamp? Otherwise we could wind up with a corrupt store. Don't we still want to put the value in the store (even if we don't forward it on to the next context) if the values are the same but the timestamp is newer? Otherwise if we get an out-of-order update with a different value, but a timestamp in between the rows with the same value, we'd incorrectly put that value into the store, e.g. the following updates: TS: 1, K: X, V: A would result in the table containing
Member
There was a problem hiding this comment.
This behavior was there also before this PR. If a out-of-order record is encountered, a log message was written, but the record was nevertheless put into the state store (cf. Could you elaborate on why a store should get corrupted because of this?
As said above, this behavior should not have been changed.
Member
There was a problem hiding this comment.
If we just put the value in the store but did not forward it, then the store would actually be corrupted, because the local state would not be consistent with downstream anymore. Not putting a record with the same value but a newer timestamp in the store and not forwarding it was the main point of this KIP. |
||
| return true; | ||
| } | ||
| }, | ||
| time, | ||
| putSensor | ||
| ); | ||
| } catch (final ProcessorStateException e) { | ||
| final String message = String.format(e.getMessage(), key, newValue); | ||
| throw new ProcessorStateException(message, e); | ||
| } | ||
| } | ||
|
|
||
| public static class RawAndDeserializedValue<ValueType> { | ||
| public final byte[] serializedValue; | ||
| public final ValueAndTimestamp<ValueType> value; | ||
| public RawAndDeserializedValue(final byte[] serializedValue, final ValueAndTimestamp<ValueType> value) { | ||
| this.serializedValue = serializedValue; | ||
| this.value = value; | ||
| } | ||
| } | ||
| } | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
req: Please add a unit test in
ProcessorNodeMetricsTest.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Done.