You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ar...@apache.org on 2021/10/14 06:05:58 UTC
[flink] branch release-1.14 updated: Fix flaky test by retrying
notifyCheckpointComplete until either commit success or timeout
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch release-1.14
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.14 by this push:
new 7578758 Fix flaky test by retrying notifyCheckpointComplete until either commit success or timeout
7578758 is described below
commit 7578758fa8c84314b8b3206629b3afa9ff41b636
Author: Dong Lin <li...@gmail.com>
AuthorDate: Thu Sep 23 20:47:56 2021 +0800
Fix flaky test by retrying notifyCheckpointComplete until either commit success or timeout
---
.../kafka/source/reader/KafkaSourceReader.java | 13 ++++++++++---
.../kafka/source/reader/KafkaSourceReaderTest.java | 20 +++++++++++++++-----
.../flink/core/testutils/CommonTestUtils.java | 21 +++++++++++++++++++--
3 files changed, 44 insertions(+), 10 deletions(-)
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReader.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReader.java
index 08f82b0..4a42a48 100644
--- a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReader.java
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReader.java
@@ -125,9 +125,18 @@ public class KafkaSourceReader<T>
return;
}
+ Map<TopicPartition, OffsetAndMetadata> committedPartitions =
+ offsetsToCommit.get(checkpointId);
+ if (committedPartitions == null) {
+ LOG.debug(
+ "Offsets for checkpoint {} either do not exist or have already been committed.",
+ checkpointId);
+ return;
+ }
+
((KafkaSourceFetcherManager<T>) splitFetcherManager)
.commitOffsets(
- offsetsToCommit.get(checkpointId),
+ committedPartitions,
(ignored, e) -> {
// The offset commit here is needed by the external monitoring. It won't
// break Flink job's correctness if we fail to commit the offset here.
@@ -144,8 +153,6 @@ public class KafkaSourceReader<T>
kafkaSourceReaderMetrics.recordSucceededCommit();
// If the finished topic partition has been committed, we remove it
// from the offsets of the finished splits map.
- Map<TopicPartition, OffsetAndMetadata> committedPartitions =
- offsetsToCommit.get(checkpointId);
committedPartitions.forEach(
(tp, offset) ->
kafkaSourceReaderMetrics.recordCommittedOffset(
diff --git a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReaderTest.java b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReaderTest.java
index 09652ea..5911031 100644
--- a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReaderTest.java
+++ b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReaderTest.java
@@ -215,11 +215,21 @@ public class KafkaSourceReaderTest extends SourceReaderTestBase<KafkaPartitionSp
// The completion of the last checkpoint should subsume all the previous checkpoitns.
assertEquals(checkpointId, reader.getOffsetsToCommit().size());
- reader.notifyCheckpointComplete(checkpointId);
- pollUntil(
- reader,
- output,
- () -> reader.getOffsetsToCommit().isEmpty(),
+
+ long lastCheckpointId = checkpointId;
+ waitUtil(
+ () -> {
+ try {
+ reader.notifyCheckpointComplete(lastCheckpointId);
+ } catch (Exception exception) {
+ throw new RuntimeException(
+ "Caught unexpected exception when polling from the reader",
+ exception);
+ }
+ return reader.getOffsetsToCommit().isEmpty();
+ },
+ Duration.ofSeconds(60),
+ Duration.ofSeconds(1),
"The offset commit did not finish before timeout.");
}
diff --git a/flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/core/testutils/CommonTestUtils.java b/flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/core/testutils/CommonTestUtils.java
index ab6f2a6..7466648 100644
--- a/flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/core/testutils/CommonTestUtils.java
+++ b/flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/core/testutils/CommonTestUtils.java
@@ -190,13 +190,15 @@ public class CommonTestUtils {
*
* @param condition the condition to wait for.
* @param timeout the maximum time to wait for the condition to become true.
+ * @param pause delay between condition checks.
* @param errorMsg the error message to include in the <code>TimeoutException</code> if the
* condition was not met before timeout.
* @throws TimeoutException if the condition is not met before timeout.
* @throws InterruptedException if the thread is interrupted.
*/
@SuppressWarnings("BusyWait")
- public static void waitUtil(Supplier<Boolean> condition, Duration timeout, String errorMsg)
+ public static void waitUtil(
+ Supplier<Boolean> condition, Duration timeout, Duration pause, String errorMsg)
throws TimeoutException, InterruptedException {
long timeoutMs = timeout.toMillis();
if (timeoutMs <= 0) {
@@ -206,10 +208,25 @@ public class CommonTestUtils {
boolean conditionResult = condition.get();
while (!conditionResult && System.currentTimeMillis() - startingTime < timeoutMs) {
conditionResult = condition.get();
- Thread.sleep(1);
+ Thread.sleep(pause.toMillis());
}
if (!conditionResult) {
throw new TimeoutException(errorMsg);
}
}
+
+ /**
+ * Wait util the given condition is met or timeout.
+ *
+ * @param condition the condition to wait for.
+ * @param timeout the maximum time to wait for the condition to become true.
+ * @param errorMsg the error message to include in the <code>TimeoutException</code> if the
+ * condition was not met before timeout.
+ * @throws TimeoutException if the condition is not met before timeout.
+ * @throws InterruptedException if the thread is interrupted.
+ */
+ public static void waitUtil(Supplier<Boolean> condition, Duration timeout, String errorMsg)
+ throws TimeoutException, InterruptedException {
+ waitUtil(condition, timeout, Duration.ofMillis(1), errorMsg);
+ }
}