[flink] 02/02: [FLINK-30239][rest] Fix the bug that FlameGraph cannot be generated due to using ImmutableSet incorrectly
This is an automated email from the ASF dual-hosted git repository. xtsong pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git commit 95be1e5abfe78d7ed8d171a779f280277f51569c Author: 1996fanrui <1996fan...@gmail.com> AuthorDate: Mon Dec 5 20:58:54 2022 +0800 [FLINK-30239][rest] Fix the bug that FlameGraph cannot be generated due to using ImmutableSet incorrectly This closes #21420 --- .../threadinfo/JobVertexThreadInfoTracker.java | 2 +- .../threadinfo/JobVertexThreadInfoTrackerTest.java | 29 +- 2 files changed, 29 insertions(+), 2 deletions(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTracker.java b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTracker.java index 57072263b73..6933655c0b4 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTracker.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTracker.java @@ -269,7 +269,7 @@ public class JobVertexThreadInfoTracker implements JobVert ExecutionAttemptID attemptId = execution.getAttemptId(); groupedAttemptIds.add(attemptId); -executionAttemptsByLocation.put(tmLocation, ImmutableSet.copyOf(groupedAttemptIds)); +executionAttemptsByLocation.put(tmLocation, groupedAttemptIds); } } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTrackerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTrackerTest.java index ec6855cafa1..f7d4ebfd288 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTrackerTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTrackerTest.java @@ -20,16 +20,21 @@ package org.apache.flink.runtime.webmonitor.threadinfo; import org.apache.flink.api.common.JobID; import org.apache.flink.runtime.clusterframework.types.ResourceID; +import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutorServiceAdapter; import org.apache.flink.runtime.executiongraph.ExecutionAttemptID; +import org.apache.flink.runtime.executiongraph.ExecutionGraph; import org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils; import org.apache.flink.runtime.executiongraph.ExecutionJobVertex; import org.apache.flink.runtime.executiongraph.ExecutionVertex; +import org.apache.flink.runtime.jobgraph.JobGraphTestUtils; import org.apache.flink.runtime.jobgraph.JobVertex; import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable; import org.apache.flink.runtime.messages.ThreadInfoSample; import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway; import org.apache.flink.runtime.resourcemanager.utils.TestingResourceManagerGateway; +import org.apache.flink.runtime.scheduler.SchedulerBase; import org.apache.flink.runtime.taskexecutor.TaskExecutorThreadInfoGateway; +import org.apache.flink.runtime.testutils.DirectScheduledExecutorService; import org.apache.flink.runtime.util.JvmUtils; import org.apache.flink.testutils.TestingUtils; import org.apache.flink.util.TestLogger; @@ -49,6 +54,7 @@ import javax.annotation.Nonnull; import java.time.Duration; import java.time.Instant; import java.time.temporal.ChronoUnit; +import java.util.Arrays; import java.util.Collection; import java.util.Collections; import java.util.HashMap; @@ -65,7 +71,9 @@ import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.function.Function; +import java.util.stream.Collectors; +import static org.apache.flink.runtime.scheduler.SchedulerTestingUtils.createScheduler; import static org.assertj.core.api.AssertionsForClassTypes.assertThat; /** Tests for the {@link JobVertexThreadInfoTracker}. */ @@ -74,6 +82,12 @@ public class JobVertexThreadInfoTrackerTest extends TestLogger { private static final int REQUEST_ID = 0; private static final ExecutionJobVertex EXECUTION_JOB_VERTEX = createExecutionJobVertex(); private static final ExecutionVertex[] TASK_VERTICES = EXECUTION_JOB_VERTEX.getTaskVertices(); +private static final Set ATTEMPT_IDS = +Arrays.stream(TASK_VERTICES) +.map( +executionVertex -> + executionVertex.getCurrentExecutionAttempt().getAttemptId()) +.collect(Collectors.toSet()); private static final JobID JOB_ID = new JobID(); private static ThreadInfoSample threadInfoSample; @@ -338,8 +352,18 @@ public class JobVertexThreadInfoTrackerTest extends TestLogger { private static
[flink] 01/02: [hotfix][rest] Minor clean-ups in JobVertexThreadInfoTracker.
This is an automated email from the ASF dual-hosted git repository. xtsong pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git commit ea9491756f8877ec54af1b55e8e144cfeb315d3d Author: 1996fanrui <1996fan...@gmail.com> AuthorDate: Wed Nov 30 00:26:00 2022 +0800 [hotfix][rest] Minor clean-ups in JobVertexThreadInfoTracker. --- .../runtime/webmonitor/threadinfo/JobVertexThreadInfoTracker.java | 6 -- .../webmonitor/threadinfo/JobVertexThreadInfoTrackerTest.java | 2 +- 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTracker.java b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTracker.java index 45b469632b1..57072263b73 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTracker.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTracker.java @@ -274,7 +274,9 @@ public class JobVertexThreadInfoTracker implements JobVert } return executionAttemptsByLocation.entrySet().stream() -.collect(Collectors.toMap(e -> e.getKey(), e -> ImmutableSet.copyOf(e.getValue(; +.collect( +Collectors.toMap( +Map.Entry::getKey, e -> ImmutableSet.copyOf(e.getValue(; } @VisibleForTesting @@ -353,7 +355,7 @@ public class JobVertexThreadInfoTracker implements JobVert vertexStatsCache.put(key, createStatsFn.apply(threadInfoStats)); resultAvailableFuture.complete(null); } else { -LOG.debug( +LOG.error( "Failed to gather a thread info sample for {}", vertex.getName(), throwable); diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTrackerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTrackerTest.java index 845b5a798f1..ec6855cafa1 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTrackerTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTrackerTest.java @@ -135,7 +135,7 @@ public class JobVertexThreadInfoTrackerTest extends TestLogger { Optional result = tracker.getVertexStats(JOB_ID, EXECUTION_JOB_VERTEX); // cached result is returned instead of unusedThreadInfoStats -assertThat(threadInfoStatsDefaultSample).isEqualTo(result.get()); +assertThat(result).isPresent().hasValue(threadInfoStatsDefaultSample); } /** Tests that cached result is NOT reused after refresh interval. */
[flink] branch master updated (bca57b7a222 -> 95be1e5abfe)
This is an automated email from the ASF dual-hosted git repository. xtsong pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/flink.git from bca57b7a222 [FLINK-30245][table-runtime] Fix the NPE thrown when filtering decimal(18, x) values after calling DecimalDataUtils.subtract method new ea9491756f8 [hotfix][rest] Minor clean-ups in JobVertexThreadInfoTracker. new 95be1e5abfe [FLINK-30239][rest] Fix the bug that FlameGraph cannot be generated due to using ImmutableSet incorrectly The 2 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: .../threadinfo/JobVertexThreadInfoTracker.java | 8 +++--- .../threadinfo/JobVertexThreadInfoTrackerTest.java | 31 -- 2 files changed, 34 insertions(+), 5 deletions(-)
[flink] 02/02: [FLINK-30239][rest] Fix the bug that FlameGraph cannot be generated due to using ImmutableSet incorrectly
This is an automated email from the ASF dual-hosted git repository. xtsong pushed a commit to branch release-1.16 in repository https://gitbox.apache.org/repos/asf/flink.git commit 0f98c6aad1f5bc8ad25f7608c419a5e396b8e8ac Author: 1996fanrui <1996fan...@gmail.com> AuthorDate: Mon Dec 5 20:58:54 2022 +0800 [FLINK-30239][rest] Fix the bug that FlameGraph cannot be generated due to using ImmutableSet incorrectly This closes #21420 --- .../threadinfo/JobVertexThreadInfoTracker.java | 2 +- .../threadinfo/JobVertexThreadInfoTrackerTest.java | 29 +- 2 files changed, 29 insertions(+), 2 deletions(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTracker.java b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTracker.java index 57072263b73..6933655c0b4 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTracker.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTracker.java @@ -269,7 +269,7 @@ public class JobVertexThreadInfoTracker implements JobVert ExecutionAttemptID attemptId = execution.getAttemptId(); groupedAttemptIds.add(attemptId); -executionAttemptsByLocation.put(tmLocation, ImmutableSet.copyOf(groupedAttemptIds)); +executionAttemptsByLocation.put(tmLocation, groupedAttemptIds); } } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTrackerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTrackerTest.java index ec6855cafa1..f7d4ebfd288 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTrackerTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTrackerTest.java @@ -20,16 +20,21 @@ package org.apache.flink.runtime.webmonitor.threadinfo; import org.apache.flink.api.common.JobID; import org.apache.flink.runtime.clusterframework.types.ResourceID; +import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutorServiceAdapter; import org.apache.flink.runtime.executiongraph.ExecutionAttemptID; +import org.apache.flink.runtime.executiongraph.ExecutionGraph; import org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils; import org.apache.flink.runtime.executiongraph.ExecutionJobVertex; import org.apache.flink.runtime.executiongraph.ExecutionVertex; +import org.apache.flink.runtime.jobgraph.JobGraphTestUtils; import org.apache.flink.runtime.jobgraph.JobVertex; import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable; import org.apache.flink.runtime.messages.ThreadInfoSample; import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway; import org.apache.flink.runtime.resourcemanager.utils.TestingResourceManagerGateway; +import org.apache.flink.runtime.scheduler.SchedulerBase; import org.apache.flink.runtime.taskexecutor.TaskExecutorThreadInfoGateway; +import org.apache.flink.runtime.testutils.DirectScheduledExecutorService; import org.apache.flink.runtime.util.JvmUtils; import org.apache.flink.testutils.TestingUtils; import org.apache.flink.util.TestLogger; @@ -49,6 +54,7 @@ import javax.annotation.Nonnull; import java.time.Duration; import java.time.Instant; import java.time.temporal.ChronoUnit; +import java.util.Arrays; import java.util.Collection; import java.util.Collections; import java.util.HashMap; @@ -65,7 +71,9 @@ import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.function.Function; +import java.util.stream.Collectors; +import static org.apache.flink.runtime.scheduler.SchedulerTestingUtils.createScheduler; import static org.assertj.core.api.AssertionsForClassTypes.assertThat; /** Tests for the {@link JobVertexThreadInfoTracker}. */ @@ -74,6 +82,12 @@ public class JobVertexThreadInfoTrackerTest extends TestLogger { private static final int REQUEST_ID = 0; private static final ExecutionJobVertex EXECUTION_JOB_VERTEX = createExecutionJobVertex(); private static final ExecutionVertex[] TASK_VERTICES = EXECUTION_JOB_VERTEX.getTaskVertices(); +private static final Set ATTEMPT_IDS = +Arrays.stream(TASK_VERTICES) +.map( +executionVertex -> + executionVertex.getCurrentExecutionAttempt().getAttemptId()) +.collect(Collectors.toSet()); private static final JobID JOB_ID = new JobID(); private static ThreadInfoSample threadInfoSample; @@ -338,8 +352,18 @@ public class JobVertexThreadInfoTrackerTest extends TestLogger { private s
[flink] 01/02: [hotfix][rest] Minor clean-ups in JobVertexThreadInfoTracker.
This is an automated email from the ASF dual-hosted git repository. xtsong pushed a commit to branch release-1.16 in repository https://gitbox.apache.org/repos/asf/flink.git commit eef91f8be7beaabae94596bfd36807a4c955f266 Author: 1996fanrui <1996fan...@gmail.com> AuthorDate: Wed Nov 30 00:26:00 2022 +0800 [hotfix][rest] Minor clean-ups in JobVertexThreadInfoTracker. --- .../runtime/webmonitor/threadinfo/JobVertexThreadInfoTracker.java | 6 -- .../webmonitor/threadinfo/JobVertexThreadInfoTrackerTest.java | 2 +- 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTracker.java b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTracker.java index 45b469632b1..57072263b73 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTracker.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTracker.java @@ -274,7 +274,9 @@ public class JobVertexThreadInfoTracker implements JobVert } return executionAttemptsByLocation.entrySet().stream() -.collect(Collectors.toMap(e -> e.getKey(), e -> ImmutableSet.copyOf(e.getValue(; +.collect( +Collectors.toMap( +Map.Entry::getKey, e -> ImmutableSet.copyOf(e.getValue(; } @VisibleForTesting @@ -353,7 +355,7 @@ public class JobVertexThreadInfoTracker implements JobVert vertexStatsCache.put(key, createStatsFn.apply(threadInfoStats)); resultAvailableFuture.complete(null); } else { -LOG.debug( +LOG.error( "Failed to gather a thread info sample for {}", vertex.getName(), throwable); diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTrackerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTrackerTest.java index 845b5a798f1..ec6855cafa1 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTrackerTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTrackerTest.java @@ -135,7 +135,7 @@ public class JobVertexThreadInfoTrackerTest extends TestLogger { Optional result = tracker.getVertexStats(JOB_ID, EXECUTION_JOB_VERTEX); // cached result is returned instead of unusedThreadInfoStats -assertThat(threadInfoStatsDefaultSample).isEqualTo(result.get()); +assertThat(result).isPresent().hasValue(threadInfoStatsDefaultSample); } /** Tests that cached result is NOT reused after refresh interval. */
[flink] branch release-1.16 updated (da9e997752c -> 0f98c6aad1f)
This is an automated email from the ASF dual-hosted git repository. xtsong pushed a change to branch release-1.16 in repository https://gitbox.apache.org/repos/asf/flink.git from da9e997752c [FLINK-30291][Connector/DynamoDB] Update docs to render DynamoDB connector docs new eef91f8be7b [hotfix][rest] Minor clean-ups in JobVertexThreadInfoTracker. new 0f98c6aad1f [FLINK-30239][rest] Fix the bug that FlameGraph cannot be generated due to using ImmutableSet incorrectly The 2 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: .../threadinfo/JobVertexThreadInfoTracker.java | 8 +++--- .../threadinfo/JobVertexThreadInfoTrackerTest.java | 31 -- 2 files changed, 34 insertions(+), 5 deletions(-)
[flink-kubernetes-operator] branch main updated: [FLINK-30307] Turn off e2e test error check temporarily
This is an automated email from the ASF dual-hosted git repository. mbalassi pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git The following commit(s) were added to refs/heads/main by this push: new b95e3e41 [FLINK-30307] Turn off e2e test error check temporarily b95e3e41 is described below commit b95e3e41fbac940fa281b69d05df32ef02591691 Author: Gabor Somogyi AuthorDate: Tue Dec 6 09:59:35 2022 +0100 [FLINK-30307] Turn off e2e test error check temporarily --- e2e-tests/utils.sh | 4 1 file changed, 4 insertions(+) diff --git a/e2e-tests/utils.sh b/e2e-tests/utils.sh index 8db0059e..81021bc3 100755 --- a/e2e-tests/utils.sh +++ b/e2e-tests/utils.sh @@ -132,6 +132,10 @@ function retry_times() { function check_operator_log_for_errors { echo "Checking for operator log errors..." + #https://issues.apache.org/jira/browse/FLINK-30310 + echo "Error checking is temporarily turned off." + return 0 + operator_pod_namespace=$(get_operator_pod_namespace) operator_pod_name=$(get_operator_pod_name) echo "Operator namespace: ${operator_pod_namespace} pod: ${operator_pod_name}"
[flink-kubernetes-operator] branch main updated: [FLINK-30151] Remove AuditUtils from error log check whitelist
This is an automated email from the ASF dual-hosted git repository. mbalassi pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git The following commit(s) were added to refs/heads/main by this push: new e0479732 [FLINK-30151] Remove AuditUtils from error log check whitelist e0479732 is described below commit e04797321f398e12f16cd0c6a7059b5037863bdf Author: Gabor Somogyi AuthorDate: Tue Dec 6 10:01:41 2022 +0100 [FLINK-30151] Remove AuditUtils from error log check whitelist --- e2e-tests/utils.sh | 1 - 1 file changed, 1 deletion(-) diff --git a/e2e-tests/utils.sh b/e2e-tests/utils.sh index 81021bc3..32cdd593 100755 --- a/e2e-tests/utils.sh +++ b/e2e-tests/utils.sh @@ -145,7 +145,6 @@ function check_operator_log_for_errors { | grep -v "Failed to submit job to session cluster" `#https://issues.apache.org/jira/browse/FLINK-30148` \ | grep -v "Error during event processing" `#https://issues.apache.org/jira/browse/FLINK-30149` \ | grep -v "REST service in session cluster is bad now" `#https://issues.apache.org/jira/browse/FLINK-30150` \ - | grep -v "AuditUtils" `#https://issues.apache.org/jira/browse/FLINK-30151` \ | grep -v "Error while patching status" `#https://issues.apache.org/jira/browse/FLINK-30283` \ | grep -e "\[\s*ERROR\s*\]" || true) if [ -z "${errors}" ]; then
[flink] branch master updated (95be1e5abfe -> 576c312d373)
This is an automated email from the ASF dual-hosted git repository. chesnay pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/flink.git from 95be1e5abfe [FLINK-30239][rest] Fix the bug that FlameGraph cannot be generated due to using ImmutableSet incorrectly add 576c312d373 [FLINK-29420][build] Upgrade Zookeeper to 3.7.1 No new revisions were added by this update. Summary of changes: .../testframe/container/FlinkTestcontainersConfigurator.java | 2 +- pom.xml | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-)
[flink-web] branch asf-site updated: Add Cassandra connector 3.0.0
This is an automated email from the ASF dual-hosted git repository. chesnay pushed a commit to branch asf-site in repository https://gitbox.apache.org/repos/asf/flink-web.git The following commit(s) were added to refs/heads/asf-site by this push: new c032a25c3 Add Cassandra connector 3.0.0 c032a25c3 is described below commit c032a25c3d2af1c38db58c3a7947eaef642a0fc2 Author: Chesnay Schepler AuthorDate: Fri Nov 25 10:29:56 2022 +0100 Add Cassandra connector 3.0.0 --- _config.yml | 13 + 1 file changed, 13 insertions(+) diff --git a/_config.yml b/_config.yml index 28182c00b..2fe52574f 100644 --- a/_config.yml +++ b/_config.yml @@ -125,6 +125,15 @@ connector_releases: url: "https://www.apache.org/dyn/closer.lua/flink/flink-connector-aws-3.0.0/flink-connector-aws-3.0.0-src.tgz"; asc_url: "https://downloads.apache.org/flink/flink-connector-aws-3.0.0/flink-connector-aws-3.0.0-src.tgz.asc"; sha512_url: "https://downloads.apache.org/flink/flink-connector-aws-3.0.0/flink-connector-aws-3.0.0-src.tgz.sha512"; + - version: "3.0.0" +source_release: + name: "Apache Flink Cassandra Connector 3.0.0" + id: "300-cassandra-connector-download-source" + flink_versions: +- "1.16" + url: "https://www.apache.org/dyn/closer.lua/flink/flink-connector-cassandra-3.0.0/flink-connector-cassandra-3.0.0-src.tgz"; + asc_url: "https://downloads.apache.org/flink/flink-connector-cassandra-3.0.0/flink-connector-cassandra-3.0.0-src.tgz.asc"; + sha512_url: "https://downloads.apache.org/flink/flink-connector-cassandra-3.0.0/flink-connector-cassandra-3.0.0-src.tgz.sha512"; flink_statefun_releases: @@ -646,6 +655,10 @@ release_archive: connector: "aws" version: 3.0.0 release_date: 2022-11-28 + - name: "Flink Cassandra Connector" +connector: "cassandra" +version: 3.0.0 +release_date: 2022-11-30 flink_shaded: -
[flink-web] branch asf-site updated: Rebuild website
This is an automated email from the ASF dual-hosted git repository. chesnay pushed a commit to branch asf-site in repository https://gitbox.apache.org/repos/asf/flink-web.git The following commit(s) were added to refs/heads/asf-site by this push: new 45ec46e09 Rebuild website 45ec46e09 is described below commit 45ec46e09ff55499687d75778a3673a046fd728d Author: Chesnay Schepler AuthorDate: Tue Dec 6 11:05:55 2022 +0100 Rebuild website --- content/downloads.html| 19 +++ content/zh/downloads.html | 19 +++ 2 files changed, 38 insertions(+) diff --git a/content/downloads.html b/content/downloads.html index 9131cd208..374ad6894 100644 --- a/content/downloads.html +++ b/content/downloads.html @@ -245,6 +245,7 @@ Flink connectors Apache Flink Elasticsearch Connector 3.0.0 Apache Flink AWS Connectors 3.0.0 + Apache Flink Cassandra Connector 3.0.0 Apache Flink Stateful Functions 3.2.0 @@ -349,6 +350,19 @@ +Apache Flink Cassandra Connector 3.0.0 + + +https://www.apache.org/dyn/closer.lua/flink/flink-connector-cassandra-3.0.0/flink-connector-cassandra-3.0.0-src.tgz"; id="300-cassandra-connector-download-source">Apache Flink Cassandra Connector 3.0.0 Source Release +(https://downloads.apache.org/flink/flink-connector-cassandra-3.0.0/flink-connector-cassandra-3.0.0-src.tgz.asc";>asc, https://downloads.apache.org/flink/flink-connector-cassandra-3.0.0/flink-connector-cassandra-3.0.0-src.tgz.sha512";>sha512) + + +This connector is compatible with these Apache Flink versions: + + + 1.16.x + + Apache Flink® Stateful Functions 3.2.0 is the latest stable release for the https://flink.apache.org/stateful-functions.html";>Stateful Functions component. @@ -1524,6 +1538,11 @@ Flink AWS Connectors 3.0.0 - 2022-11-28 (https://archive.apache.org/dist/flink/flink-connector-$-3.0.0/flink-connector-$-3.0.0-src.tgz";>Source) + +Flink Cassandra Connector 3.0.0 - 2022-11-30 +(https://archive.apache.org/dist/flink/flink-connector-$-3.0.0/flink-connector-$-3.0.0-src.tgz";>Source) + + Flink-StateFun diff --git a/content/zh/downloads.html b/content/zh/downloads.html index 885f29221..c0638f50d 100644 --- a/content/zh/downloads.html +++ b/content/zh/downloads.html @@ -243,6 +243,7 @@ Flink connectors Apache Flink Elasticsearch Connector 3.0.0 Apache Flink AWS Connectors 3.0.0 + Apache Flink Cassandra Connector 3.0.0 Apache Flink Stateful Functions 3.2.0 @@ -346,6 +347,19 @@ flink-docs-release-1.15/release-notes/flink-1.15.html">Flink 1.15 的发布说 +Apache Flink Cassandra Connector 3.0.0 + + +https://www.apache.org/dyn/closer.lua/flink/flink-connector-cassandra-3.0.0/flink-connector-cassandra-3.0.0-src.tgz"; id="300-cassandra-connector-download-source">Apache Flink Cassandra Connector 3.0.0 Source Release +(https://downloads.apache.org/flink/flink-connector-cassandra-3.0.0/flink-connector-cassandra-3.0.0-src.tgz.asc";>asc, https://downloads.apache.org/flink/flink-connector-cassandra-3.0.0/flink-connector-cassandra-3.0.0-src.tgz.sha512";>sha512) + + +This connector is compatible with these Apache Flink versions: + + + 1.16.x + + Apache Flink® Stateful Functions 3.2.0 是 https://flink.apache.org/stateful-functions.html";>Stateful Functions 组件的最新稳定版本. @@ -1475,6 +1489,11 @@ Flink AWS Connectors 3.0.0 - 2022-11-28 (https://archive.apache.org/dist/flink/flink-connector-$-3.0.0/flink-connector-$-3.0.0-src.tgz";>Source) + +Flink Cassandra Connector 3.0.0 - 2022-11-30 +(https://archive.apache.org/dist/flink/flink-connector-$-3.0.0/flink-connector-$-3.0.0-src.tgz";>Source) + + Flink-shaded
[flink] branch master updated (576c312d373 -> c6e824c955b)
This is an automated email from the ASF dual-hosted git repository. chesnay pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/flink.git from 576c312d373 [FLINK-29420][build] Upgrade Zookeeper to 3.7.1 add c6e824c955b [hotfix] Remove out-dated comment No new revisions were added by this update. Summary of changes: docs/setup_docs.sh | 2 -- 1 file changed, 2 deletions(-)
[flink] branch master updated (c6e824c955b -> 209df810f13)
This is an automated email from the ASF dual-hosted git repository. xtsong pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/flink.git from c6e824c955b [hotfix] Remove out-dated comment new a0aa4813b9f [FLINK-29870][runtime] Split ResourceActions to ResourceAllocator and ResourceEventListener. new 209df810f13 [FLINK-29870][runtime] move ResourceAllocator to ActiveResourceManager. The 2 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: .../runtime/resourcemanager/ResourceManager.java | 81 +++ .../resourcemanager/StandaloneResourceManager.java | 23 +++--- .../active/ActiveResourceManager.java | 70 + .../slotmanager/DeclarativeSlotManager.java| 25 +++--- .../slotmanager/FineGrainedSlotManager.java| 53 - .../NonSupportedResourceAllocatorImpl.java}| 29 +++ ...ResourceActions.java => ResourceAllocator.java} | 26 +++ .../slotmanager/ResourceEventListener.java}| 21 +++-- .../resourcemanager/slotmanager/SlotManager.java | 10 ++- .../slotmanager/TaskExecutorManager.java | 37 + .../resourcemanager/ResourceManagerTest.java | 31 +++- .../resourcemanager/TestingResourceManager.java| 27 --- .../TestingResourceManagerFactory.java | 20 +++-- .../active/ActiveResourceManagerTest.java | 71 - .../AbstractFineGrainedSlotManagerITCase.java | 12 +-- .../slotmanager/DeclarativeSlotManagerBuilder.java | 40 -- .../slotmanager/DeclarativeSlotManagerTest.java| 91 +++--- ...gerDefaultResourceAllocationStrategyITCase.java | 7 +- .../slotmanager/FineGrainedSlotManagerTest.java| 12 +-- .../FineGrainedSlotManagerTestBase.java| 10 ++- .../slotmanager/TaskExecutorManagerBuilder.java| 8 +- .../slotmanager/TaskExecutorManagerTest.java | 52 ++--- .../slotmanager/TestingResourceActionsBuilder.java | 72 - ...eActions.java => TestingResourceAllocator.java} | 40 -- .../TestingResourceAllocatorBuilder.java | 47 +++ .../slotmanager/TestingResourceEventListener.java | 48 .../TestingResourceEventListenerBuilder.java} | 30 --- .../slotmanager/TestingSlotManager.java| 3 +- 28 files changed, 526 insertions(+), 470 deletions(-) copy flink-runtime/src/{test/java/org/apache/flink/runtime/io/disk/NoOpFileChannelManager.java => main/java/org/apache/flink/runtime/resourcemanager/slotmanager/NonSupportedResourceAllocatorImpl.java} (54%) rename flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/{ResourceActions.java => ResourceAllocator.java} (69%) copy flink-runtime/src/main/java/org/apache/flink/runtime/{jobmaster/slotpool/DeclarativeSlotPoolFactory.java => resourcemanager/slotmanager/ResourceEventListener.java} (63%) delete mode 100644 flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/TestingResourceActionsBuilder.java rename flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/{TestingResourceActions.java => TestingResourceAllocator.java} (54%) create mode 100644 flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/TestingResourceAllocatorBuilder.java create mode 100644 flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/TestingResourceEventListener.java copy flink-runtime/src/test/java/org/apache/flink/runtime/{jobmaster/slotpool/TestingDeclarativeSlotPoolFactory.java => resourcemanager/slotmanager/TestingResourceEventListenerBuilder.java} (52%)
[flink] 02/02: [FLINK-29870][runtime] move ResourceAllocator to ActiveResourceManager.
This is an automated email from the ASF dual-hosted git repository. xtsong pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git commit 209df810f13e2fcbe5a4ca8bb015a8a5f662adc4 Author: Weihua Hu AuthorDate: Thu Nov 3 21:13:36 2022 +0800 [FLINK-29870][runtime] move ResourceAllocator to ActiveResourceManager. This closes #21233 --- .../runtime/resourcemanager/ResourceManager.java | 75 -- .../resourcemanager/StandaloneResourceManager.java | 23 +++ .../active/ActiveResourceManager.java | 70 +++- .../slotmanager/FineGrainedSlotManager.java| 25 +--- .../NonSupportedResourceAllocatorImpl.java}| 34 +- .../slotmanager/ResourceAllocator.java | 14 +++- .../slotmanager/TaskExecutorManager.java | 27 .../resourcemanager/ResourceManagerTest.java | 31 +++-- .../resourcemanager/TestingResourceManager.java| 27 .../TestingResourceManagerFactory.java | 20 +++--- .../active/ActiveResourceManagerTest.java | 71 ++-- .../slotmanager/DeclarativeSlotManagerTest.java| 18 ++ ...gerDefaultResourceAllocationStrategyITCase.java | 7 +- .../slotmanager/TaskExecutorManagerTest.java | 14 +--- .../slotmanager/TestingResourceAllocator.java | 23 +-- .../TestingResourceAllocatorBuilder.java | 17 + 16 files changed, 246 insertions(+), 250 deletions(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java index 3abd6b6b68d..5d38143974d 100755 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java @@ -169,6 +169,8 @@ public abstract class ResourceManager private final AtomicReference latestTokens = new AtomicReference<>(); +private final ResourceAllocator resourceAllocator; + public ResourceManager( RpcService rpcService, UUID leaderSessionId, @@ -234,6 +236,8 @@ public abstract class ResourceManager this.startedFuture = new CompletableFuture<>(); this.delegationTokenManager = delegationTokenManager; + +this.resourceAllocator = getResourceAllocator(); } // @@ -267,7 +271,7 @@ public abstract class ResourceManager slotManager.start( getFencingToken(), getMainThreadExecutor(), -new ResourceAllocatorImpl(), +resourceAllocator, new ResourceEventListenerImpl(), blocklistHandler::isBlockedTaskManager); @@ -533,7 +537,8 @@ public abstract class ResourceManager @Override public void disconnectTaskManager(final ResourceID resourceId, final Exception cause) { -closeTaskManagerConnection(resourceId, cause).ifPresent(ResourceManager.this::stopWorker); +closeTaskManagerConnection(resourceId, cause) +.ifPresent(ResourceManager.this::stopWorkerIfSupported); } @Override @@ -984,10 +989,11 @@ public abstract class ResourceManager taskExecutorResourceId.getStringWithMetadata(; } -final WorkerType newWorker = workerStarted(taskExecutorResourceId); +final Optional newWorkerOptional = +getWorkerNodeIfAcceptRegistration(taskExecutorResourceId); String taskExecutorAddress = taskExecutorRegistration.getTaskExecutorAddress(); -if (newWorker == null) { +if (!newWorkerOptional.isPresent()) { log.warn( "Discard registration from TaskExecutor {} at ({}) because the framework did " + "not recognize it", @@ -996,6 +1002,7 @@ public abstract class ResourceManager return new TaskExecutorRegistrationRejection( "The ResourceManager does not recognize this TaskExecutor."); } else { +WorkerType newWorker = newWorkerOptional.get(); WorkerRegistration registration = new WorkerRegistration<>( taskExecutorGateway, @@ -1159,9 +1166,8 @@ public abstract class ResourceManager } } -protected void releaseResource(InstanceID instanceId, Exception cause) { +protected WorkerType getWorkerByInstanceId(InstanceID instanceId) { WorkerType worker = null; - // TODO: Improve performance by having an index on the instanceId for (Map.Entry> entry : taskExecutors.entrySet()) { @@ -1171,18 +1177,7 @@ public abstract class ResourceManager
[flink] 01/02: [FLINK-29870][runtime] Split ResourceActions to ResourceAllocator and ResourceEventListener.
This is an automated email from the ASF dual-hosted git repository. xtsong pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git commit a0aa4813b9fe3db37cabe38fa47f1d1096c879dc Author: Weihua Hu AuthorDate: Thu Nov 3 16:10:49 2022 +0800 [FLINK-29870][runtime] Split ResourceActions to ResourceAllocator and ResourceEventListener. --- .../runtime/resourcemanager/ResourceManager.java | 12 ++-- .../slotmanager/DeclarativeSlotManager.java| 25 +++ .../slotmanager/FineGrainedSlotManager.java| 32 + ...ResourceActions.java => ResourceAllocator.java} | 16 + ...urceActions.java => ResourceEventListener.java} | 30 ++-- .../resourcemanager/slotmanager/SlotManager.java | 10 +-- .../slotmanager/TaskExecutorManager.java | 14 ++-- .../AbstractFineGrainedSlotManagerITCase.java | 12 ++-- .../slotmanager/DeclarativeSlotManagerBuilder.java | 40 +-- .../slotmanager/DeclarativeSlotManagerTest.java| 83 -- ...gerDefaultResourceAllocationStrategyITCase.java | 2 +- .../slotmanager/FineGrainedSlotManagerTest.java| 12 ++-- .../FineGrainedSlotManagerTestBase.java| 10 ++- .../slotmanager/TaskExecutorManagerBuilder.java| 8 +-- .../slotmanager/TaskExecutorManagerTest.java | 38 +- ...eActions.java => TestingResourceAllocator.java} | 25 ++- ...r.java => TestingResourceAllocatorBuilder.java} | 28 ++-- .../slotmanager/TestingResourceEventListener.java | 48 + .../TestingResourceEventListenerBuilder.java | 41 +++ .../slotmanager/TestingSlotManager.java| 3 +- 20 files changed, 284 insertions(+), 205 deletions(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java index 12684c25247..3abd6b6b68d 100755 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java @@ -56,7 +56,8 @@ import org.apache.flink.runtime.resourcemanager.exceptions.ResourceManagerExcept import org.apache.flink.runtime.resourcemanager.exceptions.UnknownTaskExecutorException; import org.apache.flink.runtime.resourcemanager.registration.JobManagerRegistration; import org.apache.flink.runtime.resourcemanager.registration.WorkerRegistration; -import org.apache.flink.runtime.resourcemanager.slotmanager.ResourceActions; +import org.apache.flink.runtime.resourcemanager.slotmanager.ResourceAllocator; +import org.apache.flink.runtime.resourcemanager.slotmanager.ResourceEventListener; import org.apache.flink.runtime.resourcemanager.slotmanager.SlotManager; import org.apache.flink.runtime.rest.messages.LogInfo; import org.apache.flink.runtime.rest.messages.ThreadDumpInfo; @@ -266,7 +267,8 @@ public abstract class ResourceManager slotManager.start( getFencingToken(), getMainThreadExecutor(), -new ResourceActionsImpl(), +new ResourceAllocatorImpl(), +new ResourceEventListenerImpl(), blocklistHandler::isBlockedTaskManager); delegationTokenManager.start(this); @@ -1334,7 +1336,7 @@ public abstract class ResourceManager } } -private class ResourceActionsImpl implements ResourceActions { +private class ResourceAllocatorImpl implements ResourceAllocator { @Override public void releaseResource(InstanceID instanceId, Exception cause) { @@ -1348,9 +1350,11 @@ public abstract class ResourceManager validateRunsInMainThread(); return startNewWorker(workerResourceSpec); } +} +private class ResourceEventListenerImpl implements ResourceEventListener { @Override -public void notifyNotEnoughResourcesAvailable( +public void notEnoughResourceAvailable( JobID jobId, Collection acquiredResources) { validateRunsInMainThread(); diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/DeclarativeSlotManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/DeclarativeSlotManager.java index 71898ae747d..f5e210de073 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/DeclarativeSlotManager.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/DeclarativeSlotManager.java @@ -69,7 +69,7 @@ public class DeclarativeSlotManager implements SlotManager { private final SlotTracker slotTracker; private final ResourceTracker resourceTracker; -private final BiFunction +private final BiFunction taskExecutorManagerFactory;
[flink-kubernetes-operator] branch main updated: [FLINK-30268] HA metadata and other related errors should not throw DeploymentFailedException
This is an automated email from the ASF dual-hosted git repository. mbalassi pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git The following commit(s) were added to refs/heads/main by this push: new 310ff307 [FLINK-30268] HA metadata and other related errors should not throw DeploymentFailedException 310ff307 is described below commit 310ff3072cd6196202ac37a171a896d3359cfc56 Author: pvary AuthorDate: Tue Dec 6 14:41:16 2022 +0100 [FLINK-30268] HA metadata and other related errors should not throw DeploymentFailedException --- .../controller/FlinkDeploymentController.java | 15 ++ .../exception/RecoveryFailureException.java| 32 + .../deployment/ApplicationReconciler.java | 4 +- .../operator/service/AbstractFlinkService.java | 6 +-- .../kubernetes/operator/TestingFlinkService.java | 4 +- .../controller/DeploymentRecoveryTest.java | 10 ++-- .../controller/FlinkDeploymentControllerTest.java | 55 ++ .../deployment/ApplicationReconcilerTest.java | 6 +-- .../ApplicationReconcilerUpgradeModeTest.java | 6 +-- .../operator/service/NativeFlinkServiceTest.java | 4 +- 10 files changed, 124 insertions(+), 18 deletions(-) diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentController.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentController.java index b681a4b5..7b9db5a3 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentController.java +++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentController.java @@ -24,6 +24,7 @@ import org.apache.flink.kubernetes.operator.api.status.JobManagerDeploymentStatu import org.apache.flink.kubernetes.operator.config.FlinkConfigManager; import org.apache.flink.kubernetes.operator.exception.DeploymentFailedException; import org.apache.flink.kubernetes.operator.exception.ReconciliationException; +import org.apache.flink.kubernetes.operator.exception.RecoveryFailureException; import org.apache.flink.kubernetes.operator.observer.deployment.FlinkDeploymentObserverFactory; import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils; import org.apache.flink.kubernetes.operator.reconciler.deployment.ReconcilerFactory; @@ -121,6 +122,8 @@ public class FlinkDeploymentController } statusRecorder.patchAndCacheStatus(flinkApp); reconcilerFactory.getOrCreate(flinkApp).reconcile(flinkApp, context); +} catch (RecoveryFailureException rfe) { +handleRecoveryFailed(flinkApp, rfe); } catch (DeploymentFailedException dfe) { handleDeploymentFailed(flinkApp, dfe); } catch (Exception e) { @@ -153,6 +156,18 @@ public class FlinkDeploymentController EventRecorder.Component.JobManagerDeployment); } +private void handleRecoveryFailed(FlinkDeployment flinkApp, RecoveryFailureException rfe) { +LOG.error("Flink recovery failed", rfe); +ReconciliationUtils.updateForReconciliationError( +flinkApp, rfe, configManager.getOperatorConfiguration()); +eventRecorder.triggerEvent( +flinkApp, +EventRecorder.Type.Warning, +rfe.getReason(), +rfe.getMessage(), +EventRecorder.Component.JobManagerDeployment); +} + @Override public Map prepareEventSources( EventSourceContext context) { diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/exception/RecoveryFailureException.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/exception/RecoveryFailureException.java new file mode 100644 index ..1b0be67b --- /dev/null +++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/exception/RecoveryFailureException.java @@ -0,0 +1,32 @@ +/* + * 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 Licen
[flink-kubernetes-operator] branch main updated: [FLINK-30154] Allow to set flinkConfiguration in SessionJob
This is an automated email from the ASF dual-hosted git repository. mbalassi pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git The following commit(s) were added to refs/heads/main by this push: new 2c207edf [FLINK-30154] Allow to set flinkConfiguration in SessionJob 2c207edf is described below commit 2c207edf5915f077b92c5b2bc2f0e6cac54daa9d Author: Gabor Somogyi AuthorDate: Tue Dec 6 14:44:40 2022 +0100 [FLINK-30154] Allow to set flinkConfiguration in SessionJob --- .../kubernetes/operator/validation/DefaultValidator.java | 12 +--- .../kubernetes/operator/validation/DefaultValidatorTest.java | 7 +++ 2 files changed, 4 insertions(+), 15 deletions(-) diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java index 0df0e593..05fdfaa8 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java +++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java @@ -61,9 +61,6 @@ public class DefaultValidator implements FlinkResourceValidator { KubernetesConfigOptions.NAMESPACE.key(), KubernetesConfigOptions.CLUSTER_ID.key() }; -private static final Set ALLOWED_FLINK_SESSION_JOB_CONF_KEYS = - Set.of(KubernetesOperatorConfigOptions.JAR_ARTIFACT_HTTP_HEADER.key()); - private static final Set ALLOWED_LOG_CONF_KEYS = Set.of(Constants.CONFIG_FILE_LOG4J_NAME, Constants.CONFIG_FILE_LOGBACK_NAME); @@ -468,14 +465,7 @@ public class DefaultValidator implements FlinkResourceValidator { return Optional.empty(); } -for (String key : flinkSessionJobConfig.keySet()) { -if (!ALLOWED_FLINK_SESSION_JOB_CONF_KEYS.contains(key)) { -return Optional.of( -String.format( -"Invalid session job flinkConfiguration key: %s. Allowed keys are %s", -key, ALLOWED_FLINK_SESSION_JOB_CONF_KEYS)); -} -} +// Exclude specific keys if they cause issues return Optional.empty(); } } diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/validation/DefaultValidatorTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/validation/DefaultValidatorTest.java index fe2e3162..a8cb5fa5 100644 --- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/validation/DefaultValidatorTest.java +++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/validation/DefaultValidatorTest.java @@ -514,12 +514,11 @@ public class DefaultValidatorTest { .setFlinkConfiguration( Map.of( KubernetesOperatorConfigOptions - .OPERATOR_RECONCILE_INTERVAL + .PERIODIC_SAVEPOINT_INTERVAL .key(), -"60")), +"1m")), flinkDeployment -> {}, -"Invalid session job flinkConfiguration key: kubernetes.operator.reconcile.interval." -+ " Allowed keys are [kubernetes.operator.user.artifacts.http.header]"); +null); testSessionJobValidateWithModifier( sessionJob -> {
[flink-connector-pulsar] 13/16: [FLINK-29830][Connector/Pulsar] Create the topic with schema before consuming messages in PulsarSinkITCase. (#21252)
This is an automated email from the ASF dual-hosted git repository. martijnvisser pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git commit a6ba99792ea4916f73cca5bee4389dcbec2bf180 Author: Yufan Sheng AuthorDate: Fri Nov 11 15:14:22 2022 +0800 [FLINK-29830][Connector/Pulsar] Create the topic with schema before consuming messages in PulsarSinkITCase. (#21252) --- .../connector/pulsar/sink/PulsarSinkITCase.java| 2 ++ .../pulsar/testutils/function/ControlSource.java | 41 ++ .../testutils/runtime/PulsarRuntimeOperator.java | 4 +++ 3 files changed, 41 insertions(+), 6 deletions(-) diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/PulsarSinkITCase.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/PulsarSinkITCase.java index 71e778c..0e5a173 100644 --- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/PulsarSinkITCase.java +++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/sink/PulsarSinkITCase.java @@ -51,6 +51,7 @@ import java.util.concurrent.ThreadLocalRandom; import static org.apache.commons.lang3.RandomStringUtils.randomAlphabetic; import static org.apache.flink.connector.pulsar.sink.writer.serializer.PulsarSerializationSchema.flinkSchema; +import static org.apache.pulsar.client.api.Schema.STRING; import static org.assertj.core.api.Assertions.assertThat; /** Tests for using PulsarSink writing to a Pulsar cluster. */ @@ -104,6 +105,7 @@ class PulsarSinkITCase { // A random topic with partition 4. String topic = randomAlphabetic(8); operator().createTopic(topic, 4); +operator().createSchema(topic, STRING); int counts = ThreadLocalRandom.current().nextInt(100, 200); ControlSource source = diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/function/ControlSource.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/function/ControlSource.java index 3684167..127e267 100644 --- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/function/ControlSource.java +++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/function/ControlSource.java @@ -31,8 +31,12 @@ import org.apache.flink.testutils.junit.SharedReference; import org.apache.flink.shaded.guava30.com.google.common.util.concurrent.Uninterruptibles; +import org.apache.pulsar.client.api.Consumer; +import org.apache.pulsar.client.api.ConsumerBuilder; import org.apache.pulsar.client.api.Message; +import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.api.SubscriptionInitialPosition; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -45,8 +49,12 @@ import java.util.List; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; import static org.apache.commons.lang3.RandomStringUtils.randomAlphanumeric; +import static org.apache.flink.connector.pulsar.common.utils.PulsarExceptionUtils.sneakyClient; +import static org.apache.pulsar.client.api.SubscriptionMode.Durable; +import static org.apache.pulsar.client.api.SubscriptionType.Exclusive; /** * This source is used for testing in Pulsar sink. We would generate a fix number of records by the @@ -183,20 +191,30 @@ public class ControlSource extends AbstractRichFunction private static class StopSignal implements Closeable { private static final Logger LOG = LoggerFactory.getLogger(StopSignal.class); -private final String topic; private final int desiredCounts; // This is a thread-safe list. private final List consumedRecords; private final AtomicLong deadline; private final ExecutorService executor; +private final Consumer consumer; +private final AtomicReference throwableException; public StopSignal( PulsarRuntimeOperator operator, String topic, int messageCounts, Duration timeout) { -this.topic = topic; this.desiredCounts = messageCounts; this.consumedRecords = Collections.synchronizedList(new ArrayList<>(messageCounts)); this.deadline = new AtomicLong(timeout.toMillis() + System.currentTimeMillis()); this.executor = Executors.newSingleThreadExecutor(); +ConsumerBuilder consumerBuilder = +operator.client() +.newConsumer(Schema.STRING) +.topic(topic) +.subscriptionName(randomAlphanumeric(10)) +.subscription
[flink-connector-pulsar] branch main updated (40c2705 -> 3199922)
This is an automated email from the ASF dual-hosted git repository. martijnvisser pushed a change to branch main in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git from 40c2705 [hotfix] Enable checkstyle only for source- and testsource directories, not for generated sources new d21147e [FLINK-28853][pulsar] Implement pausing Pulsar splits in case of watermark drift new 9cce627 [FLINK-28853][connector-base] Add allow-unaligned-source-splits config option new b0a1678 [FLINK-28853] Address PR comments / Add Kafka and Pulsar split pausing tests new 08e7b19 [hotfix][docs]: Fix minor grammar and spelling mistakes new dfb5fd8 [FLINK-29580][Connector/Pulsar] Remove pulsar.consumer.autoUpdatePartitionsIntervalSeconds option. (#21070) new 618daa2 [FLINK-29433][Connector/Pulsar] Support Auth through the builder pattern in Pulsar connector. (#21071) new 62da02a [FLINK-28085][Connector/Pulsar] Close the pending Pulsar transactions before closing pipeline (#21075) new c6a20a6 [FLINK-28820][Connector/Pulsar] Improve the writing performance for PulsarSink (#21074) new 1c88632 [FLINK-29624][Common][Connector][Filesystem] Upgrade org.apache.commons:commons-lang3 from 3.3.2 to 3.12.0 new 4f204e8 [FLINK-29709][Connector/Pulsar] Bump the Pulsar to latest 2.10.2 (#21204) new d6e2e06 [FLINK-28083][Connector/Pulsar] PulsarSource work with object-reusing DeserializationSchema. (#21205) new f6ead4c [FLINK-29860][Connectors/Pulsar] InlineElement fields should be transient (#21262) new a6ba997 [FLINK-29830][Connector/Pulsar] Create the topic with schema before consuming messages in PulsarSinkITCase. (#21252) new d8abb2a [FLINK-26027][Connector/Pulsar] Expose Pulsar producer metrics and add FLIP-33 sink metrics. (#21249) new d330e40 [FLINK-30254][Connector/Pulsar] Sync Pulsar updates and set Flink version to 1.17-SNAPSHOT new 3199922 [FLINK-30254][Connector/Pulsar] Resolve dependency convergence errors The 16 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: .github/workflows/push_pr.yml | 6 +- .github/workflows/weekly.yml | 2 +- flink-connector-pulsar/pom.xml | 13 +-- .../pulsar/common/metrics/MetricNames.java | 63 .../common/metrics/ProducerMetricsInterceptor.java | 67 .../pulsar/common/schema/PulsarSchema.java | 15 ++- .../pulsar/common/schema/PulsarSchemaUtils.java| 9 +- .../common/utils/PulsarTransactionUtils.java | 2 + .../connector/pulsar/sink/PulsarSinkBuilder.java | 33 ++ .../connector/pulsar/sink/PulsarSinkOptions.java | 16 +++ .../pulsar/sink/config/PulsarSinkConfigUtils.java | 4 - .../pulsar/sink/config/SinkConfiguration.java | 29 +++--- .../connector/pulsar/sink/writer/PulsarWriter.java | 86 +++- .../pulsar/sink/writer/router/MessageKeyHash.java | 2 +- .../sink/writer/router/TopicRoutingMode.java | 2 +- .../sink/writer/topic/TopicProducerRegister.java | 86 +++- .../pulsar/source/PulsarSourceBuilder.java | 33 ++ .../pulsar/source/PulsarSourceOptions.java | 21 ++-- .../pulsar/source/config/CursorVerification.java | 2 +- .../source/config/PulsarSourceConfigUtils.java | 5 - .../pulsar/source/config/SourceConfiguration.java | 16 ++- .../source/enumerator/PulsarSourceEnumerator.java | 8 ++ .../source/enumerator/assigner/SplitAssigner.java | 3 + .../enumerator/assigner/SplitAssignerBase.java | 5 + .../source/enumerator/cursor/StartCursor.java | 4 +- .../source/enumerator/cursor/StopCursor.java | 6 +- .../source/reader/PulsarSourceReaderFactory.java | 21 ++-- .../source/reader/emitter/PulsarRecordEmitter.java | 63 ++-- .../reader/fetcher/PulsarFetcherManagerBase.java | 41 +--- .../fetcher/PulsarOrderedFetcherManager.java | 22 ++-- .../fetcher/PulsarUnorderedFetcherManager.java | 18 ++-- .../source/reader/message/PulsarMessage.java | 74 - .../reader/message/PulsarMessageCollector.java | 60 --- .../reader/source/PulsarOrderedSourceReader.java | 19 ++-- .../reader/source/PulsarSourceReaderBase.java | 31 -- .../reader/source/PulsarUnorderedSourceReader.java | 41 +++- .../split/PulsarOrderedPartitionSplitReader.java | 13 +-- .../split/PulsarPartitionSplitReaderBase.java | 114 - .../split/PulsarUnorderedPartitionSplitReader.java | 27 ++--- .../connector/pulsar/sink/PulsarSinkITCase.java| 6 +- .../writer/topic/TopicProducerRegisterTest.java| 7 +- .../source/enumerator/cursor/StopCursorTest.java
[flink-connector-pulsar] 03/16: [FLINK-28853] Address PR comments / Add Kafka and Pulsar split pausing tests
This is an automated email from the ASF dual-hosted git repository. martijnvisser pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git commit b0a1678f91dae9b7bee35f77c79b441f365499be Author: Maximilian Michels AuthorDate: Thu Sep 1 16:15:33 2022 +0200 [FLINK-28853] Address PR comments / Add Kafka and Pulsar split pausing tests --- .../split/PulsarPartitionSplitReaderBase.java | 9 +++--- .../source/PulsarOrderedSourceReaderTest.java | 32 ++ 2 files changed, 37 insertions(+), 4 deletions(-) diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarPartitionSplitReaderBase.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarPartitionSplitReaderBase.java index c1459a6..b884c57 100644 --- a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarPartitionSplitReaderBase.java +++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarPartitionSplitReaderBase.java @@ -162,7 +162,7 @@ abstract class PulsarPartitionSplitReaderBase List newSplits = splitsChanges.splits(); Preconditions.checkArgument( -newSplits.size() == 1, "This pulsar split reader only support one split."); +newSplits.size() == 1, "This pulsar split reader only supports one split."); this.registeredSplit = newSplits.get(0); // Open stop cursor. @@ -184,9 +184,10 @@ abstract class PulsarPartitionSplitReaderBase public void pauseOrResumeSplits( Collection splitsToPause, Collection splitsToResume) { -if (splitsToPause.size() > 1 || splitsToResume.size() > 1) { -throw new IllegalStateException("This pulsar split reader only support one split."); -} +// This shouldn't happen but just in case... +Preconditions.checkState( +splitsToPause.size() + splitsToResume.size() <= 1, +"This pulsar split reader only supports one split."); if (!splitsToPause.isEmpty()) { pulsarConsumer.pause(); diff --git a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarOrderedSourceReaderTest.java b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarOrderedSourceReaderTest.java index 9806108..4ec1912 100644 --- a/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarOrderedSourceReaderTest.java +++ b/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarOrderedSourceReaderTest.java @@ -29,11 +29,13 @@ import org.apache.flink.core.testutils.CommonTestUtils; import org.apache.pulsar.client.admin.PulsarAdminException; import org.apache.pulsar.client.api.Consumer; +import org.apache.pulsar.client.api.MessageId; import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.SubscriptionInitialPosition; import org.apache.pulsar.client.api.SubscriptionType; import org.apache.pulsar.client.impl.MessageIdImpl; import org.junit.jupiter.api.TestTemplate; +import org.junit.jupiter.api.Timeout; import java.time.Duration; import java.util.Collections; @@ -130,6 +132,36 @@ class PulsarOrderedSourceReaderTest extends PulsarSourceReaderTestBase { } } +@TestTemplate +@Timeout(600) +void supportsPausingOrResumingSplits( +PulsarSourceReaderBase reader, Boundedness boundedness, String topicName) +throws Exception { +final PulsarPartitionSplit split = +createPartitionSplit(topicName, 0, boundedness, MessageId.earliest); + +reader.addSplits(Collections.singletonList(split)); + +TestingReaderOutput output = new TestingReaderOutput<>(); + +reader.pauseOrResumeSplits( +Collections.singletonList(split.splitId()), Collections.emptyList()); + +InputStatus status = reader.pollNext(output); +assertThat(status).isEqualTo(InputStatus.NOTHING_AVAILABLE); + +reader.pauseOrResumeSplits(Collections.emptyList(), Collections.singleton(split.splitId())); + +do { +status = reader.pollNext(output); +Thread.sleep(5); +} while (status != InputStatus.MORE_AVAILABLE); + +assertThat(status).isEqualTo(InputStatus.MORE_AVAILABLE); + +reader.close(); +} + private void setupSourceReader( PulsarSourceReaderBase reader, String topicName,
[flink-connector-pulsar] 14/16: [FLINK-26027][Connector/Pulsar] Expose Pulsar producer metrics and add FLIP-33 sink metrics. (#21249)
This is an automated email from the ASF dual-hosted git repository. martijnvisser pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git commit d8abb2a4cf2bfc692f2386e62220520680193381 Author: Yufan Sheng AuthorDate: Fri Nov 11 21:10:11 2022 +0800 [FLINK-26027][Connector/Pulsar] Expose Pulsar producer metrics and add FLIP-33 sink metrics. (#21249) --- .../pulsar/common/metrics/MetricNames.java | 63 .../common/metrics/ProducerMetricsInterceptor.java | 67 + .../connector/pulsar/sink/PulsarSinkOptions.java | 14 .../pulsar/sink/config/SinkConfiguration.java | 16 +++- .../connector/pulsar/sink/writer/PulsarWriter.java | 6 +- .../sink/writer/topic/TopicProducerRegister.java | 86 +- .../pulsar/source/PulsarSourceOptions.java | 14 .../pulsar/source/config/SourceConfiguration.java | 16 +++- .../source/enumerator/PulsarSourceEnumerator.java | 8 ++ .../source/enumerator/assigner/SplitAssigner.java | 3 + .../enumerator/assigner/SplitAssignerBase.java | 5 ++ .../source/reader/PulsarSourceReaderFactory.java | 6 +- .../split/PulsarOrderedPartitionSplitReader.java | 6 +- .../split/PulsarPartitionSplitReaderBase.java | 66 - .../split/PulsarUnorderedPartitionSplitReader.java | 4 +- .../connector/pulsar/sink/PulsarSinkITCase.java| 4 +- .../writer/topic/TopicProducerRegisterTest.java| 7 +- .../source/enumerator/cursor/StopCursorTest.java | 6 +- .../split/PulsarPartitionSplitReaderTestBase.java | 12 ++- .../testutils/sink/PulsarSinkTestSuiteBase.java| 37 -- 20 files changed, 389 insertions(+), 57 deletions(-) diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/metrics/MetricNames.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/metrics/MetricNames.java new file mode 100644 index 000..5156b47 --- /dev/null +++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/metrics/MetricNames.java @@ -0,0 +1,63 @@ +/* + * 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.connector.pulsar.common.metrics; + +/** The constant class for holding all the custom metrics names in Pulsar. */ +public final class MetricNames { + +private MetricNames() { +// No public constructor. +} + +public static final String PULSAR_PRODUCER_METRIC_NAME = "PulsarProducer"; +public static final String NUM_MSGS_SENT = "numMsgsSent"; +public static final String NUM_BYTES_SENT = "numBytesSent"; +public static final String NUM_SEND_FAILED = "numSendFailed"; +public static final String NUM_ACKS_RECEIVED = "numAcksReceived"; +public static final String SEND_MSGS_RATE = "sendMsgsRate"; +public static final String SEND_BYTES_RATE = "sendBytesRate"; +public static final String SEND_LATENCY_MILLIS_50_PCT = "sendLatencyMillis50pct"; +public static final String SEND_LATENCY_MILLIS_75_PCT = "sendLatencyMillis75pct"; +public static final String SEND_LATENCY_MILLIS_95_PCT = "sendLatencyMillis95pct"; +public static final String SEND_LATENCY_MILLIS_99_PCT = "sendLatencyMillis99pct"; +public static final String SEND_LATENCY_MILLIS_999_PCT = "sendLatencyMillis999pct"; +public static final String SEND_LATENCY_MILLIS_MAX = "sendLatencyMillisMax"; +public static final String TOTAL_MSGS_SENT = "totalMsgsSent"; +public static final String TOTAL_BYTES_SENT = "totalBytesSent"; +public static final String TOTAL_SEND_FAILED = "totalSendFailed"; +public static final String TOTAL_ACKS_RECEIVED = "totalAcksReceived"; +public static final String PENDING_QUEUE_SIZE = "pendingQueueSize"; + +public static final String PULSAR_CONSUMER_METRIC_NAME = "PulsarConsumer"; +public static final String NUM_MSGS_RECEIVED = "numMsgsReceived"; +public static final String NUM_BYTES_RECEIVED = "numBytesReceived"; +public static final String RATE_MSGS_RECEIVED = "rateMsgsReceived"; +public static final String RATE_BYTES_RECEIVED = "rateBytesReceived"; +public static final String N
[flink-connector-pulsar] 01/16: [FLINK-28853][pulsar] Implement pausing Pulsar splits in case of watermark drift
This is an automated email from the ASF dual-hosted git repository. martijnvisser pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git commit d21147e8bc3dab7d42ecd7b1fa48d2736ff7b918 Author: Dawid Wysakowicz AuthorDate: Fri Mar 25 09:33:33 2022 +0100 [FLINK-28853][pulsar] Implement pausing Pulsar splits in case of watermark drift --- .../source/reader/source/PulsarSourceReaderBase.java | 8 .../reader/split/PulsarPartitionSplitReaderBase.java | 16 2 files changed, 24 insertions(+) diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarSourceReaderBase.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarSourceReaderBase.java index 0122021..c910e94 100644 --- a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarSourceReaderBase.java +++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarSourceReaderBase.java @@ -32,6 +32,8 @@ import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplitState; import org.apache.pulsar.client.admin.PulsarAdmin; import org.apache.pulsar.client.api.PulsarClient; +import java.util.Collection; + /** * The common pulsar source reader for both ordered & unordered message consuming. * @@ -75,6 +77,12 @@ abstract class PulsarSourceReaderBase return splitState.toPulsarPartitionSplit(); } +@Override +public void pauseOrResumeSplits( +Collection splitsToPause, Collection splitsToResume) { +splitFetcherManager.pauseOrResumeSplits(splitsToPause, splitsToResume); +} + @Override public void close() throws Exception { // Close the all the consumers first. diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarPartitionSplitReaderBase.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarPartitionSplitReaderBase.java index 9537969..c1459a6 100644 --- a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarPartitionSplitReaderBase.java +++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarPartitionSplitReaderBase.java @@ -50,6 +50,7 @@ import javax.annotation.Nullable; import java.io.IOException; import java.time.Duration; +import java.util.Collection; import java.util.List; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeoutException; @@ -179,6 +180,21 @@ abstract class PulsarPartitionSplitReaderBase LOG.info("Register split {} consumer for current reader.", registeredSplit); } +@Override +public void pauseOrResumeSplits( +Collection splitsToPause, +Collection splitsToResume) { +if (splitsToPause.size() > 1 || splitsToResume.size() > 1) { +throw new IllegalStateException("This pulsar split reader only support one split."); +} + +if (!splitsToPause.isEmpty()) { +pulsarConsumer.pause(); +} else if (!splitsToResume.isEmpty()) { +pulsarConsumer.resume(); +} +} + @Override public void wakeUp() { // Nothing to do on this method.
[flink-connector-pulsar] 04/16: [hotfix][docs]: Fix minor grammar and spelling mistakes
This is an automated email from the ASF dual-hosted git repository. martijnvisser pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git commit 08e7b1997b209d6f484fd753d66541759de55a38 Author: Ryan Skraba AuthorDate: Wed Sep 1 10:27:07 2021 +0200 [hotfix][docs]: Fix minor grammar and spelling mistakes docs: Incomplete sentence docs: use consistent voice chore: Connectors document review chore: Reword SplitEnumerator [hotfix][docs] Use consistent intro sentence Apply grammar fixes from code review Co-authored-by: Matthias Pohl --- .../connector/pulsar/source/enumerator/cursor/StartCursor.java | 4 ++-- .../flink/connector/pulsar/source/enumerator/cursor/StopCursor.java | 6 +++--- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/cursor/StartCursor.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/cursor/StartCursor.java index 9c1d699..eae7cbf 100644 --- a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/cursor/StartCursor.java +++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/cursor/StartCursor.java @@ -28,8 +28,8 @@ import org.apache.pulsar.client.api.SubscriptionType; import java.io.Serializable; /** - * A interface for users to specify the start position of a pulsar subscription. Since it would be - * serialized into split. The implementation for this interface should be well considered. I don't + * An interface for users to specify the start position of a pulsar subscription. Since it would be + * serialized into split, the implementation for this interface should be well considered. I don't * recommend adding extra internal state for this implementation. * * This class would be used only for {@link SubscriptionType#Exclusive} and {@link diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/cursor/StopCursor.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/cursor/StopCursor.java index d44c78f..866d036 100644 --- a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/cursor/StopCursor.java +++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/cursor/StopCursor.java @@ -33,9 +33,9 @@ import org.apache.pulsar.client.api.MessageId; import java.io.Serializable; /** - * A interface for users to specify the stop position of a pulsar subscription. Since it would be - * serialized into split. The implementation for this interface should be well considered. I don't - * recommend adding extra internal state for this implementation. + * An interface for users to specify the stop position of a pulsar subscription. Since it would be + * serialized into split, the implementation for this interface should be well considered. It's not + * recommended to add extra internal state for this implementation. */ @PublicEvolving @FunctionalInterface
[flink-connector-pulsar] 05/16: [FLINK-29580][Connector/Pulsar] Remove pulsar.consumer.autoUpdatePartitionsIntervalSeconds option. (#21070)
This is an automated email from the ASF dual-hosted git repository. martijnvisser pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git commit dfb5fd86c481f09d5ac2c1abe8a2acaa0f60a862 Author: Yufan Sheng AuthorDate: Tue Oct 18 13:27:38 2022 +0800 [FLINK-29580][Connector/Pulsar] Remove pulsar.consumer.autoUpdatePartitionsIntervalSeconds option. (#21070) --- .../apache/flink/connector/pulsar/source/PulsarSourceOptions.java | 7 --- .../connector/pulsar/source/config/PulsarSourceConfigUtils.java| 5 - 2 files changed, 12 deletions(-) diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceOptions.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceOptions.java index 440e96b..d0ef6de 100644 --- a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceOptions.java +++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceOptions.java @@ -584,13 +584,6 @@ public final class PulsarSourceOptions { .defaultValue(false) .withDescription("If enabled, the consumer will automatically retry messages."); -public static final ConfigOption PULSAR_AUTO_UPDATE_PARTITIONS_INTERVAL_SECONDS = -ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "autoUpdatePartitionsIntervalSeconds") -.intType() -.defaultValue(60) -.withDescription( -"The interval (in seconds) of updating partitions. This only works if autoUpdatePartitions is enabled."); - public static final ConfigOption PULSAR_REPLICATE_SUBSCRIPTION_STATE = ConfigOptions.key(CONSUMER_CONFIG_PREFIX + "replicateSubscriptionState") .booleanType() diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/config/PulsarSourceConfigUtils.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/config/PulsarSourceConfigUtils.java index 063d588..d76ba0a 100644 --- a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/config/PulsarSourceConfigUtils.java +++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/config/PulsarSourceConfigUtils.java @@ -35,7 +35,6 @@ import java.util.concurrent.TimeUnit; import static java.util.concurrent.TimeUnit.MICROSECONDS; import static java.util.concurrent.TimeUnit.MILLISECONDS; -import static java.util.concurrent.TimeUnit.SECONDS; import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ADMIN_URL; import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_AUTH_PARAMS; import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_AUTH_PARAM_MAP; @@ -44,7 +43,6 @@ import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSA import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_ACK_RECEIPT_ENABLED; import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_ACK_TIMEOUT_MILLIS; import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_AUTO_ACK_OLDEST_CHUNKED_MESSAGE_ON_QUEUE_FULL; -import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_AUTO_UPDATE_PARTITIONS_INTERVAL_SECONDS; import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_CONSUMER_NAME; import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_CONSUMER_PROPERTIES; import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_CRYPTO_FAILURE_ACTION; @@ -121,9 +119,6 @@ public final class PulsarSourceConfigUtils { configuration.useOption(PULSAR_READ_COMPACTED, builder::readCompacted); configuration.useOption(PULSAR_PRIORITY_LEVEL, builder::priorityLevel); createDeadLetterPolicy(configuration).ifPresent(builder::deadLetterPolicy); -configuration.useOption( -PULSAR_AUTO_UPDATE_PARTITIONS_INTERVAL_SECONDS, -v -> builder.autoUpdatePartitionsInterval(v, SECONDS)); configuration.useOption(PULSAR_RETRY_ENABLE, builder::enableRetry); configuration.useOption( PULSAR_MAX_PENDING_CHUNKED_MESSAGE, builder::maxPendingChunkedMessage);
[flink-connector-pulsar] 09/16: [FLINK-29624][Common][Connector][Filesystem] Upgrade org.apache.commons:commons-lang3 from 3.3.2 to 3.12.0
This is an automated email from the ASF dual-hosted git repository. martijnvisser pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git commit 1c886323ad9edf985efbf04e602ed524d6fa6beb Author: Martijn Visser AuthorDate: Thu Oct 13 11:14:49 2022 +0200 [FLINK-29624][Common][Connector][Filesystem] Upgrade org.apache.commons:commons-lang3 from 3.3.2 to 3.12.0 * Remove Pulsar specific version of `commons-lang3` since Flink now uses a newer version * Replace usage of deprecated method `org.apache.commons.lang3.ArrayUtils#add` with `org.apache.commons.lang3.ArrayUtils#insert` * Replace usage of `org.apache.commons.lang3.StringEscapeUtils` with `org.apache.commons.text.StringEscapeUtils` --- flink-connector-pulsar/pom.xml | 10 -- 1 file changed, 10 deletions(-) diff --git a/flink-connector-pulsar/pom.xml b/flink-connector-pulsar/pom.xml index 30c0a1b..3118a9a 100644 --- a/flink-connector-pulsar/pom.xml +++ b/flink-connector-pulsar/pom.xml @@ -41,7 +41,6 @@ under the License. 0.6.1 1.7.0 - 3.11 4.1.77.Final 1.45.1 @@ -137,15 +136,6 @@ under the License. test - - - - org.apache.commons - commons-lang3 - ${pulsar-commons-lang3.version} - test - -
[flink-connector-pulsar] 11/16: [FLINK-28083][Connector/Pulsar] PulsarSource work with object-reusing DeserializationSchema. (#21205)
This is an automated email from the ASF dual-hosted git repository. martijnvisser pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git commit d6e2e06964c4cf40a5e58d4bc2f976a65b4a566b Author: Yufan Sheng AuthorDate: Tue Nov 8 14:53:22 2022 +0800 [FLINK-28083][Connector/Pulsar] PulsarSource work with object-reusing DeserializationSchema. (#21205) --- .../source/reader/PulsarSourceReaderFactory.java | 23 +++ .../source/reader/emitter/PulsarRecordEmitter.java | 63 -- .../reader/fetcher/PulsarFetcherManagerBase.java | 27 .../fetcher/PulsarOrderedFetcherManager.java | 18 +++--- .../fetcher/PulsarUnorderedFetcherManager.java | 14 ++-- .../source/reader/message/PulsarMessage.java | 74 -- .../reader/message/PulsarMessageCollector.java | 60 -- .../reader/source/PulsarOrderedSourceReader.java | 15 +++-- .../reader/source/PulsarSourceReaderBase.java | 18 ++ .../reader/source/PulsarUnorderedSourceReader.java | 13 ++-- .../split/PulsarOrderedPartitionSplitReader.java | 13 +--- .../split/PulsarPartitionSplitReaderBase.java | 33 +++--- .../split/PulsarUnorderedPartitionSplitReader.java | 11 +--- .../source/enumerator/cursor/StopCursorTest.java | 17 ++--- .../PulsarOrderedPartitionSplitReaderTest.java | 8 +-- .../split/PulsarPartitionSplitReaderTestBase.java | 74 ++ 16 files changed, 175 insertions(+), 306 deletions(-) diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/PulsarSourceReaderFactory.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/PulsarSourceReaderFactory.java index 820888a..2e83ab5 100644 --- a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/PulsarSourceReaderFactory.java +++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/PulsarSourceReaderFactory.java @@ -25,7 +25,7 @@ import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds; import org.apache.flink.connector.base.source.reader.synchronization.FutureCompletingBlockingQueue; import org.apache.flink.connector.pulsar.source.config.SourceConfiguration; import org.apache.flink.connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema; -import org.apache.flink.connector.pulsar.source.reader.message.PulsarMessage; +import org.apache.flink.connector.pulsar.source.reader.emitter.PulsarRecordEmitter; import org.apache.flink.connector.pulsar.source.reader.source.PulsarOrderedSourceReader; import org.apache.flink.connector.pulsar.source.reader.source.PulsarUnorderedSourceReader; import org.apache.flink.connector.pulsar.source.reader.split.PulsarOrderedPartitionSplitReader; @@ -33,6 +33,7 @@ import org.apache.flink.connector.pulsar.source.reader.split.PulsarUnorderedPart import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplit; import org.apache.pulsar.client.admin.PulsarAdmin; +import org.apache.pulsar.client.api.Message; import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.SubscriptionType; import org.apache.pulsar.client.api.transaction.TransactionCoordinatorClient; @@ -70,25 +71,25 @@ public final class PulsarSourceReaderFactory { // Create a message queue with the predefined source option. int queueCapacity = sourceConfiguration.getMessageQueueCapacity(); -FutureCompletingBlockingQueue>> elementsQueue = +FutureCompletingBlockingQueue>> elementsQueue = new FutureCompletingBlockingQueue<>(queueCapacity); +PulsarRecordEmitter recordEmitter = new PulsarRecordEmitter<>(deserializationSchema); + // Create different pulsar source reader by subscription type. SubscriptionType subscriptionType = sourceConfiguration.getSubscriptionType(); if (subscriptionType == SubscriptionType.Failover || subscriptionType == SubscriptionType.Exclusive) { // Create an ordered split reader supplier. -Supplier> splitReaderSupplier = +Supplier splitReaderSupplier = () -> -new PulsarOrderedPartitionSplitReader<>( -pulsarClient, -pulsarAdmin, -sourceConfiguration, -deserializationSchema); +new PulsarOrderedPartitionSplitReader( +pulsarClient, pulsarAdmin, sourceConfiguration); return new PulsarOrderedSourceReader<>( elementsQueue, splitReaderSupplier, +recordEmitter, readerContext, sourceConfigu
[flink-connector-pulsar] 15/16: [FLINK-30254][Connector/Pulsar] Sync Pulsar updates and set Flink version to 1.17-SNAPSHOT
This is an automated email from the ASF dual-hosted git repository. martijnvisser pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git commit d330e403dfe6e2b1cb5051c93d1ff77d103746a3 Author: Martijn Visser AuthorDate: Wed Nov 30 20:03:59 2022 +0100 [FLINK-30254][Connector/Pulsar] Sync Pulsar updates and set Flink version to 1.17-SNAPSHOT --- .github/workflows/push_pr.yml | 6 +++--- .github/workflows/weekly.yml | 2 +- pom.xml | 2 +- 3 files changed, 5 insertions(+), 5 deletions(-) diff --git a/.github/workflows/push_pr.yml b/.github/workflows/push_pr.yml index be18918..759de4b 100644 --- a/.github/workflows/push_pr.yml +++ b/.github/workflows/push_pr.yml @@ -22,6 +22,6 @@ jobs: compile_and_test: uses: ./.github/workflows/ci.yml with: - flink_version: 1.16.0 - flink_url: https://dist.apache.org/repos/dist/release/flink/flink-1.16.0/flink-1.16.0-bin-scala_2.12.tgz - cache_flink_binary: true \ No newline at end of file + flink_version: 1.17-SNAPSHOT + flink_url: https://s3.amazonaws.com/flink-nightly/flink-1.17-SNAPSHOT-bin-scala_2.12.tgz + cache_flink_binary: false \ No newline at end of file diff --git a/.github/workflows/weekly.yml b/.github/workflows/weekly.yml index 28aed6b..cb628f2 100644 --- a/.github/workflows/weekly.yml +++ b/.github/workflows/weekly.yml @@ -24,7 +24,7 @@ jobs: compile_and_test: strategy: matrix: -flink: [1.16-SNAPSHOT, 1.17-SNAPSHOT] +flink: [1.17-SNAPSHOT] uses: ./.github/workflows/ci.yml with: flink_version: ${{ matrix.flink }} diff --git a/pom.xml b/pom.xml index 2fca0a9..9c5040e 100644 --- a/pom.xml +++ b/pom.xml @@ -48,7 +48,7 @@ under the License. -1.16.0 +1.17-SNAPSHOT 2.13.4.20221013 1.45.1
[flink-connector-pulsar] 06/16: [FLINK-29433][Connector/Pulsar] Support Auth through the builder pattern in Pulsar connector. (#21071)
This is an automated email from the ASF dual-hosted git repository. martijnvisser pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git commit 618daa266d483f4b0d3af03ffbb47315e872df3a Author: Yufan Sheng AuthorDate: Tue Oct 18 13:29:39 2022 +0800 [FLINK-29433][Connector/Pulsar] Support Auth through the builder pattern in Pulsar connector. (#21071) --- .../connector/pulsar/sink/PulsarSinkBuilder.java | 33 ++ .../pulsar/source/PulsarSourceBuilder.java | 33 ++ 2 files changed, 66 insertions(+) diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/PulsarSinkBuilder.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/PulsarSinkBuilder.java index 332a27a..4e8f9b2 100644 --- a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/PulsarSinkBuilder.java +++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/PulsarSinkBuilder.java @@ -38,9 +38,13 @@ import org.slf4j.LoggerFactory; import java.util.Arrays; import java.util.List; +import java.util.Map; import java.util.Properties; import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ADMIN_URL; +import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_AUTH_PARAMS; +import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_AUTH_PARAM_MAP; +import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_AUTH_PLUGIN_CLASS_NAME; import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ENABLE_TRANSACTION; import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_SERVICE_URL; import static org.apache.flink.connector.pulsar.sink.PulsarSinkOptions.PULSAR_PRODUCER_NAME; @@ -243,6 +247,35 @@ public class PulsarSinkBuilder { return this; } +/** + * Configure the authentication provider to use in the Pulsar client instance. + * + * @param authPluginClassName name of the Authentication-Plugin you want to use + * @param authParamsString string which represents parameters for the Authentication-Plugin, + * e.g., "key1:val1,key2:val2" + * @return this PulsarSinkBuilder. + */ +public PulsarSinkBuilder setAuthentication( +String authPluginClassName, String authParamsString) { +configBuilder.set(PULSAR_AUTH_PLUGIN_CLASS_NAME, authPluginClassName); +configBuilder.set(PULSAR_AUTH_PARAMS, authParamsString); +return this; +} + +/** + * Configure the authentication provider to use in the Pulsar client instance. + * + * @param authPluginClassName name of the Authentication-Plugin you want to use + * @param authParams map which represents parameters for the Authentication-Plugin + * @return this PulsarSinkBuilder. + */ +public PulsarSinkBuilder setAuthentication( +String authPluginClassName, Map authParams) { +configBuilder.set(PULSAR_AUTH_PLUGIN_CLASS_NAME, authPluginClassName); +configBuilder.set(PULSAR_AUTH_PARAM_MAP, authParams); +return this; +} + /** * Set an arbitrary property for the PulsarSink and Pulsar Producer. The valid keys can be found * in {@link PulsarSinkOptions} and {@link PulsarOptions}. diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceBuilder.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceBuilder.java index 5309dd0..5aac3d2 100644 --- a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceBuilder.java +++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceBuilder.java @@ -44,11 +44,15 @@ import org.slf4j.LoggerFactory; import java.util.Arrays; import java.util.List; +import java.util.Map; import java.util.Properties; import java.util.regex.Pattern; import static java.lang.Boolean.FALSE; import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ADMIN_URL; +import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_AUTH_PARAMS; +import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_AUTH_PARAM_MAP; +import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_AUTH_PLUGIN_CLASS_NAME; import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ENABLE_TRANSACTION; import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_SERVICE_URL; import static org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_CONSUMER_NAME; @@ -369,6 +373,35 @@ public final class PulsarSourceBuilder { return self; } +/** + * Configu
[flink-connector-pulsar] 07/16: [FLINK-28085][Connector/Pulsar] Close the pending Pulsar transactions before closing pipeline (#21075)
This is an automated email from the ASF dual-hosted git repository. martijnvisser pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git commit 62da02aec6c62067bacaef932fff496553cc5c60 Author: Yufan Sheng AuthorDate: Tue Oct 18 20:23:24 2022 +0800 [FLINK-28085][Connector/Pulsar] Close the pending Pulsar transactions before closing pipeline (#21075) --- .../common/utils/PulsarTransactionUtils.java | 2 ++ .../reader/fetcher/PulsarFetcherManagerBase.java | 12 ++ .../reader/source/PulsarOrderedSourceReader.java | 3 +++ .../reader/source/PulsarSourceReaderBase.java | 7 ++ .../reader/source/PulsarUnorderedSourceReader.java | 27 ++ .../split/PulsarUnorderedPartitionSplitReader.java | 14 +++ 6 files changed, 61 insertions(+), 4 deletions(-) diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/utils/PulsarTransactionUtils.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/utils/PulsarTransactionUtils.java index a48b4d4..c8ffe63 100644 --- a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/utils/PulsarTransactionUtils.java +++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/utils/PulsarTransactionUtils.java @@ -60,6 +60,8 @@ public final class PulsarTransactionUtils { /** * This is a bug in original {@link TransactionCoordinatorClientException#unwrap(Throwable)} * method. Pulsar wraps the {@link ExecutionException} which hides the real execution exception. + * + * This bug should be fixed after the 2.10.0 release. We just keep this for safety. */ public static TransactionCoordinatorClientException unwrap( TransactionCoordinatorClientException e) { diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/fetcher/PulsarFetcherManagerBase.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/fetcher/PulsarFetcherManagerBase.java index 6abb8bd..1622eee 100644 --- a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/fetcher/PulsarFetcherManagerBase.java +++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/fetcher/PulsarFetcherManagerBase.java @@ -86,6 +86,18 @@ public abstract class PulsarFetcherManagerBase } } +/** Close the finished split related fetcher. */ +public void closeFetcher(String splitId) { +Integer fetchId = splitFetcherMapping.remove(splitId); +if (fetchId != null) { +fetcherStatus.remove(fetchId); +SplitFetcher, PulsarPartitionSplit> fetcher = fetchers.remove(fetchId); +if (fetcher != null) { +fetcher.shutdown(); +} +} +} + protected SplitFetcher, PulsarPartitionSplit> getOrCreateFetcher( String splitId) { SplitFetcher, PulsarPartitionSplit> fetcher; diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarOrderedSourceReader.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarOrderedSourceReader.java index a06d8a4..6ff0466 100644 --- a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarOrderedSourceReader.java +++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarOrderedSourceReader.java @@ -109,6 +109,9 @@ public class PulsarOrderedSourceReader extends PulsarSourceReaderBase @Override protected void onSplitFinished(Map finishedSplitIds) { +// Close all the finished splits. +closeFinishedSplits(finishedSplitIds.keySet()); + // We don't require new splits, all the splits are pre-assigned by source enumerator. if (LOG.isDebugEnabled()) { LOG.debug("onSplitFinished event: {}", finishedSplitIds); diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarSourceReaderBase.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarSourceReaderBase.java index c910e94..eee6950 100644 --- a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarSourceReaderBase.java +++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarSourceReaderBase.java @@ -33,6 +33,7 @@ import org.apache.pulsar.client.admin.PulsarAdmin; import org.apache.pulsar.client.api.PulsarClient; import java.util.Collection; +import java.util.Set; /** * The common pulsar source reader for both ordered & unordered message consuming. @@ -92,4 +93,10 @@ abstract class PulsarSource
[flink-connector-pulsar] 10/16: [FLINK-29709][Connector/Pulsar] Bump the Pulsar to latest 2.10.2 (#21204)
This is an automated email from the ASF dual-hosted git repository. martijnvisser pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git commit 4f204e8091fc870c0928b3834de691dc8b2c4dfb Author: Yufan Sheng AuthorDate: Mon Nov 7 11:44:15 2022 +0800 [FLINK-29709][Connector/Pulsar] Bump the Pulsar to latest 2.10.2 (#21204) --- flink-connector-pulsar/pom.xml| 9 - .../connector/pulsar/common/schema/PulsarSchema.java | 15 ++- .../connector/pulsar/common/schema/PulsarSchemaUtils.java | 9 +++-- .../src/main/resources/META-INF/NOTICE| 8 4 files changed, 33 insertions(+), 8 deletions(-) diff --git a/flink-connector-pulsar/pom.xml b/flink-connector-pulsar/pom.xml index 3118a9a..5881495 100644 --- a/flink-connector-pulsar/pom.xml +++ b/flink-connector-pulsar/pom.xml @@ -35,7 +35,7 @@ under the License. jar - 2.10.1 + 2.10.2 3.21.7 @@ -43,6 +43,7 @@ under the License. 1.7.0 4.1.77.Final 1.45.1 + 2.9.1 @@ -135,6 +136,12 @@ under the License. ${pulsar.version} test + + com.github.ben-manes.caffeine + caffeine + ${pulsar-caffeine.version} + test + diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/schema/PulsarSchema.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/schema/PulsarSchema.java index bb09315..fe5daca 100644 --- a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/schema/PulsarSchema.java +++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/schema/PulsarSchema.java @@ -155,6 +155,9 @@ public final class PulsarSchema implements Serializable { oos.writeUTF(entry.getKey()); oos.writeUTF(entry.getValue()); } + +// Timestamp +oos.writeLong(schemaInfo.getTimestamp()); } private void readObject(ObjectInputStream ois) throws ClassNotFoundException, IOException { @@ -177,7 +180,17 @@ public final class PulsarSchema implements Serializable { properties.put(ois.readUTF(), ois.readUTF()); } -this.schemaInfo = new SchemaInfoImpl(name, schemaBytes, type, properties); +// Timestamp +long timestamp = ois.readLong(); + +this.schemaInfo = +SchemaInfoImpl.builder() +.name(name) +.schema(schemaBytes) +.type(type) +.properties(properties) +.timestamp(timestamp) +.build(); this.schema = createSchema(schemaInfo); } diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/schema/PulsarSchemaUtils.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/schema/PulsarSchemaUtils.java index 4b1f7ee..00bf2cf 100644 --- a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/schema/PulsarSchemaUtils.java +++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/common/schema/PulsarSchemaUtils.java @@ -181,8 +181,13 @@ public final class PulsarSchemaUtils { Map properties = new HashMap<>(schemaInfo.getProperties()); properties.put(CLASS_INFO_PLACEHOLDER, typeClass.getName()); -return new SchemaInfoImpl( -schemaInfo.getName(), schemaInfo.getSchema(), schemaInfo.getType(), properties); +return SchemaInfoImpl.builder() +.name(schemaInfo.getName()) +.schema(schemaInfo.getSchema()) +.type(schemaInfo.getType()) +.properties(properties) +.timestamp(schemaInfo.getTimestamp()) +.build(); } @SuppressWarnings("unchecked") diff --git a/flink-sql-connector-pulsar/src/main/resources/META-INF/NOTICE b/flink-sql-connector-pulsar/src/main/resources/META-INF/NOTICE index 151da33..1aee6f4 100644 --- a/flink-sql-connector-pulsar/src/main/resources/META-INF/NOTICE +++ b/flink-sql-connector-pulsar/src/main/resources/META-INF/NOTICE @@ -6,10 +6,10 @@ The Apache Software Foundation (http://www.apache.org/). This project bundles the following dependencies under the Apache Software License 2.0 (http://www.apache.org/licenses/LICENSE-2.0.txt) -- org.apache.pulsar:bouncy-castle-bc:pkg:2.10.1 -- org.apache.pulsar:pulsar-client-admin-api:2.10.1 -- org.apache.pulsar:pulsar-client-all:2.10.1 -- org.apache.pulsar:pulsar-client-api:2.10.1 +- or
[flink-connector-pulsar] 12/16: [FLINK-29860][Connectors/Pulsar] InlineElement fields should be transient (#21262)
This is an automated email from the ASF dual-hosted git repository. martijnvisser pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git commit f6ead4c4ef3c195381c7f5090764fbda881ebcc4 Author: noelo AuthorDate: Thu Nov 10 12:27:55 2022 +0100 [FLINK-29860][Connectors/Pulsar] InlineElement fields should be transient (#21262) --- .../flink/connector/pulsar/sink/writer/router/MessageKeyHash.java | 2 +- .../flink/connector/pulsar/sink/writer/router/TopicRoutingMode.java | 2 +- .../apache/flink/connector/pulsar/source/config/CursorVerification.java | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/router/MessageKeyHash.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/router/MessageKeyHash.java index bbac99e..3ad092d 100644 --- a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/router/MessageKeyHash.java +++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/router/MessageKeyHash.java @@ -62,7 +62,7 @@ public enum MessageKeyHash implements DescribedEnum { }; private final String name; -private final InlineElement desc; +private final transient InlineElement desc; MessageKeyHash(String name, InlineElement desc) { this.name = name; diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/router/TopicRoutingMode.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/router/TopicRoutingMode.java index c327435..f251f8f 100644 --- a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/router/TopicRoutingMode.java +++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/writer/router/TopicRoutingMode.java @@ -67,7 +67,7 @@ public enum TopicRoutingMode implements DescribedEnum { code(TopicRouter.class.getSimpleName(; private final String name; -private final InlineElement desc; +private final transient InlineElement desc; TopicRoutingMode(String name, InlineElement desc) { this.name = name; diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/config/CursorVerification.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/config/CursorVerification.java index fc70bdd..d604ceb 100644 --- a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/config/CursorVerification.java +++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/config/CursorVerification.java @@ -35,7 +35,7 @@ public enum CursorVerification implements DescribedEnum { /** Print a warn message and start consuming from the valid offset. */ WARN_ON_MISMATCH(text("Print a warn message and start consuming from the valid offset.")); -private final InlineElement desc; +private final transient InlineElement desc; CursorVerification(InlineElement desc) { this.desc = desc;
[flink-connector-pulsar] 02/16: [FLINK-28853][connector-base] Add allow-unaligned-source-splits config option
This is an automated email from the ASF dual-hosted git repository. martijnvisser pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git commit 9cce62760a86b3dc3c4c8aef45d7e1444a436914 Author: Sebastian Mattheis AuthorDate: Sun Jul 24 12:09:48 2022 +0200 [FLINK-28853][connector-base] Add allow-unaligned-source-splits config option --- .../pulsar/source/reader/fetcher/PulsarFetcherManagerBase.java | 6 -- .../pulsar/source/reader/fetcher/PulsarOrderedFetcherManager.java | 6 -- .../pulsar/source/reader/fetcher/PulsarUnorderedFetcherManager.java | 6 -- .../pulsar/source/reader/source/PulsarOrderedSourceReader.java | 3 ++- .../pulsar/source/reader/source/PulsarUnorderedSourceReader.java| 3 ++- 5 files changed, 16 insertions(+), 8 deletions(-) diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/fetcher/PulsarFetcherManagerBase.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/fetcher/PulsarFetcherManagerBase.java index 1f9a35f..6abb8bd 100644 --- a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/fetcher/PulsarFetcherManagerBase.java +++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/fetcher/PulsarFetcherManagerBase.java @@ -19,6 +19,7 @@ package org.apache.flink.connector.pulsar.source.reader.fetcher; import org.apache.flink.annotation.Internal; +import org.apache.flink.configuration.Configuration; import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds; import org.apache.flink.connector.base.source.reader.SourceReaderBase; import org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher; @@ -57,8 +58,9 @@ public abstract class PulsarFetcherManagerBase */ protected PulsarFetcherManagerBase( FutureCompletingBlockingQueue>> elementsQueue, -Supplier, PulsarPartitionSplit>> splitReaderSupplier) { -super(elementsQueue, splitReaderSupplier); +Supplier, PulsarPartitionSplit>> splitReaderSupplier, +Configuration configuration) { +super(elementsQueue, splitReaderSupplier, configuration); } /** diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/fetcher/PulsarOrderedFetcherManager.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/fetcher/PulsarOrderedFetcherManager.java index f8b89ee..103dc62 100644 --- a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/fetcher/PulsarOrderedFetcherManager.java +++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/fetcher/PulsarOrderedFetcherManager.java @@ -19,6 +19,7 @@ package org.apache.flink.connector.pulsar.source.reader.fetcher; import org.apache.flink.annotation.Internal; +import org.apache.flink.configuration.Configuration; import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds; import org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher; import org.apache.flink.connector.base.source.reader.splitreader.SplitReader; @@ -49,8 +50,9 @@ public class PulsarOrderedFetcherManager extends PulsarFetcherManagerBase public PulsarOrderedFetcherManager( FutureCompletingBlockingQueue>> elementsQueue, -Supplier, PulsarPartitionSplit>> splitReaderSupplier) { -super(elementsQueue, splitReaderSupplier); +Supplier, PulsarPartitionSplit>> splitReaderSupplier, +Configuration configuration) { +super(elementsQueue, splitReaderSupplier, configuration); } public void acknowledgeMessages(Map cursorsToCommit) { diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/fetcher/PulsarUnorderedFetcherManager.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/fetcher/PulsarUnorderedFetcherManager.java index 1523b9a..449e992 100644 --- a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/fetcher/PulsarUnorderedFetcherManager.java +++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/fetcher/PulsarUnorderedFetcherManager.java @@ -19,6 +19,7 @@ package org.apache.flink.connector.pulsar.source.reader.fetcher; import org.apache.flink.annotation.Internal; +import org.apache.flink.configuration.Configuration; import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds; import org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher; import org.apache.flink.connector.base.source.reader.splitreader.SplitReader; @@ -49,8 +50,9 @@ public class PulsarUnorderedFetcherManager extends PulsarFetcherManagerBase>> elementsQueue, -Supplier, PulsarPartitionSpl
[flink-connector-pulsar] 08/16: [FLINK-28820][Connector/Pulsar] Improve the writing performance for PulsarSink (#21074)
This is an automated email from the ASF dual-hosted git repository. martijnvisser pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git commit c6a20a66bf22383c114a6ae23f70413d6736aa51 Author: Yufan Sheng AuthorDate: Tue Oct 18 21:32:37 2022 +0800 [FLINK-28820][Connector/Pulsar] Improve the writing performance for PulsarSink (#21074) --- .../connector/pulsar/sink/PulsarSinkOptions.java | 2 + .../pulsar/sink/config/PulsarSinkConfigUtils.java | 4 -- .../pulsar/sink/config/SinkConfiguration.java | 13 .../connector/pulsar/sink/writer/PulsarWriter.java | 82 +- 4 files changed, 35 insertions(+), 66 deletions(-) diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/PulsarSinkOptions.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/PulsarSinkOptions.java index dafb8aa..0433bb0 100644 --- a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/PulsarSinkOptions.java +++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/PulsarSinkOptions.java @@ -127,6 +127,8 @@ public final class PulsarSinkOptions { "The allowed transaction recommit times if we meet some retryable exception." + " This is used in Pulsar Transaction."); +/** @deprecated This config option was removed for better performance. */ +@Deprecated public static final ConfigOption PULSAR_MAX_PENDING_MESSAGES_ON_PARALLELISM = ConfigOptions.key(SINK_CONFIG_PREFIX + "maxPendingMessages") .intType() diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/config/PulsarSinkConfigUtils.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/config/PulsarSinkConfigUtils.java index 5f935a4..61cfd5a 100644 --- a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/config/PulsarSinkConfigUtils.java +++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/config/PulsarSinkConfigUtils.java @@ -84,10 +84,6 @@ public final class PulsarSinkConfigUtils { PULSAR_SEND_TIMEOUT_MS, Math::toIntExact, ms -> builder.sendTimeout(ms, MILLISECONDS)); -configuration.useOption(PULSAR_MAX_PENDING_MESSAGES, builder::maxPendingMessages); -configuration.useOption( -PULSAR_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS, -builder::maxPendingMessagesAcrossPartitions); configuration.useOption( PULSAR_BATCHING_MAX_PUBLISH_DELAY_MICROS, s -> builder.batchingMaxPublishDelay(s, MICROSECONDS)); diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/config/SinkConfiguration.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/config/SinkConfiguration.java index 768b730..d6a6ee3 100644 --- a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/config/SinkConfiguration.java +++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/sink/config/SinkConfiguration.java @@ -32,7 +32,6 @@ import org.apache.pulsar.client.api.Schema; import java.util.Objects; import static org.apache.flink.connector.pulsar.sink.PulsarSinkOptions.PULSAR_BATCHING_MAX_MESSAGES; -import static org.apache.flink.connector.pulsar.sink.PulsarSinkOptions.PULSAR_MAX_PENDING_MESSAGES_ON_PARALLELISM; import static org.apache.flink.connector.pulsar.sink.PulsarSinkOptions.PULSAR_MAX_RECOMMIT_TIMES; import static org.apache.flink.connector.pulsar.sink.PulsarSinkOptions.PULSAR_MESSAGE_KEY_HASH; import static org.apache.flink.connector.pulsar.sink.PulsarSinkOptions.PULSAR_TOPIC_METADATA_REFRESH_INTERVAL; @@ -51,7 +50,6 @@ public class SinkConfiguration extends PulsarConfiguration { private final int partitionSwitchSize; private final MessageKeyHash messageKeyHash; private final boolean enableSchemaEvolution; -private final int maxPendingMessages; private final int maxRecommitTimes; public SinkConfiguration(Configuration configuration) { @@ -63,7 +61,6 @@ public class SinkConfiguration extends PulsarConfiguration { this.partitionSwitchSize = getInteger(PULSAR_BATCHING_MAX_MESSAGES); this.messageKeyHash = get(PULSAR_MESSAGE_KEY_HASH); this.enableSchemaEvolution = get(PULSAR_WRITE_SCHEMA_EVOLUTION); -this.maxPendingMessages = get(PULSAR_MAX_PENDING_MESSAGES_ON_PARALLELISM); this.maxRecommitTimes = get(PULSAR_MAX_RECOMMIT_TIMES); } @@ -111,14 +108,6 @@ public class SinkConfiguration extends PulsarConfiguration { return enableSchemaEvolution; } -/** - * Pulsar message is sent asynchronously. Set this option for limiting the pending messages
[flink-connector-pulsar] 16/16: [FLINK-30254][Connector/Pulsar] Resolve dependency convergence errors
This is an automated email from the ASF dual-hosted git repository. martijnvisser pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git commit 319992293faaf1c678b37904c3996c990f6e279d Author: Martijn Visser AuthorDate: Wed Nov 30 20:11:23 2022 +0100 [FLINK-30254][Connector/Pulsar] Resolve dependency convergence errors --- pom.xml | 11 +-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/pom.xml b/pom.xml index 9c5040e..64472a7 100644 --- a/pom.xml +++ b/pom.xml @@ -54,10 +54,10 @@ under the License. 1.45.1 1.69 1.4.0 -9.4.44.v20210927 +9.4.48.v20220622 0.8.1 4.13.2 -5.8.1 +5.9.1 3.23.1 0.22.0 1.17.2 @@ -315,6 +315,13 @@ under the License. ${junit5.version} + + +org.junit.platform +junit-platform-commons +1.9.1 + + org.checkerframework
[flink-kubernetes-operator] branch main updated (2c207edf -> 41b9f4fc)
This is an automated email from the ASF dual-hosted git repository. mbalassi pushed a change to branch main in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git from 2c207edf [FLINK-30154] Allow to set flinkConfiguration in SessionJob add 41b9f4fc [FLINK-30280] Document logging configuration behaviour No new revisions were added by this update. Summary of changes: helm/flink-kubernetes-operator/values.yaml | 8 +++- 1 file changed, 7 insertions(+), 1 deletion(-)
[flink-kubernetes-operator] branch main updated: [FLINK-30287] Delete job manager deployment first when cleaning up standalone cluster
This is an automated email from the ASF dual-hosted git repository. morhidi pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git The following commit(s) were added to refs/heads/main by this push: new f3db414f [FLINK-30287] Delete job manager deployment first when cleaning up standalone cluster f3db414f is described below commit f3db414ff2673d69aa113d63c996912fb9d57bc4 Author: Steven Zhang AuthorDate: Fri Dec 2 11:28:23 2022 -0800 [FLINK-30287] Delete job manager deployment first when cleaning up standalone cluster --- .../flink/kubernetes/operator/service/StandaloneFlinkService.java | 8 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/StandaloneFlinkService.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/StandaloneFlinkService.java index a9a5c21f..ad3e190e 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/StandaloneFlinkService.java +++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/StandaloneFlinkService.java @@ -134,20 +134,20 @@ public class StandaloneFlinkService extends AbstractFlinkService { final String clusterId = meta.getName(); final String namespace = meta.getNamespace(); -LOG.info("Deleting Flink Standalone cluster TM resources"); +LOG.info("Deleting Flink Standalone cluster JM resources"); kubernetesClient .apps() .deployments() .inNamespace(namespace) - .withName(StandaloneKubernetesUtils.getTaskManagerDeploymentName(clusterId)) + .withName(StandaloneKubernetesUtils.getJobManagerDeploymentName(clusterId)) .delete(); -LOG.info("Deleting Flink Standalone cluster JM resources"); +LOG.info("Deleting Flink Standalone cluster TM resources"); kubernetesClient .apps() .deployments() .inNamespace(namespace) - .withName(StandaloneKubernetesUtils.getJobManagerDeploymentName(clusterId)) + .withName(StandaloneKubernetesUtils.getTaskManagerDeploymentName(clusterId)) .delete(); if (deleteHaConfigmaps) {
[flink-kubernetes-operator] branch main updated: [FLINK-30046] Support for operator docker image digest
This is an automated email from the ASF dual-hosted git repository. morhidi pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git The following commit(s) were added to refs/heads/main by this push: new 03654167 [FLINK-30046] Support for operator docker image digest 03654167 is described below commit 0365416773390ae7a4946ba7180abb9e66b0dd86 Author: Marton Balassi AuthorDate: Tue Dec 6 16:25:07 2022 +0100 [FLINK-30046] Support for operator docker image digest --- docs/content/docs/operations/helm.md | 1 + helm/flink-kubernetes-operator/templates/_helpers.tpl| 11 +++ helm/flink-kubernetes-operator/templates/flink-operator.yaml | 4 ++-- helm/flink-kubernetes-operator/values.yaml | 2 ++ 4 files changed, 16 insertions(+), 2 deletions(-) diff --git a/docs/content/docs/operations/helm.md b/docs/content/docs/operations/helm.md index 1aaca026..3c5969ba 100644 --- a/docs/content/docs/operations/helm.md +++ b/docs/content/docs/operations/helm.md @@ -63,6 +63,7 @@ The configurable parameters of the Helm chart and which default values as detail | image.repository | The image repository of flink-kubernetes-operator. | ghcr.io/apache/flink-kubernetes-operator | | image.pullPolicy | The image pull policy of flink-kubernetes-operator. | IfNotPresent | | image.tag | The image tag of flink-kubernetes-operator. | latest | +| image.digest | The image tag of flink-kubernetes-operator. If set then it takes precedence and the image tag will be ignored. | | | replicas | Operator replica count. Must be 1 unless leader election is configured. | 1 | | strategy.type | Operator pod upgrade strategy. Must be Recreate unless leader election is configured. | Recreate | | rbac.create | Whether to enable RBAC to create for said namespaces. | true | diff --git a/helm/flink-kubernetes-operator/templates/_helpers.tpl b/helm/flink-kubernetes-operator/templates/_helpers.tpl index aa53cb3d..3456ab46 100644 --- a/helm/flink-kubernetes-operator/templates/_helpers.tpl +++ b/helm/flink-kubernetes-operator/templates/_helpers.tpl @@ -67,6 +67,17 @@ Selector labels app.kubernetes.io/name: {{ include "flink-operator.name" . }} {{- end }} +{{/* +Create the path of the operator image to use +*/}} +{{- define "flink-operator.imagePath" -}} +{{- if .Values.image.digest }} +{{- .Values.image.repository }}@{{ .Values.image.digest }} +{{- else }} +{{- .Values.image.repository }}:{{ default .Chart.AppVersion .Values.image.tag }} +{{- end }} +{{- end }} + {{/* Create the name of the operator role to use */}} diff --git a/helm/flink-kubernetes-operator/templates/flink-operator.yaml b/helm/flink-kubernetes-operator/templates/flink-operator.yaml index e8caa574..f94c2198 100644 --- a/helm/flink-kubernetes-operator/templates/flink-operator.yaml +++ b/helm/flink-kubernetes-operator/templates/flink-operator.yaml @@ -63,7 +63,7 @@ spec: serviceAccountName: {{ include "flink-operator.serviceAccountName" . }} containers: - name: {{ .Chart.Name }} - image: "{{ .Values.image.repository }}:{{ .Values.image.tag | default .Chart.AppVersion }}" + image: {{ include "flink-operator.imagePath" . }} imagePullPolicy: {{ .Values.image.pullPolicy }} command: ["/docker-entrypoint.sh", "operator"] ports: @@ -127,7 +127,7 @@ spec: {{- end }} {{- if eq (include "flink-operator.webhook-enabled" .) "true" }} - name: flink-webhook - image: "{{ .Values.image.repository }}:{{ .Values.image.tag | default .Chart.AppVersion }}" + image: {{ include "flink-operator.imagePath" . }} imagePullPolicy: {{ .Values.image.pullPolicy }} command: ["/docker-entrypoint.sh", "webhook"] env: diff --git a/helm/flink-kubernetes-operator/values.yaml b/helm/flink-kubernetes-operator/values.yaml index e55ba519..32517cc4 100644 --- a/helm/flink-kubernetes-operator/values.yaml +++ b/helm/flink-kubernetes-operator/values.yaml @@ -26,6 +26,8 @@ image: repository: flink-kubernetes-operator pullPolicy: IfNotPresent tag: latest + # If image digest is set then it takes precedence and the image tag will be ignored + # digest: "" imagePullSecrets: []
[flink-kubernetes-operator] branch main updated: [FLINK-30287] add unit test
This is an automated email from the ASF dual-hosted git repository. morhidi pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git The following commit(s) were added to refs/heads/main by this push: new ca8c5112 [FLINK-30287] add unit test ca8c5112 is described below commit ca8c5112af4173c68caa91f6470bac360fc0374f Author: Steven Zhang AuthorDate: Tue Dec 6 15:15:26 2022 -0800 [FLINK-30287] add unit test --- .../flink/kubernetes/operator/service/StandaloneFlinkServiceTest.java | 4 1 file changed, 4 insertions(+) diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/StandaloneFlinkServiceTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/StandaloneFlinkServiceTest.java index ff68c6fc..b406220f 100644 --- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/StandaloneFlinkServiceTest.java +++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/StandaloneFlinkServiceTest.java @@ -77,9 +77,13 @@ public class StandaloneFlinkServiceTest { assertEquals(2, deployments.size()); +var requestsBeforeDelete = mockServer.getRequestCount(); flinkStandaloneService.deleteClusterDeployment( flinkDeployment.getMetadata(), flinkDeployment.getStatus(), false); +assertEquals(2, mockServer.getRequestCount() - requestsBeforeDelete); + assertTrue(mockServer.getLastRequest().getPath().contains("taskmanager")); + deployments = kubernetesClient.apps().deployments().list().getItems(); assertEquals(0, deployments.size());
[flink-kubernetes-operator] branch main updated: Add OLM bundle generator
This is an automated email from the ASF dual-hosted git repository. morhidi pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git The following commit(s) were added to refs/heads/main by this push: new 66953e01 Add OLM bundle generator 66953e01 is described below commit 66953e01dcfa9ed1b31a7b6b854373cce477a8db Author: ted chang AuthorDate: Wed Nov 30 13:37:13 2022 -0800 Add OLM bundle generator Signed-off-by: ted chang --- tools/olm/AutomatedOLMBuildAndDeploy.sh| 86 tools/olm/README.md| 144 ++ ...-kubernetes-operator.clusterserviceversion.yaml | 216 + tools/olm/csv-template/bases/initContainer.yaml| 40 tools/olm/csv-template/bundle.Dockerfile | 37 tools/olm/csv-template/catalog.Dockerfile | 33 tools/olm/docker-entry.sh | 114 +++ tools/olm/env.sh | 42 tools/olm/generate-olm-bundle.sh | 153 +++ tools/olm/install-prereq.sh| 97 + tools/olm/utils.Dockerfile | 43 11 files changed, 1005 insertions(+) diff --git a/tools/olm/AutomatedOLMBuildAndDeploy.sh b/tools/olm/AutomatedOLMBuildAndDeploy.sh new file mode 100755 index ..359388a5 --- /dev/null +++ b/tools/olm/AutomatedOLMBuildAndDeploy.sh @@ -0,0 +1,86 @@ +#!/bin/bash + +# 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. + +set +x +# Script that Automates the OLM Creation and test of Flink Operator + +# Run the generate-olm-bundle.sh script +./generate-olm-bundle.sh + +# Reset Cluster +kind delete cluster +kind create cluster +kubectl cluster-info --context kind-kind +# operator-sdk olm install +curl -sL https://github.com/operator-framework/operator-lifecycle-manager/releases/download/v0.22.0/install.sh | bash -s v0.22.0 + +# Deploy the catalog src +cat
[flink-table-store] branch master updated: [FLINK-30288] Use PredicateVisitor to convert predicate for orc
This is an automated email from the ASF dual-hosted git repository. lzljs3620320 pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink-table-store.git The following commit(s) were added to refs/heads/master by this push: new f32ce6e9 [FLINK-30288] Use PredicateVisitor to convert predicate for orc f32ce6e9 is described below commit f32ce6e99f9ced265abbd57ce7af4c2e184c20cf Author: shammon AuthorDate: Wed Dec 7 11:32:11 2022 +0800 [FLINK-30288] Use PredicateVisitor to convert predicate for orc This closes #419 --- .../table/store/format/orc/OrcFileFormat.java | 7 +- .../table/store/format/orc/OrcFilterConverter.java | 246 - .../format/orc/OrcPredicateFunctionVisitor.java| 215 ++ .../store/format/orc/OrcFilterConverterTest.java | 4 +- 4 files changed, 220 insertions(+), 252 deletions(-) diff --git a/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/OrcFileFormat.java b/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/OrcFileFormat.java index 663260ab..2c68889f 100644 --- a/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/OrcFileFormat.java +++ b/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/OrcFileFormat.java @@ -87,10 +87,9 @@ public class OrcFileFormat extends FileFormat { if (filters != null) { for (Predicate pred : filters) { -OrcFilters.Predicate orcPred = OrcFilterConverter.toOrcPredicate(pred); -if (orcPred != null) { -orcPredicates.add(orcPred); -} +Optional orcPred = +pred.visit(OrcPredicateFunctionVisitor.VISITOR); +orcPred.ifPresent(orcPredicates::add); } } diff --git a/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/OrcFilterConverter.java b/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/OrcFilterConverter.java deleted file mode 100644 index acccd161.. --- a/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/OrcFilterConverter.java +++ /dev/null @@ -1,246 +0,0 @@ -/* - * 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.table.store.format.orc; - -import org.apache.flink.orc.OrcFilters; -import org.apache.flink.orc.OrcFilters.Predicate; -import org.apache.flink.table.data.DecimalData; -import org.apache.flink.table.data.TimestampData; -import org.apache.flink.table.store.file.predicate.CompoundPredicate; -import org.apache.flink.table.store.file.predicate.Equal; -import org.apache.flink.table.store.file.predicate.GreaterOrEqual; -import org.apache.flink.table.store.file.predicate.GreaterThan; -import org.apache.flink.table.store.file.predicate.IsNotNull; -import org.apache.flink.table.store.file.predicate.IsNull; -import org.apache.flink.table.store.file.predicate.LeafFunction; -import org.apache.flink.table.store.file.predicate.LeafPredicate; -import org.apache.flink.table.store.file.predicate.LessOrEqual; -import org.apache.flink.table.store.file.predicate.LessThan; -import org.apache.flink.table.store.file.predicate.NotEqual; -import org.apache.flink.table.store.file.predicate.Or; -import org.apache.flink.table.types.logical.LogicalType; -import org.apache.flink.util.function.TriFunction; - -import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableMap; - -import org.apache.hadoop.hive.ql.io.sarg.PredicateLeaf; - -import javax.annotation.Nullable; - -import java.io.Serializable; -import java.sql.Date; -import java.time.LocalDate; -import java.util.function.Function; - -/** Utility class that provides helper methods to work with Orc Filter PushDown. */ -public class OrcFilterConverter { - -private static final ImmutableMap< -Class, Function> -FILTERS = -new ImmutableMap.Builder< -Class, -Function>() -.put(IsNull.class, OrcFilterConverter::
[flink] branch master updated: [FLINK-30189][runtime] HsSubpartitionFileReader may load data that has been consumed from memory
This is an automated email from the ASF dual-hosted git repository. xtsong 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 4c67f8fca52 [FLINK-30189][runtime] HsSubpartitionFileReader may load data that has been consumed from memory 4c67f8fca52 is described below commit 4c67f8fca529a72389d69990307bbf78fcd3d99d Author: Weijie Guo AuthorDate: Tue Dec 6 19:24:36 2022 +0800 [FLINK-30189][runtime] HsSubpartitionFileReader may load data that has been consumed from memory This closes #21415 --- .../hybrid/HsSubpartitionFileReaderImpl.java | 25 ++- .../hybrid/HsSubpartitionFileReaderImplTest.java | 48 -- 2 files changed, 50 insertions(+), 23 deletions(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/HsSubpartitionFileReaderImpl.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/HsSubpartitionFileReaderImpl.java index e40e917cfa1..c190690d45c 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/HsSubpartitionFileReaderImpl.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/HsSubpartitionFileReaderImpl.java @@ -265,19 +265,24 @@ public class HsSubpartitionFileReaderImpl implements HsSubpartitionFileReader { private Optional checkAndGetFirstBufferIndexOrError(int expectedBufferIndex) throws Throwable { -if (loadedBuffers.isEmpty()) { -return Optional.empty(); -} - BufferIndexOrError peek = loadedBuffers.peek(); - -if (peek.getThrowable().isPresent()) { -throw peek.getThrowable().get(); -} else if (peek.getIndex() != expectedBufferIndex) { -return Optional.empty(); +while (peek != null) { +if (peek.getThrowable().isPresent()) { +throw peek.getThrowable().get(); +} else if (peek.getIndex() == expectedBufferIndex) { +break; +} else if (peek.getIndex() > expectedBufferIndex) { +return Optional.empty(); +} else if (peek.getIndex() < expectedBufferIndex) { +// Because the update of consumption progress may be delayed, there is a +// very small probability to load the buffer that has been consumed from memory. +// Skip these buffers directly to avoid repeated consumption. +loadedBuffers.poll(); +peek = loadedBuffers.peek(); +} } -return Optional.of(peek); +return Optional.ofNullable(peek); } private void moveFileOffsetToBuffer(int bufferIndex) throws IOException { diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/HsSubpartitionFileReaderImplTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/HsSubpartitionFileReaderImplTest.java index cccf9f5d93e..b8f5d206a78 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/HsSubpartitionFileReaderImplTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/HsSubpartitionFileReaderImplTest.java @@ -398,22 +398,20 @@ class HsSubpartitionFileReaderImplTest { // if no preload data in file reader, return Optional.empty. assertThat(subpartitionFileReader.consumeBuffer(0)).isNotPresent(); -// buffers in file: (0-0, 0-1) -writeDataToFile(0, 0, 0, 2); +// buffers in file: (0-0, 0-1, 0-2) +writeDataToFile(0, 0, 0, 3); -Queue memorySegments = createsMemorySegments(2); +Queue memorySegments = createsMemorySegments(3); subpartitionFileReader.prepareForScheduling(); // trigger reading, add buffer to queue. subpartitionFileReader.readBuffers(memorySegments, (ignore) -> {}); -// if nextBufferToConsume is not equal to peek elements index, return Optional.empty. -assertThat(subpartitionFileReader.consumeBuffer(10)).isNotPresent(); - +// if nextBufferToConsume is equal to peek elements index. assertThat(subpartitionFileReader.consumeBuffer(0)) .hasValueSatisfying( (bufferAndBacklog -> { assertThat(bufferAndBacklog.getNextDataType()) -.isEqualTo(DataType.EVENT_BUFFER); +.isEqualTo(DataType.DATA_BUFFER); assertThat(bufferAndBacklog.getSequenceNumber()).isEqualTo(0); // first buffer's data is 0. assertThat( @@ -424,6 +422,26 @@ class HsSubpartitionFileReaderImplTest {
[flink] branch release-1.16 updated: [FLINK-30189][runtime] HsSubpartitionFileReader may load data that has been consumed from memory
This is an automated email from the ASF dual-hosted git repository. xtsong pushed a commit to branch release-1.16 in repository https://gitbox.apache.org/repos/asf/flink.git The following commit(s) were added to refs/heads/release-1.16 by this push: new a9e65bc2377 [FLINK-30189][runtime] HsSubpartitionFileReader may load data that has been consumed from memory a9e65bc2377 is described below commit a9e65bc2377ee7a4b3599b58a58ff0301b79c5d8 Author: Weijie Guo AuthorDate: Tue Dec 6 19:24:36 2022 +0800 [FLINK-30189][runtime] HsSubpartitionFileReader may load data that has been consumed from memory This closes #21415 --- .../hybrid/HsSubpartitionFileReaderImpl.java | 25 ++- .../hybrid/HsSubpartitionFileReaderImplTest.java | 48 -- 2 files changed, 50 insertions(+), 23 deletions(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/HsSubpartitionFileReaderImpl.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/HsSubpartitionFileReaderImpl.java index e6dc7122c5e..88d730b14ea 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/HsSubpartitionFileReaderImpl.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/HsSubpartitionFileReaderImpl.java @@ -261,19 +261,24 @@ public class HsSubpartitionFileReaderImpl implements HsSubpartitionFileReader { private Optional checkAndGetFirstBufferIndexOrError(int expectedBufferIndex) throws Throwable { -if (loadedBuffers.isEmpty()) { -return Optional.empty(); -} - BufferIndexOrError peek = loadedBuffers.peek(); - -if (peek.getThrowable().isPresent()) { -throw peek.getThrowable().get(); -} else if (peek.getIndex() != expectedBufferIndex) { -return Optional.empty(); +while (peek != null) { +if (peek.getThrowable().isPresent()) { +throw peek.getThrowable().get(); +} else if (peek.getIndex() == expectedBufferIndex) { +break; +} else if (peek.getIndex() > expectedBufferIndex) { +return Optional.empty(); +} else if (peek.getIndex() < expectedBufferIndex) { +// Because the update of consumption progress may be delayed, there is a +// very small probability to load the buffer that has been consumed from memory. +// Skip these buffers directly to avoid repeated consumption. +loadedBuffers.poll(); +peek = loadedBuffers.peek(); +} } -return Optional.of(peek); +return Optional.ofNullable(peek); } private void moveFileOffsetToBuffer(int bufferIndex) throws IOException { diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/HsSubpartitionFileReaderImplTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/HsSubpartitionFileReaderImplTest.java index 03a560589d6..411a3131975 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/HsSubpartitionFileReaderImplTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/HsSubpartitionFileReaderImplTest.java @@ -398,22 +398,20 @@ class HsSubpartitionFileReaderImplTest { // if no preload data in file reader, return Optional.empty. assertThat(subpartitionFileReader.consumeBuffer(0)).isNotPresent(); -// buffers in file: (0-0, 0-1) -writeDataToFile(0, 0, 0, 2); +// buffers in file: (0-0, 0-1, 0-2) +writeDataToFile(0, 0, 0, 3); -Queue memorySegments = createsMemorySegments(2); +Queue memorySegments = createsMemorySegments(3); subpartitionFileReader.prepareForScheduling(); // trigger reading, add buffer to queue. subpartitionFileReader.readBuffers(memorySegments, (ignore) -> {}); -// if nextBufferToConsume is not equal to peek elements index, return Optional.empty. -assertThat(subpartitionFileReader.consumeBuffer(10)).isNotPresent(); - +// if nextBufferToConsume is equal to peek elements index. assertThat(subpartitionFileReader.consumeBuffer(0)) .hasValueSatisfying( (bufferAndBacklog -> { assertThat(bufferAndBacklog.getNextDataType()) -.isEqualTo(DataType.EVENT_BUFFER); +.isEqualTo(DataType.DATA_BUFFER); assertThat(bufferAndBacklog.getSequenceNumber()).isEqualTo(0); // first buffer's data is 0. assertThat( @@ -424,6 +422,26 @@ class HsSubpartitionFileReaderImplTest {