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 2021/09/06 07:54:10 UTC
[flink] branch release-1.14 updated (9c8d21f -> 5a68bb8)
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a change to branch release-1.14
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 9c8d21f [FLINK-24096][checkpoint] Skip the complete notification for checkpoints before task startup
new bf71226 [FLINK-24090][docs] Added Troubleshooting section with ignoring in-flight data explanation into unaligned checkpoints page
new 6fa6033 [refactor][streaming] Improved description of the calculation of the processing time delay
new 08481d9 [refactor][streaming] Busy waiting in checkScheduledTimestamps replaced by CompletableFuture
new 5a68bb8 [FLINK-23960][streaming] checkScheduledTimestamps changed according to semantic of the calculation processing time delay
The 4 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:
.../docs/ops/state/unaligned_checkpoints.md | 12 ++++
.../runtime/tasks/ProcessingTimeServiceUtil.java | 10 +++-
.../runtime/operators/StreamTaskTimerTest.java | 66 +++++++++++-----------
3 files changed, 52 insertions(+), 36 deletions(-)
[flink] 03/04: [refactor][streaming] Busy waiting in
checkScheduledTimestamps replaced by CompletableFuture
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.14
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 08481d95c0cd014ee48746352fad19d29e2eecce
Author: Anton Kalashnikov <ka...@yandex.ru>
AuthorDate: Wed Sep 1 15:32:35 2021 +0200
[refactor][streaming] Busy waiting in checkScheduledTimestamps replaced by CompletableFuture
---
.../runtime/operators/StreamTaskTimerTest.java | 55 ++++++++++------------
1 file changed, 24 insertions(+), 31 deletions(-)
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/StreamTaskTimerTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/StreamTaskTimerTest.java
index cab02b2..765075d 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/StreamTaskTimerTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/StreamTaskTimerTest.java
@@ -38,15 +38,17 @@ import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.Timeout;
-import javax.annotation.Nullable;
-
import java.time.Duration;
+import java.util.ArrayList;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicReference;
import static org.hamcrest.Matchers.instanceOf;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThat;
-import static org.junit.Assert.fail;
/** Tests for the timer service of {@link org.apache.flink.streaming.runtime.tasks.StreamTask}. */
@SuppressWarnings("serial")
@@ -107,26 +109,22 @@ public class StreamTaskTimerTest extends TestLogger {
@Test
public void checkScheduledTimestamps() throws Exception {
- final AtomicReference<Throwable> errorRef = new AtomicReference<>();
-
- final long t1 = System.currentTimeMillis();
- final long t2 = System.currentTimeMillis() - 200;
- final long t3 = System.currentTimeMillis() + 100;
- final long t4 = System.currentTimeMillis() + 200;
+ ValidatingProcessingTimeCallback.numInSequence = 0;
+ long currentTimeMillis = System.currentTimeMillis();
+ ArrayList<ValidatingProcessingTimeCallback> timeCallbacks = new ArrayList<>();
- timeService.registerTimer(t1, new ValidatingProcessingTimeCallback(errorRef, t1, 0));
- timeService.registerTimer(t2, new ValidatingProcessingTimeCallback(errorRef, t2, 1));
- timeService.registerTimer(t3, new ValidatingProcessingTimeCallback(errorRef, t3, 2));
- timeService.registerTimer(t4, new ValidatingProcessingTimeCallback(errorRef, t4, 3));
+ timeCallbacks.add(new ValidatingProcessingTimeCallback(currentTimeMillis, 0));
+ timeCallbacks.add(new ValidatingProcessingTimeCallback(currentTimeMillis - 200, 1));
+ timeCallbacks.add(new ValidatingProcessingTimeCallback(currentTimeMillis + 100, 2));
+ timeCallbacks.add(new ValidatingProcessingTimeCallback(currentTimeMillis + 200, 3));
- long deadline = System.currentTimeMillis() + 20000;
- while (errorRef.get() == null
- && ValidatingProcessingTimeCallback.numInSequence < 4
- && System.currentTimeMillis() < deadline) {
- Thread.sleep(100);
+ for (ValidatingProcessingTimeCallback timeCallback : timeCallbacks) {
+ timeService.registerTimer(timeCallback.expectedTimestamp, timeCallback);
}
- verifyNoException(errorRef.get());
+ for (ValidatingProcessingTimeCallback timeCallback : timeCallbacks) {
+ timeCallback.assertExpectedValues();
+ }
assertEquals(4, ValidatingProcessingTimeCallback.numInSequence);
}
@@ -134,16 +132,12 @@ public class StreamTaskTimerTest extends TestLogger {
static int numInSequence;
- private final AtomicReference<Throwable> errorRef;
+ private final CompletableFuture<Void> finished = new CompletableFuture<>();
private final long expectedTimestamp;
private final int expectedInSequence;
- private ValidatingProcessingTimeCallback(
- AtomicReference<Throwable> errorRef,
- long expectedTimestamp,
- int expectedInSequence) {
- this.errorRef = errorRef;
+ private ValidatingProcessingTimeCallback(long expectedTimestamp, int expectedInSequence) {
this.expectedTimestamp = expectedTimestamp;
this.expectedInSequence = expectedInSequence;
}
@@ -154,16 +148,15 @@ public class StreamTaskTimerTest extends TestLogger {
assertEquals(expectedTimestamp, timestamp);
assertEquals(expectedInSequence, numInSequence);
numInSequence++;
+ finished.complete(null);
} catch (Throwable t) {
- errorRef.compareAndSet(null, t);
+ finished.completeExceptionally(t);
}
}
- }
- private static void verifyNoException(@Nullable Throwable exception) {
- if (exception != null) {
- exception.printStackTrace();
- fail(exception.getMessage());
+ private void assertExpectedValues()
+ throws ExecutionException, InterruptedException, TimeoutException {
+ finished.get(20, TimeUnit.SECONDS);
}
}
[flink] 01/04: [FLINK-24090][docs] Added Troubleshooting section
with ignoring in-flight data explanation into unaligned checkpoints page
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.14
in repository https://gitbox.apache.org/repos/asf/flink.git
commit bf71226f86bfbd5d07cf1c9bf56f782abdd18e8d
Author: Anton Kalashnikov <ka...@yandex.ru>
AuthorDate: Thu Sep 2 16:23:36 2021 +0200
[FLINK-24090][docs] Added Troubleshooting section with ignoring in-flight data explanation into unaligned checkpoints page
---
docs/content/docs/ops/state/unaligned_checkpoints.md | 12 ++++++++++++
1 file changed, 12 insertions(+)
diff --git a/docs/content/docs/ops/state/unaligned_checkpoints.md b/docs/content/docs/ops/state/unaligned_checkpoints.md
index b47e43e..591c838 100644
--- a/docs/content/docs/ops/state/unaligned_checkpoints.md
+++ b/docs/content/docs/ops/state/unaligned_checkpoints.md
@@ -118,3 +118,15 @@ operators. Flink implements the broadcast state by checkpointing only a single c
from subtask 0 of the stateful operator. Upon restore, we send that copy to all of the
operators. Therefore it might happen that an operator will get the state with changes applied for a
record that it will soon consume from its checkpointed channels.
+
+## Troubleshooting
+
+### Corrupted in-flight data
+{{< hint warning >}}
+Actions described below are a last resort as they will lead to data loss.
+{{< /hint >}}
+In case of the in-flight data corrupted or by another reason when the job should be restored without the in-flight data,
+it is possible to use [recover-without-channel-state.checkpoint-id]({{< ref "docs/deployment/config" >}}#execution-checkpointing-recover-without-channel-state-checkpoint) property.
+This property requires to specify a checkpoint id for which in-flight data will be ignored.
+Do not set this property, unless a corruption inside the persisted in-flight data has lead to an otherwise unrecoverable situation.
+The property can be applied only after the job will be redeployed which means this operation makes sense only if [externalized checkpoint]({{< ref "docs/dev/datastream/fault-tolerance/checkpointing" >}}#enabling-and-configuring-checkpointing) is enabled.
[flink] 02/04: [refactor][streaming] Improved description of the
calculation of the processing time delay
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.14
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 6fa603327c7a2aeda0c2416c36e763949832da02
Author: Anton Kalashnikov <ka...@yandex.ru>
AuthorDate: Wed Sep 1 15:30:57 2021 +0200
[refactor][streaming] Improved description of the calculation of the processing time delay
---
.../streaming/runtime/tasks/ProcessingTimeServiceUtil.java | 10 +++++++---
1 file changed, 7 insertions(+), 3 deletions(-)
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/ProcessingTimeServiceUtil.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/ProcessingTimeServiceUtil.java
index c718f49..3733d8a 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/ProcessingTimeServiceUtil.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/ProcessingTimeServiceUtil.java
@@ -26,6 +26,8 @@ public class ProcessingTimeServiceUtil {
/**
* Returns the remaining delay of the processing time specified by {@code processingTimestamp}.
+ * This delay guarantees that the timer will be fired at least 1ms after the time it's
+ * registered for.
*
* @param processingTimestamp the processing time in milliseconds
* @param currentTimestamp the current processing timestamp; it usually uses {@link
@@ -36,9 +38,11 @@ public class ProcessingTimeServiceUtil {
// Two cases of timers here:
// (1) future/now timers(processingTimestamp >= currentTimestamp): delay the firing of the
- // timer by 1 ms to align the semantics with watermark. A watermark T says we won't see
- // elements in the future with a timestamp smaller or equal to T. With processing time, we
- // therefore need to delay firing the timer by one ms.
+ // timer by 1 ms to align the semantics with watermark. A watermark T says we
+ // won't see elements in the future with a timestamp smaller or equal to T. Without this
+ // 1ms delay, if we had fired the timer for T at the timestamp T, it would be possible
+ // that we would process another record for timestamp == T in the same millisecond, but
+ // after the timer for the timsetamp T has already been fired.
// (2) past timers(processingTimestamp < currentTimestamp): do not need to delay the firing
// because currentTimestamp is larger than processingTimestamp pluses the 1ms offset.
// TODO. The processing timers' performance can be further improved.
[flink] 04/04: [FLINK-23960][streaming] checkScheduledTimestamps
changed according to semantic of the calculation processing time delay
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.14
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 5a68bb88d925614da957156ee55cdc91d2985a56
Author: Anton Kalashnikov <ka...@yandex.ru>
AuthorDate: Fri Sep 3 19:54:08 2021 +0200
[FLINK-23960][streaming] checkScheduledTimestamps changed according to semantic of the calculation processing time delay
---
.../flink/streaming/runtime/operators/StreamTaskTimerTest.java | 9 ++++++++-
1 file changed, 8 insertions(+), 1 deletion(-)
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/StreamTaskTimerTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/StreamTaskTimerTest.java
index 765075d..f62dcea 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/StreamTaskTimerTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/StreamTaskTimerTest.java
@@ -113,7 +113,14 @@ public class StreamTaskTimerTest extends TestLogger {
long currentTimeMillis = System.currentTimeMillis();
ArrayList<ValidatingProcessingTimeCallback> timeCallbacks = new ArrayList<>();
- timeCallbacks.add(new ValidatingProcessingTimeCallback(currentTimeMillis, 0));
+ /*
+ It is not possible to test registering timer for currentTimeMillis or value slightly greater than
+ currentTimeMillis because if the during registerTimer the internal currentTime is equal
+ to this value then according to current logic the time will be increased for 1ms while
+ `currentTimeMillis - 200` is always transform to 0, so it can lead to reordering. See
+ comment in {@link ProcessingTimeServiceUtil#getRecordProcessingTimeDelay(long, long)}.
+ */
+ timeCallbacks.add(new ValidatingProcessingTimeCallback(currentTimeMillis - 1, 0));
timeCallbacks.add(new ValidatingProcessingTimeCallback(currentTimeMillis - 200, 1));
timeCallbacks.add(new ValidatingProcessingTimeCallback(currentTimeMillis + 100, 2));
timeCallbacks.add(new ValidatingProcessingTimeCallback(currentTimeMillis + 200, 3));