Skip to content

KIP-557: Add Emit On Change Support #8254

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 38 commits into from
May 12, 2020
Merged

KIP-557: Add Emit On Change Support #8254

Show file tree
Hide file tree
Changes from 15 commits
Commits
Show all changes
38 commits
Select commit Hold shift + click to select a range
80de5bb
[KIP-557] Add emit on change support
ConcurrencyPractitioner Mar 8, 2020
1aaaa01
Adding some class modifications
ConcurrencyPractitioner Mar 14, 2020
ec46f57
Adding class
ConcurrencyPractitioner Mar 14, 2020
2c19bf7
Merge branch 'trunk' into EMIT-ON-CHANGE
ConcurrencyPractitioner Mar 14, 2020
e3ffa93
Adding test
ConcurrencyPractitioner Mar 14, 2020
a14725a
Merge branch 'EMIT-ON-CHANGE' of https://github.com/ConcurrencyPracti…
ConcurrencyPractitioner Mar 14, 2020
6d3c7ab
Adding sensor
ConcurrencyPractitioner Mar 15, 2020
81c19bd
Add working sensor
ConcurrencyPractitioner Mar 15, 2020
209d376
Bumping processor level
ConcurrencyPractitioner Mar 15, 2020
8fef99c
Update streams/src/test/java/org/apache/kafka/streams/kstream/interna…
ConcurrencyPractitioner Apr 2, 2020
1d7c4ad
Update streams/src/main/java/org/apache/kafka/streams/kstream/interna…
ConcurrencyPractitioner Apr 2, 2020
d062bfa
Update streams/src/main/java/org/apache/kafka/streams/state/internals…
ConcurrencyPractitioner Apr 2, 2020
d0737f5
Fixing compilation errors and other issues
ConcurrencyPractitioner Apr 2, 2020
afdde96
Removing unneeded class
ConcurrencyPractitioner Apr 2, 2020
35c16b1
Removing all vestiges of new class
ConcurrencyPractitioner Apr 2, 2020
9e7649b
Adding more inclusive lambda
ConcurrencyPractitioner Apr 3, 2020
21a7345
Update streams/src/main/java/org/apache/kafka/streams/kstream/interna…
ConcurrencyPractitioner Apr 7, 2020
c4257e1
Adding some modifications
ConcurrencyPractitioner Apr 11, 2020
6ff8b65
Merge branch 'EMIT-ON-CHANGE' of https://github.com/ConcurrencyPracti…
ConcurrencyPractitioner Apr 11, 2020
860d41b
Pushing changes
ConcurrencyPractitioner Apr 11, 2020
2871c6b
Realiging diffs
ConcurrencyPractitioner Apr 11, 2020
367ebaf
Adding tests
ConcurrencyPractitioner Apr 11, 2020
449efd9
Adding test
ConcurrencyPractitioner Apr 13, 2020
9cd3726
Adding fixed getWithBinary TEst
ConcurrencyPractitioner Apr 15, 2020
cdad348
Fixing description
ConcurrencyPractitioner Apr 15, 2020
48cc4d9
Making some test changes
ConcurrencyPractitioner Apr 17, 2020
de0fc6d
Merge branch 'trunk' into EMIT-ON-CHANGE
ConcurrencyPractitioner Apr 23, 2020
d396fd4
Making some modifications to test
ConcurrencyPractitioner Apr 23, 2020
7d93107
Merge branch 'EMIT-ON-CHANGE' of https://github.com/ConcurrencyPracti…
ConcurrencyPractitioner Apr 23, 2020
07d40af
Merge branch 'trunk' of https://github.com/apache/kafka into EMIT-ON-…
ConcurrencyPractitioner Apr 24, 2020
4779b27
Fixing details
ConcurrencyPractitioner Apr 24, 2020
ddbf2cf
Catching bad modification
ConcurrencyPractitioner Apr 25, 2020
197ddd2
Resolving most comments
ConcurrencyPractitioner Apr 29, 2020
527ba28
Deleting massive amount of print statements
ConcurrencyPractitioner Apr 29, 2020
bae2860
Final removal
ConcurrencyPractitioner Apr 29, 2020
bf5532f
Resolving last comment
ConcurrencyPractitioner Apr 29, 2020
d9aa12d
Addressing some last comments
ConcurrencyPractitioner May 11, 2020
2a3d93d
minor test fix
vvcephei May 12, 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
32 changes: 26 additions & 6 deletions streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableSource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,15 +21,19 @@
import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.ProcessorSupplier;
import org.apache.kafka.streams.processor.StateStore;
import org.apache.kafka.streams.processor.internals.InternalProcessorContext;
import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
import org.apache.kafka.streams.state.TimestampedKeyValueStore;
import org.apache.kafka.streams.state.ValueAndTimestamp;
import org.apache.kafka.streams.state.internals.MeteredTimestampedKeyValueStore;
import org.apache.kafka.streams.state.internals.WrappedStateStore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Objects;

import static org.apache.kafka.streams.processor.internals.metrics.TaskMetrics.droppedRecordsSensorOrSkippedRecordsSensor;
import static org.apache.kafka.streams.processor.internals.metrics.ProcessorNodeMetrics.skippedIdempotentUpdatesSensor;

public class KTableSource<K, V> implements ProcessorSupplier<K, V> {
private static final Logger LOG = LoggerFactory.getLogger(KTableSource.class);
Expand Down Expand Up @@ -74,10 +78,11 @@ public boolean materialized() {

private class KTableSourceProcessor extends AbstractProcessor<K, V> {

private TimestampedKeyValueStore<K, V> store;
private MeteredTimestampedKeyValueStore<K, V> store;
private TimestampedTupleForwarder<K, V> tupleForwarder;
private StreamsMetricsImpl metrics;
private Sensor droppedRecordsSensor;
private Sensor skippedIdempotentUpdatesSensor = null;

@SuppressWarnings("unchecked")
@Override
Expand All @@ -86,12 +91,21 @@ public void init(final ProcessorContext context) {
metrics = (StreamsMetricsImpl) context.metrics();
droppedRecordsSensor = droppedRecordsSensorOrSkippedRecordsSensor(Thread.currentThread().getName(), context.taskId().toString(), metrics);
if (queryableName != null) {
store = (TimestampedKeyValueStore<K, V>) context.getStateStore(queryableName);
final StateStore stateStore = context.getStateStore(queryableName);
try {
store = ((WrappedStateStore<MeteredTimestampedKeyValueStore<K, V>, K, V>) stateStore).wrapped();
} catch (final ClassCastException e) {
throw new IllegalStateException("Unexpected store type: " + stateStore.getClass() + " for store: " + queryableName, e);
}
tupleForwarder = new TimestampedTupleForwarder<>(
store,
context,
new TimestampedCacheFlushListener<>(context),
sendOldValues);
skippedIdempotentUpdatesSensor = skippedIdempotentUpdatesSensor(Thread.currentThread().getName(),
context.taskId().toString(),
((InternalProcessorContext) context).currentNode().name(),
metrics);
Comment thread
ConcurrencyPractitioner marked this conversation as resolved.
Outdated
}
}

Expand All @@ -108,7 +122,8 @@ public void process(final K key, final V value) {
}

if (queryableName != null) {
final ValueAndTimestamp<V> oldValueAndTimestamp = store.get(key);
final MeteredTimestampedKeyValueStore<K, V>.RawAndDeserializedValue<V> tuple = store.getWithBinary(key);
final ValueAndTimestamp<V> oldValueAndTimestamp = tuple.value;
final V oldValue;
if (oldValueAndTimestamp != null) {
oldValue = oldValueAndTimestamp.value();
Expand All @@ -119,8 +134,13 @@ public void process(final K key, final V value) {
} else {
oldValue = null;
}
store.put(key, ValueAndTimestamp.make(value, context().timestamp()));
tupleForwarder.maybeForward(key, value, oldValue);
final boolean isDifferentValue =
store.putIfDifferentValues(key, ValueAndTimestamp.make(value, context().timestamp()), tuple.serializedValue);

@cadonna cadonna Apr 7, 2020

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.

Q: If the record is out-of-order, shouldn't you put the value into the state store regardless of whether it is different from the old value or not? I think this would be more correct because the old value is actually the idempotent update to the new out-of-order value. This is one of the rare cases where we can correct the order. WDYT?

@vvcephei vvcephei Apr 7, 2020

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.

Thanks @cadonna. I'd like to ask for clarification:

Do you mean, given the following sequence of input records:

A: 2 @time3
A: 2 @time4
A: 2 @time1

We would take the following actions:

put(A, 2 @time3) and forward the update
skip idempotent update (A is still 2 @time3)
put(A, 2 @time1) and forward the update (so now A is still 2, but the timestamp is updated to 1)

This might be controversial. I believe that when we "complete" the implementation of timestamp semantics in Streams, the current proposal is to actually drop disordered updates. In that context, it might seem like a step in the wrong direction to specifically implement the code to preserve them when we would otherwise drop them.

On the other hand, the current semantics do seem to be better maintained if we take your suggestion. Perhaps we should let the future take care of itself and implement correct semantics as we currently understand them. I.e., I'm +1 on your suggestion.

@cadonna cadonna Apr 8, 2020

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.

Yes, this is what I mean and the current semantics were my motivation to ask the question.

@ConcurrencyPractitioner ConcurrencyPractitioner Apr 11, 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.

These are good points! However, if this is the case, then we probably need to compare timestamps as well (i.e. to check if one is smaller than the other). We just need to modify the comparator in the serializer.

@ConcurrencyPractitioner ConcurrencyPractitioner Apr 11, 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.

Also, added some changes to accomodate this in the test.

if (isDifferentValue) {
tupleForwarder.maybeForward(key, value, oldValue);
} else {
skippedIdempotentUpdatesSensor.record();
}
} else {
context().forward(key, new Change<>(value, null));
}
Expand Down
22 changes: 22 additions & 0 deletions .../main/java/org/apache/kafka/streams/processor/internals/metrics/ProcessorNodeMetrics.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,12 @@ private ProcessorNodeMetrics() {}
private static final String SUPPRESSION_EMIT_RATE_DESCRIPTION =
RATE_DESCRIPTION_PREFIX + SUPPRESSION_EMIT_DESCRIPTION + RATE_DESCRIPTION_SUFFIX;

private static final String IDEMPOTENT_UPDATE_SKIP = "idempotent-update-skip";
private static final String IDEMPOTENT_UPDATE_SKIP_DESCRIPTION = "skipped idempotent updates";
private static final String IDEMPOTENT_UPDATE_SKIP_TOTAL_DESCRIPTION = TOTAL_DESCRIPTION + IDEMPOTENT_UPDATE_SKIP_DESCRIPTION;
private static final String IDEMPOTENT_UPDATE_SKIP_RATE_DESCRIPTION =
RATE_DESCRIPTION_PREFIX + IDEMPOTENT_UPDATE_SKIP + RATE_DESCRIPTION_SUFFIX;

private static final String PROCESS = "process";
private static final String PROCESS_DESCRIPTION = "calls to process";
private static final String PROCESS_TOTAL_DESCRIPTION = TOTAL_DESCRIPTION + PROCESS_DESCRIPTION;
Expand Down Expand Up @@ -108,6 +114,22 @@ public static Sensor suppressionEmitSensor(final String threadId,
);
}

public static Sensor skippedIdempotentUpdatesSensor(final String threadId,
final String taskId,
final String processorNodeId,
final StreamsMetricsImpl streamsMetrics) {
return throughputSensor(
threadId,
taskId,
processorNodeId,
IDEMPOTENT_UPDATE_SKIP,
IDEMPOTENT_UPDATE_SKIP_RATE_DESCRIPTION,
IDEMPOTENT_UPDATE_SKIP_TOTAL_DESCRIPTION,
RecordingLevel.DEBUG,
streamsMetrics
);
}

Comment on lines +117 to +132

@cadonna cadonna Apr 7, 2020

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.

req: Please add a unit test in ProcessorNodeMetricsTest.

@ConcurrencyPractitioner ConcurrencyPractitioner Apr 13, 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.

Done.

public static Sensor processSensor(final String threadId,
final String taskId,
final String processorNodeId,
Expand Down
8 changes: 4 additions & 4 deletions streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,9 +54,9 @@ public class MeteredKeyValueStore<K, V>

private final String metricsScope;
protected final Time time;
private Sensor putSensor;
protected Sensor putSensor;
private Sensor putIfAbsentSensor;
private Sensor getSensor;
protected Sensor getSensor;
private Sensor deleteSensor;
private Sensor putAllSensor;
private Sensor allSensor;
Expand Down Expand Up @@ -203,11 +203,11 @@ public void close() {
streamsMetrics.removeAllStoreLevelSensors(threadId, taskId, name());
}

private V outerValue(final byte[] value) {
protected V outerValue(final byte[] value) {
return value != null ? serdes.valueFrom(value) : null;
}

private Bytes keyBytes(final K key) {
protected Bytes keyBytes(final K key) {
return Bytes.wrap(serdes.rawKey(key));
}

Expand Down
50 changes: 48 additions & 2 deletions ...c/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -35,7 +38,7 @@
* @param <V>
*/
public class MeteredTimestampedKeyValueStore<K, V>
extends MeteredKeyValueStore<K, ValueAndTimestamp<V>>
extends MeteredKeyValueStore<K, ValueAndTimestamp<V>>
implements TimestampedKeyValueStore<K, V> {

MeteredTimestampedKeyValueStore(final KeyValueStore<Bytes, byte[]> inner,
Expand All @@ -53,4 +56,47 @@ 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) {

@cadonna cadonna Apr 7, 2020

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.

req: Please add unit tests for this method.

@ConcurrencyPractitioner ConcurrencyPractitioner Apr 11, 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.

Done.

try {
final byte[] serializedValue = wrapped().get(keyBytes(key));
return new RawAndDeserializedValue<V>(serializedValue,
maybeMeasureLatency(() -> outerValue(serializedValue), time, getSensor));

@vvcephei vvcephei Apr 2, 2020

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.

Ah, missed this the last time though. We should also perform the wrapped().get inside the lambda, so that the time to perform the get is included in the latency.

@ConcurrencyPractitioner ConcurrencyPractitioner Apr 3, 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.

Cool, I will make the change.

} catch (final ProcessorStateException e) {
final String message = String.format(e.getMessage(), key);
throw new ProcessorStateException(message, e);
}
}

public boolean putIfDifferentValues(final K key,

@cadonna cadonna Apr 7, 2020

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.

req: Please add unit tests for this method.

@ConcurrencyPractitioner ConcurrencyPractitioner Apr 11, 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.

Done.

final ValueAndTimestamp<V> newValue,
final byte[] oldSerializedValue) {
try {
return maybeMeasureLatency(
() -> {
final byte[] newSerializedValue = serdes.rawValue(newValue);
if (ValueAndTimestampSerializer.maskTimestampAndCompareValues(oldSerializedValue, newSerializedValue)) {
return false;
} else {
wrapped().put(keyBytes(key), newSerializedValue);

@rodesai rodesai May 14, 2020 •
edited
Loading

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.

@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
TS: 3, K: X, V: A
TS: 2, K: X, V: B

would result in the table containing K: X, V: B, which is wrong.

@cadonna cadonna May 14, 2020 •
edited
Loading

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 when the timestamp of the newer value is lower, 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.

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.

kafka/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableSource.java

Line 122 in 7624e62

store.put(key, ValueAndTimestamp.make(value, context().timestamp()));
). The only thing that changed is that if the serialized value of the new record is equal to the serialized value of the old value and the timestamp of the new record is equal or newer, we drop the record because it is a idempotent update.
Could you elaborate on why a store should get corrupted because of this?

would result in the table containing K: X, V: B, which is wrong.

As said above, this behavior should not have been changed.

@cadonna cadonna May 14, 2020 •
edited
Loading

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.

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?

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 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;
}
}
}
19 changes: 19 additions & 0 deletions ...s/src/main/java/org/apache/kafka/streams/state/internals/ValueAndTimestampSerializer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,25 @@ public class ValueAndTimestampSerializer<V> implements Serializer<ValueAndTimest
timestampSerializer = new LongSerializer();
}

public static boolean maskTimestampAndCompareValues(final byte[] left, final byte[] right) {
Comment thread
vvcephei marked this conversation as resolved.
Outdated

@vvcephei vvcephei May 11, 2020

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.

Ah, upon reading that last test, I realized that I previously overlooked when you added the timestamp comparison to this method. We should change the method name for maintainability. It no longer just "masks the timestamp and compares the values". Can we instead call it "compareValuesAndCheckForIncreasingTimestamp" or something? I can almost guarantee that one or more people will be badly misled by the current method name.

// adapted from Arrays.equals

@cadonna cadonna Apr 7, 2020

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.

prop: Delete this comment since it is not very useful.

if (left == right)
return true;
if (left == null || right == null)
return false;

final int length = left.length;
if (right.length != length)
return false;

// skip the timestamp when comparing just the values
for (int i = Long.BYTES; i < length; i++)
if (left[i] != right[i])
return false;

@cadonna cadonna Apr 7, 2020

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.

prop: Please use braces to delimit the scopes of the loop and the ifs.


return true;

@cadonna cadonna Apr 7, 2020

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.

prop: Instead of adding the comment, factor out this code to a method named skipTimestampAndCompareValues().

}

@Override
public void configure(final Map<String, ?> configs,
final boolean isKey) {
Expand Down
38 changes: 38 additions & 0 deletions streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableSourceTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,8 +16,10 @@
*/
package org.apache.kafka.streams.kstream.internals;

import org.apache.kafka.common.serialization.IntegerDeserializer;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.streams.KeyValueTimestamp;
import org.apache.kafka.streams.StreamsBuilder;
Expand All @@ -32,7 +34,9 @@
import org.apache.kafka.streams.processor.internals.InternalTopologyBuilder;
import org.apache.kafka.streams.processor.internals.testutil.LogCaptureAppender;
import org.apache.kafka.streams.state.ValueAndTimestamp;
import org.apache.kafka.streams.test.TestRecord;
import org.apache.kafka.streams.TestInputTopic;
import org.apache.kafka.streams.TestOutputTopic;
import org.apache.kafka.test.MockProcessor;
import org.apache.kafka.test.MockProcessorSupplier;
import org.apache.kafka.test.StreamsTestUtils;
Expand Down Expand Up @@ -85,6 +89,40 @@ public void testKTable() {
supplier.theCapturedProcessor().processed);
}

@Test
public void testKTableSourceEmitOnChange() {
Comment on lines +95 to +96

@cadonna cadonna Apr 7, 2020

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.

Q: Why do you need the TopologyTestDriver here? I see that the other test use it. I guess you could simply instantiate a KTableSource, get the processor from it, and test directly the processor.

@ConcurrencyPractitioner ConcurrencyPractitioner Apr 11, 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 think that we can just follow the previous test format, so I think there's not too much need to change it at the moment.

final StreamsBuilder builder = new StreamsBuilder();
final String topic1 = "topic1";

builder.table(topic1, Consumed.with(Serdes.String(), Serdes.Integer()), Materialized.as("store"))
.toStream()
.to("output");

try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {
final TestInputTopic<String, Integer> inputTopic =
driver.createInputTopic(topic1, new StringSerializer(), new IntegerSerializer());
final TestOutputTopic<String, Integer> outputTopic =
driver.createOutputTopic("output", new StringDeserializer(), new IntegerDeserializer());

inputTopic.pipeInput("A", 1, 10L);
inputTopic.pipeInput("B", 2, 11L);
inputTopic.pipeInput("A", 1, 12L);
inputTopic.pipeInput("B", 3, 13L);

assertEquals(
1.0,
getMetricByName(driver.metrics(), "idempotent-update-skip-total", "stream-processor-node-metrics").metricValue()
);

assertEquals(
asList(new TestRecord<>("A", 1, Instant.ofEpochMilli(10L)),
new TestRecord<>("B", 2, Instant.ofEpochMilli(11L)),
new TestRecord<>("B", 3, Instant.ofEpochMilli(13L))),
outputTopic.readRecordsToList()
);
}
}
Comment thread
ConcurrencyPractitioner marked this conversation as resolved.

@Test
public void kTableShouldLogAndMeterOnSkippedRecordsWithBuiltInMetrics0100To24() {
kTableShouldLogAndMeterOnSkippedRecords(StreamsConfig.METRICS_0100_TO_24);
Expand Down