Skip to content

KAFKA-8065: restore original input record timestamp in forward() #6393

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 1 commit into from
Mar 9, 2019
Merged

KAFKA-8065: restore original input record timestamp in forward() #6393

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
12 changes: 8 additions & 4 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 @@ -146,12 +146,15 @@ public <K, V> void forward(final K key,
public <K, V> void forward(final K key,
final V value,
final To to) {
toInternal.update(to);
if (toInternal.hasTimestamp()) {
recordContext.setTimestamp(toInternal.timestamp());
}
final ProcessorNode previousNode = currentNode();
final long currentTimestamp = recordContext.timestamp;

try {
toInternal.update(to);
if (toInternal.hasTimestamp()) {
recordContext.setTimestamp(toInternal.timestamp());
}

final String sendTo = toInternal.child();
if (sendTo == null) {
final List<ProcessorNode<K, V>> children = (List<ProcessorNode<K, V>>) currentNode().children();
Expand All @@ -167,6 +170,7 @@ public <K, V> void forward(final K key,
forward(child, key, value);
}
} finally {
recordContext.timestamp = currentTimestamp;

@vvcephei vvcephei Mar 8, 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.

Should we consider saving/restoring the whole record context, just in case downstream processors have changed other parts of the context, not just the timestamp?

@mjsax mjsax Mar 8, 2019

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

Headers work differently. Users would mutate the input record header so they understand that the header changes.

We can still address it (I agree that the current header API is not great) but I would exclude it from this PR. Feel free to create a ticket for it.

@vvcephei vvcephei Mar 8, 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.

Ah, this is a good point. I was thinking that downstream processors were directly manipulating the record context, but if they do that, it's clearly their responsibility to restore it when they're done.

I think I won't bother with a ticket right now, it seems like the mutable-headers thing is actually just part of the key and value also being mutable. Plus, there's a bunch of other stuff that's not well defined about header handling in Streams. It seems like altogether too much to scope in a ticket without investing some serious design work up front.

I think the current change is fine as-is. Thanks!

setCurrentNode(previousNode);
}
}
Expand Down
55 changes: 55 additions & 0 deletions ...ams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -348,6 +348,32 @@ public void shouldConsiderModifiedTimeStamps() {
assertNextOutputRecord(OUTPUT_TOPIC_1, "key3", "value3", partition, 40L);
}

@Test
public void shouldConsiderModifiedTimeStampsForMultipleProcessors() {
final int partition = 10;
driver = new TopologyTestDriver(createMultiProcessorTimestampTopology(partition), props);

driver.pipeInput(recordFactory.create(INPUT_TOPIC_1, "key1", "value1", 10L));
assertNextOutputRecord(OUTPUT_TOPIC_1, "key1", "value1", partition, 10L);
assertNextOutputRecord(OUTPUT_TOPIC_2, "key1", "value1", partition, 20L);
assertNextOutputRecord(OUTPUT_TOPIC_1, "key1", "value1", partition, 15L);

@mjsax mjsax Mar 7, 2019 •
edited
Loading

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

Without the fix, this fails, because output record ts is 25.

And the following timestamps below would be: 35, 37, 47 (it's accumulating also for "deep" modifications)

assertNextOutputRecord(OUTPUT_TOPIC_2, "key1", "value1", partition, 20L);
assertNextOutputRecord(OUTPUT_TOPIC_1, "key1", "value1", partition, 12L);
assertNextOutputRecord(OUTPUT_TOPIC_2, "key1", "value1", partition, 22L);
assertNoOutputRecord(OUTPUT_TOPIC_1);
assertNoOutputRecord(OUTPUT_TOPIC_2);

driver.pipeInput(recordFactory.create(INPUT_TOPIC_1, "key2", "value2", 20L));
assertNextOutputRecord(OUTPUT_TOPIC_1, "key2", "value2", partition, 20L);
assertNextOutputRecord(OUTPUT_TOPIC_2, "key2", "value2", partition, 30L);
assertNextOutputRecord(OUTPUT_TOPIC_1, "key2", "value2", partition, 25L);
assertNextOutputRecord(OUTPUT_TOPIC_2, "key2", "value2", partition, 30L);
assertNextOutputRecord(OUTPUT_TOPIC_1, "key2", "value2", partition, 22L);
assertNextOutputRecord(OUTPUT_TOPIC_2, "key2", "value2", partition, 32L);
assertNoOutputRecord(OUTPUT_TOPIC_1);
assertNoOutputRecord(OUTPUT_TOPIC_2);
}

@Test
public void shouldConsiderHeaders() {
final int partition = 10;
Expand Down Expand Up @@ -489,6 +515,16 @@ private Topology createTimestampTopology(final int partition) {
.addSink("sink", OUTPUT_TOPIC_1, constantPartitioner(partition), "processor");
}

private Topology createMultiProcessorTimestampTopology(final int partition) {
return topology
.addSource("source", STRING_DESERIALIZER, STRING_DESERIALIZER, INPUT_TOPIC_1)
.addProcessor("processor", define(new FanOutTimestampProcessor("child1", "child2")), "source")
.addProcessor("child1", define(new ForwardingProcessor()), "processor")
.addProcessor("child2", define(new TimestampProcessor()), "processor")
.addSink("sink1", OUTPUT_TOPIC_1, constantPartitioner(partition), "child1")
.addSink("sink2", OUTPUT_TOPIC_2, constantPartitioner(partition), "child2");
}

private Topology createMultiplexingTopology() {
return topology
.addSource("source", STRING_DESERIALIZER, STRING_DESERIALIZER, INPUT_TOPIC_1)
Expand Down Expand Up @@ -582,6 +618,25 @@ public void process(final String key, final String value) {
}
}

protected static class FanOutTimestampProcessor extends AbstractProcessor<String, String> {
private final String firstChild;
private final String secondChild;

FanOutTimestampProcessor(final String firstChild,
final String secondChild) {
this.firstChild = firstChild;
this.secondChild = secondChild;
}

@Override
public void process(final String key, final String value) {
context().forward(key, value);
context().forward(key, value, To.child(firstChild).withTimestamp(context().timestamp() + 5));
context().forward(key, value, To.child(secondChild));
context().forward(key, value, To.all().withTimestamp(context().timestamp() + 2));
}
}

protected static class AddHeaderProcessor extends AbstractProcessor<String, String> {
@Override
public void process(final String key, final String value) {
Expand Down