[
https://issues.apache.org/jira/browse/FLINK-3647?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15346356#comment-15346356
]
ASF GitHub Bot commented on FLINK-3647:
---------------------------------------
Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/flink/pull/2124#discussion_r68226912
--- Diff:
flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/testutils/MockRuntimeContext.java
---
@@ -186,16 +199,19 @@ public DistributedCache getDistributedCache() {
public <T> ReducingState<T> getReducingState(ReducingStateDescriptor<T>
stateProperties) {
throw new UnsupportedOperationException();
}
-
+
+ public long getCurrentProcessingTime() {
+ Preconditions.checkNotNull(timerService, "The processing time
timer has not been initialized.");
+ return timerService.getCurrentProcessingTime();
+ }
+
@Override
public ScheduledFuture<?> registerTimer(final long time, final
Triggerable target) {
- if (timer == null) {
- timer = Executors.newSingleThreadScheduledExecutor();
- }
+ Preconditions.checkNotNull(timerService, "The processing time
timer has not been initialized.");
- final long delay = Math.max(time - System.currentTimeMillis(),
0);
+ final long delay = Math.max(time -
timerService.getCurrentProcessingTime(), 0);
--- End diff --
`timerService.registerTimer()` takes an absolute timestamp not a delay now.
> Change StreamSource to use Processing-Time Clock Service
> --------------------------------------------------------
>
> Key: FLINK-3647
> URL: https://issues.apache.org/jira/browse/FLINK-3647
> Project: Flink
> Issue Type: Sub-task
> Components: Streaming
> Reporter: Aljoscha Krettek
> Assignee: Kostas Kloudas
>
> Currently, the {{StreamSource.AutomaticWatermarkContext}} has it's own timer
> service. This should be changed to use the Clock service introduced in
> FLINK-3646 to make watermark emission testable by providing a custom Clock.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
