You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by sn...@apache.org on 2024/03/27 08:13:53 UTC
(flink) branch master updated: [FLINK-34941][checkpoint] Add `CollectIteratorAssert#matchesRecordsFromSource` and `SinkTestSuiteBase#checkResultWithSemantic` with former signature
This is an automated email from the ASF dual-hosted git repository.
snuyanzin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 7d390a982f1 [FLINK-34941][checkpoint] Add `CollectIteratorAssert#matchesRecordsFromSource` and `SinkTestSuiteBase#checkResultWithSemantic` with former signature
7d390a982f1 is described below
commit 7d390a982f1f6c42b98ce85ed38f60221bb7b526
Author: Sergey Nuyanzin <sn...@gmail.com>
AuthorDate: Wed Mar 27 09:13:44 2024 +0100
[FLINK-34941][checkpoint] Add `CollectIteratorAssert#matchesRecordsFromSource` and `SinkTestSuiteBase#checkResultWithSemantic` with former signature
---
.../org/apache/flink/streaming/api/CheckpointingMode.java | 14 +++++++++++++-
.../connector/testframe/testsuites/SinkTestSuiteBase.java | 15 +++++++++++++++
.../connector/testframe/utils/CollectIteratorAssert.java | 14 ++++++++++++++
3 files changed, 42 insertions(+), 1 deletion(-)
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/CheckpointingMode.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/CheckpointingMode.java
index 0ff0ce89c76..17e20dbe037 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/CheckpointingMode.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/CheckpointingMode.java
@@ -77,5 +77,17 @@ public enum CheckpointingMode {
* scenarios, where a sustained very-low latency (such as few milliseconds) is needed, and where
* occasional duplicate messages (on recovery) do not matter.
*/
- AT_LEAST_ONCE
+ AT_LEAST_ONCE;
+
+ public static org.apache.flink.core.execution.CheckpointingMode convertToCheckpointingMode(
+ org.apache.flink.streaming.api.CheckpointingMode semantic) {
+ switch (semantic) {
+ case EXACTLY_ONCE:
+ return org.apache.flink.core.execution.CheckpointingMode.EXACTLY_ONCE;
+ case AT_LEAST_ONCE:
+ return org.apache.flink.core.execution.CheckpointingMode.AT_LEAST_ONCE;
+ default:
+ throw new IllegalArgumentException("Unsupported semantic: " + semantic);
+ }
+ }
}
diff --git a/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testframe/testsuites/SinkTestSuiteBase.java b/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testframe/testsuites/SinkTestSuiteBase.java
index a92d9b67309..895469d7e83 100644
--- a/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testframe/testsuites/SinkTestSuiteBase.java
+++ b/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testframe/testsuites/SinkTestSuiteBase.java
@@ -87,6 +87,7 @@ import static org.apache.flink.runtime.testutils.CommonTestUtils.terminateJob;
import static org.apache.flink.runtime.testutils.CommonTestUtils.waitForAllTaskRunning;
import static org.apache.flink.runtime.testutils.CommonTestUtils.waitForJobStatus;
import static org.apache.flink.runtime.testutils.CommonTestUtils.waitUntilCondition;
+import static org.apache.flink.streaming.api.CheckpointingMode.convertToCheckpointingMode;
import static org.apache.flink.util.Preconditions.checkNotNull;
/**
@@ -513,6 +514,20 @@ public abstract class SinkTestSuiteBase<T extends Comparable<T>> {
});
}
+ /**
+ * This method is required for downstream projects e.g. Flink connectors extending this test for
+ * the case when there should be supported Flink versions below 1.20. Could be removed together
+ * with dropping support for Flink 1.19.
+ */
+ @Deprecated
+ protected void checkResultWithSemantic(
+ ExternalSystemDataReader<T> reader,
+ List<T> testData,
+ org.apache.flink.streaming.api.CheckpointingMode semantic)
+ throws Exception {
+ checkResultWithSemantic(reader, testData, convertToCheckpointingMode(semantic));
+ }
+
/** Compare the metrics. */
private boolean compareSinkMetrics(
MetricQuerier metricQuerier,
diff --git a/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testframe/utils/CollectIteratorAssert.java b/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testframe/utils/CollectIteratorAssert.java
index 1ae093ebe81..c1e44242f3b 100644
--- a/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testframe/utils/CollectIteratorAssert.java
+++ b/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testframe/utils/CollectIteratorAssert.java
@@ -27,6 +27,8 @@ import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
+import static org.apache.flink.streaming.api.CheckpointingMode.convertToCheckpointingMode;
+
/**
* This assertion used to compare records in the collect iterator to the target test data with
* different semantic(AT_LEAST_ONCE, EXACTLY_ONCE).
@@ -51,6 +53,18 @@ public class CollectIteratorAssert<T>
return this;
}
+ /**
+ * This method is required for downstream projects e.g. Flink connectors extending this test for
+ * the case when there should be supported Flink versions below 1.20. Could be removed together
+ * with dropping support for Flink 1.19.
+ */
+ @Deprecated
+ public void matchesRecordsFromSource(
+ List<List<T>> recordsBySplitsFromSource,
+ org.apache.flink.streaming.api.CheckpointingMode semantic) {
+ matchesRecordsFromSource(recordsBySplitsFromSource, convertToCheckpointingMode(semantic));
+ }
+
public void matchesRecordsFromSource(
List<List<T>> recordsBySplitsFromSource, CheckpointingMode semantic) {
for (List<T> recordsFromSplit : recordsBySplitsFromSource) {