You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2021/02/21 07:17:37 UTC
[GitHub] [kafka] guozhangwang opened a new pull request #10170: KAFKA-12323: Set timestamp in record context when punctuate
guozhangwang opened a new pull request #10170:
URL: https://github.com/apache/kafka/pull/10170
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on a change in pull request #10170: KAFKA-12323: Set timestamp in record context when punctuate
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on a change in pull request #10170:
URL: https://github.com/apache/kafka/pull/10170#discussion_r580793422
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
##########
@@ -792,7 +790,16 @@ public void punctuate(final ProcessorNode<?, ?, ?, ?> node,
throw new IllegalStateException(String.format("%sCurrent node is not null", logPrefix));
}
- updateProcessorContext(node, time.milliseconds(), null);
+ // when punctuating, we need to preserve the timestamp (this can be either system time or event time)
+ // while other record context are set as dummy: null topic, -1 partition, -1 offset and empty header
+ final ProcessorRecordContext recordContext = new ProcessorRecordContext(
+ timestamp,
+ -1L,
+ -1,
+ null,
+ new RecordHeaders()
Review comment:
I'm following the existing behavior here: if the record context is `null`, we also return a `return new RecordHeaders();`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on pull request #10170: KAFKA-12323: Set timestamp in record context when punctuate
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on pull request #10170:
URL: https://github.com/apache/kafka/pull/10170#issuecomment-784755502
Cherry-picked to 2.8 / 2.7, cc RM @vvcephei
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on pull request #10170: KAFKA-12323: Set timestamp in record context when punctuate
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on pull request #10170:
URL: https://github.com/apache/kafka/pull/10170#issuecomment-782812913
@ableegoldman @vvcephei for reviews, thanks!
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mjsax commented on a change in pull request #10170: KAFKA-12323: Set timestamp in record context when punctuate
Posted by GitBox <gi...@apache.org>.
mjsax commented on a change in pull request #10170:
URL: https://github.com/apache/kafka/pull/10170#discussion_r580546373
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
##########
@@ -1874,6 +1873,66 @@ public void close() {}
assertEquals(2, punctuatedWallClockTime.size());
}
+ @Test
+ public void shouldPunctuateWithTimestampPreservedInProcessorContext() {
Review comment:
I think we should add one more test for `STREAM_TIME` punctuation?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mjsax commented on a change in pull request #10170: KAFKA-12323: Set timestamp in record context when punctuate
Posted by GitBox <gi...@apache.org>.
mjsax commented on a change in pull request #10170:
URL: https://github.com/apache/kafka/pull/10170#discussion_r582202700
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
##########
@@ -1874,6 +1873,85 @@ public void close() {}
assertEquals(2, punctuatedWallClockTime.size());
}
+ @Test
+ public void shouldPunctuateWithTimestampPreservedInProcessorContext() {
+ final org.apache.kafka.streams.kstream.TransformerSupplier<Object, Object, KeyValue<Object, Object>> punctuateProcessor =
+ () -> new org.apache.kafka.streams.kstream.Transformer<Object, Object, KeyValue<Object, Object>>() {
+ @Override
+ public void init(final org.apache.kafka.streams.processor.ProcessorContext context) {
+ context.schedule(Duration.ofMillis(100L), PunctuationType.WALL_CLOCK_TIME, timestamp -> context.forward("key", "value"));
+ context.schedule(Duration.ofMillis(100L), PunctuationType.STREAM_TIME, timestamp -> context.forward("key", "value"));
+ }
+
+ @Override
+ public KeyValue<Object, Object> transform(final Object key, final Object value) {
+ return null;
+ }
+
+ @Override
+ public void close() {}
+ };
+
+ final List<Long> peekedContextTime = new ArrayList<>();
+ final org.apache.kafka.streams.processor.ProcessorSupplier<Object, Object> peekProcessor =
+ () -> new org.apache.kafka.streams.processor.AbstractProcessor<Object, Object>() {
+ @Override
+ public void process(final Object key, final Object value) {
+ peekedContextTime.add(context.timestamp());
+ }
+ };
+
+ internalStreamsBuilder.stream(Collections.singleton(topic1), consumed)
+ .transform(punctuateProcessor)
+ .process(peekProcessor);
+ internalStreamsBuilder.buildAndOptimizeTopology();
+
+ final long currTime = mockTime.milliseconds();
+ final StreamThread thread = createStreamThread(CLIENT_ID, config, false);
+
+ thread.setState(StreamThread.State.STARTING);
+ thread.rebalanceListener().onPartitionsRevoked(Collections.emptySet());
+ final List<TopicPartition> assignedPartitions = new ArrayList<>();
+
+ final Map<TaskId, Set<TopicPartition>> activeTasks = new HashMap<>();
+
+ // assign single partition
+ assignedPartitions.add(t1p1);
+ activeTasks.put(task1, Collections.singleton(t1p1));
+
+ thread.taskManager().handleAssignment(activeTasks, emptyMap());
+
+ clientSupplier.consumer.assign(assignedPartitions);
+ clientSupplier.consumer.updateBeginningOffsets(Collections.singletonMap(t1p1, 0L));
+ thread.rebalanceListener().onPartitionsAssigned(assignedPartitions);
+
+ thread.runOnce();
+ assertEquals(0, peekedContextTime.size());
+
+ mockTime.sleep(100L);
+ thread.runOnce();
+
+ assertEquals(1, peekedContextTime.size());
+ assertEquals(currTime + 100L, peekedContextTime.get(0).longValue());
+
+ clientSupplier.consumer.addRecord(new ConsumerRecord<>(
+ topic1,
+ 1,
+ 0L,
+ 100L,
+ TimestampType.CREATE_TIME,
+ ConsumerRecord.NULL_CHECKSUM,
+ "K".getBytes().length,
+ "V".getBytes().length,
+ "K".getBytes(),
+ "V".getBytes()));
+
+ thread.runOnce();
+
+ assertEquals(2, peekedContextTime.size());
+ assertEquals(0L, peekedContextTime.get(1).longValue());
Review comment:
So to make sure we actually use "stream-time" we should change the test to actually punctuation twice? Would you mind doing a follow-up PR?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang merged pull request #10170: KAFKA-12323: Set timestamp in record context when punctuate
Posted by GitBox <gi...@apache.org>.
guozhangwang merged pull request #10170:
URL: https://github.com/apache/kafka/pull/10170
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on a change in pull request #10170: KAFKA-12323: Set timestamp in record context when punctuate
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on a change in pull request #10170:
URL: https://github.com/apache/kafka/pull/10170#discussion_r581618213
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
##########
@@ -1874,6 +1873,85 @@ public void close() {}
assertEquals(2, punctuatedWallClockTime.size());
}
+ @Test
+ public void shouldPunctuateWithTimestampPreservedInProcessorContext() {
+ final org.apache.kafka.streams.kstream.TransformerSupplier<Object, Object, KeyValue<Object, Object>> punctuateProcessor =
+ () -> new org.apache.kafka.streams.kstream.Transformer<Object, Object, KeyValue<Object, Object>>() {
+ @Override
+ public void init(final org.apache.kafka.streams.processor.ProcessorContext context) {
+ context.schedule(Duration.ofMillis(100L), PunctuationType.WALL_CLOCK_TIME, timestamp -> context.forward("key", "value"));
+ context.schedule(Duration.ofMillis(100L), PunctuationType.STREAM_TIME, timestamp -> context.forward("key", "value"));
+ }
+
+ @Override
+ public KeyValue<Object, Object> transform(final Object key, final Object value) {
+ return null;
+ }
+
+ @Override
+ public void close() {}
+ };
+
+ final List<Long> peekedContextTime = new ArrayList<>();
+ final org.apache.kafka.streams.processor.ProcessorSupplier<Object, Object> peekProcessor =
+ () -> new org.apache.kafka.streams.processor.AbstractProcessor<Object, Object>() {
+ @Override
+ public void process(final Object key, final Object value) {
+ peekedContextTime.add(context.timestamp());
+ }
+ };
+
+ internalStreamsBuilder.stream(Collections.singleton(topic1), consumed)
+ .transform(punctuateProcessor)
+ .process(peekProcessor);
+ internalStreamsBuilder.buildAndOptimizeTopology();
+
+ final long currTime = mockTime.milliseconds();
+ final StreamThread thread = createStreamThread(CLIENT_ID, config, false);
+
+ thread.setState(StreamThread.State.STARTING);
+ thread.rebalanceListener().onPartitionsRevoked(Collections.emptySet());
+ final List<TopicPartition> assignedPartitions = new ArrayList<>();
+
+ final Map<TaskId, Set<TopicPartition>> activeTasks = new HashMap<>();
+
+ // assign single partition
+ assignedPartitions.add(t1p1);
+ activeTasks.put(task1, Collections.singleton(t1p1));
+
+ thread.taskManager().handleAssignment(activeTasks, emptyMap());
+
+ clientSupplier.consumer.assign(assignedPartitions);
+ clientSupplier.consumer.updateBeginningOffsets(Collections.singletonMap(t1p1, 0L));
+ thread.rebalanceListener().onPartitionsAssigned(assignedPartitions);
+
+ thread.runOnce();
+ assertEquals(0, peekedContextTime.size());
+
+ mockTime.sleep(100L);
+ thread.runOnce();
+
+ assertEquals(1, peekedContextTime.size());
+ assertEquals(currTime + 100L, peekedContextTime.get(0).longValue());
+
+ clientSupplier.consumer.addRecord(new ConsumerRecord<>(
+ topic1,
+ 1,
+ 0L,
+ 100L,
+ TimestampType.CREATE_TIME,
+ ConsumerRecord.NULL_CHECKSUM,
+ "K".getBytes().length,
+ "V".getBytes().length,
+ "K".getBytes(),
+ "V".getBytes()));
+
+ thread.runOnce();
+
+ assertEquals(2, peekedContextTime.size());
+ assertEquals(0L, peekedContextTime.get(1).longValue());
Review comment:
It is not actually.. when we schedule at init we always add the scheduled event with the current timestamp
```
case STREAM_TIME:
// align punctuation to 0L, punctuate as soon as we have data
return schedule(0L, interval, type, punctuator);
case WALL_CLOCK_TIME:
// align punctuation to now, punctuate after interval has elapsed
return schedule(time.milliseconds() + interval, interval, type, punctuator);
```
hence for stream time punctuation, the first one always triggers right when we received the first record.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mjsax commented on a change in pull request #10170: KAFKA-12323: Set timestamp in record context when punctuate
Posted by GitBox <gi...@apache.org>.
mjsax commented on a change in pull request #10170:
URL: https://github.com/apache/kafka/pull/10170#discussion_r580544814
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
##########
@@ -792,7 +790,16 @@ public void punctuate(final ProcessorNode<?, ?, ?, ?> node,
throw new IllegalStateException(String.format("%sCurrent node is not null", logPrefix));
}
- updateProcessorContext(node, time.milliseconds(), null);
+ // when punctuating, we need to preserve the timestamp (this can be either system time or event time)
+ // while other record context are set as dummy: null topic, -1 partition, -1 offset and empty header
+ final ProcessorRecordContext recordContext = new ProcessorRecordContext(
+ timestamp,
+ -1L,
+ -1,
+ null,
+ new RecordHeaders()
Review comment:
Why not pass `null` ?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on a change in pull request #10170: KAFKA-12323: Set timestamp in record context when punctuate
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on a change in pull request #10170:
URL: https://github.com/apache/kafka/pull/10170#discussion_r581615084
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
##########
@@ -1805,18 +1806,16 @@ public void shouldPunctuateActiveTask() {
final List<Long> punctuatedStreamTime = new ArrayList<>();
final List<Long> punctuatedWallClockTime = new ArrayList<>();
final org.apache.kafka.streams.processor.ProcessorSupplier<Object, Object> punctuateProcessor =
- () -> new org.apache.kafka.streams.processor.Processor<Object, Object>() {
+ () -> new org.apache.kafka.streams.processor.AbstractProcessor<Object, Object>() {
@Override
public void init(final org.apache.kafka.streams.processor.ProcessorContext context) {
context.schedule(Duration.ofMillis(100L), PunctuationType.STREAM_TIME, punctuatedStreamTime::add);
context.schedule(Duration.ofMillis(100L), PunctuationType.WALL_CLOCK_TIME, punctuatedWallClockTime::add);
+ context.schedule(Duration.ofMillis(200L), PunctuationType.WALL_CLOCK_TIME, timestamp -> context.forward("key", "value"));
Review comment:
No it is not intentional, removed.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mjsax commented on a change in pull request #10170: KAFKA-12323: Set timestamp in record context when punctuate
Posted by GitBox <gi...@apache.org>.
mjsax commented on a change in pull request #10170:
URL: https://github.com/apache/kafka/pull/10170#discussion_r580546901
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
##########
@@ -1874,6 +1873,66 @@ public void close() {}
assertEquals(2, punctuatedWallClockTime.size());
}
+ @Test
+ public void shouldPunctuateWithTimestampPreservedInProcessorContext() {
+ final org.apache.kafka.streams.kstream.TransformerSupplier<Object, Object, KeyValue<Object, Object>> punctuateProcessor =
+ () -> new org.apache.kafka.streams.kstream.Transformer<Object, Object, KeyValue<Object, Object>>() {
+ @Override
+ public void init(final org.apache.kafka.streams.processor.ProcessorContext context) {
+ context.schedule(Duration.ofMillis(100L), PunctuationType.WALL_CLOCK_TIME, timestamp -> context.forward("key", "value"));
+ }
+
+ @Override
+ public KeyValue<Object, Object> transform(final Object key, final Object value) {
+ return null;
+ }
+
+ @Override
+ public void close() {}
+ };
+
+ final List<Long> peekedContextTime = new ArrayList<>();
+ final org.apache.kafka.streams.processor.ProcessorSupplier<Object, Object> peekProcessor =
+ () -> new org.apache.kafka.streams.processor.AbstractProcessor<Object, Object>() {
+ @Override
+ public void process(final Object key, final Object value) {
+ peekedContextTime.add(context.timestamp());
+ }
+ };
+
+ internalStreamsBuilder.stream(Collections.singleton(topic1), consumed)
+ .transform(punctuateProcessor)
+ .process(peekProcessor);
+ internalStreamsBuilder.buildAndOptimizeTopology();
+
+ final StreamThread thread = createStreamThread(CLIENT_ID, config, false);
+
+ thread.setState(StreamThread.State.STARTING);
+ thread.rebalanceListener().onPartitionsRevoked(Collections.emptySet());
+ final List<TopicPartition> assignedPartitions = new ArrayList<>();
+
+ final Map<TaskId, Set<TopicPartition>> activeTasks = new HashMap<>();
+
+ // assign single partition
+ assignedPartitions.add(t1p1);
+ activeTasks.put(task1, Collections.singleton(t1p1));
+
+ thread.taskManager().handleAssignment(activeTasks, emptyMap());
+
+ clientSupplier.consumer.assign(assignedPartitions);
+ clientSupplier.consumer.updateBeginningOffsets(Collections.singletonMap(t1p1, 0L));
+ thread.rebalanceListener().onPartitionsAssigned(assignedPartitions);
+
+ thread.runOnce();
+ assertEquals(0, peekedContextTime.size());
+
+ mockTime.sleep(100L);
+ thread.runOnce();
+
+ assertEquals(1, peekedContextTime.size());
+ assertNotNull(peekedContextTime.get(0));
Review comment:
We should verify the actual timestamp.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mjsax commented on a change in pull request #10170: KAFKA-12323: Set timestamp in record context when punctuate
Posted by GitBox <gi...@apache.org>.
mjsax commented on a change in pull request #10170:
URL: https://github.com/apache/kafka/pull/10170#discussion_r582202700
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
##########
@@ -1874,6 +1873,85 @@ public void close() {}
assertEquals(2, punctuatedWallClockTime.size());
}
+ @Test
+ public void shouldPunctuateWithTimestampPreservedInProcessorContext() {
+ final org.apache.kafka.streams.kstream.TransformerSupplier<Object, Object, KeyValue<Object, Object>> punctuateProcessor =
+ () -> new org.apache.kafka.streams.kstream.Transformer<Object, Object, KeyValue<Object, Object>>() {
+ @Override
+ public void init(final org.apache.kafka.streams.processor.ProcessorContext context) {
+ context.schedule(Duration.ofMillis(100L), PunctuationType.WALL_CLOCK_TIME, timestamp -> context.forward("key", "value"));
+ context.schedule(Duration.ofMillis(100L), PunctuationType.STREAM_TIME, timestamp -> context.forward("key", "value"));
+ }
+
+ @Override
+ public KeyValue<Object, Object> transform(final Object key, final Object value) {
+ return null;
+ }
+
+ @Override
+ public void close() {}
+ };
+
+ final List<Long> peekedContextTime = new ArrayList<>();
+ final org.apache.kafka.streams.processor.ProcessorSupplier<Object, Object> peekProcessor =
+ () -> new org.apache.kafka.streams.processor.AbstractProcessor<Object, Object>() {
+ @Override
+ public void process(final Object key, final Object value) {
+ peekedContextTime.add(context.timestamp());
+ }
+ };
+
+ internalStreamsBuilder.stream(Collections.singleton(topic1), consumed)
+ .transform(punctuateProcessor)
+ .process(peekProcessor);
+ internalStreamsBuilder.buildAndOptimizeTopology();
+
+ final long currTime = mockTime.milliseconds();
+ final StreamThread thread = createStreamThread(CLIENT_ID, config, false);
+
+ thread.setState(StreamThread.State.STARTING);
+ thread.rebalanceListener().onPartitionsRevoked(Collections.emptySet());
+ final List<TopicPartition> assignedPartitions = new ArrayList<>();
+
+ final Map<TaskId, Set<TopicPartition>> activeTasks = new HashMap<>();
+
+ // assign single partition
+ assignedPartitions.add(t1p1);
+ activeTasks.put(task1, Collections.singleton(t1p1));
+
+ thread.taskManager().handleAssignment(activeTasks, emptyMap());
+
+ clientSupplier.consumer.assign(assignedPartitions);
+ clientSupplier.consumer.updateBeginningOffsets(Collections.singletonMap(t1p1, 0L));
+ thread.rebalanceListener().onPartitionsAssigned(assignedPartitions);
+
+ thread.runOnce();
+ assertEquals(0, peekedContextTime.size());
+
+ mockTime.sleep(100L);
+ thread.runOnce();
+
+ assertEquals(1, peekedContextTime.size());
+ assertEquals(currTime + 100L, peekedContextTime.get(0).longValue());
+
+ clientSupplier.consumer.addRecord(new ConsumerRecord<>(
+ topic1,
+ 1,
+ 0L,
+ 100L,
+ TimestampType.CREATE_TIME,
+ ConsumerRecord.NULL_CHECKSUM,
+ "K".getBytes().length,
+ "V".getBytes().length,
+ "K".getBytes(),
+ "V".getBytes()));
+
+ thread.runOnce();
+
+ assertEquals(2, peekedContextTime.size());
+ assertEquals(0L, peekedContextTime.get(1).longValue());
Review comment:
So to make sure we actually use "stream-time" we should change the test to actually punctuation twice? Would you mind doing a follow-up PR? Might also be worth not use a different timestamp compare to "wall-clock time" to make sure we don't by accident pass in current wall-clock.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mjsax commented on a change in pull request #10170: KAFKA-12323: Set timestamp in record context when punctuate
Posted by GitBox <gi...@apache.org>.
mjsax commented on a change in pull request #10170:
URL: https://github.com/apache/kafka/pull/10170#discussion_r581324176
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
##########
@@ -1805,18 +1806,16 @@ public void shouldPunctuateActiveTask() {
final List<Long> punctuatedStreamTime = new ArrayList<>();
final List<Long> punctuatedWallClockTime = new ArrayList<>();
final org.apache.kafka.streams.processor.ProcessorSupplier<Object, Object> punctuateProcessor =
- () -> new org.apache.kafka.streams.processor.Processor<Object, Object>() {
+ () -> new org.apache.kafka.streams.processor.AbstractProcessor<Object, Object>() {
@Override
public void init(final org.apache.kafka.streams.processor.ProcessorContext context) {
context.schedule(Duration.ofMillis(100L), PunctuationType.STREAM_TIME, punctuatedStreamTime::add);
context.schedule(Duration.ofMillis(100L), PunctuationType.WALL_CLOCK_TIME, punctuatedWallClockTime::add);
+ context.schedule(Duration.ofMillis(200L), PunctuationType.WALL_CLOCK_TIME, timestamp -> context.forward("key", "value"));
Review comment:
What is the purpose of adding this line? Seem this case is covered in the newly added test below?
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
##########
@@ -1874,6 +1873,85 @@ public void close() {}
assertEquals(2, punctuatedWallClockTime.size());
}
+ @Test
+ public void shouldPunctuateWithTimestampPreservedInProcessorContext() {
+ final org.apache.kafka.streams.kstream.TransformerSupplier<Object, Object, KeyValue<Object, Object>> punctuateProcessor =
+ () -> new org.apache.kafka.streams.kstream.Transformer<Object, Object, KeyValue<Object, Object>>() {
+ @Override
+ public void init(final org.apache.kafka.streams.processor.ProcessorContext context) {
+ context.schedule(Duration.ofMillis(100L), PunctuationType.WALL_CLOCK_TIME, timestamp -> context.forward("key", "value"));
+ context.schedule(Duration.ofMillis(100L), PunctuationType.STREAM_TIME, timestamp -> context.forward("key", "value"));
+ }
+
+ @Override
+ public KeyValue<Object, Object> transform(final Object key, final Object value) {
+ return null;
+ }
+
+ @Override
+ public void close() {}
+ };
+
+ final List<Long> peekedContextTime = new ArrayList<>();
+ final org.apache.kafka.streams.processor.ProcessorSupplier<Object, Object> peekProcessor =
+ () -> new org.apache.kafka.streams.processor.AbstractProcessor<Object, Object>() {
+ @Override
+ public void process(final Object key, final Object value) {
+ peekedContextTime.add(context.timestamp());
+ }
+ };
+
+ internalStreamsBuilder.stream(Collections.singleton(topic1), consumed)
+ .transform(punctuateProcessor)
+ .process(peekProcessor);
+ internalStreamsBuilder.buildAndOptimizeTopology();
+
+ final long currTime = mockTime.milliseconds();
+ final StreamThread thread = createStreamThread(CLIENT_ID, config, false);
+
+ thread.setState(StreamThread.State.STARTING);
+ thread.rebalanceListener().onPartitionsRevoked(Collections.emptySet());
+ final List<TopicPartition> assignedPartitions = new ArrayList<>();
+
+ final Map<TaskId, Set<TopicPartition>> activeTasks = new HashMap<>();
+
+ // assign single partition
+ assignedPartitions.add(t1p1);
+ activeTasks.put(task1, Collections.singleton(t1p1));
+
+ thread.taskManager().handleAssignment(activeTasks, emptyMap());
+
+ clientSupplier.consumer.assign(assignedPartitions);
+ clientSupplier.consumer.updateBeginningOffsets(Collections.singletonMap(t1p1, 0L));
+ thread.rebalanceListener().onPartitionsAssigned(assignedPartitions);
+
+ thread.runOnce();
+ assertEquals(0, peekedContextTime.size());
+
+ mockTime.sleep(100L);
+ thread.runOnce();
+
+ assertEquals(1, peekedContextTime.size());
+ assertEquals(currTime + 100L, peekedContextTime.get(0).longValue());
+
+ clientSupplier.consumer.addRecord(new ConsumerRecord<>(
+ topic1,
+ 1,
+ 0L,
+ 100L,
+ TimestampType.CREATE_TIME,
+ ConsumerRecord.NULL_CHECKSUM,
+ "K".getBytes().length,
+ "V".getBytes().length,
+ "K".getBytes(),
+ "V".getBytes()));
+
+ thread.runOnce();
+
+ assertEquals(2, peekedContextTime.size());
+ assertEquals(0L, peekedContextTime.get(1).longValue());
Review comment:
Why is the expected value `0L` ? Should it not be `100L` (ie, the input record's timestamp that dictates stream-time)?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org