You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by pn...@apache.org on 2024/08/21 06:47:33 UTC
(flink) branch release-1.19 updated (204db68ac41 -> 5da82f8ca4e)
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a change to branch release-1.19
in repository https://gitbox.apache.org/repos/asf/flink.git
from 204db68ac41 [FLINK-36004] Deep copy the BinaryArrayData in ArrayDataSerializer#copy while using custom ArrayData (#25196) (#25209)
new f10d5e3be03 [hotfix] Refactor SourceOperatorSlitWatermarkAlignmentTest and support pausing splits in MockSourceReader
new 75828335870 [hotfix][test] Do not add three splits with the same splitId in testMetrics
new cc91a024a70 [FLINK-35886][task] Define RelativeClock interface
new 7cd94768252 [FLINK-35886][task] Support markStart and markEnd listeners in TimerGauge
new 9ecc8c74c3c [FLINK-35886][task] Implement PausableRelativeClock
new 0637e96e295 [FLINK-35886][task] Use RelativeClock in WatermarksWithIdleness
new 25250a47e10 [FLINK-35886][task] Expose Clock from ProcessingTimeService
new dedd467f5fe [FLINK-35886][task] Hide backpressure from idleness detection in TimestampsAndWatermarksOperator
new 74d47efaf9d [FLINK-35886][task] Hide backpressure and watermark alignment from idleness detection in SourceOperator
new 70b0b695d13 [FLINK-35886][task] Hide backpressure from idleness detection in WatermarkAssignerOperator
new 45a2e04e414 [FLINK-35886][source] Do not track already finished splits in watermark alignment
new 5da82f8ca4e [FLINK-35886][source] Exclude japi checks for the API changes
The 12 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../eventtime/WatermarkGeneratorSupplier.java | 12 +-
.../eventtime/WatermarkStrategyWithIdleness.java | 4 +-
.../common/eventtime/WatermarksWithIdleness.java | 25 +-
.../java/org/apache/flink/util/clock/Clock.java | 2 +-
.../org/apache/flink/util/clock/RelativeClock.java | 17 +-
.../common/eventtime/WatermarkStrategyTest.java | 85 +------
.../connector/source/mocks/MockSourceReader.java | 39 ++-
.../flink/table/toolbox/TestSourceFunction.java | 17 +-
.../runtime/NeverFireProcessingTimeService.java | 6 +-
.../apache/flink/runtime/metrics/TimerGauge.java | 60 ++++-
.../runtime/metrics/groups/TaskIOMetricGroup.java | 10 +
.../CheckpointCoordinatorMasterHooksTest.java | 4 +-
.../CheckpointCoordinatorTestingUtils.java | 4 +-
.../flink/runtime/metrics/TimerGaugeTest.java | 58 +++++
.../streaming/api/operators/SourceOperator.java | 32 ++-
.../source/NoOpTimestampsAndWatermarks.java | 6 +
.../source/ProgressiveTimestampsAndWatermarks.java | 80 ++++++-
.../operators/source/TimestampsAndWatermarks.java | 50 +++-
.../source/TimestampsAndWatermarksContext.java | 11 +-
.../operators/source/WatermarkToDataOutput.java | 3 +
.../api/operators/util/PausableRelativeClock.java | 96 ++++++++
.../operators/TimestampsAndWatermarksOperator.java | 36 ++-
.../runtime/tasks/ProcessingTimeService.java | 10 +
.../runtime/tasks/ProcessingTimeServiceImpl.java | 5 +-
.../runtime/tasks/SystemProcessingTimeService.java | 6 +-
.../runtime/tasks/TestProcessingTimeService.java | 66 +++++-
.../SourceOperatorSplitWatermarkAlignmentTest.java | 264 ++++++++++++++++++---
.../operators/util/PausableRelativeClockTest.java | 80 +++++++
.../TimestampsAndWatermarksOperatorTest.java | 44 ++++
.../runtime/tasks/MultipleInputStreamTaskTest.java | 4 +-
.../util/AbstractStreamOperatorTestHarness.java | 4 +
.../factories/TestValuesRuntimeFunctions.java | 17 +-
.../codegen/WatermarkGeneratorCodeGenTest.scala | 5 +-
.../wmassigners/WatermarkAssignerOperator.java | 21 +-
.../wmassigners/WatermarkAssignerOperatorTest.java | 29 +++
pom.xml | 4 +
36 files changed, 1023 insertions(+), 193 deletions(-)
copy flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/table/options/ConfigurationValidator.java => flink-core/src/main/java/org/apache/flink/util/clock/RelativeClock.java (62%)
create mode 100644 flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/util/PausableRelativeClock.java
create mode 100644 flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/util/PausableRelativeClockTest.java
(flink) 12/12: [FLINK-35886][source] Exclude japi checks for the API changes
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch release-1.19
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 5da82f8ca4e5dbfa395b6c04a28eab7726d3b1c5
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Tue Aug 20 12:45:22 2024 +0200
[FLINK-35886][source] Exclude japi checks for the API changes
---
pom.xml | 4 ++++
1 file changed, 4 insertions(+)
diff --git a/pom.xml b/pom.xml
index e7f7b7e018f..bb9454c3bf6 100644
--- a/pom.xml
+++ b/pom.xml
@@ -2355,6 +2355,10 @@ under the License.
<exclude>@org.apache.flink.annotation.Experimental</exclude>
<exclude>@org.apache.flink.annotation.Internal</exclude>
<!-- MARKER: start exclusions; these will be wiped by tools/releasing/update_japicmp_configuration.sh -->
+ <!-- New method has been added to the interface, but this interface shouldn't be implemented by users, only used/called, so this doesn't break API compatibility. -->
+ <exclude>org.apache.flink.api.common.eventtime.WatermarkGeneratorSupplier$Context</exclude>
+ <!-- Base interface has been extracted from the existing Clock class. This interface shouldn't affect user code. -->
+ <exclude>org.apache.flink.util.clock.Clock</exclude>
<!-- MARKER: end exclusions -->
</excludes>
<accessModifier>public</accessModifier>
(flink) 07/12: [FLINK-35886][task] Expose Clock from ProcessingTimeService
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch release-1.19
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 25250a47e105ccdfa18a7e34fc3108245ffd83c0
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Fri Aug 9 10:31:39 2024 +0200
[FLINK-35886][task] Expose Clock from ProcessingTimeService
Operators have to be serializable and they will also need an access
to Clock to construct ProgressBlockingRelativeClock. Because we
also want to be able to provide for testing purposes ManualClock
we have to find a way how Operators could obtain a Clock instance.
Exposing Clock from ProcessingTimeService sounds like a good place
as it also will provide a since source of processing time for
potential users (for example ProgressBlockingRelativeClock and
firing timers)
---
.../runtime/NeverFireProcessingTimeService.java | 6 +-
.../runtime/tasks/ProcessingTimeService.java | 10 ++++
.../runtime/tasks/ProcessingTimeServiceImpl.java | 5 +-
.../runtime/tasks/SystemProcessingTimeService.java | 6 +-
.../runtime/tasks/TestProcessingTimeService.java | 66 +++++++++++++++++++---
5 files changed, 79 insertions(+), 14 deletions(-)
diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/NeverFireProcessingTimeService.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/NeverFireProcessingTimeService.java
index 3f1860549d8..2bb2f06c55c 100644
--- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/NeverFireProcessingTimeService.java
+++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/NeverFireProcessingTimeService.java
@@ -19,6 +19,8 @@ package org.apache.flink.state.api.runtime;
import org.apache.flink.annotation.Internal;
import org.apache.flink.streaming.runtime.tasks.TimerService;
+import org.apache.flink.util.clock.Clock;
+import org.apache.flink.util.clock.SystemClock;
import org.apache.flink.util.concurrent.NeverCompleteFuture;
import java.util.concurrent.CompletableFuture;
@@ -33,8 +35,8 @@ public final class NeverFireProcessingTimeService implements TimerService {
private AtomicBoolean shutdown = new AtomicBoolean(true);
@Override
- public long getCurrentProcessingTime() {
- return System.currentTimeMillis();
+ public Clock getClock() {
+ return SystemClock.getInstance();
}
@Override
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/ProcessingTimeService.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/ProcessingTimeService.java
index bd551e854fb..743f48b38a3 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/ProcessingTimeService.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/ProcessingTimeService.java
@@ -17,6 +17,7 @@
package org.apache.flink.streaming.runtime.tasks;
+import org.apache.flink.util.clock.Clock;
import org.apache.flink.util.concurrent.ScheduledExecutor;
import java.util.concurrent.CompletableFuture;
@@ -32,6 +33,15 @@ import java.util.concurrent.TimeUnit;
*/
public interface ProcessingTimeService
extends org.apache.flink.api.common.operators.ProcessingTimeService {
+
+ @Override
+ default long getCurrentProcessingTime() {
+ return getClock().absoluteTimeMillis();
+ }
+
+ /** Returns {@link Clock} associated with this timer service. */
+ Clock getClock();
+
/**
* Registers a task to be executed repeatedly at a fixed rate.
*
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/ProcessingTimeServiceImpl.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/ProcessingTimeServiceImpl.java
index 3a42ed9f8b3..e3b5b9ed475 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/ProcessingTimeServiceImpl.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/ProcessingTimeServiceImpl.java
@@ -19,6 +19,7 @@
package org.apache.flink.streaming.runtime.tasks;
import org.apache.flink.annotation.Internal;
+import org.apache.flink.util.clock.Clock;
import org.apache.flink.util.concurrent.NeverCompleteFuture;
import java.util.concurrent.CompletableFuture;
@@ -53,8 +54,8 @@ class ProcessingTimeServiceImpl implements ProcessingTimeService {
}
@Override
- public long getCurrentProcessingTime() {
- return timerService.getCurrentProcessingTime();
+ public Clock getClock() {
+ return timerService.getClock();
}
@Override
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SystemProcessingTimeService.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SystemProcessingTimeService.java
index fee6a646475..73722249340 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SystemProcessingTimeService.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SystemProcessingTimeService.java
@@ -20,6 +20,8 @@ package org.apache.flink.streaming.runtime.tasks;
import org.apache.flink.annotation.Internal;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.time.Deadline;
+import org.apache.flink.util.clock.Clock;
+import org.apache.flink.util.clock.SystemClock;
import org.apache.flink.util.concurrent.NeverCompleteFuture;
import org.slf4j.Logger;
@@ -86,8 +88,8 @@ public class SystemProcessingTimeService implements TimerService {
}
@Override
- public long getCurrentProcessingTime() {
- return System.currentTimeMillis();
+ public Clock getClock() {
+ return SystemClock.getInstance();
}
/**
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/TestProcessingTimeService.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/TestProcessingTimeService.java
index 8229076393a..2b54390d986 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/TestProcessingTimeService.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/TestProcessingTimeService.java
@@ -19,7 +19,9 @@ package org.apache.flink.streaming.runtime.tasks;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Preconditions;
+import org.apache.flink.util.clock.Clock;
+import java.time.Duration;
import java.util.Comparator;
import java.util.HashSet;
import java.util.PriorityQueue;
@@ -30,6 +32,7 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
/**
@@ -38,14 +41,14 @@ import java.util.concurrent.atomic.AtomicReference;
*/
public class TestProcessingTimeService implements TimerService {
- private volatile long currentTime = Long.MIN_VALUE;
-
private volatile boolean isTerminated;
private volatile boolean isQuiesced;
// sorts the timers by timestamp so that they are processed in the correct order.
private final PriorityQueue<Tuple2<Long, CallbackTask>> priorityQueue;
+ private final ManualMSClock clock = new ManualMSClock(Long.MIN_VALUE);
+
public TestProcessingTimeService() {
this.priorityQueue =
new PriorityQueue<>(
@@ -60,14 +63,19 @@ public class TestProcessingTimeService implements TimerService {
}
public void advance(long delta) throws Exception {
- setCurrentTime(this.currentTime + delta);
+ clock.advanceTime(Duration.ofMillis(delta));
+ maybeFireTimers();
}
public void setCurrentTime(long timestamp) throws Exception {
- this.currentTime = timestamp;
+ clock.setCurrentTime(timestamp, TimeUnit.MILLISECONDS);
+ maybeFireTimers();
+ }
+ private void maybeFireTimers() throws Exception {
if (!isQuiesced) {
- while (!priorityQueue.isEmpty() && currentTime >= priorityQueue.peek().f0) {
+ while (!priorityQueue.isEmpty()
+ && getCurrentProcessingTime() >= priorityQueue.peek().f0) {
Tuple2<Long, CallbackTask> entry = priorityQueue.poll();
CallbackTask callbackTask = entry.f1;
@@ -88,8 +96,8 @@ public class TestProcessingTimeService implements TimerService {
}
@Override
- public long getCurrentProcessingTime() {
- return currentTime;
+ public Clock getClock() {
+ return clock;
}
@Override
@@ -121,7 +129,8 @@ public class TestProcessingTimeService implements TimerService {
PeriodicCallbackTask periodicCallbackTask = new PeriodicCallbackTask(callback, period);
priorityQueue.offer(
- Tuple2.<Long, CallbackTask>of(currentTime + initialDelay, periodicCallbackTask));
+ Tuple2.<Long, CallbackTask>of(
+ getCurrentProcessingTime() + initialDelay, periodicCallbackTask));
return periodicCallbackTask;
}
@@ -264,4 +273,45 @@ public class TestProcessingTimeService implements TimerService {
return currentTimestamp + period;
}
}
+
+ /**
+ * Similar to {@link org.apache.flink.util.clock.ManualClock}, but with ms precision and thus
+ * greater range. This is needed to support registering and firing timers with {@link
+ * Long#MAX_VALUE}.
+ */
+ private static class ManualMSClock extends Clock {
+ private final AtomicLong currentTime;
+
+ public ManualMSClock(long startTime) {
+ this.currentTime = new AtomicLong(startTime);
+ }
+
+ @Override
+ public long absoluteTimeMillis() {
+ return currentTime.get();
+ }
+
+ @Override
+ public long relativeTimeMillis() {
+ return currentTime.get();
+ }
+
+ @Override
+ public long relativeTimeNanos() {
+ return currentTime.get() * 1_000_000;
+ }
+
+ /**
+ * Advances the time by the given duration. Time can also move backwards by supplying a
+ * negative value. This method performs no overflow check.
+ */
+ public void advanceTime(Duration duration) {
+ currentTime.addAndGet(duration.toMillis());
+ }
+
+ /** Sets the time to the given value. */
+ public void setCurrentTime(long time, TimeUnit timeUnit) {
+ currentTime.set(timeUnit.toMillis(time));
+ }
+ }
}
(flink) 03/12: [FLINK-35886][task] Define RelativeClock interface
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch release-1.19
in repository https://gitbox.apache.org/repos/asf/flink.git
commit cc91a024a70350d84e1f816f73c87e8a6cffda96
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Mon Aug 5 16:41:19 2024 +0200
[FLINK-35886][task] Define RelativeClock interface
---
.../java/org/apache/flink/util/clock/Clock.java | 2 +-
.../util/clock/{Clock.java => RelativeClock.java} | 27 +++++-----------------
2 files changed, 7 insertions(+), 22 deletions(-)
diff --git a/flink-core/src/main/java/org/apache/flink/util/clock/Clock.java b/flink-core/src/main/java/org/apache/flink/util/clock/Clock.java
index 29a05aacf28..879cbfca9c6 100644
--- a/flink-core/src/main/java/org/apache/flink/util/clock/Clock.java
+++ b/flink-core/src/main/java/org/apache/flink/util/clock/Clock.java
@@ -37,7 +37,7 @@ import org.apache.flink.annotation.PublicEvolving;
* the system clock, so it never jumps. Relative time behaves similar to {@link System#nanoTime()}.
*/
@PublicEvolving
-public abstract class Clock {
+public abstract class Clock implements RelativeClock {
/** Gets the current absolute time, in milliseconds. */
public abstract long absoluteTimeMillis();
diff --git a/flink-core/src/main/java/org/apache/flink/util/clock/Clock.java b/flink-core/src/main/java/org/apache/flink/util/clock/RelativeClock.java
similarity index 50%
copy from flink-core/src/main/java/org/apache/flink/util/clock/Clock.java
copy to flink-core/src/main/java/org/apache/flink/util/clock/RelativeClock.java
index 29a05aacf28..990de585c7a 100644
--- a/flink-core/src/main/java/org/apache/flink/util/clock/Clock.java
+++ b/flink-core/src/main/java/org/apache/flink/util/clock/RelativeClock.java
@@ -21,30 +21,15 @@ package org.apache.flink.util.clock;
import org.apache.flink.annotation.PublicEvolving;
/**
- * A clock that gives access to time. This clock returns two flavors of time:
- *
- * <h3>Absolute Time</h3>
- *
- * <p>This refers to real world wall clock time, and it is typically derived from a system clock. It
- * is subject to clock drift and inaccuracy, and can jump if the system clock is adjusted. Absolute
- * time behaves similar to {@link System#currentTimeMillis()}.
- *
- * <h3>Relative Time</h3>
- *
- * <p>This time advances at the same speed as the <i>absolute time</i>, but the timestamps can only
- * be referred to relative to each other. The timestamps have no absolute meaning and cannot be
- * compared across JVM processes. The source for the timestamps is not affected by adjustments to
- * the system clock, so it never jumps. Relative time behaves similar to {@link System#nanoTime()}.
+ * A clock that gives access to relative time, similar to System#nanoTime(), however the progress of
+ * the relative time doesn't have to reflect the progress of a wall clock. Concrete classes can
+ * specify a different contract in that regard. Returned time must be monotonically increasing.
*/
@PublicEvolving
-public abstract class Clock {
-
- /** Gets the current absolute time, in milliseconds. */
- public abstract long absoluteTimeMillis();
-
+public interface RelativeClock {
/** Gets the current relative time, in milliseconds. */
- public abstract long relativeTimeMillis();
+ long relativeTimeMillis();
/** Gets the current relative time, in nanoseconds. */
- public abstract long relativeTimeNanos();
+ long relativeTimeNanos();
}
(flink) 06/12: [FLINK-35886][task] Use RelativeClock in WatermarksWithIdleness
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch release-1.19
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 0637e96e295e0cb542606f1e3081dc64f85aee5c
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Mon Aug 5 16:52:41 2024 +0200
[FLINK-35886][task] Use RelativeClock in WatermarksWithIdleness
---
.../eventtime/WatermarkGeneratorSupplier.java | 12 ++-
.../eventtime/WatermarkStrategyWithIdleness.java | 4 +-
.../common/eventtime/WatermarksWithIdleness.java | 25 +++++--
.../common/eventtime/WatermarkStrategyTest.java | 85 +++-------------------
.../flink/table/toolbox/TestSourceFunction.java | 17 ++++-
.../CheckpointCoordinatorMasterHooksTest.java | 4 +-
.../CheckpointCoordinatorTestingUtils.java | 4 +-
.../source/TimestampsAndWatermarksContext.java | 7 ++
.../operators/TimestampsAndWatermarksOperator.java | 17 ++++-
.../factories/TestValuesRuntimeFunctions.java | 17 ++++-
.../codegen/WatermarkGeneratorCodeGenTest.scala | 5 +-
11 files changed, 104 insertions(+), 93 deletions(-)
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/eventtime/WatermarkGeneratorSupplier.java b/flink-core/src/main/java/org/apache/flink/api/common/eventtime/WatermarkGeneratorSupplier.java
index 0b0a1fd3ed8..4d149d39259 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/eventtime/WatermarkGeneratorSupplier.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/eventtime/WatermarkGeneratorSupplier.java
@@ -20,6 +20,7 @@ package org.apache.flink.api.common.eventtime;
import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.util.clock.RelativeClock;
import java.io.Serializable;
@@ -39,7 +40,7 @@ public interface WatermarkGeneratorSupplier<T> extends Serializable {
/**
* Additional information available to {@link #createWatermarkGenerator(Context)}. This can be
- * access to {@link org.apache.flink.metrics.MetricGroup MetricGroups}, for example.
+ * access to {@link MetricGroup MetricGroups}, for example.
*/
interface Context {
@@ -54,5 +55,14 @@ public interface WatermarkGeneratorSupplier<T> extends Serializable {
* @see MetricGroup
*/
MetricGroup getMetricGroup();
+
+ /**
+ * Returns a {@link RelativeClock} that hides periods when input was not active and {@link
+ * WatermarkGenerator} could not have been executed due to execution being blocked by the
+ * runtime. For example a backpressure or watermark alignment blocking the progress.
+ *
+ * @see RelativeClock
+ */
+ RelativeClock getInputActivityClock();
}
}
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/eventtime/WatermarkStrategyWithIdleness.java b/flink-core/src/main/java/org/apache/flink/api/common/eventtime/WatermarkStrategyWithIdleness.java
index 660a7890277..7c0ae861952 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/eventtime/WatermarkStrategyWithIdleness.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/eventtime/WatermarkStrategyWithIdleness.java
@@ -42,7 +42,9 @@ final class WatermarkStrategyWithIdleness<T> implements WatermarkStrategy<T> {
public WatermarkGenerator<T> createWatermarkGenerator(
WatermarkGeneratorSupplier.Context context) {
return new WatermarksWithIdleness<>(
- baseStrategy.createWatermarkGenerator(context), idlenessTimeout);
+ baseStrategy.createWatermarkGenerator(context),
+ idlenessTimeout,
+ context.getInputActivityClock());
}
@Override
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/eventtime/WatermarksWithIdleness.java b/flink-core/src/main/java/org/apache/flink/api/common/eventtime/WatermarksWithIdleness.java
index 3d61217ce04..a4d5639a5e1 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/eventtime/WatermarksWithIdleness.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/eventtime/WatermarksWithIdleness.java
@@ -21,6 +21,7 @@ package org.apache.flink.api.common.eventtime;
import org.apache.flink.annotation.Public;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.util.clock.Clock;
+import org.apache.flink.util.clock.RelativeClock;
import org.apache.flink.util.clock.SystemClock;
import java.time.Duration;
@@ -42,19 +43,27 @@ public class WatermarksWithIdleness<T> implements WatermarkGenerator<T> {
private boolean isIdleNow = false;
+ /**
+ * This is not used anymore, but it's technically part of the {@link Public} API. Please use
+ * {@link #WatermarksWithIdleness(WatermarkGenerator, Duration, RelativeClock)} instead.
+ */
+ @Deprecated
+ public WatermarksWithIdleness(WatermarkGenerator<T> watermarks, Duration idleTimeout) {
+ this(watermarks, idleTimeout, SystemClock.getInstance());
+ }
+
/**
* Creates a new WatermarksWithIdleness generator to the given generator idleness detection with
* the given timeout.
*
* @param watermarks The original watermark generator.
* @param idleTimeout The timeout for the idleness detection.
+ * @param clock The clock that will be used to measure idleness period. It is expected that this
+ * clock will hide periods when this {@link WatermarkGenerator} has been blocked from making
+ * any progress despite availability of records on the input.
*/
- public WatermarksWithIdleness(WatermarkGenerator<T> watermarks, Duration idleTimeout) {
- this(watermarks, idleTimeout, SystemClock.getInstance());
- }
-
- @VisibleForTesting
- WatermarksWithIdleness(WatermarkGenerator<T> watermarks, Duration idleTimeout, Clock clock) {
+ public WatermarksWithIdleness(
+ WatermarkGenerator<T> watermarks, Duration idleTimeout, RelativeClock clock) {
checkNotNull(idleTimeout, "idleTimeout");
checkArgument(
!(idleTimeout.isZero() || idleTimeout.isNegative()),
@@ -88,7 +97,7 @@ public class WatermarksWithIdleness<T> implements WatermarkGenerator<T> {
static final class IdlenessTimer {
/** The clock used to measure elapsed time. */
- private final Clock clock;
+ private final RelativeClock clock;
/** Counter to detect change. No problem if it overflows. */
private long counter;
@@ -105,7 +114,7 @@ public class WatermarksWithIdleness<T> implements WatermarkGenerator<T> {
/** The duration before the output is marked as idle. */
private final long maxIdleTimeNanos;
- IdlenessTimer(Clock clock, Duration idleTimeout) {
+ IdlenessTimer(RelativeClock clock, Duration idleTimeout) {
this.clock = clock;
long idleNanos;
diff --git a/flink-core/src/test/java/org/apache/flink/api/common/eventtime/WatermarkStrategyTest.java b/flink-core/src/test/java/org/apache/flink/api/common/eventtime/WatermarkStrategyTest.java
index df9d24853d8..34f51a8d7a9 100644
--- a/flink-core/src/test/java/org/apache/flink/api/common/eventtime/WatermarkStrategyTest.java
+++ b/flink-core/src/test/java/org/apache/flink/api/common/eventtime/WatermarkStrategyTest.java
@@ -20,18 +20,15 @@ package org.apache.flink.api.common.eventtime;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.java.ClosureCleaner;
-import org.apache.flink.metrics.CharacterFilter;
-import org.apache.flink.metrics.Counter;
-import org.apache.flink.metrics.Gauge;
-import org.apache.flink.metrics.Histogram;
-import org.apache.flink.metrics.Meter;
import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
+import org.apache.flink.util.clock.RelativeClock;
+import org.apache.flink.util.clock.SystemClock;
import org.junit.Test;
import java.io.Serializable;
import java.time.Duration;
-import java.util.Map;
import static org.hamcrest.CoreMatchers.equalTo;
import static org.hamcrest.CoreMatchers.instanceOf;
@@ -188,82 +185,20 @@ public class WatermarkStrategyTest {
}
static TimestampAssignerSupplier.Context assignerContext() {
- return new TimestampAssignerSupplier.Context() {
- @Override
- public MetricGroup getMetricGroup() {
- return new DummyMetricGroup();
- }
- };
+ return UnregisteredMetricsGroup::new;
}
static WatermarkGeneratorSupplier.Context generatorContext() {
return new WatermarkGeneratorSupplier.Context() {
@Override
public MetricGroup getMetricGroup() {
- return new DummyMetricGroup();
+ return new UnregisteredMetricsGroup();
}
- };
- }
-
- /**
- * A dummy {@link MetricGroup} to be used when a group is required as an argument but not
- * actually used.
- */
- public static class DummyMetricGroup implements MetricGroup {
-
- @Override
- public Counter counter(String name) {
- return null;
- }
-
- @Override
- public <C extends Counter> C counter(String name, C counter) {
- return null;
- }
-
- @Override
- public <T, G extends Gauge<T>> G gauge(String name, G gauge) {
- return null;
- }
-
- @Override
- public <H extends Histogram> H histogram(String name, H histogram) {
- return null;
- }
-
- @Override
- public <M extends Meter> M meter(String name, M meter) {
- return null;
- }
-
- @Override
- public MetricGroup addGroup(String name) {
- return null;
- }
- @Override
- public MetricGroup addGroup(String key, String value) {
- return null;
- }
-
- @Override
- public String[] getScopeComponents() {
- return new String[0];
- }
-
- @Override
- public Map<String, String> getAllVariables() {
- return null;
- }
-
- @Override
- public String getMetricIdentifier(String metricName) {
- return null;
- }
-
- @Override
- public String getMetricIdentifier(String metricName, CharacterFilter filter) {
- return null;
- }
+ @Override
+ public RelativeClock getInputActivityClock() {
+ return SystemClock.getInstance();
+ }
+ };
}
}
diff --git a/flink-end-to-end-tests/flink-sql-client-test/src/main/java/org/apache/flink/table/toolbox/TestSourceFunction.java b/flink-end-to-end-tests/flink-sql-client-test/src/main/java/org/apache/flink/table/toolbox/TestSourceFunction.java
index 347db1589c8..3fe4dce03aa 100644
--- a/flink-end-to-end-tests/flink-sql-client-test/src/main/java/org/apache/flink/table/toolbox/TestSourceFunction.java
+++ b/flink-end-to-end-tests/flink-sql-client-test/src/main/java/org/apache/flink/table/toolbox/TestSourceFunction.java
@@ -20,13 +20,17 @@ package org.apache.flink.table.toolbox;
import org.apache.flink.api.common.eventtime.Watermark;
import org.apache.flink.api.common.eventtime.WatermarkGenerator;
+import org.apache.flink.api.common.eventtime.WatermarkGeneratorSupplier;
import org.apache.flink.api.common.eventtime.WatermarkOutput;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.StringData;
import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.util.clock.RelativeClock;
+import org.apache.flink.util.clock.SystemClock;
import java.util.ArrayList;
import java.util.Arrays;
@@ -60,7 +64,18 @@ public class TestSourceFunction implements SourceFunction<RowData> {
@Override
public void run(SourceContext<RowData> ctx) {
WatermarkGenerator<RowData> generator =
- watermarkStrategy.createWatermarkGenerator(() -> null);
+ watermarkStrategy.createWatermarkGenerator(
+ new WatermarkGeneratorSupplier.Context() {
+ @Override
+ public MetricGroup getMetricGroup() {
+ return null;
+ }
+
+ @Override
+ public RelativeClock getInputActivityClock() {
+ return SystemClock.getInstance();
+ }
+ });
WatermarkOutput output = new TestWatermarkOutput(ctx);
int rowDataSize = DATA.get(0).size();
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorMasterHooksTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorMasterHooksTest.java
index 75e637dd078..a7332d07194 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorMasterHooksTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorMasterHooksTest.java
@@ -19,8 +19,8 @@
package org.apache.flink.runtime.checkpoint;
import org.apache.flink.api.common.JobID;
-import org.apache.flink.api.common.eventtime.WatermarkStrategyTest.DummyMetricGroup;
import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
import org.apache.flink.runtime.executiongraph.ExecutionGraph;
import org.apache.flink.runtime.executiongraph.ExecutionGraphCheckpointPlanCalculatorContext;
@@ -500,7 +500,7 @@ class CheckpointCoordinatorMasterHooksTest {
new ExecutionGraphCheckpointPlanCalculatorContext(graph),
graph.getVerticesTopologically(),
false),
- new CheckpointStatsTracker(1, new DummyMetricGroup(), new JobID()));
+ new CheckpointStatsTracker(1, new UnregisteredMetricsGroup(), new JobID()));
}
private static <T> T mockGeneric(Class<?> clazz) {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTestingUtils.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTestingUtils.java
index aac4ab51b36..829445287d9 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTestingUtils.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTestingUtils.java
@@ -19,11 +19,11 @@
package org.apache.flink.runtime.checkpoint;
import org.apache.flink.api.common.JobID;
-import org.apache.flink.api.common.eventtime.WatermarkStrategyTest.DummyMetricGroup;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.core.fs.FSDataInputStream;
import org.apache.flink.core.fs.Path;
import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
import org.apache.flink.runtime.OperatorIDPair;
import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutorServiceAdapter;
@@ -781,7 +781,7 @@ public class CheckpointCoordinatorTestingUtils {
private boolean allowCheckpointsAfterTasksFinished;
private CheckpointStatsTracker checkpointStatsTracker =
- new CheckpointStatsTracker(1, new DummyMetricGroup(), new JobID());
+ new CheckpointStatsTracker(1, new UnregisteredMetricsGroup(), new JobID());
private BiFunction<
Set<ExecutionJobVertex>,
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/TimestampsAndWatermarksContext.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/TimestampsAndWatermarksContext.java
index 4685691385d..8ec6cc8c01f 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/TimestampsAndWatermarksContext.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/TimestampsAndWatermarksContext.java
@@ -22,6 +22,8 @@ import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.eventtime.TimestampAssignerSupplier;
import org.apache.flink.api.common.eventtime.WatermarkGeneratorSupplier;
import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.util.clock.RelativeClock;
+import org.apache.flink.util.clock.SystemClock;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -43,4 +45,9 @@ public final class TimestampsAndWatermarksContext
public MetricGroup getMetricGroup() {
return metricGroup;
}
+
+ @Override
+ public RelativeClock getInputActivityClock() {
+ return SystemClock.getInstance();
+ }
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/TimestampsAndWatermarksOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/TimestampsAndWatermarksOperator.java
index 570cb6b90b3..02f72f3ba32 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/TimestampsAndWatermarksOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/TimestampsAndWatermarksOperator.java
@@ -21,14 +21,18 @@ import org.apache.flink.api.common.eventtime.NoWatermarksGenerator;
import org.apache.flink.api.common.eventtime.TimestampAssigner;
import org.apache.flink.api.common.eventtime.Watermark;
import org.apache.flink.api.common.eventtime.WatermarkGenerator;
+import org.apache.flink.api.common.eventtime.WatermarkGeneratorSupplier;
import org.apache.flink.api.common.eventtime.WatermarkOutput;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
import org.apache.flink.streaming.api.operators.ChainingStrategy;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.operators.Output;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus;
+import org.apache.flink.util.clock.RelativeClock;
+import org.apache.flink.util.clock.SystemClock;
import static org.apache.flink.api.common.operators.ProcessingTimeService.ProcessingTimeCallback;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -80,7 +84,18 @@ public class TimestampsAndWatermarksOperator<T> extends AbstractStreamOperator<T
timestampAssigner = watermarkStrategy.createTimestampAssigner(this::getMetricGroup);
watermarkGenerator =
emitProgressiveWatermarks
- ? watermarkStrategy.createWatermarkGenerator(this::getMetricGroup)
+ ? watermarkStrategy.createWatermarkGenerator(
+ new WatermarkGeneratorSupplier.Context() {
+ @Override
+ public MetricGroup getMetricGroup() {
+ return this.getMetricGroup();
+ }
+
+ @Override
+ public RelativeClock getInputActivityClock() {
+ return SystemClock.getInstance();
+ }
+ })
: new NoWatermarksGenerator<>();
wmOutput = new WatermarkEmitter(output);
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesRuntimeFunctions.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesRuntimeFunctions.java
index acc4cf4a0cd..91908faefe5 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesRuntimeFunctions.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesRuntimeFunctions.java
@@ -20,6 +20,7 @@ package org.apache.flink.table.planner.factories;
import org.apache.flink.api.common.eventtime.Watermark;
import org.apache.flink.api.common.eventtime.WatermarkGenerator;
+import org.apache.flink.api.common.eventtime.WatermarkGeneratorSupplier;
import org.apache.flink.api.common.eventtime.WatermarkOutput;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.io.RichOutputFormat;
@@ -30,6 +31,7 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataInputViewStreamWrapper;
import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
+import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
@@ -54,6 +56,8 @@ import org.apache.flink.test.util.SuccessException;
import org.apache.flink.types.Row;
import org.apache.flink.types.RowKind;
import org.apache.flink.types.RowUtils;
+import org.apache.flink.util.clock.RelativeClock;
+import org.apache.flink.util.clock.SystemClock;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
@@ -247,7 +251,18 @@ final class TestValuesRuntimeFunctions {
ByteArrayInputStream bais = new ByteArrayInputStream(elementsSerialized);
final DataInputView input = new DataInputViewStreamWrapper(bais);
WatermarkGenerator<RowData> generator =
- watermarkStrategy.createWatermarkGenerator(() -> null);
+ watermarkStrategy.createWatermarkGenerator(
+ new WatermarkGeneratorSupplier.Context() {
+ @Override
+ public MetricGroup getMetricGroup() {
+ return null;
+ }
+
+ @Override
+ public RelativeClock getInputActivityClock() {
+ return SystemClock.getInstance();
+ }
+ });
WatermarkOutput output = new TestValuesWatermarkOutput(ctx);
final Object lock = ctx.getCheckpointLock();
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/codegen/WatermarkGeneratorCodeGenTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/codegen/WatermarkGeneratorCodeGenTest.scala
index 98265f67299..75ae617a4f2 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/codegen/WatermarkGeneratorCodeGenTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/codegen/WatermarkGeneratorCodeGenTest.scala
@@ -31,9 +31,10 @@ import org.apache.flink.table.runtime.generated.WatermarkGenerator
import org.apache.flink.table.types.logical.{IntType, TimestampType}
import org.apache.flink.table.utils.CatalogManagerMocks
import org.apache.flink.testutils.junit.extensions.parameterized.{ParameterizedTestExtension, Parameters}
+import org.apache.flink.util.clock.{RelativeClock, SystemClock}
-import org.junit.jupiter.api.{Test, TestTemplate}
import org.junit.jupiter.api.Assertions.{assertEquals, assertTrue}
+import org.junit.jupiter.api.TestTemplate
import org.junit.jupiter.api.extension.ExtendWith
import java.lang.{Integer => JInt, Long => JLong}
@@ -167,6 +168,8 @@ class WatermarkGeneratorCodeGenTest(useDefinedConstructor: Boolean) {
val newReferences = generated.getReferences :+
new WatermarkGeneratorSupplier.Context {
override def getMetricGroup: MetricGroup = null
+
+ override def getInputActivityClock: RelativeClock = SystemClock.getInstance()
}
generated.newInstance(Thread.currentThread().getContextClassLoader, newReferences)
} else {
(flink) 10/12: [FLINK-35886][task] Hide backpressure from idleness detection in WatermarkAssignerOperator
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch release-1.19
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 70b0b695d13e54bb440b07c00761bd2788e7cfdd
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Fri Aug 9 11:10:40 2024 +0200
[FLINK-35886][task] Hide backpressure from idleness detection in WatermarkAssignerOperator
---
.../wmassigners/WatermarkAssignerOperator.java | 21 ++++++++++++++--
.../wmassigners/WatermarkAssignerOperatorTest.java | 29 ++++++++++++++++++++++
2 files changed, 48 insertions(+), 2 deletions(-)
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/wmassigners/WatermarkAssignerOperator.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/wmassigners/WatermarkAssignerOperator.java
index 5fd7c8a22e1..486ee5274e9 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/wmassigners/WatermarkAssignerOperator.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/wmassigners/WatermarkAssignerOperator.java
@@ -24,6 +24,7 @@ import org.apache.flink.api.common.functions.util.FunctionUtils;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
import org.apache.flink.streaming.api.operators.ChainingStrategy;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.util.PausableRelativeClock;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
@@ -70,6 +71,9 @@ public class WatermarkAssignerOperator extends AbstractStreamOperator<RowData>
private transient long lastIdleCheckProcessedElements = -1;
+ /** {@link PausableRelativeClock} that will be paused in case of backpressure. */
+ private transient PausableRelativeClock inputActivityClock;
+
/**
* Create a watermark assigner operator.
*
@@ -95,6 +99,12 @@ public class WatermarkAssignerOperator extends AbstractStreamOperator<RowData>
@Override
public void open() throws Exception {
super.open();
+ inputActivityClock = new PausableRelativeClock(getProcessingTimeService().getClock());
+ getContainingTask()
+ .getEnvironment()
+ .getMetricGroup()
+ .getIOMetricGroup()
+ .registerBackPressureListener(inputActivityClock);
// watermark and timestamp should start from 0
this.currentWatermark = 0;
@@ -154,7 +164,9 @@ public class WatermarkAssignerOperator extends AbstractStreamOperator<RowData>
@Override
public void onProcessingTime(long timestamp) throws Exception {
// timestamp and now can be off in case TM is heavily overloaded.
+ // now and inputActivityNow are using different clocks and can have very different values.
long now = getProcessingTimeService().getCurrentProcessingTime();
+ long inputActivityNow = inputActivityClock.relativeTimeMillis();
if (watermarkInterval > 0 && lastWatermarkPeriodicEmitTime + watermarkInterval <= now) {
lastWatermarkPeriodicEmitTime = now;
@@ -162,13 +174,13 @@ public class WatermarkAssignerOperator extends AbstractStreamOperator<RowData>
}
if (processedElements != lastIdleCheckProcessedElements) {
- timeSinceLastIdleCheck = now;
+ timeSinceLastIdleCheck = inputActivityNow;
lastIdleCheckProcessedElements = processedElements;
}
if (isIdlenessEnabled()
&& currentStatus.equals(WatermarkStatus.ACTIVE)
- && timeSinceLastIdleCheck + idleTimeout <= now) {
+ && timeSinceLastIdleCheck + idleTimeout <= inputActivityNow) {
// mark the channel as idle to ignore watermarks from this channel
emitWatermarkStatus(WatermarkStatus.IDLE);
}
@@ -213,6 +225,11 @@ public class WatermarkAssignerOperator extends AbstractStreamOperator<RowData>
@Override
public void close() throws Exception {
+ getContainingTask()
+ .getEnvironment()
+ .getMetricGroup()
+ .getIOMetricGroup()
+ .unregisterBackPressureListener(inputActivityClock);
FunctionUtils.closeFunction(watermarkGenerator);
super.close();
}
diff --git a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/wmassigners/WatermarkAssignerOperatorTest.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/wmassigners/WatermarkAssignerOperatorTest.java
index 9cd558629bc..2a5419eafbb 100644
--- a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/wmassigners/WatermarkAssignerOperatorTest.java
+++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/wmassigners/WatermarkAssignerOperatorTest.java
@@ -20,6 +20,7 @@ package org.apache.flink.table.runtime.operators.wmassigners;
import org.apache.flink.api.common.functions.OpenContext;
import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.runtime.metrics.groups.TaskIOMetricGroup;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus;
@@ -139,6 +140,34 @@ public class WatermarkAssignerOperatorTest extends WatermarkAssignerOperatorTest
assertThat(extractWatermarkStatuses(output)).doesNotContain(WatermarkStatus.IDLE);
}
+ @Test
+ public void testIdleTimeoutUnderBackpressure() throws Exception {
+ long idleTimeout = 100;
+
+ OneInputStreamOperatorTestHarness<RowData, RowData> testHarness =
+ createTestHarness(0, WATERMARK_GENERATOR, idleTimeout);
+ testHarness.getExecutionConfig().setAutoWatermarkInterval(idleTimeout);
+ testHarness.open();
+
+ TaskIOMetricGroup taskIOMetricGroup =
+ testHarness.getEnvironment().getMetricGroup().getIOMetricGroup();
+ taskIOMetricGroup.getHardBackPressuredTimePerSecond().markStart();
+
+ stepProcessingTime(testHarness, 0, idleTimeout * 10, idleTimeout / 10);
+ assertThat(testHarness.getOutput()).isEmpty();
+
+ taskIOMetricGroup.getHardBackPressuredTimePerSecond().markEnd();
+ taskIOMetricGroup.getSoftBackPressuredTimePerSecond().markStart();
+
+ stepProcessingTime(testHarness, idleTimeout * 10, idleTimeout * 20, idleTimeout / 10);
+ assertThat(testHarness.getOutput()).isEmpty();
+
+ taskIOMetricGroup.getSoftBackPressuredTimePerSecond().markEnd();
+
+ stepProcessingTime(testHarness, idleTimeout * 20, idleTimeout * 30, idleTimeout / 10);
+ assertThat(testHarness.getOutput()).containsExactly(WatermarkStatus.IDLE);
+ }
+
private void stepProcessingTime(
OneInputStreamOperatorTestHarness<?, ?> testHarness,
long fromInclusive,
(flink) 01/12: [hotfix] Refactor SourceOperatorSlitWatermarkAlignmentTest and support pausing splits in MockSourceReader
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch release-1.19
in repository https://gitbox.apache.org/repos/asf/flink.git
commit f10d5e3be03f8427bc3dd295ea08638234fdf534
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Mon Aug 5 15:24:46 2024 +0200
[hotfix] Refactor SourceOperatorSlitWatermarkAlignmentTest and support pausing splits in MockSourceReader
---
.../connector/source/mocks/MockSourceReader.java | 25 ++++++++-
.../SourceOperatorSplitWatermarkAlignmentTest.java | 64 +++++++++-------------
2 files changed, 47 insertions(+), 42 deletions(-)
diff --git a/flink-core/src/test/java/org/apache/flink/api/connector/source/mocks/MockSourceReader.java b/flink-core/src/test/java/org/apache/flink/api/connector/source/mocks/MockSourceReader.java
index 2c1d4b40867..facfcf346cd 100644
--- a/flink-core/src/test/java/org/apache/flink/api/connector/source/mocks/MockSourceReader.java
+++ b/flink-core/src/test/java/org/apache/flink/api/connector/source/mocks/MockSourceReader.java
@@ -26,11 +26,15 @@ import org.apache.flink.core.io.InputStatus;
import javax.annotation.concurrent.GuardedBy;
import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashSet;
import java.util.List;
+import java.util.Set;
import java.util.concurrent.CompletableFuture;
/** A mock {@link SourceReader} for unit tests. */
public class MockSourceReader implements SourceReader<Integer, MockSourceSplit> {
+ private final Set<String> pausedSplits = new HashSet<>();
private final List<MockSourceSplit> assignedSplits = new ArrayList<>();
private final List<SourceEvent> receivedSourceEvents = new ArrayList<>();
private final List<Long> completedCheckpoints = new ArrayList<>();
@@ -100,11 +104,16 @@ public class MockSourceReader implements SourceReader<Integer, MockSourceSplit>
|| waitingForSplitsBehaviour == WaitingForSplits.DO_NOT_WAIT_FOR_SPLITS;
currentSplitIndex = 0;
// Find first splits with available records.
- while (currentSplitIndex < assignedSplits.size()
- && !assignedSplits.get(currentSplitIndex).isAvailable()) {
- finished &= assignedSplits.get(currentSplitIndex).isFinished();
+ for (MockSourceSplit assignedSplit : assignedSplits) {
+ finished &= assignedSplit.isFinished();
+ if (!pausedSplits.contains(assignedSplit.splitId())) {
+ if (assignedSplit.isAvailable()) {
+ break;
+ }
+ }
currentSplitIndex++;
}
+
// Read from the split with available record.
if (currentSplitIndex < assignedSplits.size()) {
if (idle) {
@@ -151,6 +160,12 @@ public class MockSourceReader implements SourceReader<Integer, MockSourceSplit>
markAvailable();
}
+ public void pauseOrResumeSplits(
+ Collection<String> splitsToPause, Collection<String> splitsToResume) {
+ pausedSplits.removeAll(splitsToResume);
+ pausedSplits.addAll(splitsToPause);
+ }
+
@Override
public void notifyNoMoreSplits() {
splitsAssignmentState = SplitsAssignmentState.NO_MORE_SPLITS;
@@ -224,4 +239,8 @@ public class MockSourceReader implements SourceReader<Integer, MockSourceSplit>
public List<Long> getAbortedCheckpoints() {
return abortedCheckpoints;
}
+
+ public Set<String> getPausedSplits() {
+ return pausedSplits;
+ }
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorSplitWatermarkAlignmentTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorSplitWatermarkAlignmentTest.java
index de80d7f4ef5..8be17a021d5 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorSplitWatermarkAlignmentTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorSplitWatermarkAlignmentTest.java
@@ -24,6 +24,7 @@ import org.apache.flink.api.common.eventtime.WatermarkGenerator;
import org.apache.flink.api.common.eventtime.WatermarkOutput;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.connector.source.mocks.MockSourceReader;
+import org.apache.flink.api.connector.source.mocks.MockSourceReader.WaitingForSplits;
import org.apache.flink.api.connector.source.mocks.MockSourceSplit;
import org.apache.flink.api.connector.source.mocks.MockSourceSplitSerializer;
import org.apache.flink.configuration.Configuration;
@@ -42,46 +43,27 @@ import org.apache.flink.streaming.runtime.tasks.TestProcessingTimeService;
import org.apache.flink.streaming.util.MockOutput;
import org.apache.flink.streaming.util.MockStreamConfig;
+import org.assertj.core.api.Condition;
import org.junit.jupiter.api.Test;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
-import java.util.Collection;
-import java.util.HashSet;
-import java.util.Set;
import static org.assertj.core.api.Assertions.assertThat;
/** Unit test for split alignment in {@link SourceOperator}. */
-public class SourceOperatorSplitWatermarkAlignmentTest {
- public static final WatermarkGenerator<Integer> WATERMARK_GENERATOR =
- new WatermarkGenerator<Integer>() {
-
- private long maxWatermark = Long.MIN_VALUE;
-
- @Override
- public void onEvent(Integer event, long eventTimestamp, WatermarkOutput output) {
- if (eventTimestamp > maxWatermark) {
- this.maxWatermark = eventTimestamp;
- output.emitWatermark(new Watermark(maxWatermark));
- }
- }
-
- @Override
- public void onPeriodicEmit(WatermarkOutput output) {
- output.emitWatermark(new Watermark(maxWatermark));
- }
- };
+class SourceOperatorSplitWatermarkAlignmentTest {
@Test
public void testSplitWatermarkAlignment() throws Exception {
- final SplitAligningSourceReader sourceReader = new SplitAligningSourceReader();
+ MockSourceReader sourceReader =
+ new MockSourceReader(WaitingForSplits.DO_NOT_WAIT_FOR_SPLITS, false, true);
SourceOperator<Integer, MockSourceSplit> operator =
new TestingSourceOperator<>(
sourceReader,
- WatermarkStrategy.forGenerator(ctx -> WATERMARK_GENERATOR)
+ WatermarkStrategy.forGenerator(ctx -> new TestWatermarkGenerator())
.withTimestampAssigner((r, l) -> r)
.withWatermarkAlignment("group-1", Duration.ofMillis(1)),
new TestProcessingTimeService(),
@@ -97,8 +79,8 @@ public class SourceOperatorSplitWatermarkAlignmentTest {
operator.initializeState(new StreamTaskStateInitializerImpl(env, new MemoryStateBackend()));
operator.open();
- final MockSourceSplit split1 = new MockSourceSplit(0, 0, 10);
- final MockSourceSplit split2 = new MockSourceSplit(1, 10, 20);
+ MockSourceSplit split1 = new MockSourceSplit(0, 0, 10);
+ MockSourceSplit split2 = new MockSourceSplit(1, 10, 20);
split1.addRecord(5);
split1.addRecord(11);
split2.addRecord(3);
@@ -107,25 +89,25 @@ public class SourceOperatorSplitWatermarkAlignmentTest {
operator.handleOperatorEvent(
new AddSplitEvent<>(
Arrays.asList(split1, split2), new MockSourceSplitSerializer()));
- final CollectingDataOutput<Integer> dataOutput = new CollectingDataOutput<>();
+ CollectingDataOutput<Integer> dataOutput = new CollectingDataOutput<>();
operator.emitNext(dataOutput); // split 1 emits 5
operator.handleOperatorEvent(
new WatermarkAlignmentEvent(4)); // pause by coordinator message
- assertThat(sourceReader.pausedSplits).containsExactly("0");
+ assertThat(sourceReader.getPausedSplits()).containsExactly("0");
operator.handleOperatorEvent(new WatermarkAlignmentEvent(5));
- assertThat(sourceReader.pausedSplits).isEmpty();
+ assertThat(sourceReader.getPausedSplits()).isEmpty();
operator.emitNext(dataOutput); // split 1 emits 11
operator.emitNext(dataOutput); // split 2 emits 3
- assertThat(sourceReader.pausedSplits).containsExactly("0");
+ assertThat(sourceReader.getPausedSplits()).containsExactly("0");
operator.emitNext(dataOutput); // split 2 emits 6
- assertThat(sourceReader.pausedSplits).containsExactly("0", "1");
+ assertThat(sourceReader.getPausedSplits()).containsExactly("0", "1");
}
private Environment getTestingEnvironment() {
@@ -139,17 +121,21 @@ public class SourceOperatorSplitWatermarkAlignmentTest {
new TestTaskStateManager());
}
- private static class SplitAligningSourceReader extends MockSourceReader {
- Set<String> pausedSplits = new HashSet<>();
+ private static class TestWatermarkGenerator implements WatermarkGenerator<Integer> {
- public SplitAligningSourceReader() {
- super(WaitingForSplits.DO_NOT_WAIT_FOR_SPLITS, false, true);
+ private long maxWatermark = Long.MIN_VALUE;
+
+ @Override
+ public void onEvent(Integer event, long eventTimestamp, WatermarkOutput output) {
+ if (eventTimestamp > maxWatermark) {
+ this.maxWatermark = eventTimestamp;
+ output.emitWatermark(new Watermark(maxWatermark));
+ }
}
- public void pauseOrResumeSplits(
- Collection<String> splitsToPause, Collection<String> splitsToResume) {
- pausedSplits.removeAll(splitsToResume);
- pausedSplits.addAll(splitsToPause);
+ @Override
+ public void onPeriodicEmit(WatermarkOutput output) {
+ output.emitWatermark(new Watermark(maxWatermark));
}
}
}
(flink) 05/12: [FLINK-35886][task] Implement PausableRelativeClock
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch release-1.19
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 9ecc8c74c3c10e1c9e611188b92d45b0b61de2f5
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Mon Aug 5 17:36:38 2024 +0200
[FLINK-35886][task] Implement PausableRelativeClock
---
.../api/operators/util/PausableRelativeClock.java | 96 ++++++++++++++++++++++
.../operators/util/PausableRelativeClockTest.java | 80 ++++++++++++++++++
2 files changed, 176 insertions(+)
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/util/PausableRelativeClock.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/util/PausableRelativeClock.java
new file mode 100644
index 00000000000..f7cc275b86d
--- /dev/null
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/util/PausableRelativeClock.java
@@ -0,0 +1,96 @@
+/*
+ * 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.flink.streaming.api.operators.util;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.metrics.TimerGauge;
+import org.apache.flink.util.clock.Clock;
+import org.apache.flink.util.clock.RelativeClock;
+
+import javax.annotation.concurrent.ThreadSafe;
+
+import static org.apache.flink.util.Preconditions.checkState;
+
+/**
+ * A {@link RelativeClock} whose time progress with respect to the wall clock can be paused and
+ * un-paused. It can be paused multiple times. If it is paused N times, it has to be un-paused also
+ * N times to resume progress.
+ */
+@Internal
+@ThreadSafe
+public class PausableRelativeClock implements RelativeClock, TimerGauge.StartStopListener {
+ private final Clock baseClock;
+
+ private long accumulativeBlockedNanoTime;
+ private long currentBlockedNanoTimeStart;
+ /** How many times this clock has been paused. */
+ private long pausedCounter;
+
+ public PausableRelativeClock(Clock baseClock) {
+ this.baseClock = baseClock;
+ }
+
+ @Override
+ public long relativeTimeMillis() {
+ return relativeTimeNanos() / 1_000_000;
+ }
+
+ @Override
+ public synchronized long relativeTimeNanos() {
+ long now = baseClock.relativeTimeNanos();
+ // we offset relativeTimeNanos by the time this clock has been paused.
+ return now - getBlockedTime(now);
+ }
+
+ /** @return how long this {@link PausableRelativeClock} has been paused so far. */
+ private long getBlockedTime(long now) {
+ long blockedTime = accumulativeBlockedNanoTime;
+ if (pausedCounter != 0) {
+ // If we are paused right now, add the time since when we were paused
+ blockedTime += now - currentBlockedNanoTimeStart;
+ }
+ return blockedTime;
+ }
+
+ public synchronized void pause() {
+ if (pausedCounter == 0) {
+ currentBlockedNanoTimeStart = baseClock.relativeTimeNanos();
+ }
+ pausedCounter++;
+ }
+
+ public synchronized void unPause() {
+ checkState(pausedCounter >= 1);
+ pausedCounter--;
+ if (pausedCounter == 0) {
+ accumulativeBlockedNanoTime +=
+ baseClock.relativeTimeNanos() - currentBlockedNanoTimeStart;
+ }
+ }
+
+ @Override
+ public void markStart() {
+ pause();
+ }
+
+ @Override
+ public void markEnd() {
+ unPause();
+ }
+}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/util/PausableRelativeClockTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/util/PausableRelativeClockTest.java
new file mode 100644
index 00000000000..a7ff6fef822
--- /dev/null
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/util/PausableRelativeClockTest.java
@@ -0,0 +1,80 @@
+/*
+ * 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.flink.streaming.api.operators.util;
+
+import org.apache.flink.util.clock.ManualClock;
+
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class PausableRelativeClockTest {
+ private static final long TIME_STEP = 14;
+
+ @Test
+ void simpleTest() {
+ ManualClock baseClock = new ManualClock();
+ PausableRelativeClock relativeClock = new PausableRelativeClock(baseClock);
+
+ long startMillis = relativeClock.relativeTimeMillis();
+ long startNanos = relativeClock.relativeTimeNanos();
+ baseClock.advanceTime(Duration.ofMillis(TIME_STEP));
+ long endNanos = relativeClock.relativeTimeNanos();
+ long endMillis = relativeClock.relativeTimeMillis();
+
+ long durationNanos = endNanos - startNanos;
+ long durationMillis = endMillis - startMillis;
+
+ assertThat((durationNanos) / 1_000_000).isEqualTo(TIME_STEP);
+ assertThat(durationMillis).isEqualTo(TIME_STEP);
+ }
+
+ @Test
+ void pausedTest() throws Exception {
+ ManualClock baseClock = new ManualClock();
+ PausableRelativeClock relativeClock = new PausableRelativeClock(baseClock);
+
+ long startMillis = relativeClock.relativeTimeMillis();
+ long startNanos = relativeClock.relativeTimeNanos();
+
+ baseClock.advanceTime(Duration.ofMillis(TIME_STEP));
+ relativeClock.pause();
+ baseClock.advanceTime(Duration.ofMillis(TIME_STEP)); // doesn't count
+ relativeClock.pause();
+ baseClock.advanceTime(Duration.ofMillis(TIME_STEP)); // doesn't count
+ relativeClock.unPause();
+ baseClock.advanceTime(Duration.ofMillis(TIME_STEP)); // doesn't count
+ relativeClock.unPause();
+ baseClock.advanceTime(Duration.ofMillis(TIME_STEP));
+ relativeClock.pause();
+ baseClock.advanceTime(Duration.ofMillis(TIME_STEP)); // doesn't count
+ // leave the clock blocked
+
+ long endNanos = relativeClock.relativeTimeNanos();
+ long endMillis = relativeClock.relativeTimeMillis();
+
+ long durationNanos = endNanos - startNanos;
+ long durationMillis = endMillis - startMillis;
+
+ assertThat((durationNanos) / 1_000_000).isEqualTo(TIME_STEP * 2);
+ assertThat(durationMillis).isEqualTo(TIME_STEP * 2);
+ }
+}
(flink) 08/12: [FLINK-35886][task] Hide backpressure from idleness detection in TimestampsAndWatermarksOperator
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch release-1.19
in repository https://gitbox.apache.org/repos/asf/flink.git
commit dedd467f5fe48abdad175451155961ca2bb5a00f
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Tue Aug 6 17:52:12 2024 +0200
[FLINK-35886][task] Hide backpressure from idleness detection in TimestampsAndWatermarksOperator
---
.../runtime/metrics/groups/TaskIOMetricGroup.java | 10 +++++
.../operators/TimestampsAndWatermarksOperator.java | 23 ++++++++++-
.../TimestampsAndWatermarksOperatorTest.java | 44 ++++++++++++++++++++++
.../util/AbstractStreamOperatorTestHarness.java | 4 ++
4 files changed, 79 insertions(+), 2 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/TaskIOMetricGroup.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/TaskIOMetricGroup.java
index a571b078c2e..25868633f30 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/TaskIOMetricGroup.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/TaskIOMetricGroup.java
@@ -302,6 +302,16 @@ public class TaskIOMetricGroup extends ProxyMetricGroup<TaskMetricGroup> {
return mailboxSize;
}
+ public void registerBackPressureListener(TimerGauge.StartStopListener backPressureListener) {
+ hardBackPressuredTimePerSecond.registerListener(backPressureListener);
+ softBackPressuredTimePerSecond.registerListener(backPressureListener);
+ }
+
+ public void unregisterBackPressureListener(TimerGauge.StartStopListener backPressureListener) {
+ hardBackPressuredTimePerSecond.unregisterListener(backPressureListener);
+ softBackPressuredTimePerSecond.unregisterListener(backPressureListener);
+ }
+
// ============================================================================================
// Metric Reuse
// ============================================================================================
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/TimestampsAndWatermarksOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/TimestampsAndWatermarksOperator.java
index 02f72f3ba32..a636ea9ab82 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/TimestampsAndWatermarksOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/TimestampsAndWatermarksOperator.java
@@ -29,10 +29,10 @@ import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
import org.apache.flink.streaming.api.operators.ChainingStrategy;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.api.operators.util.PausableRelativeClock;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus;
import org.apache.flink.util.clock.RelativeClock;
-import org.apache.flink.util.clock.SystemClock;
import static org.apache.flink.api.common.operators.ProcessingTimeService.ProcessingTimeCallback;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -70,6 +70,9 @@ public class TimestampsAndWatermarksOperator<T> extends AbstractStreamOperator<T
/** Whether to emit intermediate watermarks or only one final watermark at the end of input. */
private final boolean emitProgressiveWatermarks;
+ /** {@link PausableRelativeClock} that will be paused in case of backpressure. */
+ private transient PausableRelativeClock inputActivityClock;
+
public TimestampsAndWatermarksOperator(
WatermarkStrategy<T> watermarkStrategy, boolean emitProgressiveWatermarks) {
this.watermarkStrategy = checkNotNull(watermarkStrategy);
@@ -80,6 +83,12 @@ public class TimestampsAndWatermarksOperator<T> extends AbstractStreamOperator<T
@Override
public void open() throws Exception {
super.open();
+ inputActivityClock = new PausableRelativeClock(getProcessingTimeService().getClock());
+ getContainingTask()
+ .getEnvironment()
+ .getMetricGroup()
+ .getIOMetricGroup()
+ .registerBackPressureListener(inputActivityClock);
timestampAssigner = watermarkStrategy.createTimestampAssigner(this::getMetricGroup);
watermarkGenerator =
@@ -93,7 +102,7 @@ public class TimestampsAndWatermarksOperator<T> extends AbstractStreamOperator<T
@Override
public RelativeClock getInputActivityClock() {
- return SystemClock.getInstance();
+ return inputActivityClock;
}
})
: new NoWatermarksGenerator<>();
@@ -107,6 +116,16 @@ public class TimestampsAndWatermarksOperator<T> extends AbstractStreamOperator<T
}
}
+ @Override
+ public void close() throws Exception {
+ getContainingTask()
+ .getEnvironment()
+ .getMetricGroup()
+ .getIOMetricGroup()
+ .unregisterBackPressureListener(inputActivityClock);
+ super.close();
+ }
+
@Override
public void processElement(final StreamRecord<T> element) throws Exception {
final T event = element.getValue();
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/TimestampsAndWatermarksOperatorTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/TimestampsAndWatermarksOperatorTest.java
index b7f34619a30..6162c2d6f1a 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/TimestampsAndWatermarksOperatorTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/TimestampsAndWatermarksOperatorTest.java
@@ -24,6 +24,7 @@ import org.apache.flink.api.common.eventtime.WatermarkGenerator;
import org.apache.flink.api.common.eventtime.WatermarkOutput;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.runtime.metrics.groups.TaskIOMetricGroup;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus;
import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
@@ -31,10 +32,13 @@ import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
import org.junit.Test;
import java.io.Serializable;
+import java.time.Duration;
import static org.apache.flink.streaming.util.StreamRecordMatchers.streamRecord;
import static org.apache.flink.streaming.util.WatermarkMatchers.legacyWatermark;
+import static org.hamcrest.Matchers.contains;
import static org.hamcrest.Matchers.empty;
+import static org.hamcrest.Matchers.hasSize;
import static org.hamcrest.Matchers.is;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertThat;
@@ -235,6 +239,46 @@ public class TimestampsAndWatermarksOperatorTest {
}
}
+ @Test
+ public void watermarksWithIdlenessUnderBackpressure() throws Exception {
+ long idleTimeout = 100;
+
+ TimestampsAndWatermarksOperator<Tuple2<Boolean, Long>> operator =
+ new TimestampsAndWatermarksOperator<>(
+ WatermarkStrategy.forGenerator((ctx) -> new PunctuatedWatermarkGenerator())
+ .withTimestampAssigner((ctx) -> new TupleExtractor())
+ .withIdleness(Duration.ofMillis(idleTimeout)),
+ true);
+
+ OneInputStreamOperatorTestHarness<Tuple2<Boolean, Long>, Tuple2<Boolean, Long>>
+ testHarness = new OneInputStreamOperatorTestHarness<>(operator);
+ testHarness.open();
+
+ TaskIOMetricGroup taskIOMetricGroup =
+ testHarness.getEnvironment().getMetricGroup().getIOMetricGroup();
+ taskIOMetricGroup.getHardBackPressuredTimePerSecond().markStart();
+
+ for (int i = 0; i < 10; i++) {
+ testHarness.advanceTime(idleTimeout);
+ }
+ assertThat(testHarness.getOutput(), hasSize(0));
+
+ taskIOMetricGroup.getHardBackPressuredTimePerSecond().markEnd();
+ taskIOMetricGroup.getSoftBackPressuredTimePerSecond().markStart();
+
+ for (int i = 10; i < 20; i++) {
+ testHarness.advanceTime(idleTimeout);
+ }
+ assertThat(testHarness.getOutput(), hasSize(0));
+
+ taskIOMetricGroup.getSoftBackPressuredTimePerSecond().markEnd();
+
+ for (int i = 20; i < 30; i++) {
+ testHarness.advanceTime(idleTimeout);
+ }
+ assertThat(testHarness.getOutput(), contains(WatermarkStatus.IDLE));
+ }
+
private static <T> OneInputStreamOperatorTestHarness<T, T> createTestHarness(
WatermarkStrategy<T> watermarkStrategy) throws Exception {
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/AbstractStreamOperatorTestHarness.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/AbstractStreamOperatorTestHarness.java
index fc5a71cd61a..aa58c4ea8cc 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/AbstractStreamOperatorTestHarness.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/AbstractStreamOperatorTestHarness.java
@@ -762,6 +762,10 @@ public class AbstractStreamOperatorTestHarness<OUT> implements AutoCloseable {
return factory;
}
+ public void advanceTime(long delta) throws Exception {
+ processingTimeService.advance(delta);
+ }
+
public void setProcessingTime(long time) throws Exception {
processingTimeService.setCurrentTime(time);
}
(flink) 02/12: [hotfix][test] Do not add three splits with the same splitId in testMetrics
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch release-1.19
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 75828335870718cc6865349939c2df7d5a58f09d
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Fri Aug 16 14:46:32 2024 +0200
[hotfix][test] Do not add three splits with the same splitId in testMetrics
---
.../flink/streaming/runtime/tasks/MultipleInputStreamTaskTest.java | 4 +---
1 file changed, 1 insertion(+), 3 deletions(-)
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTaskTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTaskTest.java
index fe3a6ad4c1d..7897797b4b8 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTaskTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/MultipleInputStreamTaskTest.java
@@ -465,9 +465,7 @@ public class MultipleInputStreamTaskTest {
int numRecords3 = 2;
// add source splits before processing any elements, so the MockSourceReader does not
// end prematurely
- for (int x = 0; x < numRecords2; x++) {
- addSourceRecords(testHarness, 1, 42);
- }
+ addSourceRecords(testHarness, 1, 42, 43, 44);
for (int x = 0; x < numRecords1; x++) {
testHarness.processElement(new StreamRecord<>("hello"), 0, 0);
}
(flink) 09/12: [FLINK-35886][task] Hide backpressure and watermark alignment from idleness detection in SourceOperator
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch release-1.19
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 74d47efaf9d15078763ccae831fac005f42508cb
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Mon Aug 5 15:26:29 2024 +0200
[FLINK-35886][task] Hide backpressure and watermark alignment from idleness detection in SourceOperator
---
.../streaming/api/operators/SourceOperator.java | 27 +++-
.../source/NoOpTimestampsAndWatermarks.java | 6 +
.../source/ProgressiveTimestampsAndWatermarks.java | 79 ++++++++--
.../operators/source/TimestampsAndWatermarks.java | 47 +++++-
.../source/TimestampsAndWatermarksContext.java | 8 +-
.../SourceOperatorSplitWatermarkAlignmentTest.java | 167 +++++++++++++++++++++
6 files changed, 311 insertions(+), 23 deletions(-)
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java
index d9a04853107..b26edc77c63 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java
@@ -37,6 +37,7 @@ import org.apache.flink.metrics.groups.SourceReaderMetricGroup;
import org.apache.flink.runtime.io.AvailabilityProvider;
import org.apache.flink.runtime.io.network.api.StopMode;
import org.apache.flink.runtime.metrics.groups.InternalSourceReaderMetricGroup;
+import org.apache.flink.runtime.metrics.groups.TaskIOMetricGroup;
import org.apache.flink.runtime.operators.coordination.OperatorEvent;
import org.apache.flink.runtime.operators.coordination.OperatorEventGateway;
import org.apache.flink.runtime.operators.coordination.OperatorEventHandler;
@@ -52,6 +53,7 @@ import org.apache.flink.runtime.state.StateInitializationContext;
import org.apache.flink.runtime.state.StateSnapshotContext;
import org.apache.flink.streaming.api.graph.StreamConfig;
import org.apache.flink.streaming.api.operators.source.TimestampsAndWatermarks;
+import org.apache.flink.streaming.api.operators.util.PausableRelativeClock;
import org.apache.flink.streaming.api.operators.util.SimpleVersionedListState;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.io.DataInputStatus;
@@ -197,6 +199,12 @@ public class SourceOperator<OUT, SplitT extends SourceSplit> extends AbstractStr
private final CanEmitBatchOfRecordsChecker canEmitBatchOfRecords;
+ /**
+ * {@link PausableRelativeClock} tracking activity of the operator's main input. It's paused on
+ * backpressure. Note, each split output has its own independent {@link PausableRelativeClock}.
+ */
+ private transient PausableRelativeClock mainInputActivityClock;
+
public SourceOperator(
FunctionWithException<SourceReaderContext, SourceReader<OUT, SplitT>, Exception>
readerFactory,
@@ -322,6 +330,11 @@ public class SourceOperator<OUT, SplitT extends SourceSplit> extends AbstractStr
@Override
public void open() throws Exception {
+ mainInputActivityClock = new PausableRelativeClock(getProcessingTimeService().getClock());
+ TaskIOMetricGroup taskIOMetricGroup =
+ getContainingTask().getEnvironment().getMetricGroup().getIOMetricGroup();
+ taskIOMetricGroup.registerBackPressureListener(mainInputActivityClock);
+
initReader();
// in the future when we this one is migrated to the "eager initialization" operator
@@ -332,11 +345,14 @@ public class SourceOperator<OUT, SplitT extends SourceSplit> extends AbstractStr
watermarkStrategy,
sourceMetricGroup,
getProcessingTimeService(),
- getExecutionConfig().getAutoWatermarkInterval());
+ getExecutionConfig().getAutoWatermarkInterval(),
+ mainInputActivityClock,
+ getProcessingTimeService().getClock(),
+ taskIOMetricGroup);
} else {
eventTimeLogic =
TimestampsAndWatermarks.createNoOpEventTimeLogic(
- watermarkStrategy, sourceMetricGroup);
+ watermarkStrategy, sourceMetricGroup, mainInputActivityClock);
}
// restore the state if necessary.
@@ -396,6 +412,12 @@ public class SourceOperator<OUT, SplitT extends SourceSplit> extends AbstractStr
@Override
public void close() throws Exception {
+ getContainingTask()
+ .getEnvironment()
+ .getMetricGroup()
+ .getIOMetricGroup()
+ .unregisterBackPressureListener(mainInputActivityClock);
+
if (sourceReader != null) {
sourceReader.close();
}
@@ -675,6 +697,7 @@ public class SourceOperator<OUT, SplitT extends SourceSplit> extends AbstractStr
Collection<String> splitsToPause, Collection<String> splitsToResume) {
try {
sourceReader.pauseOrResumeSplits(splitsToPause, splitsToResume);
+ eventTimeLogic.pauseOrResumeSplits(splitsToPause, splitsToResume);
} catch (UnsupportedOperationException e) {
if (!allowUnalignedSourceSplits) {
throw e;
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/NoOpTimestampsAndWatermarks.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/NoOpTimestampsAndWatermarks.java
index 3d21512a143..b2debd089b5 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/NoOpTimestampsAndWatermarks.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/NoOpTimestampsAndWatermarks.java
@@ -27,6 +27,8 @@ import org.apache.flink.streaming.runtime.io.PushingAsyncDataInput;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.runtime.tasks.ExceptionInChainedOperatorException;
+import java.util.Collection;
+
import static org.apache.flink.util.Preconditions.checkNotNull;
/**
@@ -68,6 +70,10 @@ public class NoOpTimestampsAndWatermarks<T> implements TimestampsAndWatermarks<T
// do nothing
}
+ @Override
+ public void pauseOrResumeSplits(
+ Collection<String> splitsToPause, Collection<String> splitsToResume) {}
+
// ------------------------------------------------------------------------
/**
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/ProgressiveTimestampsAndWatermarks.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/ProgressiveTimestampsAndWatermarks.java
index 98507c096b9..ca0c5f47b9b 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/ProgressiveTimestampsAndWatermarks.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/ProgressiveTimestampsAndWatermarks.java
@@ -27,16 +27,23 @@ import org.apache.flink.api.common.eventtime.WatermarkOutput;
import org.apache.flink.api.common.eventtime.WatermarkOutputMultiplexer;
import org.apache.flink.api.connector.source.ReaderOutput;
import org.apache.flink.api.connector.source.SourceOutput;
+import org.apache.flink.runtime.metrics.groups.TaskIOMetricGroup;
+import org.apache.flink.streaming.api.operators.util.PausableRelativeClock;
import org.apache.flink.streaming.runtime.io.PushingAsyncDataInput;
import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
+import org.apache.flink.util.clock.Clock;
+import org.apache.flink.util.clock.RelativeClock;
import javax.annotation.Nullable;
import java.time.Duration;
+import java.util.Collection;
+import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.concurrent.ScheduledFuture;
+import static java.util.Objects.requireNonNull;
import static org.apache.flink.util.Preconditions.checkState;
/**
@@ -53,12 +60,18 @@ public class ProgressiveTimestampsAndWatermarks<T> implements TimestampsAndWater
private final WatermarkGeneratorSupplier<T> watermarksFactory;
- private final WatermarkGeneratorSupplier.Context watermarksContext;
+ private final TimestampsAndWatermarksContextProvider watermarksContextProvider;
private final ProcessingTimeService timeService;
private final long periodicWatermarkInterval;
+ private final RelativeClock mainInputActivityClock;
+
+ private final Clock clock;
+
+ private final TaskIOMetricGroup taskIOMetricGroup;
+
@Nullable private SplitLocalOutputs<T> currentPerSplitOutputs;
@Nullable private StreamingReaderOutput<T> currentMainOutput;
@@ -68,14 +81,20 @@ public class ProgressiveTimestampsAndWatermarks<T> implements TimestampsAndWater
public ProgressiveTimestampsAndWatermarks(
TimestampAssigner<T> timestampAssigner,
WatermarkGeneratorSupplier<T> watermarksFactory,
- WatermarkGeneratorSupplier.Context watermarksContext,
+ TimestampsAndWatermarksContextProvider watermarksContextProvider,
ProcessingTimeService timeService,
- Duration periodicWatermarkInterval) {
+ Duration periodicWatermarkInterval,
+ RelativeClock mainInputActivityClock,
+ Clock clock,
+ TaskIOMetricGroup taskIOMetricGroup) {
this.timestampAssigner = timestampAssigner;
this.watermarksFactory = watermarksFactory;
- this.watermarksContext = watermarksContext;
+ this.watermarksContextProvider = watermarksContextProvider;
this.timeService = timeService;
+ this.mainInputActivityClock = mainInputActivityClock;
+ this.clock = clock;
+ this.taskIOMetricGroup = taskIOMetricGroup;
long periodicWatermarkIntervalMillis;
try {
@@ -106,7 +125,8 @@ public class ProgressiveTimestampsAndWatermarks<T> implements TimestampsAndWater
IdlenessManager idlenessManager = new IdlenessManager(watermarkOutput);
final WatermarkGenerator<T> watermarkGenerator =
- watermarksFactory.createWatermarkGenerator(watermarksContext);
+ watermarksFactory.createWatermarkGenerator(
+ watermarksContextProvider.create(mainInputActivityClock));
currentPerSplitOutputs =
new SplitLocalOutputs<>(
@@ -115,7 +135,9 @@ public class ProgressiveTimestampsAndWatermarks<T> implements TimestampsAndWater
watermarkUpdateListener,
timestampAssigner,
watermarksFactory,
- watermarksContext);
+ watermarksContextProvider,
+ clock,
+ taskIOMetricGroup);
currentMainOutput =
new StreamingReaderOutput<>(
@@ -162,6 +184,12 @@ public class ProgressiveTimestampsAndWatermarks<T> implements TimestampsAndWater
}
}
+ @Override
+ public void pauseOrResumeSplits(
+ Collection<String> splitsToPause, Collection<String> splitsToResume) {
+ currentPerSplitOutputs.pauseOrResumeSplits(splitsToPause, splitsToResume);
+ }
+
// ------------------------------------------------------------------------
private static final class StreamingReaderOutput<T> extends SourceOutputWithWatermarks<T>
@@ -203,11 +231,14 @@ public class ProgressiveTimestampsAndWatermarks<T> implements TimestampsAndWater
private final WatermarkOutputMultiplexer watermarkMultiplexer;
private final Map<String, SourceOutputWithWatermarks<T>> localOutputs;
+ private final Map<String, PausableRelativeClock> inputActivityClocks = new HashMap<>();
private final PushingAsyncDataInput.DataOutput<T> recordOutput;
private final TimestampAssigner<T> timestampAssigner;
private final WatermarkGeneratorSupplier<T> watermarksFactory;
- private final WatermarkGeneratorSupplier.Context watermarkContext;
+ private final TimestampsAndWatermarksContextProvider watermarksContextProvider;
private final WatermarkUpdateListener watermarkUpdateListener;
+ private final Clock clock;
+ private final TaskIOMetricGroup taskIOMetricGroup;
private SplitLocalOutputs(
PushingAsyncDataInput.DataOutput<T> recordOutput,
@@ -215,13 +246,17 @@ public class ProgressiveTimestampsAndWatermarks<T> implements TimestampsAndWater
WatermarkUpdateListener watermarkUpdateListener,
TimestampAssigner<T> timestampAssigner,
WatermarkGeneratorSupplier<T> watermarksFactory,
- WatermarkGeneratorSupplier.Context watermarkContext) {
+ TimestampsAndWatermarksContextProvider watermarksContextProvider,
+ Clock clock,
+ TaskIOMetricGroup taskIOMetricGroup) {
this.recordOutput = recordOutput;
this.timestampAssigner = timestampAssigner;
this.watermarksFactory = watermarksFactory;
- this.watermarkContext = watermarkContext;
+ this.watermarksContextProvider = watermarksContextProvider;
this.watermarkUpdateListener = watermarkUpdateListener;
+ this.clock = clock;
+ this.taskIOMetricGroup = taskIOMetricGroup;
this.watermarkMultiplexer = new WatermarkOutputMultiplexer(watermarkOutput);
this.localOutputs =
@@ -234,6 +269,7 @@ public class ProgressiveTimestampsAndWatermarks<T> implements TimestampsAndWater
return previous;
}
+ PausableRelativeClock inputActivityClock = createInputActivityClock(splitId);
watermarkMultiplexer.registerNewOutput(
splitId,
watermark ->
@@ -243,7 +279,8 @@ public class ProgressiveTimestampsAndWatermarks<T> implements TimestampsAndWater
final WatermarkOutput periodicOutput = watermarkMultiplexer.getDeferredOutput(splitId);
final WatermarkGenerator<T> watermarks =
- watermarksFactory.createWatermarkGenerator(watermarkContext);
+ watermarksFactory.createWatermarkGenerator(
+ watermarksContextProvider.create(inputActivityClock));
final SourceOutputWithWatermarks<T> localOutput =
SourceOutputWithWatermarks.createWithSeparateOutputs(
@@ -257,9 +294,21 @@ public class ProgressiveTimestampsAndWatermarks<T> implements TimestampsAndWater
return localOutput;
}
+ private PausableRelativeClock createInputActivityClock(String splitId) {
+ // Dedicated inputActivityClock for a particular split. It will be paused both in case
+ // of back pressure and when split is paused due to watermark alignment.
+ PausableRelativeClock inputActivityClock = new PausableRelativeClock(clock);
+ inputActivityClocks.put(splitId, inputActivityClock);
+ taskIOMetricGroup.registerBackPressureListener(inputActivityClock);
+ return inputActivityClock;
+ }
+
void releaseOutputForSplit(String splitId) {
localOutputs.remove(splitId);
watermarkMultiplexer.unregisterOutput(splitId);
+ PausableRelativeClock inputActivityClock =
+ requireNonNull(inputActivityClocks.remove(splitId));
+ taskIOMetricGroup.unregisterBackPressureListener(inputActivityClock);
}
void emitPeriodicWatermark() {
@@ -273,6 +322,16 @@ public class ProgressiveTimestampsAndWatermarks<T> implements TimestampsAndWater
}
watermarkMultiplexer.onPeriodicEmit();
}
+
+ public void pauseOrResumeSplits(
+ Collection<String> splitsToPause, Collection<String> splitsToResume) {
+ for (String splitId : splitsToPause) {
+ inputActivityClocks.get(splitId).pause();
+ }
+ for (String splitId : splitsToResume) {
+ inputActivityClocks.get(splitId).unPause();
+ }
+ }
}
/**
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/TimestampsAndWatermarks.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/TimestampsAndWatermarks.java
index 84bb7457601..b6e1a95acaf 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/TimestampsAndWatermarks.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/TimestampsAndWatermarks.java
@@ -24,10 +24,14 @@ import org.apache.flink.api.common.eventtime.WatermarkOutput;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.connector.source.ReaderOutput;
import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.runtime.metrics.groups.TaskIOMetricGroup;
import org.apache.flink.streaming.runtime.io.PushingAsyncDataInput;
import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
+import org.apache.flink.util.clock.Clock;
+import org.apache.flink.util.clock.RelativeClock;
import java.time.Duration;
+import java.util.Collection;
/**
* Basic interface for the timestamp extraction and watermark generation logic for the {@link
@@ -83,6 +87,8 @@ public interface TimestampsAndWatermarks<T> {
/** Emit a watermark immediately. */
void emitImmediateWatermark(long wallClockTimestamp);
+ void pauseOrResumeSplits(Collection<String> splitsToPause, Collection<String> splitsToResume);
+
// ------------------------------------------------------------------------
// factories
// ------------------------------------------------------------------------
@@ -91,27 +97,52 @@ public interface TimestampsAndWatermarks<T> {
WatermarkStrategy<E> watermarkStrategy,
MetricGroup metrics,
ProcessingTimeService timeService,
- long periodicWatermarkIntervalMillis) {
+ long periodicWatermarkIntervalMillis,
+ RelativeClock mainInputActivityClock,
+ Clock clock,
+ TaskIOMetricGroup taskIOMetricGroup) {
- final TimestampsAndWatermarksContext context = new TimestampsAndWatermarksContext(metrics);
- final TimestampAssigner<E> timestampAssigner =
- watermarkStrategy.createTimestampAssigner(context);
+ TimestampsAndWatermarksContextProvider contextProvider =
+ new TimestampsAndWatermarksContextProvider(metrics);
+ TimestampAssigner<E> timestampAssigner =
+ watermarkStrategy.createTimestampAssigner(
+ contextProvider.create(mainInputActivityClock));
return new ProgressiveTimestampsAndWatermarks<>(
timestampAssigner,
watermarkStrategy,
- context,
+ contextProvider,
timeService,
- Duration.ofMillis(periodicWatermarkIntervalMillis));
+ Duration.ofMillis(periodicWatermarkIntervalMillis),
+ mainInputActivityClock,
+ clock,
+ taskIOMetricGroup);
}
static <E> TimestampsAndWatermarks<E> createNoOpEventTimeLogic(
- WatermarkStrategy<E> watermarkStrategy, MetricGroup metrics) {
+ WatermarkStrategy<E> watermarkStrategy,
+ MetricGroup metrics,
+ RelativeClock inputActivityClock) {
- final TimestampsAndWatermarksContext context = new TimestampsAndWatermarksContext(metrics);
+ final TimestampsAndWatermarksContext context =
+ new TimestampsAndWatermarksContext(metrics, inputActivityClock);
final TimestampAssigner<E> timestampAssigner =
watermarkStrategy.createTimestampAssigner(context);
return new NoOpTimestampsAndWatermarks<>(timestampAssigner);
}
+
+ /** Helper class to construct {@link TimestampsAndWatermarksContext}. */
+ @Internal
+ class TimestampsAndWatermarksContextProvider {
+ private final MetricGroup metrics;
+
+ public TimestampsAndWatermarksContextProvider(MetricGroup metrics) {
+ this.metrics = metrics;
+ }
+
+ public TimestampsAndWatermarksContext create(RelativeClock inputActivityClock) {
+ return new TimestampsAndWatermarksContext(metrics, inputActivityClock);
+ }
+ }
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/TimestampsAndWatermarksContext.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/TimestampsAndWatermarksContext.java
index 8ec6cc8c01f..6f543e0392c 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/TimestampsAndWatermarksContext.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/TimestampsAndWatermarksContext.java
@@ -23,7 +23,6 @@ import org.apache.flink.api.common.eventtime.TimestampAssignerSupplier;
import org.apache.flink.api.common.eventtime.WatermarkGeneratorSupplier;
import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.util.clock.RelativeClock;
-import org.apache.flink.util.clock.SystemClock;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -36,9 +35,12 @@ public final class TimestampsAndWatermarksContext
implements TimestampAssignerSupplier.Context, WatermarkGeneratorSupplier.Context {
private final MetricGroup metricGroup;
+ private final RelativeClock inputActivityClock;
- public TimestampsAndWatermarksContext(MetricGroup metricGroup) {
+ public TimestampsAndWatermarksContext(
+ MetricGroup metricGroup, RelativeClock inputActivityClock) {
this.metricGroup = checkNotNull(metricGroup);
+ this.inputActivityClock = inputActivityClock;
}
@Override
@@ -48,6 +50,6 @@ public final class TimestampsAndWatermarksContext
@Override
public RelativeClock getInputActivityClock() {
- return SystemClock.getInstance();
+ return inputActivityClock;
}
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorSplitWatermarkAlignmentTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorSplitWatermarkAlignmentTest.java
index 8be17a021d5..1a199b23123 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorSplitWatermarkAlignmentTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorSplitWatermarkAlignmentTest.java
@@ -29,6 +29,7 @@ import org.apache.flink.api.connector.source.mocks.MockSourceSplit;
import org.apache.flink.api.connector.source.mocks.MockSourceSplitSerializer;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.execution.Environment;
+import org.apache.flink.runtime.metrics.groups.TaskIOMetricGroup;
import org.apache.flink.runtime.operators.coordination.MockOperatorEventGateway;
import org.apache.flink.runtime.operators.testutils.MockInputSplitProvider;
import org.apache.flink.runtime.source.event.AddSplitEvent;
@@ -40,6 +41,7 @@ import org.apache.flink.streaming.api.operators.source.TestingSourceOperator;
import org.apache.flink.streaming.runtime.tasks.SourceOperatorStreamTask;
import org.apache.flink.streaming.runtime.tasks.StreamMockEnvironment;
import org.apache.flink.streaming.runtime.tasks.TestProcessingTimeService;
+import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus;
import org.apache.flink.streaming.util.MockOutput;
import org.apache.flink.streaming.util.MockStreamConfig;
@@ -110,6 +112,144 @@ class SourceOperatorSplitWatermarkAlignmentTest {
assertThat(sourceReader.getPausedSplits()).containsExactly("0", "1");
}
+ @Test
+ void testBackpressureAndIdleness() throws Exception {
+ long idleTimeout = 100;
+ MockSourceReader sourceReader =
+ new MockSourceReader(WaitingForSplits.DO_NOT_WAIT_FOR_SPLITS, false, true);
+ TestProcessingTimeService processingTimeService = new TestProcessingTimeService();
+ SourceOperator<Integer, MockSourceSplit> operator =
+ createAndOpenSourceOperatorWithIdleness(
+ sourceReader, processingTimeService, idleTimeout);
+
+ /**
+ * Intention behind this setup is that split0 emits a couple of records, while we keep
+ * advancing processing time and keep firing timers. Normally split1 would switch to idle
+ * first (it hasn't emitted any records), which would cause a watermark from split0 to be
+ * emitted and then WatermarkStatus.IDLE should be emitted after split0 also switches to
+ * idle. However we assert that neither watermark no idle status this doesn't happen due to
+ * the back pressure status.
+ */
+ MockSourceSplit split0 = new MockSourceSplit(0, 0, 10).addRecord(42).addRecord(44);
+ MockSourceSplit split1 = new MockSourceSplit(1, 10, 20);
+ operator.handleOperatorEvent(
+ new AddSplitEvent<>(
+ Arrays.asList(split0, split1), new MockSourceSplitSerializer()));
+
+ CollectingDataOutput<Integer> dataOutput = new CollectingDataOutput<>();
+
+ // Output is initialised by the SourceOperator on the first emitNext invocation
+ operator.emitNext(dataOutput);
+
+ TaskIOMetricGroup taskIOMetricGroup =
+ operator.getContainingTask().getEnvironment().getMetricGroup().getIOMetricGroup();
+ taskIOMetricGroup.getHardBackPressuredTimePerSecond().markStart();
+
+ for (int i = 0; i < 10; i++) {
+ processingTimeService.advance(idleTimeout);
+ operator.emitNext(dataOutput);
+ }
+ assertThat(dataOutput.getEvents()).doesNotContain(WatermarkStatus.IDLE);
+ assertThat(dataOutput.getEvents()).doNotHave(new AnyWatermark());
+
+ taskIOMetricGroup.getHardBackPressuredTimePerSecond().markEnd();
+ taskIOMetricGroup.getSoftBackPressuredTimePerSecond().markStart();
+
+ for (int i = 0; i < 10; i++) {
+ processingTimeService.advance(idleTimeout);
+ }
+ assertThat(dataOutput.getEvents()).doesNotContain(WatermarkStatus.IDLE);
+ assertThat(dataOutput.getEvents()).doNotHave(new AnyWatermark());
+
+ taskIOMetricGroup.getSoftBackPressuredTimePerSecond().markEnd();
+
+ for (int i = 0; i < 10; i++) {
+ processingTimeService.advance(idleTimeout);
+ }
+
+ assertThat(dataOutput.getEvents()).contains(WatermarkStatus.IDLE);
+ assertThat(dataOutput.getEvents()).doNotHave(new AnyWatermark());
+ }
+
+ @Test
+ void testSplitWatermarkAlignmentAndIdleness() throws Exception {
+ long idleTimeout = 100;
+ MockSourceReader sourceReader =
+ new MockSourceReader(WaitingForSplits.DO_NOT_WAIT_FOR_SPLITS, false, true);
+ TestProcessingTimeService processingTimeService = new TestProcessingTimeService();
+ SourceOperator<Integer, MockSourceSplit> operator =
+ createAndOpenSourceOperatorWithIdleness(
+ sourceReader, processingTimeService, idleTimeout);
+
+ MockSourceSplit split0 = new MockSourceSplit(0, 0, 10);
+ MockSourceSplit split1 = new MockSourceSplit(1, 10, 20);
+ int maxAllowedWatermark = 4;
+ int maxEmittedWatermark = maxAllowedWatermark + 1;
+ // the intention is that only first record from split0 gets emitted, then split0 gets
+ // blocked and record (maxEmittedWatermark + 100) is never emitted from split0
+ split0.addRecord(maxEmittedWatermark).addRecord(maxEmittedWatermark + 100);
+ split1.addRecord(3)
+ .addRecord(3)
+ .addRecord(3)
+ .addRecord(3)
+ .addRecord(3)
+ .addRecord(3)
+ .addRecord(3);
+ split1.addRecord(maxEmittedWatermark + 100);
+
+ operator.handleOperatorEvent(
+ new AddSplitEvent<>(
+ Arrays.asList(split0, split1), new MockSourceSplitSerializer()));
+ CollectingDataOutput<Integer> dataOutput = new CollectingDataOutput<>();
+
+ operator.emitNext(dataOutput); // split0 emits first (and only) record (maxEmittedWatermark)
+
+ operator.handleOperatorEvent(
+ new WatermarkAlignmentEvent(maxAllowedWatermark)); // blocks split0
+ assertThat(sourceReader.getPausedSplits()).containsExactly("0");
+
+ while (operator.isAvailable()) {
+ // We are advancing a couple of times by (idleTimeout - 1) to make sure the active input
+ // never switches idle, while giving plenty of time for the blocked split0 to evaluate
+ // it's idle state
+ processingTimeService.advance(idleTimeout - 1);
+ operator.emitNext(dataOutput); // split1 keeps emitting records
+ }
+ // in the end, all records are emitted from split1. This shouldn't cause the watermark to
+ // get bumped above maxEmittedWatermark, as split0 shouldn't be idle and it is still
+ // blocked.
+ assertThat(sourceReader.getPausedSplits()).containsExactly("0", "1");
+ assertThat(dataOutput.getEvents()).doNotHave(new WatermarkAbove(maxEmittedWatermark));
+ }
+
+ private SourceOperator<Integer, MockSourceSplit> createAndOpenSourceOperatorWithIdleness(
+ MockSourceReader sourceReader,
+ TestProcessingTimeService processingTimeService,
+ long idleTimeout)
+ throws Exception {
+
+ SourceOperator<Integer, MockSourceSplit> operator =
+ new TestingSourceOperator<>(
+ sourceReader,
+ WatermarkStrategy.forGenerator(ctx -> new TestWatermarkGenerator())
+ .withTimestampAssigner((r, l) -> r)
+ .withWatermarkAlignment("group-1", Duration.ofMillis(1))
+ .withIdleness(Duration.ofMillis(idleTimeout)),
+ processingTimeService,
+ new MockOperatorEventGateway(),
+ 1,
+ 5,
+ true);
+ Environment env = getTestingEnvironment();
+ operator.setup(
+ new SourceOperatorStreamTask<Integer>(env),
+ new MockStreamConfig(new Configuration(), 1),
+ new MockOutput<>(new ArrayList<>()));
+ operator.initializeState(new StreamTaskStateInitializerImpl(env, new MemoryStateBackend()));
+ operator.open();
+ return operator;
+ }
+
private Environment getTestingEnvironment() {
return new StreamMockEnvironment(
new Configuration(),
@@ -138,4 +278,31 @@ class SourceOperatorSplitWatermarkAlignmentTest {
output.emitWatermark(new Watermark(maxWatermark));
}
}
+
+ /** Condition checking if there is no watermark above a certain value among StreamElements. */
+ public static class WatermarkAbove extends Condition<Object> {
+ public WatermarkAbove(int maxEmittedWatermark) {
+ super(
+ event -> {
+ if (!(event
+ instanceof org.apache.flink.streaming.api.watermark.Watermark)) {
+ return false;
+ }
+ org.apache.flink.streaming.api.watermark.Watermark w =
+ (org.apache.flink.streaming.api.watermark.Watermark) event;
+ return w.getTimestamp() > maxEmittedWatermark;
+ },
+ "watermark value of greater than %d",
+ maxEmittedWatermark);
+ }
+ }
+
+ /** Condition checking if there is any watermark among StreamElements. */
+ public static class AnyWatermark extends Condition<Object> {
+ public AnyWatermark() {
+ super(
+ event -> event instanceof org.apache.flink.streaming.api.watermark.Watermark,
+ "any watermark");
+ }
+ }
}
(flink) 11/12: [FLINK-35886][source] Do not track already finished splits in watermark alignment
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch release-1.19
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 45a2e04e4140e8d4789246b0c9e874f5c850cf89
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Wed Aug 14 12:57:02 2024 +0200
[FLINK-35886][source] Do not track already finished splits in watermark alignment
---
.../connector/source/mocks/MockSourceReader.java | 14 +++++++++
.../streaming/api/operators/SourceOperator.java | 5 ++++
.../source/ProgressiveTimestampsAndWatermarks.java | 1 +
.../operators/source/TimestampsAndWatermarks.java | 3 ++
.../operators/source/WatermarkToDataOutput.java | 3 ++
.../SourceOperatorSplitWatermarkAlignmentTest.java | 33 ++++++++++++++++++++++
6 files changed, 59 insertions(+)
diff --git a/flink-core/src/test/java/org/apache/flink/api/connector/source/mocks/MockSourceReader.java b/flink-core/src/test/java/org/apache/flink/api/connector/source/mocks/MockSourceReader.java
index facfcf346cd..b85a2e075bd 100644
--- a/flink-core/src/test/java/org/apache/flink/api/connector/source/mocks/MockSourceReader.java
+++ b/flink-core/src/test/java/org/apache/flink/api/connector/source/mocks/MockSourceReader.java
@@ -28,6 +28,7 @@ import javax.annotation.concurrent.GuardedBy;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashSet;
+import java.util.Iterator;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
@@ -92,6 +93,7 @@ public class MockSourceReader implements SourceReader<Integer, MockSourceSplit>
@Override
public InputStatus pollNext(ReaderOutput<Integer> sourceOutput) throws Exception {
+ releaseFinishedSplits(sourceOutput);
if (waitingForSplitsBehaviour == WaitingForSplits.WAIT_FOR_INITIAL
&& splitsAssignmentState == SplitsAssignmentState.NO_SPLITS_ASSIGNED) {
@@ -141,6 +143,18 @@ public class MockSourceReader implements SourceReader<Integer, MockSourceSplit>
}
}
+ private void releaseFinishedSplits(ReaderOutput<Integer> sourceOutput) {
+ Iterator<MockSourceSplit> assignedSplitsIterator = assignedSplits.iterator();
+ while (assignedSplitsIterator.hasNext()) {
+ MockSourceSplit assignedSplit = assignedSplitsIterator.next();
+ if (assignedSplit.isFinished()) {
+ sourceOutput.releaseOutputForSplit(assignedSplit.splitId());
+ assignedSplitsIterator.remove();
+ pausedSplits.remove(assignedSplit.splitId());
+ }
+ }
+ }
+
@Override
public List<MockSourceSplit> snapshotState(long checkpointId) {
return assignedSplits;
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java
index b26edc77c63..ea6e39f4e66 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java
@@ -663,6 +663,11 @@ public class SourceOperator<OUT, SplitT extends SourceSplit> extends AbstractStr
}
}
+ @Override
+ public void splitFinished(String splitId) {
+ splitCurrentWatermarks.remove(splitId);
+ }
+
/**
* Finds the splits that are beyond the current max watermark and pauses them. At the same time,
* splits that have been paused and where the global watermark caught up are resumed.
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/ProgressiveTimestampsAndWatermarks.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/ProgressiveTimestampsAndWatermarks.java
index ca0c5f47b9b..5b96c5dc0c9 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/ProgressiveTimestampsAndWatermarks.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/ProgressiveTimestampsAndWatermarks.java
@@ -304,6 +304,7 @@ public class ProgressiveTimestampsAndWatermarks<T> implements TimestampsAndWater
}
void releaseOutputForSplit(String splitId) {
+ watermarkUpdateListener.splitFinished(splitId);
localOutputs.remove(splitId);
watermarkMultiplexer.unregisterOutput(splitId);
PausableRelativeClock inputActivityClock =
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/TimestampsAndWatermarks.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/TimestampsAndWatermarks.java
index b6e1a95acaf..91e2cb968dd 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/TimestampsAndWatermarks.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/TimestampsAndWatermarks.java
@@ -62,6 +62,9 @@ public interface TimestampsAndWatermarks<T> {
/** Notifies about changes to per split watermarks. */
void updateCurrentSplitWatermark(String splitId, long watermark);
+
+ /** Notifies that split has finished. */
+ void splitFinished(String splitId);
}
/**
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/WatermarkToDataOutput.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/WatermarkToDataOutput.java
index d4d81b64f45..4fcf46ca9f9 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/WatermarkToDataOutput.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/WatermarkToDataOutput.java
@@ -53,6 +53,9 @@ public final class WatermarkToDataOutput implements WatermarkOutput {
@Override
public void updateCurrentSplitWatermark(String splitId, long watermark) {}
+
+ @Override
+ public void splitFinished(String splitId) {}
});
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorSplitWatermarkAlignmentTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorSplitWatermarkAlignmentTest.java
index 1a199b23123..93c7072842e 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorSplitWatermarkAlignmentTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorSplitWatermarkAlignmentTest.java
@@ -38,6 +38,7 @@ import org.apache.flink.runtime.state.TestTaskStateManager;
import org.apache.flink.runtime.state.memory.MemoryStateBackend;
import org.apache.flink.streaming.api.operators.source.CollectingDataOutput;
import org.apache.flink.streaming.api.operators.source.TestingSourceOperator;
+import org.apache.flink.streaming.runtime.io.DataInputStatus;
import org.apache.flink.streaming.runtime.tasks.SourceOperatorStreamTask;
import org.apache.flink.streaming.runtime.tasks.StreamMockEnvironment;
import org.apache.flink.streaming.runtime.tasks.TestProcessingTimeService;
@@ -222,6 +223,38 @@ class SourceOperatorSplitWatermarkAlignmentTest {
assertThat(dataOutput.getEvents()).doNotHave(new WatermarkAbove(maxEmittedWatermark));
}
+ @Test
+ void testSplitWatermarkAlignmentWithFinishedSplit() throws Exception {
+ long idleTimeout = 100;
+ MockSourceReader sourceReader =
+ new MockSourceReader(WaitingForSplits.DO_NOT_WAIT_FOR_SPLITS, false, true);
+ TestProcessingTimeService processingTimeService = new TestProcessingTimeService();
+ SourceOperator<Integer, MockSourceSplit> operator =
+ createAndOpenSourceOperatorWithIdleness(
+ sourceReader, processingTimeService, idleTimeout);
+
+ MockSourceSplit split0 = new MockSourceSplit(0, 0, 1);
+ MockSourceSplit split1 = new MockSourceSplit(1, 10, 20);
+ int maxAllowedWatermark = 4;
+ int maxEmittedWatermark = maxAllowedWatermark + 1;
+ // the intention is that only first record from split0 gets emitted, then split0 gets
+ // blocked and record (maxEmittedWatermark + 100) is never emitted from split0
+ split0.addRecord(maxEmittedWatermark);
+ split1.addRecord(3);
+
+ operator.handleOperatorEvent(
+ new AddSplitEvent<>(
+ Arrays.asList(split0, split1), new MockSourceSplitSerializer()));
+ CollectingDataOutput<Integer> dataOutput = new CollectingDataOutput<>();
+
+ while (operator.emitNext(dataOutput) == DataInputStatus.MORE_AVAILABLE) {
+ // split0 emits its only record and is finished/released
+ }
+ operator.handleOperatorEvent(
+ new WatermarkAlignmentEvent(maxAllowedWatermark)); // blocks split0
+ assertThat(sourceReader.getPausedSplits()).isEmpty();
+ }
+
private SourceOperator<Integer, MockSourceSplit> createAndOpenSourceOperatorWithIdleness(
MockSourceReader sourceReader,
TestProcessingTimeService processingTimeService,
(flink) 04/12: [FLINK-35886][task] Support markStart and markEnd listeners in TimerGauge
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch release-1.19
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 7cd9476825237dac441812cdca178f689f757b14
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Tue Aug 6 16:04:23 2024 +0200
[FLINK-35886][task] Support markStart and markEnd listeners in TimerGauge
---
.../apache/flink/runtime/metrics/TimerGauge.java | 60 ++++++++++++++++++----
.../flink/runtime/metrics/TimerGaugeTest.java | 58 +++++++++++++++++++++
2 files changed, 108 insertions(+), 10 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/TimerGauge.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/TimerGauge.java
index 645121694cf..afab5a73a0a 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/TimerGauge.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/TimerGauge.java
@@ -24,6 +24,9 @@ import org.apache.flink.metrics.View;
import org.apache.flink.util.clock.Clock;
import org.apache.flink.util.clock.SystemClock;
+import java.util.ArrayList;
+import java.util.Collection;
+
/**
* {@link TimerGauge} measures how much time is spent in a given state, with entry into that state
* being signaled by {@link #markStart()}. Measuring is stopped by {@link #markEnd()}. This class in
@@ -37,6 +40,8 @@ public class TimerGauge implements Gauge<Long>, View {
private final Clock clock;
+ private final Collection<StartStopListener> startStopListeners = new ArrayList<>();
+
/** The time-span over which the average is calculated. */
private final int timeSpanInSeconds;
/** Circular array containing the history of values. */
@@ -82,21 +87,43 @@ public class TimerGauge implements Gauge<Long>, View {
this.values = new long[this.timeSpanInSeconds / UPDATE_INTERVAL_SECONDS];
}
+ public synchronized void registerListener(StartStopListener listener) {
+ if (currentMeasurementStartTS != 0) {
+ listener.markStart();
+ }
+ startStopListeners.add(listener);
+ }
+
+ public synchronized void unregisterListener(StartStopListener listener) {
+ if (currentMeasurementStartTS != 0) {
+ listener.markEnd();
+ }
+ startStopListeners.remove(listener);
+ }
+
public synchronized void markStart() {
- if (currentMeasurementStartTS == 0) {
- currentUpdateTS = clock.absoluteTimeMillis();
- currentMeasurementStartTS = currentUpdateTS;
+ if (currentMeasurementStartTS != 0) {
+ return;
+ }
+ currentUpdateTS = clock.absoluteTimeMillis();
+ currentMeasurementStartTS = currentUpdateTS;
+ for (StartStopListener startStopListener : startStopListeners) {
+ startStopListener.markStart();
}
}
public synchronized void markEnd() {
- if (currentMeasurementStartTS != 0) {
- long currentMeasurement = clock.absoluteTimeMillis() - currentMeasurementStartTS;
- currentCount += currentMeasurement;
- accumulatedCount += currentMeasurement;
- currentMaxSingleMeasurement = Math.max(currentMaxSingleMeasurement, currentMeasurement);
- currentUpdateTS = 0;
- currentMeasurementStartTS = 0;
+ if (currentMeasurementStartTS == 0) {
+ return;
+ }
+ long currentMeasurement = clock.absoluteTimeMillis() - currentMeasurementStartTS;
+ currentCount += currentMeasurement;
+ accumulatedCount += currentMeasurement;
+ currentMaxSingleMeasurement = Math.max(currentMaxSingleMeasurement, currentMeasurement);
+ currentUpdateTS = 0;
+ currentMeasurementStartTS = 0;
+ for (StartStopListener startStopListener : startStopListeners) {
+ startStopListener.markEnd();
}
}
@@ -164,4 +191,17 @@ public class TimerGauge implements Gauge<Long>, View {
public synchronized boolean isMeasuring() {
return currentMeasurementStartTS != 0;
}
+
+ /**
+ * Listens for {@link TimerGauge#markStart()} and {@link TimerGauge#markEnd()} events.
+ *
+ * <p>Beware! As it is right now, {@link StartStopListener} is notified under the {@link
+ * TimerGauge}'s lock, so those callbacks should be very short, without long call stacks that
+ * acquire more locks. Otherwise, a potential for deadlocks can be introduced.
+ */
+ public interface StartStopListener {
+ void markStart();
+
+ void markEnd();
+ }
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/TimerGaugeTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/TimerGaugeTest.java
index 7416e3292f7..ebc2d7c313a 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/TimerGaugeTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/TimerGaugeTest.java
@@ -28,6 +28,7 @@ package org.apache.flink.runtime.metrics;
import org.apache.flink.metrics.View;
import org.apache.flink.util.clock.ManualClock;
+import org.apache.flink.util.clock.SystemClock;
import org.junit.jupiter.api.Test;
@@ -148,4 +149,61 @@ class TimerGaugeTest {
assertThat(gauge.getMaxSingleMeasurement()).isEqualTo(SLEEP / 2);
assertThat(gauge.getAccumulatedCount()).isEqualTo(3 * SLEEP + SLEEP / 2);
}
+
+ @Test
+ void testListeners() {
+ TimerGauge gauge = new TimerGauge(SystemClock.getInstance(), View.UPDATE_INTERVAL_SECONDS);
+ TestStartStopListener listener1 = new TestStartStopListener();
+ TestStartStopListener listener2 = new TestStartStopListener();
+
+ gauge.registerListener(listener1);
+
+ gauge.markStart();
+ listener1.assertCounts(1, 0);
+ gauge.markEnd();
+ listener1.assertCounts(1, 1);
+
+ gauge.markStart();
+ gauge.registerListener(listener2);
+ listener1.assertCounts(2, 1);
+ listener2.assertCounts(1, 0);
+ gauge.markEnd();
+ listener1.assertCounts(2, 2);
+ listener2.assertCounts(1, 1);
+
+ gauge.markStart();
+ gauge.unregisterListener(listener1);
+ listener1.assertCounts(3, 3);
+ listener2.assertCounts(2, 1);
+
+ gauge.markEnd();
+ listener2.assertCounts(2, 2);
+
+ gauge.unregisterListener(listener2);
+
+ gauge.markStart();
+ gauge.markEnd();
+ listener1.assertCounts(3, 3);
+ listener2.assertCounts(2, 2);
+ }
+
+ static class TestStartStopListener implements TimerGauge.StartStopListener {
+ long startCount;
+ long endCount;
+
+ @Override
+ public void markStart() {
+ startCount++;
+ }
+
+ @Override
+ public void markEnd() {
+ endCount++;
+ }
+
+ public void assertCounts(long expectedStart, long expectedEnd) {
+ assertThat(startCount).isEqualTo(expectedStart);
+ assertThat(endCount).isEqualTo(expectedEnd);
+ }
+ }
}