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 2020/12/13 22:33:16 UTC
[GitHub] [kafka] rohitrmd opened a new pull request #9744: Kafka 10062 - Add a method to retrieve the current timestamp as known by the Streams app
rohitrmd opened a new pull request #9744:
URL: https://github.com/apache/kafka/pull/9744
KIP 622: https://cwiki.apache.org/confluence/display/KAFKA/KIP-622%3A+Add+currentSystemTimeMs+and+currentStreamTimeMs+to+ProcessorContext
jira: https://issues.apache.org/jira/browse/KAFKA-10062
### 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] rohitrmd commented on a change in pull request #9744: KAFKA-10062: Add a method to retrieve the current timestamp as known by the Streams app
Posted by GitBox <gi...@apache.org>.
rohitrmd commented on a change in pull request #9744:
URL: https://github.com/apache/kafka/pull/9744#discussion_r551055293
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
##########
@@ -494,7 +494,9 @@ public void process(final Record<Integer, Integer> record) {
// e2e latency = 10
task.addRecords(partition1, singletonList(getConsumerRecord(0, 0L)));
- task.process(10L);
+ time = new MockTime(0L, 10L, 0L);
Review comment:
The reason I had to change StreamTaskTest is because of following issue:
StreamTask -> process(wallClockTime) method updates processor context's system time in this method: https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java#L685
AbstractProcessorContect has another systemTime field: https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractProcessorContext.java#L48 which is updated in streamTask's process method. I removed this field from AbstractProcessorContect as we want to fetch the time from StreamTask. But the latency is measured [here](https://github.com/apache/kafka/blob/b9dfc196aaee91f4fa23fe6563261c6f2d01ac33/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java#L284) which is streamTask's time filed now which is not updated.
I couldn't find any other way to change time field in StreamTask unless I add setTime method in it. If i don't change time in StreamTask object, subsequent metric tests fail as stream-time is not updated.
Do you have recommendation about how i can fix it without adding setTime method?
----------------------------------------------------------------
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 #9744: KAFKA-10062: Add a method to retrieve the current timestamp as known by the Streams app
Posted by GitBox <gi...@apache.org>.
mjsax commented on a change in pull request #9744:
URL: https://github.com/apache/kafka/pull/9744#discussion_r556206559
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
##########
@@ -494,7 +494,9 @@ public void process(final Record<Integer, Integer> record) {
// e2e latency = 10
task.addRecords(partition1, singletonList(getConsumerRecord(0, 0L)));
- task.process(10L);
+ time = new MockTime(0L, 10L, 0L);
Review comment:
In L159, we define a global mock time object:
```
private MockTime time = new MockTime();
```
that is passed into `StreamTask`. You can call `time.sleep(...)` to advance the mock 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
[GitHub] [kafka] mjsax commented on a change in pull request #9744: KAFKA-10062: Add a method to retrieve the current timestamp as known by the Streams app
Posted by GitBox <gi...@apache.org>.
mjsax commented on a change in pull request #9744:
URL: https://github.com/apache/kafka/pull/9744#discussion_r549547331
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/ForwardingDisabledProcessorContext.java
##########
@@ -166,4 +166,14 @@ public long timestamp() {
public Map<String, Object> appConfigsWithPrefix(final String prefix) {
return delegate.appConfigsWithPrefix(prefix);
}
+
+ @Override
+ public long currentSystemTimeMs() {
+ throw new UnsupportedOperationException("this method is not supported in ForwardingDisabledProcessor context");
Review comment:
Why do we throw here? It seems to be safe to get the time from `delegate` object? We only disable `forward` but nothing else. (same below)
----------------------------------------------------------------
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 #9744: KAFKA-10062: Add a method to retrieve the current timestamp as known by the Streams app
Posted by GitBox <gi...@apache.org>.
mjsax commented on a change in pull request #9744:
URL: https://github.com/apache/kafka/pull/9744#discussion_r570716520
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractProcessorContext.java
##########
@@ -45,7 +45,6 @@
private boolean initialized;
protected ProcessorRecordContext recordContext;
protected ProcessorNode<?, ?, ?, ?> currentNode;
- private long currentSystemTimeMs;
Review comment:
The main motivation to add `ProcessorContext#currentSystemTime()` was to be able to return the mocked wall-clock time in `TopologyTestDriver`. Even if we return the cached time from `AbstractProcessorContext`, we will be able to return the mocked time, as we update the cached time based on the mocked time in `TopologyTestDriver`.
- `InternalMockProcessorContext` is just for our own unit testing -- it's fine to add the new `Time` field, it's not a public facing change anyway
- Originally we changed `ProcessorContextImpl` because we remove `AbstractProcessorContext#currentSystemTime()` -- as suggested by Guozhang, we should keep the cached time in `AbstractProcessorContext()`, and thus we don't need `ProcessorContextImpl#currentSystemTime()` any longer.
- `GlobalProcessorContextImpl` is a different code path, and thus the changes of this PR are fine
> If we add new cachedSystemTimeMs field in AbstractProcessorContext, when do you want to return this field?
Yes, we want to return this field. (We get this behaviro by adding back `AbstractProcessorContext#currentSystemTime()` (and the cached time in this class) and removing `ProcessorContextImpl#currentSystemTime()`.
> Are earlier changes not valid to return time from StreamTask?
Yes, I think we can revert all changes from `StreamTask`.
Does this make sense?
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractProcessorContext.java
##########
@@ -45,7 +45,6 @@
private boolean initialized;
protected ProcessorRecordContext recordContext;
protected ProcessorNode<?, ?, ?, ?> currentNode;
- private long currentSystemTimeMs;
Review comment:
The main motivation to add `ProcessorContext#currentSystemTime()` was to be able to return the mocked wall-clock time in `TopologyTestDriver`. Even if we return the cached time from `AbstractProcessorContext`, we will be able to return the mocked time, as we update the cached time based on the mocked time in `TopologyTestDriver`.
- `InternalMockProcessorContext` is just for our own unit testing -- it's fine to add the new `Time` field, it's not a public facing change anyway
- Originally we changed `ProcessorContextImpl` because we remove `AbstractProcessorContext#currentSystemTime()` -- as suggested by Guozhang, we should keep the cached time in `AbstractProcessorContext()`, and thus we don't need `ProcessorContextImpl#currentSystemTime()` any longer.
- `GlobalProcessorContextImpl` is a different code path, and thus the changes of this PR are fine
> If we add new cachedSystemTimeMs field in AbstractProcessorContext, when do you want to return this field?
Yes, we want to return this field. (We get this behavior by adding back `AbstractProcessorContext#currentSystemTime()` (and the cached time in this class) and removing `ProcessorContextImpl#currentSystemTime()`.
> Are earlier changes not valid to return time from StreamTask?
Yes, I think we can revert all changes from `StreamTask`.
Does this make sense?
----------------------------------------------------------------
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] rohitrmd commented on a change in pull request #9744: KAFKA-10062: Add a method to retrieve the current timestamp as known by the Streams app
Posted by GitBox <gi...@apache.org>.
rohitrmd commented on a change in pull request #9744:
URL: https://github.com/apache/kafka/pull/9744#discussion_r563247511
##########
File path: streams/src/test/java/org/apache/kafka/test/MockInternalProcessorContext.java
##########
@@ -55,14 +55,9 @@ public MockInternalProcessorContext(final Properties config, final TaskId taskId
super(config, taskId, stateDir);
}
- @Override
- public void setSystemTimeMs(long timeMs) {
Review comment:
@mjsax should i try to do this in another 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] mjsax commented on a change in pull request #9744: KAFKA-10062: Add a method to retrieve the current timestamp as known by the Streams app
Posted by GitBox <gi...@apache.org>.
mjsax commented on a change in pull request #9744:
URL: https://github.com/apache/kafka/pull/9744#discussion_r549546234
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/ProcessorContext.java
##########
@@ -289,4 +289,17 @@ Cancellable schedule(final Duration interval,
*/
Map<String, Object> appConfigsWithPrefix(final String prefix);
+ /**
+ * Returns current cached wall-clock system timestamp in milliseconds.
Review comment:
nit: `Return` without `s` -- we use imperative to write JavaDocs.
I would remove `cached` and add a dedicated second sentence about it. Also `wall-clock time` and `system time` are synonymous and thus `wall-clock system time` sounds a little odd.
What about:
```
Return the current system timestamp (also called wall-clock time) in milliseconds.
<p>
Note: this method returns the internally cached system timestamp from the Kafka Stream runtime. Thus, it may return a different value compared to `System.currentTimeMillis()`.
<p>
For a global processor, Kafka Streams does not cache system time and thus calling this method will return the same value as `System.currentTimeMillis()`.
@return the current system timestamp in milliseconds
```
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/ForwardingDisabledProcessorContext.java
##########
@@ -166,4 +166,14 @@ public long timestamp() {
public Map<String, Object> appConfigsWithPrefix(final String prefix) {
return delegate.appConfigsWithPrefix(prefix);
}
+
+ @Override
+ public long currentSystemTimeMs() {
+ throw new UnsupportedOperationException("this method is not supported in ForwardingDisabledProcessor context");
Review comment:
Why do we through here? It seems to be safe to get the time from `delegate` object? We only disable `forward` but nothing else. (same below)
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalProcessorContextImpl.java
##########
@@ -115,6 +116,16 @@ public void commit() {
//no-op
}
+ @Override
+ public long currentSystemTimeMs() {
+ return Time.SYSTEM.milliseconds();
+ }
+
+ @Override
+ public long currentStreamTimeMs() {
+ throw new UnsupportedOperationException("this method is not supported in global processor context.");
Review comment:
This makes sense, but we might want to document this somewhere else, ie, in the corresponding docs. To be fair, not sure atm, if we have much content about it and/or where it add it?
We should for sure document it in the JavaDocs, ie, the `ProcessorContext` interface. (cf my comment above)
About the error message:
```
throw new UnsupportedOperationException("There is no concept of stream-time for a global processor.");
```
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/ProcessorContext.java
##########
@@ -289,4 +289,17 @@ Cancellable schedule(final Duration interval,
*/
Map<String, Object> appConfigsWithPrefix(final String prefix);
+ /**
+ * Returns current cached wall-clock system timestamp in milliseconds.
+ *
+ * @return the current cached wall-clock system timestamp in milliseconds
+ */
+ long currentSystemTimeMs();
+
+ /**
+ * Returns the maximum timestamp of any record yet processed by the task.
Review comment:
What about:
```
Return the current stream-time in milliseconds.
<p>
Stream-time is the maximum observed {@link TimestampExtractor record timestamp} so far
(including the currently processed record), i.e., it can be considered a high-watermark.
Stream-time is tracked on a per-task basis and is preserved across restarts and during task migration.
<p>
Note: this method is not supported for global processors (cf. {@link Topology#addGlobalStore(...)}
and {@link StreamsBuilder#addGlobalStore(...)}, because there is no concept of stream-time for this case.
Calling this method in a global processor with result in an {@link UnsupportedOperationException}.
@return the current stream-time in milliseconds
```
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreToProcessorContextAdapter.java
##########
@@ -160,4 +160,14 @@ public long timestamp() {
public Map<String, Object> appConfigsWithPrefix(final String prefix) {
return delegate.appConfigsWithPrefix(prefix);
}
+
+ @Override
+ public long currentSystemTimeMs() {
+ throw new UnsupportedOperationException("this method is not supported in StoreToProcessorContextAdapter");
Review comment:
We should align the error message to the ones from above: `"StateStores can't access system time."` (same below).
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
##########
@@ -1111,13 +1111,21 @@ RecordCollector recordCollector() {
return recordCollector;
}
+ public long streamTime() {
Review comment:
Why is this `public` now? It seems it's only newly called in `ProcessorContextImpl` that is in the same package and thus it can stay (default) package-private.
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
##########
@@ -70,7 +70,8 @@
// visible for testing
static final byte LATEST_MAGIC_BYTE = 1;
- private final Time time;
+ // This time is wall clock time
Review comment:
Yes, `Time` is always wall-clock time -- comment seems unnecessary
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
##########
@@ -1111,13 +1111,21 @@ RecordCollector recordCollector() {
return recordCollector;
}
+ public long streamTime() {
+ return partitionGroup.streamTime();
+ }
+
+ public long currentSystemTimeMs() {
+ return time.milliseconds();
+ }
+
// below are visible for testing only
int numBuffered() {
return partitionGroup.numBuffered();
}
- long streamTime() {
- return partitionGroup.streamTime();
+ void setTime(final Time time) {
Review comment:
I don't think we need this method (cf. my other comment in the test code where it is used).
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorContextImplTest.java
##########
@@ -555,6 +559,12 @@ public void shouldThrowUnsupportedOperationExceptionOnRecordContext() {
);
}
+ @Test
+ public void shouldMatchSystemAndStreamTime() {
Review comment:
This should be two tests, one for each method.
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
##########
@@ -70,7 +70,8 @@
// visible for testing
static final byte LATEST_MAGIC_BYTE = 1;
- private final Time time;
+ // This time is wall clock time
+ private Time time;
Review comment:
I think we should keep this as `final` (compare my other comments).
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreToProcessorContextAdapterTest.java
##########
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.processor.internals;
+
+import org.apache.kafka.streams.processor.ProcessorContext;
+import org.apache.kafka.streams.processor.StateStoreContext;
+import org.easymock.EasyMockRunner;
+import org.easymock.Mock;
+import org.easymock.MockType;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+@RunWith(EasyMockRunner.class)
+public class StoreToProcessorContextAdapterTest {
Review comment:
Thanks for adding a test!!!
Would you mind to add tests for all other methods of `StoreToProcessorContextAdapter`, too, to get full test coverage as a side improvement?
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
##########
@@ -494,7 +494,9 @@ public void process(final Record<Integer, Integer> record) {
// e2e latency = 10
task.addRecords(partition1, singletonList(getConsumerRecord(0, 0L)));
- task.process(10L);
+ time = new MockTime(0L, 10L, 0L);
+ task.setTime(time);
+ task.process(time.milliseconds());
Review comment:
Why do we not keep `task.process(10L);` ? Similar below.
##########
File path: streams/src/test/java/org/apache/kafka/test/InternalMockProcessorContext.java
##########
@@ -369,6 +370,16 @@ public long timestamp() {
return recordContext.timestamp();
}
+ @Override
+ public long currentSystemTimeMs() {
+ return Time.SYSTEM.milliseconds();
Review comment:
We should pass in a `Time` parameter into the constructor of `InternalMockProcessorContext` and return `time.milliseconds()` here so we can mock the time.
If this method is unused, it's also ok to just throw an exception for now and we add mocking capabilities later when needed.
(Same below.)
##########
File path: streams/src/test/java/org/apache/kafka/test/NoOpProcessorContext.java
##########
@@ -120,6 +120,16 @@ public Cancellable schedule(final Duration interval,
@Override
public void commit() {}
+ @Override
+ public long currentSystemTimeMs() {
+ throw new UnsupportedOperationException("this method is not supported in NoOpProcessorContext");
Review comment:
Nit: I guess we could add support if needed. Maybe better say `"Not implemented yet."` ? (Same below.)
##########
File path: streams/src/test/java/org/apache/kafka/test/MockInternalProcessorContext.java
##########
@@ -55,14 +55,9 @@ public MockInternalProcessorContext(final Properties config, final TaskId taskId
super(config, taskId, stateDir);
}
- @Override
- public void setSystemTimeMs(long timeMs) {
Review comment:
The purpose of this class seems to be to allow mocking the time -- given that we add proper time support now, I am wondering if we can actually delete the whole class?
##########
File path: streams/test-utils/src/main/java/org/apache/kafka/streams/processor/MockProcessorContext.java
##########
@@ -261,6 +261,16 @@ public TaskId taskId() {
return config.originalsWithPrefix(prefix);
}
+ @Override
+ public long currentSystemTimeMs() {
+ return timestamp;
Review comment:
`timestamp` is the current record timestamp and thus should not be returned here (same below).
I seems, we need to actually extend this test-util class, what is a small change to the KIP. Note, that test-utils are public API and thus those changes also need a KIP. You don't need to worry about your KIP, we can just update it accordingly.
Seems, we need to have two more internal fields, one for each time. The public API change would be to add corresponding setters. We might also rename `setTimestamp` to `setRecordTimestamp` for clarity, we we would have three setters for record, system, and stream time.
##########
File path: streams/test-utils/src/test/java/org/apache/kafka/streams/TopologyTestDriverTest.java
##########
@@ -1878,4 +1877,33 @@ public void shouldRespectTaskIdling() {
);
}
}
+
+ @Test
+ public void shouldMatchSystemTime() {
Review comment:
I think we can remove those two new tests.
If you really want to test it, you should rather add a `Processor` to the Topology that gets the current system and stream time and verify it this way, ie, we should rather add an integration test than a unit test.
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalProcessorContextImpl.java
##########
@@ -115,6 +116,16 @@ public void commit() {
//no-op
}
+ @Override
+ public long currentSystemTimeMs() {
+ return Time.SYSTEM.milliseconds();
Review comment:
We should pass in a `Time` reference to the `GlobalProcessorContextImpl` and call `time.milliseconds()` here -- otherwise, we cannot mock the time (cf. `TopologyTestDriver`) for testing
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
##########
@@ -1111,13 +1111,21 @@ RecordCollector recordCollector() {
return recordCollector;
}
+ public long streamTime() {
+ return partitionGroup.streamTime();
+ }
+
+ public long currentSystemTimeMs() {
Review comment:
As above: seems this can be package-private?
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/AbstractProcessorContextTest.java
##########
@@ -179,6 +179,10 @@ public void appConfigsShouldReturnUnrecognizedValues() {
equalTo("user-supplied-value"));
}
+ @Test(expected = UnsupportedOperationException.class)
+ public void shouldThrowOnCurrentStreamTime() {
Review comment:
This test seems not to be useful, as it tests the `TestProcessorContext` from below.
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
##########
@@ -494,7 +494,9 @@ public void process(final Record<Integer, Integer> record) {
// e2e latency = 10
task.addRecords(partition1, singletonList(getConsumerRecord(0, 0L)));
- task.process(10L);
+ time = new MockTime(0L, 10L, 0L);
Review comment:
We have already a global `MockTime` object in this test that we should reused instead of creating a new one. This way, we can avoid to add `setTime` and keep the `Time` member variable `final`. -- Or is there a reason why we cannot reuse the global `MockTime` object?
Also, I don't really see any call to the newly added methods in this test, so why do we change this test method at all?
##########
File path: streams/src/test/java/org/apache/kafka/test/InternalMockProcessorContext.java
##########
@@ -116,7 +117,7 @@ public InternalMockProcessorContext(final StreamsMetricsImpl streamsMetrics) {
);
}
- public InternalMockProcessorContext(final File stateDir,
+ public InternalMockProcessorContext(final File stateDir,
Review comment:
nit: revert insert of additional space
##########
File path: streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java
##########
@@ -505,7 +506,7 @@ private void setupTask(final StreamsConfig streamsConfig,
streamsMetrics
);
- final InternalProcessorContext context = new ProcessorContextImpl(
Review comment:
I don't think we need this change. The new test you added below, seem to re-test `ProcessorContextImpl` that is already tested in `ProcessorContextImplTest` and thus we don't need those.
----------------------------------------------------------------
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 #9744: KAFKA-10062: Add a method to retrieve the current timestamp as known by the Streams app
Posted by GitBox <gi...@apache.org>.
mjsax commented on a change in pull request #9744:
URL: https://github.com/apache/kafka/pull/9744#discussion_r556207151
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextUtils.java
##########
@@ -39,9 +39,7 @@ private ProcessorContextUtils() {}
* removing the need for this method.
*/
public static long getCurrentSystemTime(final ProcessorContext context) {
Review comment:
nit (not introduced in this PR, but would be a nice side cleanup): rename to `currentSystemTime(...)`
----------------------------------------------------------------
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] rohitrmd commented on a change in pull request #9744: KAFKA-10062: Add a method to retrieve the current timestamp as known by the Streams app
Posted by GitBox <gi...@apache.org>.
rohitrmd commented on a change in pull request #9744:
URL: https://github.com/apache/kafka/pull/9744#discussion_r571532911
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractProcessorContext.java
##########
@@ -45,7 +45,6 @@
private boolean initialized;
protected ProcessorRecordContext recordContext;
protected ProcessorNode<?, ?, ?, ?> currentNode;
- private long currentSystemTimeMs;
Review comment:
@mjsax yes, thank you for the explanation. made changes as per review.
----------------------------------------------------------------
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] rohitrmd commented on pull request #9744: KAFKA-10062: Add a method to retrieve the current timestamp as known by the Streams app
Posted by GitBox <gi...@apache.org>.
rohitrmd commented on pull request #9744:
URL: https://github.com/apache/kafka/pull/9744#issuecomment-786156966
@mjsax have i missed 228 deadline for this 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] rohitrmd commented on a change in pull request #9744: KAFKA-10062: Add a method to retrieve the current timestamp as known by the Streams app
Posted by GitBox <gi...@apache.org>.
rohitrmd commented on a change in pull request #9744:
URL: https://github.com/apache/kafka/pull/9744#discussion_r566587951
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractProcessorContext.java
##########
@@ -45,7 +45,6 @@
private boolean initialized;
protected ProcessorRecordContext recordContext;
protected ProcessorNode<?, ?, ?, ?> currentNode;
- private long currentSystemTimeMs;
Review comment:
@mjsax do you recommend adding method to AbstractProcessorContext to set cachedSystemTimeMs (setCachedSystemTimeMs) or you want me to add setSystemTimeMs back to [InternalProcessorContext](https://github.com/apache/kafka/pull/9744/files#diff-34daeb287c7e79c8ccd757daa4e17d6ab585d54844f6e5e8676853762a08cedcL49) and set the system time like it was done before?
----------------------------------------------------------------
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 #9744: KAFKA-10062: Add a method to retrieve the current timestamp as known by the Streams app
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on a change in pull request #9744:
URL: https://github.com/apache/kafka/pull/9744#discussion_r556718772
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractProcessorContext.java
##########
@@ -45,7 +45,6 @@
private boolean initialized;
protected ProcessorRecordContext recordContext;
protected ProcessorNode<?, ?, ?, ?> currentNode;
- private long currentSystemTimeMs;
Review comment:
I think its rationale is actually to "not advance" the system time while in the middle of processing a record, i.e. the `system time` would just be the time when we start processing / punctuating, and it would not change throughout the process / punctuate.
----------------------------------------------------------------
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 #9744: KAFKA-10062: Add a method to retrieve the current timestamp as known by the Streams app
Posted by GitBox <gi...@apache.org>.
mjsax commented on a change in pull request #9744:
URL: https://github.com/apache/kafka/pull/9744#discussion_r556207831
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractProcessorContext.java
##########
@@ -45,7 +45,6 @@
private boolean initialized;
protected ProcessorRecordContext recordContext;
protected ProcessorNode<?, ?, ?, ?> currentNode;
- private long currentSystemTimeMs;
Review comment:
@guozhangwang -- I am just wondering if removing this variable (what I think conceptually makes sense) might have an perf impact as it seems `currentSystemTimeMs` acts as some kind of cache -- and instead of using the cached time, we would call `time.currentTimeMillis()` in `ProcessorContextImpl#forwardInternal()` now.
Thoughts?
----------------------------------------------------------------
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] rohitrmd commented on a change in pull request #9744: KAFKA-10062: Add a method to retrieve the current timestamp as known by the Streams app
Posted by GitBox <gi...@apache.org>.
rohitrmd commented on a change in pull request #9744:
URL: https://github.com/apache/kafka/pull/9744#discussion_r587996458
##########
File path: streams/test-utils/src/main/java/org/apache/kafka/streams/processor/MockProcessorContext.java
##########
@@ -353,11 +371,19 @@ public void setHeaders(final Headers headers) {
* The context exposes this metadata for use in the processor. Normally, they are set by the Kafka Streams framework,
* but for the purpose of driving unit tests, you can set it directly. Setting this attribute doesn't affect the others.
*
- * @param timestamp A record timestamp
+ * @param recordTimestamp A record timestamp
*/
@SuppressWarnings({"WeakerAccess", "unused"})
- public void setTimestamp(final long timestamp) {
- this.timestamp = timestamp;
+ public void setRecordTimestamp(final long recordTimestamp) {
+ this.recordTimestamp = recordTimestamp;
+ }
+
+ public void setCurrentSystemTimeMs(final long currentSystemTimeMs) {
+ this.currentSystemTimeMs = currentSystemTimeMs;
+ }
+
+ public void setCurrentStreamTimeMs(final long currentStreamTimeMs) {
+ this.currentStreamTimeMs = currentStreamTimeMs;
Review comment:
@mjsax added the method setTimestamp back which sets record timestamp and added Deprecated annotation. Also updated the [kip](https://cwiki.apache.org/confluence/display/KAFKA/KIP-622%3A+Add+currentSystemTimeMs+and+currentStreamTimeMs+to+ProcessorContext).
----------------------------------------------------------------
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] rohitrmd edited a comment on pull request #9744: Kafka 10062 - Add a method to retrieve the current timestamp as known by the Streams app
Posted by GitBox <gi...@apache.org>.
rohitrmd edited a comment on pull request #9744:
URL: https://github.com/apache/kafka/pull/9744#issuecomment-744644019
@mjsax @vvcephei @cadonna StreamTaskTest is failing because of following issue:
StreamTask -> process(wallClockTime) method updates processor context's system time in this method: https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java#L685
AbstractProcessorContect has another systemTime field: https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractProcessorContext.java#L48 which is updated in streamTask's process method.
Now the change in pr fetches system time from StreamTask's time field:
https://github.com/apache/kafka/pull/9744/files#diff-a76674468cda8772230fb8411717cf9068b1a363a792f32c602fb2ec5ba9efd7R1120
This time field is not updated during [process](https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java#L665) call. Hence tests fail.
What do you recommend in this case? Get rid of [currentSystemTimeMs](https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractProcessorContext.java#L48) from AbstractProcessorContext? If we get rid of this field, then how do we update StreamTask's [time](https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java#L73) field as it's immutable.
----------------------------------------------------------------
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 merged pull request #9744: KAFKA-10062: Add a method to retrieve the current timestamp as known by the Streams app
Posted by GitBox <gi...@apache.org>.
mjsax merged pull request #9744:
URL: https://github.com/apache/kafka/pull/9744
----------------------------------------------------------------
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 #9744: KAFKA-10062: Add a method to retrieve the current timestamp as known by the Streams app
Posted by GitBox <gi...@apache.org>.
mjsax commented on a change in pull request #9744:
URL: https://github.com/apache/kafka/pull/9744#discussion_r564992729
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/ProcessorContext.java
##########
@@ -289,4 +291,36 @@ Cancellable schedule(final Duration interval,
*/
Map<String, Object> appConfigsWithPrefix(final String prefix);
+ /**
+ * Return the current system timestamp (also called wall-clock time) in milliseconds.
+ *
+ * <p>
+ * Note: this method returns the internally cached system timestamp from the Kafka Stream runtime.
+ * Thus, it may return a different value compared to `System.currentTimeMillis()`.
Review comment:
nit:
```
`System.currentTimeMillis()` -> {@code System.currentTimeMillis()}
or: {link System#currentTimeMillis()}
```
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractProcessorContext.java
##########
@@ -45,7 +45,6 @@
private boolean initialized;
protected ProcessorRecordContext recordContext;
protected ProcessorNode<?, ?, ?, ?> currentNode;
- private long currentSystemTimeMs;
Review comment:
@rohitrmd did you see Guozhang comment? Seems we should not remove this variable (what is a little unfortunate, but I guess desirable behavior) -- maybe we can rename it to `cachedSystemTimeMs`?
I guess we want the caching for the `GlobalProcessorContext`, too? (For this case, we would need to update the JavaDocs of the `ProcessorContext` interface accordingly.)
We should also add a test for the corresponding contest-test classes that they don't advance system time automatically, but only in the advance is triggered explicitly.
##########
File path: streams/src/test/java/org/apache/kafka/test/MockInternalProcessorContext.java
##########
@@ -55,14 +55,9 @@ public MockInternalProcessorContext(final Properties config, final TaskId taskId
super(config, taskId, stateDir);
}
- @Override
- public void setSystemTimeMs(long timeMs) {
Review comment:
I guess a follow up PR works, too.
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/ProcessorContext.java
##########
@@ -289,4 +291,36 @@ Cancellable schedule(final Duration interval,
*/
Map<String, Object> appConfigsWithPrefix(final String prefix);
+ /**
+ * Return the current system timestamp (also called wall-clock time) in milliseconds.
+ *
+ * <p>
+ * Note: this method returns the internally cached system timestamp from the Kafka Stream runtime.
+ * Thus, it may return a different value compared to `System.currentTimeMillis()`.
+ * <p>
+ *
+ * For a global processor, Kafka Streams does not cache system time and thus calling this method will return
+ * the same value as `System.currentTimeMillis()`.
Review comment:
as above
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/ProcessorContext.java
##########
@@ -289,4 +291,36 @@ Cancellable schedule(final Duration interval,
*/
Map<String, Object> appConfigsWithPrefix(final String prefix);
+ /**
+ * Return the current system timestamp (also called wall-clock time) in milliseconds.
+ *
+ * <p>
+ * Note: this method returns the internally cached system timestamp from the Kafka Stream runtime.
+ * Thus, it may return a different value compared to `System.currentTimeMillis()`.
+ * <p>
+ *
+ * For a global processor, Kafka Streams does not cache system time and thus calling this method will return
+ * the same value as `System.currentTimeMillis()`.
+ *
+ * @return the current system timestamp in milliseconds
+ */
+ long currentSystemTimeMs();
+
+ /**
+ * Return the current stream-time in milliseconds.
+ *
+ * <p>
+ * Stream-time is the maximum observed {@link TimestampExtractor record timestamp} so far
+ * (including the currently processed record), i.e., it can be considered a high-watermark.
+ * Stream-time is tracked on a per-task basis and is preserved across restarts and during task migration.
+ * <p>
+ *
+ * Note: this method is not supported for global processors (cf. {@link Topology#addGlobalStore} (...)
+ * and {@link StreamsBuilder#addGlobalStore} (...),
+ * because there is no concept of stream-time for this case.
+ * Calling this method in a global processor with result in an {@link UnsupportedOperationException}.
Review comment:
typo: `with` -> `will`
----------------------------------------------------------------
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 #9744: KAFKA-10062: Add a method to retrieve the current timestamp as known by the Streams app
Posted by GitBox <gi...@apache.org>.
mjsax commented on a change in pull request #9744:
URL: https://github.com/apache/kafka/pull/9744#discussion_r556207831
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractProcessorContext.java
##########
@@ -45,7 +45,6 @@
private boolean initialized;
protected ProcessorRecordContext recordContext;
protected ProcessorNode<?, ?, ?, ?> currentNode;
- private long currentSystemTimeMs;
Review comment:
@guozhangwang -- I am just wondering if removing this variable (what I think conceptually make sense) might have an perf impact as it seems `currentSystemTimeMs` acts as some kind of cache -- and instead of using the cached time, we would call `time.currentTimeMillis()` in `ProcessorContextImpl#forwardInternal()` now.
Thoughts?
----------------------------------------------------------------
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] rohitrmd commented on a change in pull request #9744: KAFKA-10062: Add a method to retrieve the current timestamp as known by the Streams app
Posted by GitBox <gi...@apache.org>.
rohitrmd commented on a change in pull request #9744:
URL: https://github.com/apache/kafka/pull/9744#discussion_r567356968
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractProcessorContext.java
##########
@@ -45,7 +45,6 @@
private boolean initialized;
protected ProcessorRecordContext recordContext;
protected ProcessorNode<?, ?, ?, ?> currentNode;
- private long currentSystemTimeMs;
Review comment:
@mjsax also I am confused about what time do we want to return when currentSystemTimeMs() is called from the ProcessorContext hierarchy. If it's called from AbstractProcessorContext object reference, should we return cachedSystemTimeMs, and if it's called from ProcessorContextImpl object reference, should we return streamTask.currentSystemTimeMs()? Or we want completely new method to return the cached time when ProcessorContextImpl#forwardInternal() is called?
----------------------------------------------------------------
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] rohitrmd commented on a change in pull request #9744: KAFKA-10062: Add a method to retrieve the current timestamp as known by the Streams app
Posted by GitBox <gi...@apache.org>.
rohitrmd commented on a change in pull request #9744:
URL: https://github.com/apache/kafka/pull/9744#discussion_r566587951
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractProcessorContext.java
##########
@@ -45,7 +45,6 @@
private boolean initialized;
protected ProcessorRecordContext recordContext;
protected ProcessorNode<?, ?, ?, ?> currentNode;
- private long currentSystemTimeMs;
Review comment:
@mjsax do you recommend adding method to AbstractProcessorContext to set cachedSystemTimeMs (setCachedSystemTimeMs) or you want me to add setSystemTimeMs back to [InternalProcessorContext](https://github.com/apache/kafka/pull/9744/files#diff-34daeb287c7e79c8ccd757daa4e17d6ab585d54844f6e5e8676853762a08cedcL49)?
----------------------------------------------------------------
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] rohitrmd commented on pull request #9744: KAFKA-10062: Add a method to retrieve the current timestamp as known by the Streams app
Posted by GitBox <gi...@apache.org>.
rohitrmd commented on pull request #9744:
URL: https://github.com/apache/kafka/pull/9744#issuecomment-775357283
@mjsax made changes as per the review and resolved merge conflicts.
----------------------------------------------------------------
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 #9744: KAFKA-10062: Add a method to retrieve the current timestamp as known by the Streams app
Posted by GitBox <gi...@apache.org>.
mjsax commented on a change in pull request #9744:
URL: https://github.com/apache/kafka/pull/9744#discussion_r586727033
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/ProcessorContext.java
##########
@@ -289,4 +291,33 @@ Cancellable schedule(final Duration interval,
*/
Map<String, Object> appConfigsWithPrefix(final String prefix);
+ /**
+ * Return the current system timestamp (also called wall-clock time) in milliseconds.
+ *
+ * <p>
+ * Note: this method returns the internally cached system timestamp from the Kafka Stream runtime.
+ * Thus, it may return a different value compared to {@code System.currentTimeMillis()} .
+ * <p>
Review comment:
nit. can be removed
##########
File path: streams/test-utils/src/main/java/org/apache/kafka/streams/processor/MockProcessorContext.java
##########
@@ -353,11 +371,19 @@ public void setHeaders(final Headers headers) {
* The context exposes this metadata for use in the processor. Normally, they are set by the Kafka Streams framework,
* but for the purpose of driving unit tests, you can set it directly. Setting this attribute doesn't affect the others.
*
- * @param timestamp A record timestamp
+ * @param recordTimestamp A record timestamp
*/
@SuppressWarnings({"WeakerAccess", "unused"})
- public void setTimestamp(final long timestamp) {
- this.timestamp = timestamp;
+ public void setRecordTimestamp(final long recordTimestamp) {
+ this.recordTimestamp = recordTimestamp;
+ }
+
+ public void setCurrentSystemTimeMs(final long currentSystemTimeMs) {
+ this.currentSystemTimeMs = currentSystemTimeMs;
+ }
+
+ public void setCurrentStreamTimeMs(final long currentStreamTimeMs) {
+ this.currentStreamTimeMs = currentStreamTimeMs;
Review comment:
This class is public API, so we cannot remove `setTimestamp` but can only deprecate it.
We also need to update the KIP to mention the deprecation and the newly added methods.
----------------------------------------------------------------
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 #9744: KAFKA-10062: Add a method to retrieve the current timestamp as known by the Streams app
Posted by GitBox <gi...@apache.org>.
mjsax commented on a change in pull request #9744:
URL: https://github.com/apache/kafka/pull/9744#discussion_r568389885
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractProcessorContext.java
##########
@@ -45,7 +45,6 @@
private boolean initialized;
protected ProcessorRecordContext recordContext;
protected ProcessorNode<?, ?, ?, ?> currentNode;
- private long currentSystemTimeMs;
Review comment:
If there is an advantage to add the method to the `InternalProcessorContext ` interface, sure why not. Don't think it's a big difference, and because it all internal code, we can change it anytime in the future.
I think we should always return the cached 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
[GitHub] [kafka] vvcephei commented on pull request #9744: KAFKA-10062: Add a method to retrieve the current timestamp as known by the Streams app
Posted by GitBox <gi...@apache.org>.
vvcephei commented on pull request #9744:
URL: https://github.com/apache/kafka/pull/9744#issuecomment-786290505
Hi @rohitrmd ,
I was coincidentally just reviewing the list of KIPs that made the cutoff on Feb 3rd. Sadly, yes, this feature will not make 2.8, but we should push forward and get this merged into trunk. FYI, I'll go ahead and drop it from the 2.8 release plan and update the KIP to target 3.0.
FWIW, it's not uncommon to see some unrelated test failures. It'll be up to @mjsax to go ahead and merge or re-run the tests, based on his best judgement.
Thanks for this feature!
-John
----------------------------------------------------------------
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] rohitrmd commented on a change in pull request #9744: KAFKA-10062: Add a method to retrieve the current timestamp as known by the Streams app
Posted by GitBox <gi...@apache.org>.
rohitrmd commented on a change in pull request #9744:
URL: https://github.com/apache/kafka/pull/9744#discussion_r587996458
##########
File path: streams/test-utils/src/main/java/org/apache/kafka/streams/processor/MockProcessorContext.java
##########
@@ -353,11 +371,19 @@ public void setHeaders(final Headers headers) {
* The context exposes this metadata for use in the processor. Normally, they are set by the Kafka Streams framework,
* but for the purpose of driving unit tests, you can set it directly. Setting this attribute doesn't affect the others.
*
- * @param timestamp A record timestamp
+ * @param recordTimestamp A record timestamp
*/
@SuppressWarnings({"WeakerAccess", "unused"})
- public void setTimestamp(final long timestamp) {
- this.timestamp = timestamp;
+ public void setRecordTimestamp(final long recordTimestamp) {
+ this.recordTimestamp = recordTimestamp;
+ }
+
+ public void setCurrentSystemTimeMs(final long currentSystemTimeMs) {
+ this.currentSystemTimeMs = currentSystemTimeMs;
+ }
+
+ public void setCurrentStreamTimeMs(final long currentStreamTimeMs) {
+ this.currentStreamTimeMs = currentStreamTimeMs;
Review comment:
@mjsax added the method setTimestamp back which sets record timestamp and added Deprecated annotation.
----------------------------------------------------------------
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] rohitrmd commented on pull request #9744: KAFKA-10062: Add a method to retrieve the current timestamp as known by the Streams app
Posted by GitBox <gi...@apache.org>.
rohitrmd commented on pull request #9744:
URL: https://github.com/apache/kafka/pull/9744#issuecomment-796309030
thank you @mjsax for all the help and guidance.
----------------------------------------------------------------
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] rohitrmd commented on a change in pull request #9744: KAFKA-10062: Add a method to retrieve the current timestamp as known by the Streams app
Posted by GitBox <gi...@apache.org>.
rohitrmd commented on a change in pull request #9744:
URL: https://github.com/apache/kafka/pull/9744#discussion_r551055293
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
##########
@@ -494,7 +494,9 @@ public void process(final Record<Integer, Integer> record) {
// e2e latency = 10
task.addRecords(partition1, singletonList(getConsumerRecord(0, 0L)));
- task.process(10L);
+ time = new MockTime(0L, 10L, 0L);
Review comment:
@mjsax The reason I had to change StreamTaskTest is because of following issue:
StreamTask -> process(wallClockTime) method updates processor context's system time in this method: https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java#L685
AbstractProcessorContect has another systemTime field: https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractProcessorContext.java#L48 which is updated in streamTask's process method. I removed this field from AbstractProcessorContect as we want to fetch the time from StreamTask. But the latency is measured [here](https://github.com/apache/kafka/blob/b9dfc196aaee91f4fa23fe6563261c6f2d01ac33/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java#L284) which is streamTask's time filed now which is not updated.
I couldn't find any other way to change time field in StreamTask unless I add setTime method in it. If i don't change time in StreamTask object, subsequent metric tests fail as stream-time is not updated.
Do you have recommendation about how i can fix it without adding setTime method?
----------------------------------------------------------------
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] rohitrmd commented on pull request #9744: Kafka 10062 - Add a method to retrieve the current timestamp as known by the Streams app
Posted by GitBox <gi...@apache.org>.
rohitrmd commented on pull request #9744:
URL: https://github.com/apache/kafka/pull/9744#issuecomment-744644019
@mjsax @vvcephei @cadonna StreamTaskTest is failing because of following issue:
StreamTask -> process(wallClockTime) method updates processor context's system time in this method: https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java#L685
AbstractProcessorContect has another systemTime field: https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractProcessorContext.java#L48 which is updated in streamTask's process method.
Now the change in pr fetches system time from StreamTask's time field:
https://github.com/apache/kafka/pull/9744/files#diff-a76674468cda8772230fb8411717cf9068b1a363a792f32c602fb2ec5ba9efd7R1120
This time field is not updated during [process](https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java#L665) call. Hence tests fail.
What do you recommend in this case? Get rid of [currentSystemTimeMs](https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractProcessorContext.java#L48) from AbstractProcessorContext? If we get rid of this field, then how do we update StreamTask's [time](https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java#L73) field as it's immutable. Also how do we update StreamTask's time field?
----------------------------------------------------------------
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 #9744: KAFKA-10062: Add a method to retrieve the current timestamp as known by the Streams app
Posted by GitBox <gi...@apache.org>.
mjsax commented on a change in pull request #9744:
URL: https://github.com/apache/kafka/pull/9744#discussion_r570716520
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractProcessorContext.java
##########
@@ -45,7 +45,6 @@
private boolean initialized;
protected ProcessorRecordContext recordContext;
protected ProcessorNode<?, ?, ?, ?> currentNode;
- private long currentSystemTimeMs;
Review comment:
The main motivation to add `ProcessorContext#currentSystemTime()` was to be able to return the mocked wall-clock time in `TopologyTestDriver`. Even if we return the cached time from `AbstractProcessorContext`, we will be able to return the mocked time, as we update the cached time based on the mocked time in `TopologyTestDriver`.
- `InternalMockProcessorContext` is just for our own unit testing -- it's fine to add the new `Time` field, it's not a public facing change anyway
- Originally we changed `ProcessorContextImpl` because we remove `AbstractProcessorContext#currentSystemTime()` -- as suggested by Guozhang, we should keep the cached time in `AbstractProcessorContext()`, and thus we don't need `ProcessorContextImpl#currentSystemTime()` any longer.
- `GlobalProcessorContextImpl` is a different code path, and thus the changes of this PR are fine
> If we add new cachedSystemTimeMs field in AbstractProcessorContext, when do you want to return this field?
Yes, we want to return this field. (We get this behaviro by adding back `AbstractProcessorContext#currentSystemTime()` (and the cached time in this class) and removing `ProcessorContextImpl#currentSystemTime()`.
> Are earlier changes not valid to return time from StreamTask?
Yes, I think we can revert all changes from `StreamTask`.
Does this make sense?
----------------------------------------------------------------
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] rohitrmd commented on a change in pull request #9744: KAFKA-10062: Add a method to retrieve the current timestamp as known by the Streams app
Posted by GitBox <gi...@apache.org>.
rohitrmd commented on a change in pull request #9744:
URL: https://github.com/apache/kafka/pull/9744#discussion_r570605728
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractProcessorContext.java
##########
@@ -45,7 +45,6 @@
private boolean initialized;
protected ProcessorRecordContext recordContext;
protected ProcessorNode<?, ?, ?, ?> currentNode;
- private long currentSystemTimeMs;
Review comment:
@mjsax can you please explain again what is expected now as for me this is contrary to initial KIP. What I have understood from KIP, we wanted to return system time from Stream Task. Considering changes in this pr,
1. We added one time field in InternalMockProcessorContext which we return when currentSystemTime() is called.
2. When ProcessorContextImpl's currentSystemTime() method is called, we return time from streamTask's time field.
3. We also added time field in GlobalProcessorContextImpl which we return from currentSystemTime().
4. If we add new cachedSystemTimeMs field in AbstractProcessorContext, when do you want to return this field? Are earlier changes not valid to return time from StreamTask?
----------------------------------------------------------------
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] rohitrmd commented on a change in pull request #9744: KAFKA-10062: Add a method to retrieve the current timestamp as known by the Streams app
Posted by GitBox <gi...@apache.org>.
rohitrmd commented on a change in pull request #9744:
URL: https://github.com/apache/kafka/pull/9744#discussion_r566587951
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractProcessorContext.java
##########
@@ -45,7 +45,6 @@
private boolean initialized;
protected ProcessorRecordContext recordContext;
protected ProcessorNode<?, ?, ?, ?> currentNode;
- private long currentSystemTimeMs;
Review comment:
@mjsax do you recommend adding method to AbstractProcessorContext to set cachedSystemTimeMs (setCachedSystemTimeMs)?
----------------------------------------------------------------
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] rohitrmd commented on pull request #9744: KAFKA-10062: Add a method to retrieve the current timestamp as known by the Streams app
Posted by GitBox <gi...@apache.org>.
rohitrmd commented on pull request #9744:
URL: https://github.com/apache/kafka/pull/9744#issuecomment-776306678
@mjsax 2 1. tests failing are not related to my changes. not sure how to fix them. 2. Are there kip updates i need to do?
----------------------------------------------------------------
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] rohitrmd commented on pull request #9744: KAFKA-10062: Add a method to retrieve the current timestamp as known by the Streams app
Posted by GitBox <gi...@apache.org>.
rohitrmd commented on pull request #9744:
URL: https://github.com/apache/kafka/pull/9744#issuecomment-751068878
@mjsax made following changes to the pr:
1. Got rid of [currentSystemTimeMs](https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractProcessorContext.java#L48) from AbstractProcessorContext, that in effect required changing other [GloblaProcessorContextImpl](https://github.com/apache/kafka/pull/9744/files#diff-66f1420c00b334fbe657e3a2c6fb5f8bcd60407dd7f87d9ed6bb1365958e01e4R121) to return current system time. This was needed because of [this](https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/SourceNode.java#L87). MockProcessorContext and MockInternalProcessorContext also required similar changes.
2. In StreamTask, had to add the method to set system [time](https://github.com/apache/kafka/pull/9744/files#diff-a76674468cda8772230fb8411717cf9068b1a363a792f32c602fb2ec5ba9efd7R1127) for test purposes only. As we are getting wall clock time from StreamTask now instead of ProcessorContext, this change was needed for these [tests](https://github.com/apache/kafka/pull/9744/files#diff-d2d5426c1d8b525415735ad409e82d4ff34ea3712a8217e4a99b421e470d4c19R497-R547).
----------------------------------------------------------------
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 pull request #9744: KAFKA-10062: Add a method to retrieve the current timestamp as known by the Streams app
Posted by GitBox <gi...@apache.org>.
mjsax commented on pull request #9744:
URL: https://github.com/apache/kafka/pull/9744#issuecomment-796231276
Merged to `trunk`.
Congrats @rohitrmd and thank you for your contribution!
----------------------------------------------------------------
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 #9744: KAFKA-10062: Add a method to retrieve the current timestamp as known by the Streams app
Posted by GitBox <gi...@apache.org>.
mjsax commented on a change in pull request #9744:
URL: https://github.com/apache/kafka/pull/9744#discussion_r570716520
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractProcessorContext.java
##########
@@ -45,7 +45,6 @@
private boolean initialized;
protected ProcessorRecordContext recordContext;
protected ProcessorNode<?, ?, ?, ?> currentNode;
- private long currentSystemTimeMs;
Review comment:
The main motivation to add `ProcessorContext#currentSystemTime()` was to be able to return the mocked wall-clock time in `TopologyTestDriver`. Even if we return the cached time from `AbstractProcessorContext`, we will be able to return the mocked time, as we update the cached time based on the mocked time in `TopologyTestDriver`.
- `InternalMockProcessorContext` is just for our own unit testing -- it's fine to add the new `Time` field, it's not a public facing change anyway
- Originally we changed `ProcessorContextImpl` because we remove `AbstractProcessorContext#currentSystemTime()` -- as suggested by Guozhang, we should keep the cached time in `AbstractProcessorContext()`, and thus we don't need `ProcessorContextImpl#currentSystemTime()` any longer.
- `GlobalProcessorContextImpl` is a different code path, and thus the changes of this PR are fine
> If we add new cachedSystemTimeMs field in AbstractProcessorContext, when do you want to return this field?
Yes, we want to return this field. (We get this behavior by adding back `AbstractProcessorContext#currentSystemTime()` (and the cached time in this class) and removing `ProcessorContextImpl#currentSystemTime()`.
> Are earlier changes not valid to return time from StreamTask?
Yes, I think we can revert all changes from `StreamTask`.
Does this make sense?
----------------------------------------------------------------
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 #9744: KAFKA-10062: Add a method to retrieve the current timestamp as known by the Streams app
Posted by GitBox <gi...@apache.org>.
mjsax commented on a change in pull request #9744:
URL: https://github.com/apache/kafka/pull/9744#discussion_r568389885
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractProcessorContext.java
##########
@@ -45,7 +45,6 @@
private boolean initialized;
protected ProcessorRecordContext recordContext;
protected ProcessorNode<?, ?, ?, ?> currentNode;
- private long currentSystemTimeMs;
Review comment:
If there is an advantage to add the method to the `InternalProcessorContext ` interface, sure why not. Don't think it's a big difference, and because it all internal code, we can change it anytime in the future.
I think we should always return the cached 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
[GitHub] [kafka] rohitrmd commented on a change in pull request #9744: KAFKA-10062: Add a method to retrieve the current timestamp as known by the Streams app
Posted by GitBox <gi...@apache.org>.
rohitrmd commented on a change in pull request #9744:
URL: https://github.com/apache/kafka/pull/9744#discussion_r570605728
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractProcessorContext.java
##########
@@ -45,7 +45,6 @@
private boolean initialized;
protected ProcessorRecordContext recordContext;
protected ProcessorNode<?, ?, ?, ?> currentNode;
- private long currentSystemTimeMs;
Review comment:
@mjsax can you please explain again what is expected now as for me this is contrary to initial KIP. What I have understood from KIP, we wanted to return system time from Stream Task. Considering changes in this pr,
1. We added one time field in InternalMockProcessorContext which we return when currentSystemTime() is called.
2. When ProcessorContextImpl's currentSystemTime() method is called, we return time from streamTask's time field.
3. We also added time field in GlobalProcessorContextImpl which we return from currentSystemTime().
4. If we add new cachedSystemTimeMs field in AbstractProcessorContext, when do you want to return this field? Are earlier changes not valid to return time from StreamTask?
----------------------------------------------------------------
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] rohitrmd commented on pull request #9744: KAFKA-10062: Add a method to retrieve the current timestamp as known by the Streams app
Posted by GitBox <gi...@apache.org>.
rohitrmd commented on pull request #9744:
URL: https://github.com/apache/kafka/pull/9744#issuecomment-766302189
@mjsax made changes as per the review. Can you please re-review?
----------------------------------------------------------------
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 #9744: KAFKA-10062: Add a method to retrieve the current timestamp as known by the Streams app
Posted by GitBox <gi...@apache.org>.
mjsax commented on a change in pull request #9744:
URL: https://github.com/apache/kafka/pull/9744#discussion_r590766321
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/ProcessorContext.java
##########
@@ -289,4 +291,33 @@ Cancellable schedule(final Duration interval,
*/
Map<String, Object> appConfigsWithPrefix(final String prefix);
+ /**
+ * Return the current system timestamp (also called wall-clock time) in milliseconds.
+ *
+ * <p>
+ * Note: this method returns the internally cached system timestamp from the Kafka Stream runtime.
+ * Thus, it may return a different value compared to {@code System.currentTimeMillis()}
Review comment:
```suggestion
* Thus, it may return a different value compared to {@code System.currentTimeMillis()}.
```
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/ProcessorContext.java
##########
@@ -289,4 +291,33 @@ Cancellable schedule(final Duration interval,
*/
Map<String, Object> appConfigsWithPrefix(final String prefix);
+ /**
+ * Return the current system timestamp (also called wall-clock time) in milliseconds.
+ *
+ * <p>
+ * Note: this method returns the internally cached system timestamp from the Kafka Stream runtime.
+ * Thus, it may return a different value compared to {@code System.currentTimeMillis()}
+ * <p>
Review comment:
```suggestion
```
##########
File path: streams/test-utils/src/main/java/org/apache/kafka/streams/processor/MockProcessorContext.java
##########
@@ -354,10 +372,31 @@ public void setHeaders(final Headers headers) {
* but for the purpose of driving unit tests, you can set it directly. Setting this attribute doesn't affect the others.
*
* @param timestamp A record timestamp
+ * @deprecated Use {@link MockProcessorContext#setRecordTimestamp(long)} instead.
Review comment:
```suggestion
* @deprecated Since 3.0.0; use {@link MockProcessorContext#setRecordTimestamp(long)} instead.
```
----------------------------------------------------------------
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 pull request #9744: KAFKA-10062: Add a method to retrieve the current timestamp as known by the Streams app
Posted by GitBox <gi...@apache.org>.
mjsax commented on pull request #9744:
URL: https://github.com/apache/kafka/pull/9744#issuecomment-772905139
@rohitrmd -- Seems we have till Friday to get this PR merged. Otherwise, it would slip the 2.8.0 release.
Can you address the last comments and rebase the PR to resolve merge conflicts? Or do you have any pending questions?
----------------------------------------------------------------
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