This is an automated email from the ASF dual-hosted git repository. arvid pushed a commit to branch release-1.13 in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.13 by this push: new e75111d [FLINK-20928] Fix flaky test by retrying notifyCheckpointComplete until either commit success or timeout e75111d is described below commit e75111dffc6a7bf05ade246cc5485d50cad8162f Author: Dong Lin <lindon...@gmail.com> AuthorDate: Fri Oct 15 09:46:06 2021 +0800 [FLINK-20928] 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 0304315..16cab86 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 @@ -206,11 +206,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 4b72f9b..76a5f28 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) { @@ -204,10 +206,25 @@ public class CommonTestUtils { } long startingTime = System.currentTimeMillis(); while (!condition.get() && System.currentTimeMillis() - startingTime < timeoutMs) { - Thread.sleep(1); + Thread.sleep(pause.toMillis()); } if (!condition.get()) { 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); + } }