(flink) branch master updated: [FLINK-34247][doc] Fix the wrong zh link for flink-configuration-file and repeated flink configuration file
This is an automated email from the ASF dual-hosted git repository.
fanrui 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 d44e940e9d2 [FLINK-34247][doc] Fix the wrong zh link for
flink-configuration-file and repeated flink configuration file
d44e940e9d2 is described below
commit d44e940e9d28518115cf6e8d5ba4f963faf30f52
Author: Rui Fan <1996fan...@gmail.com>
AuthorDate: Mon Feb 5 13:21:34 2024 +0800
[FLINK-34247][doc] Fix the wrong zh link for flink-configuration-file and
repeated flink configuration file
---
docs/content.zh/docs/connectors/table/filesystem.md| 4 ++--
docs/content.zh/docs/deployment/advanced/historyserver.md | 2 +-
docs/content.zh/docs/deployment/cli.md | 4 ++--
docs/content.zh/docs/deployment/filesystems/azure.md | 6 +++---
docs/content.zh/docs/deployment/filesystems/gcs.md | 6 +++---
docs/content.zh/docs/deployment/filesystems/oss.md | 6 +++---
docs/content.zh/docs/deployment/filesystems/s3.md | 12 ++--
docs/content.zh/docs/deployment/ha/kubernetes_ha.md| 2 +-
docs/content.zh/docs/deployment/ha/zookeeper_ha.md | 4 ++--
docs/content.zh/docs/deployment/memory/mem_migration.md| 4 ++--
docs/content.zh/docs/deployment/metric_reporters.md| 2 +-
.../deployment/resource-providers/native_kubernetes.md | 4 ++--
.../deployment/resource-providers/standalone/docker.md | 8
.../deployment/resource-providers/standalone/kubernetes.md | 2 +-
.../deployment/resource-providers/standalone/overview.md | 8
docs/content.zh/docs/deployment/resource-providers/yarn.md | 2 +-
docs/content.zh/docs/deployment/security/security-ssl.md | 2 +-
docs/content.zh/docs/deployment/trace_reporters.md | 2 +-
docs/content.zh/docs/dev/datastream/execution/parallel.md | 2 +-
.../docs/dev/datastream/fault-tolerance/checkpointing.md | 2 +-
.../docs/dev/datastream/fault-tolerance/state_backends.md | 2 +-
docs/content.zh/docs/dev/table/catalogs.md | 4 ++--
.../docs/dev/table/hive-compatibility/hiveserver2.md | 4 ++--
docs/content.zh/docs/dev/table/sql-gateway/overview.md | 4 ++--
docs/content.zh/docs/dev/table/sqlClient.md| 2 +-
docs/content.zh/docs/ops/debugging/flame_graphs.md | 2 +-
docs/content.zh/docs/ops/metrics.md| 4 ++--
docs/content.zh/docs/ops/rest_api.md | 2 +-
docs/content.zh/docs/ops/state/state_backends.md | 14 +++---
docs/content.zh/docs/ops/state/task_failure_recovery.md| 8
.../docs/try-flink/flink-operations-playground.md | 4 ++--
.../deployment/resource-providers/standalone/docker.md | 2 +-
docs/content/docs/deployment/resource-providers/yarn.md| 2 +-
.../docs/dev/datastream/fault-tolerance/state_backends.md | 2 +-
docs/content/docs/dev/table/sqlClient.md | 2 +-
docs/content/docs/ops/state/task_failure_recovery.md | 4 ++--
36 files changed, 73 insertions(+), 73 deletions(-)
diff --git a/docs/content.zh/docs/connectors/table/filesystem.md
b/docs/content.zh/docs/connectors/table/filesystem.md
index 17d05045a20..584f5009081 100644
--- a/docs/content.zh/docs/connectors/table/filesystem.md
+++ b/docs/content.zh/docs/connectors/table/filesystem.md
@@ -241,8 +241,8 @@ CREATE TABLE MyUserTableWithFilepath (
**注意：** 对于 bulk formats 数据 (parquet、orc、avro)，滚动策略与 checkpoint 间隔（pending
状态的文件会在下个 checkpoint 完成）控制了 part 文件的大小和个数。
-**注意：** 对于 row formats 数据 (csv、json)，如果想使得分区文件更快在文件系统中可见，可以设置
`sink.rolling-policy.file-size` 或 `sink.rolling-policy.rollover-interval` 属性以及在
[Flink 配置文件]({{< ref "docs/deployment/config#flink-configuration-file" >}}) 中的
`execution.checkpointing.interval` 属性。
-对于其他 formats (avro、orc)，可以只设置 [Flink 配置文件]({{< ref
"docs/deployment/config#flink-configuration-file" >}}) 中的
`execution.checkpointing.interval` 属性。
+**注意：** 对于 row formats 数据 (csv、json)，如果想使得分区文件更快在文件系统中可见，可以设置
`sink.rolling-policy.file-size` 或 `sink.rolling-policy.rollover-interval` 属性以及在
[Flink 配置文件]({{< ref "docs/deployment/config#flink-配置文件" >}}) 中的
`execution.checkpointing.interval` 属性。
+对于其他 formats (avro、orc)，可以只设置 [Flink 配置文件]({{< ref
"docs/deployment/config#flink-配置文件" >}}) 中的 `execution.checkpointing.interval`
属性。
diff --git a/docs/content.zh/docs/deployment/advanced/historyserver.md
b/docs/content.zh/docs/deployment/advanced/historyserver.md
index 16950b4a06b..bef18431585 100644
--- a/docs/content.zh/docs/deployment/advanced/historyserver.md
+++ b/docs/content.zh/docs/deployment/advanced/historyserver.md
@@ -58,7 +58,7 @@ bin/historyserver.sh (start|start-foreground|stop)
**JobManager**
-已完成作业的存档在 JobManager 上进行，将已存档的作业信息上传到文件系统目录中。你可以在 [Flink 配置文件]({{< ref
"docs/deployment/config#flink-configuration-fil
(flink) branch master updated: [FLINK-34343][rpc] Use actor path when rejecting early messages
This is an automated email from the ASF dual-hosted git repository.
chesnay 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 839f298c383 [FLINK-34343][rpc] Use actor path when rejecting early
messages
839f298c383 is described below
commit 839f298c3838f2f4981e271554b82fae770747d8
Author: Chesnay Schepler
AuthorDate: Sat Feb 3 11:20:08 2024 +0100
[FLINK-34343][rpc] Use actor path when rejecting early messages
---
.../src/main/java/org/apache/flink/runtime/rpc/pekko/PekkoRpcActor.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git
a/flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/PekkoRpcActor.java
b/flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/PekkoRpcActor.java
index f3724fba0e5..dc4e342f35a 100644
---
a/flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/PekkoRpcActor.java
+++
b/flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/PekkoRpcActor.java
@@ -179,7 +179,7 @@ class PekkoRpcActor
extends AbstractActor {
new EndpointNotStartedException(
String.format(
"Discard message %s, because the rpc
endpoint %s has not been started yet.",
-message, rpcEndpoint.getAddress(;
+message, getSelf().path(;
}
}
(flink) branch release-1.18 updated: [FLINK-34343][rpc] Use actor path when rejecting early messages
This is an automated email from the ASF dual-hosted git repository.
chesnay pushed a commit to branch release-1.18
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.18 by this push:
new ee9945cd785 [FLINK-34343][rpc] Use actor path when rejecting early
messages
ee9945cd785 is described below
commit ee9945cd785577d0f68092823b71abbb53d127f8
Author: Chesnay Schepler
AuthorDate: Sat Feb 3 11:20:08 2024 +0100
[FLINK-34343][rpc] Use actor path when rejecting early messages
---
.../src/main/java/org/apache/flink/runtime/rpc/pekko/PekkoRpcActor.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git
a/flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/PekkoRpcActor.java
b/flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/PekkoRpcActor.java
index f3724fba0e5..dc4e342f35a 100644
---
a/flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/PekkoRpcActor.java
+++
b/flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/pekko/PekkoRpcActor.java
@@ -179,7 +179,7 @@ class PekkoRpcActor
extends AbstractActor {
new EndpointNotStartedException(
String.format(
"Discard message %s, because the rpc
endpoint %s has not been started yet.",
-message, rpcEndpoint.getAddress(;
+message, getSelf().path(;
}
}
(flink) branch release-1.17 updated: [FLINK-34343][rpc] Use actor path when rejecting early messages
This is an automated email from the ASF dual-hosted git repository.
chesnay pushed a commit to branch release-1.17
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.17 by this push:
new 3f22b6363e6 [FLINK-34343][rpc] Use actor path when rejecting early
messages
3f22b6363e6 is described below
commit 3f22b6363e6cad4352821f42907ec8a2a181e675
Author: Chesnay Schepler
AuthorDate: Sat Feb 3 11:20:08 2024 +0100
[FLINK-34343][rpc] Use actor path when rejecting early messages
---
.../src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActor.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git
a/flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActor.java
b/flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActor.java
index 8029956bbd9..59d30e16564 100644
---
a/flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActor.java
+++
b/flink-rpc/flink-rpc-akka/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActor.java
@@ -179,7 +179,7 @@ class AkkaRpcActor
extends AbstractActor {
new EndpointNotStartedException(
String.format(
"Discard message %s, because the rpc
endpoint %s has not been started yet.",
-message, rpcEndpoint.getAddress(;
+message, getSelf().path(;
}
}
(flink) 02/02: [FLINK-33441] Remove unused json plan
This is an automated email from the ASF dual-hosted git repository.
dwysakowicz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 3d0cddef4188995ea3d0731d6a88e7a6af56f44e
Author: bvarghese1
AuthorDate: Tue Jan 16 18:20:34 2024 -0800
[FLINK-33441] Remove unused json plan
---
.../UnionJsonPlanTest_jsonplan/testUnion.out | 157 -
1 file changed, 157 deletions(-)
diff --git
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/UnionJsonPlanTest_jsonplan/testUnion.out
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/UnionJsonPlanTest_jsonplan/testUnion.out
deleted file mode 100644
index 17fba7f6a3d..000
---
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/UnionJsonPlanTest_jsonplan/testUnion.out
+++ /dev/null
@@ -1,157 +0,0 @@
-{
- "flinkVersion" : "",
- "nodes" : [ {
-"id" : 1,
-"type" : "stream-exec-table-source-scan_1",
-"scanTableSource" : {
- "table" : {
-"identifier" : "`default_catalog`.`default_database`.`MyTable`",
-"resolvedTable" : {
- "schema" : {
-"columns" : [ {
- "name" : "a",
- "dataType" : "BIGINT"
-}, {
- "name" : "b",
- "dataType" : "INT NOT NULL"
-}, {
- "name" : "c",
- "dataType" : "VARCHAR(2147483647)"
-}, {
- "name" : "d",
- "dataType" : "TIMESTAMP(3)"
-} ],
-"watermarkSpecs" : [ ]
- },
- "partitionKeys" : [ ],
- "options" : {
-"bounded" : "false",
-"connector" : "values"
- }
-}
- },
- "abilities" : [ {
-"type" : "ProjectPushDown",
-"projectedFields" : [ [ 0 ], [ 1 ] ],
-"producedType" : "ROW<`a` BIGINT, `b` INT NOT NULL> NOT NULL"
- }, {
-"type" : "ReadingMetadata",
-"metadataKeys" : [ ],
-"producedType" : "ROW<`a` BIGINT, `b` INT NOT NULL> NOT NULL"
- } ]
-},
-"outputType" : "ROW<`a` BIGINT, `b` INT NOT NULL>",
-"description" : "TableSourceScan(table=[[default_catalog,
default_database, MyTable, project=[a, b], metadata=[]]], fields=[a, b])",
-"inputProperties" : [ ]
- }, {
-"id" : 2,
-"type" : "stream-exec-table-source-scan_1",
-"scanTableSource" : {
- "table" : {
-"identifier" : "`default_catalog`.`default_database`.`MyTable2`",
-"resolvedTable" : {
- "schema" : {
-"columns" : [ {
- "name" : "d",
- "dataType" : "BIGINT"
-}, {
- "name" : "e",
- "dataType" : "INT NOT NULL"
-} ],
-"watermarkSpecs" : [ ]
- },
- "partitionKeys" : [ ],
- "options" : {
-"bounded" : "false",
-"connector" : "values"
- }
-}
- }
-},
-"outputType" : "ROW<`d` BIGINT, `e` INT NOT NULL>",
-"description" : "TableSourceScan(table=[[default_catalog,
default_database, MyTable2]], fields=[d, e])",
-"inputProperties" : [ ]
- }, {
-"id" : 3,
-"type" : "stream-exec-union_1",
-"inputProperties" : [ {
- "requiredDistribution" : {
-"type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
-}, {
- "requiredDistribution" : {
-"type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
-} ],
-"outputType" : "ROW<`a` BIGINT, `b` INT NOT NULL>",
-"description" : "Union(all=[true], union=[a, b])"
- }, {
-"id" : 4,
-"type" : "stream-exec-sink_1",
-"configuration" : {
- "table.exec.sink.keyed-shuffle" : "AUTO",
- "table.exec.sink.not-null-enforcer" : "ERROR",
- "table.exec.sink.rowtime-inserter" : "ENABLED",
- "table.exec.sink.type-length-enforcer" : "IGNORE",
- "table.exec.sink.upsert-materialize" : "AUTO"
-},
-"dynamicTableSink" : {
- "table" : {
-"identifier" : "`default_catalog`.`default_database`.`MySink`",
-"resolvedTable" : {
- "schema" : {
-"columns" : [ {
- "name" : "a",
- "dataType" : "BIGINT"
-}, {
- "name" : "b",
- "dataType" : "INT"
-} ],
-"watermarkSpecs" : [ ]
- },
- "partitionKeys" : [ ],
- "options" : {
-"connector" : "values",
-"table-sink-class" : "DEFAULT"
- }
-}
- }
-},
-"inputChangelogMode" : [ "INSERT" ],
-"inputProperties" : [ {
- "requiredDistribution" : {
-"type" : "UNKNOWN"
- },
- "damBehavior" : "PIPELINED",
- "priority" : 0
-} ],
-"outputType" : "ROW
(flink) 01/02: [FLINK-33441] Move Union restore tests
This is an automated email from the ASF dual-hosted git repository.
dwysakowicz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit f5d889fb70688bc3e29fc979dc233d2c9674fa96
Author: bvarghese1
AuthorDate: Tue Jan 16 18:16:42 2024 -0800
[FLINK-33441] Move Union restore tests
- Moving from nodes.exec.testutil to nodes.exec.stream
- Related commit: f362dcc
---
.../plan/nodes/exec/{testutils => stream}/UnionRestoreTest.java | 4 ++--
.../plan/nodes/exec/{testutils => stream}/UnionTestPrograms.java | 3 +--
2 files changed, 3 insertions(+), 4 deletions(-)
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/testutils/UnionRestoreTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/UnionRestoreTest.java
similarity index 90%
rename from
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/testutils/UnionRestoreTest.java
rename to
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/UnionRestoreTest.java
index ca27c175fc6..b271e0221ad 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/testutils/UnionRestoreTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/UnionRestoreTest.java
@@ -16,9 +16,9 @@
* limitations under the License.
*/
-package org.apache.flink.table.planner.plan.nodes.exec.testutils;
+package org.apache.flink.table.planner.plan.nodes.exec.stream;
-import org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecUnion;
+import
org.apache.flink.table.planner.plan.nodes.exec.testutils.RestoreTestBase;
import org.apache.flink.table.test.program.TableTestProgram;
import java.util.Arrays;
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/testutils/UnionTestPrograms.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/UnionTestPrograms.java
similarity index 98%
rename from
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/testutils/UnionTestPrograms.java
rename to
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/UnionTestPrograms.java
index 562199588b5..e40ee481fce 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/testutils/UnionTestPrograms.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/UnionTestPrograms.java
@@ -16,9 +16,8 @@
* limitations under the License.
*/
-package org.apache.flink.table.planner.plan.nodes.exec.testutils;
+package org.apache.flink.table.planner.plan.nodes.exec.stream;
-import org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecUnion;
import org.apache.flink.table.test.program.SinkTestStep;
import org.apache.flink.table.test.program.SourceTestStep;
import org.apache.flink.table.test.program.TableTestProgram;
(flink) branch master updated (839f298c383 -> 3d0cddef418)
This is an automated email from the ASF dual-hosted git repository.
dwysakowicz pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from 839f298c383 [FLINK-34343][rpc] Use actor path when rejecting early
messages
new f5d889fb706 [FLINK-33441] Move Union restore tests
new 3d0cddef418 [FLINK-33441] Remove unused json plan
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:
.../{testutils => stream}/UnionRestoreTest.java| 4 +-
.../{testutils => stream}/UnionTestPrograms.java | 3 +-
.../UnionJsonPlanTest_jsonplan/testUnion.out | 157 -
3 files changed, 3 insertions(+), 161 deletions(-)
rename
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/{testutils
=> stream}/UnionRestoreTest.java (90%)
rename
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/{testutils
=> stream}/UnionTestPrograms.java (98%)
delete mode 100644
flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/UnionJsonPlanTest_jsonplan/testUnion.out
(flink) branch master updated (3d0cddef418 -> 79cccd7103a)
This is an automated email from the ASF dual-hosted git repository. mapohl pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/flink.git from 3d0cddef418 [FLINK-33441] Remove unused json plan new 39e2eaf6805 [hotfix][core] Makes ExecutorUtils#gracefulShutdown return any outstanding tasks new 79cccd7103a [FLINK-34007][k8s] Adds graceful shutdown logic to KubernetesLeaderElector 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: .../java/org/apache/flink/util/ExecutorUtils.java | 14 ++ .../kubeclient/resources/KubernetesLeaderElector.java | 19 ++- 2 files changed, 24 insertions(+), 9 deletions(-)
(flink) 02/02: [FLINK-34007][k8s] Adds graceful shutdown logic to KubernetesLeaderElector
This is an automated email from the ASF dual-hosted git repository.
mapohl pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 79cccd7103a304bfa07104dcafd1f65a032c88ce
Author: Matthias Pohl
AuthorDate: Fri Feb 2 14:47:42 2024 +0100
[FLINK-34007][k8s] Adds graceful shutdown logic to KubernetesLeaderElector
We need to make sure that any ongoing leadership event is properly handled.
shutdownNow and the Precondition are too aggressive.
---
.../kubeclient/resources/KubernetesLeaderElector.java | 19 ++-
1 file changed, 14 insertions(+), 5 deletions(-)
diff --git
a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesLeaderElector.java
b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesLeaderElector.java
index 0bf65ddbac2..728f46abcb6 100644
---
a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesLeaderElector.java
+++
b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesLeaderElector.java
@@ -21,7 +21,7 @@ package org.apache.flink.kubernetes.kubeclient.resources;
import org.apache.flink.annotation.VisibleForTesting;
import
org.apache.flink.kubernetes.configuration.KubernetesLeaderElectionConfiguration;
import org.apache.flink.kubernetes.utils.KubernetesUtils;
-import org.apache.flink.util.Preconditions;
+import org.apache.flink.util.ExecutorUtils;
import org.apache.flink.util.concurrent.ExecutorThreadFactory;
import org.apache.flink.util.concurrent.FutureUtils;
@@ -41,6 +41,7 @@ import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
import static
org.apache.flink.kubernetes.utils.Constants.LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY;
@@ -160,10 +161,18 @@ public class KubernetesLeaderElector {
// code that handles the leader loss
cancelCurrentLeaderElectionSession();
-final List outstandingTasks =
executorService.shutdownNow();
-Preconditions.checkState(
-outstandingTasks.isEmpty(),
-"All tasks that handle the leadership revocation should
have been executed.");
+// the shutdown of the executor needs to happen gracefully for
scenarios where the
+// release is called in the executorService. Interrupting this
logic will result in the
+// leadership-lost event not being sent to the client.
+final List outStandingTasks =
+ExecutorUtils.gracefulShutdown(30, TimeUnit.SECONDS,
executorService);
+
+if (!outStandingTasks.isEmpty()) {
+LOG.warn(
+"{} events were not processed before stopping the {}
instance.",
+outStandingTasks.size(),
+KubernetesLeaderElector.class.getSimpleName());
+}
}
}
(flink) 01/02: [hotfix][core] Makes ExecutorUtils#gracefulShutdown return any outstanding tasks
This is an automated email from the ASF dual-hosted git repository.
mapohl pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 39e2eaf68055803956909ab995ebe73aefd308a5
Author: Matthias Pohl
AuthorDate: Fri Feb 2 14:34:47 2024 +0100
[hotfix][core] Makes ExecutorUtils#gracefulShutdown return any outstanding
tasks
---
.../src/main/java/org/apache/flink/util/ExecutorUtils.java | 14 ++
1 file changed, 10 insertions(+), 4 deletions(-)
diff --git a/flink-core/src/main/java/org/apache/flink/util/ExecutorUtils.java
b/flink-core/src/main/java/org/apache/flink/util/ExecutorUtils.java
index 939ffd6624c..99e2f3317e5 100644
--- a/flink-core/src/main/java/org/apache/flink/util/ExecutorUtils.java
+++ b/flink-core/src/main/java/org/apache/flink/util/ExecutorUtils.java
@@ -21,6 +21,8 @@ package org.apache.flink.util;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.ArrayList;
+import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
@@ -38,8 +40,9 @@ public class ExecutorUtils {
* @param timeout to wait for the termination of all ExecutorServices
* @param unit of the timeout
* @param executorServices to shut down
+ * @return Tasks that were not executed prior to a {@link
ExecutorService#shutdownNow()}.
*/
-public static void gracefulShutdown(
+public static List gracefulShutdown(
long timeout, TimeUnit unit, ExecutorService... executorServices) {
for (ExecutorService executorService : executorServices) {
executorService.shutdown();
@@ -50,22 +53,23 @@ public class ExecutorUtils {
long timeLeft = unit.toMillis(timeout);
boolean hasTimeLeft = timeLeft > 0L;
+final List outstandingTasks = new ArrayList<>();
for (ExecutorService executorService : executorServices) {
if (wasInterrupted || !hasTimeLeft) {
-executorService.shutdownNow();
+outstandingTasks.addAll(executorService.shutdownNow());
} else {
try {
if (!executorService.awaitTermination(timeLeft,
TimeUnit.MILLISECONDS)) {
LOG.warn(
"ExecutorService did not terminate in time.
Shutting it down now.");
-executorService.shutdownNow();
+outstandingTasks.addAll(executorService.shutdownNow());
}
} catch (InterruptedException e) {
LOG.warn(
"Interrupted while shutting down executor
services. Shutting all "
+ "remaining ExecutorServices down now.",
e);
-executorService.shutdownNow();
+outstandingTasks.addAll(executorService.shutdownNow());
wasInterrupted = true;
@@ -76,6 +80,8 @@ public class ExecutorUtils {
hasTimeLeft = timeLeft > 0L;
}
}
+
+return outstandingTasks;
}
/**
(flink) branch release-1.18 updated: [FLINK-34324][test] Makes all s3 related operations being declared and called in a single location
This is an automated email from the ASF dual-hosted git repository.
mapohl pushed a commit to branch release-1.18
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.18 by this push:
new a15515ebc0e [FLINK-34324][test] Makes all s3 related operations being
declared and called in a single location
a15515ebc0e is described below
commit a15515ebc0e4c59ea0642e745e942591c28b3a3c
Author: Matthias Pohl
AuthorDate: Wed Jan 31 15:02:24 2024 +0100
[FLINK-34324][test] Makes all s3 related operations being declared and
called in a single location
---
.../test-scripts/test_file_sink.sh | 111 +++--
1 file changed, 59 insertions(+), 52 deletions(-)
diff --git a/flink-end-to-end-tests/test-scripts/test_file_sink.sh
b/flink-end-to-end-tests/test-scripts/test_file_sink.sh
index 711f74b6672..5ed1fda2c68 100755
--- a/flink-end-to-end-tests/test-scripts/test_file_sink.sh
+++ b/flink-end-to-end-tests/test-scripts/test_file_sink.sh
@@ -20,53 +20,16 @@
OUT_TYPE="${1:-local}" # other type: s3
SINK_TO_TEST="${2:-"StreamingFileSink"}"
-S3_PREFIX=temp/test_file_sink-$(uuidgen)
-OUTPUT_PATH="$TEST_DATA_DIR/$S3_PREFIX"
-S3_OUTPUT_PATH="s3://$IT_CASE_S3_BUCKET/$S3_PREFIX"
source "$(dirname "$0")"/common.sh
-if [ "${OUT_TYPE}" == "s3" ]; then
- source "$(dirname "$0")"/common_s3.sh
-else
- echo "S3 environment is not loaded for non-s3 test runs (test run type:
$OUT_TYPE)."
-fi
-
-# randomly set up openSSL with dynamically/statically linked libraries
-OPENSSL_LINKAGE=$(if (( RANDOM % 2 )) ; then echo "dynamic"; else echo
"static"; fi)
-echo "Executing test with ${OPENSSL_LINKAGE} openSSL linkage (random selection
between 'dynamic' and 'static')"
-
-s3_setup hadoop
-set_conf_ssl "mutual" "OPENSSL" "${OPENSSL_LINKAGE}"
-set_config_key "metrics.fetcher.update-interval" "2000"
-# this test relies on global failovers
-set_config_key "jobmanager.execution.failover-strategy" "full"
-
+# OUTPUT_PATH is a local folder that can be used as a download folder for
remote data
+# the helper functions will access this folder
+RANDOM_PREFIX="temp/test_file_sink-$(uuidgen)"
+OUTPUT_PATH="$TEST_DATA_DIR/${RANDOM_PREFIX}"
mkdir -p $OUTPUT_PATH
-if [ "${OUT_TYPE}" == "local" ]; then
- echo "Use local output"
- JOB_OUTPUT_PATH=${OUTPUT_PATH}
-elif [ "${OUT_TYPE}" == "s3" ]; then
- echo "Use s3 output"
- JOB_OUTPUT_PATH=${S3_OUTPUT_PATH}
- set_config_key "state.checkpoints.dir"
"s3://$IT_CASE_S3_BUCKET/$S3_PREFIX-chk"
- mkdir -p "$OUTPUT_PATH-chk"
-else
- echo "Unknown output type: ${OUT_TYPE}"
- exit 1
-fi
-
-# make sure we delete the file at the end
-function out_cleanup {
- s3_delete_by_full_path_prefix "$S3_PREFIX"
- s3_delete_by_full_path_prefix "${S3_PREFIX}-chk"
- rollback_openssl_lib
-}
-if [ "${OUT_TYPE}" == "s3" ]; then
- on_exit out_cleanup
-fi
-
-TEST_PROGRAM_JAR="${END_TO_END_DIR}/flink-file-sink-test/target/FileSinkProgram.jar"
+# JOB_OUTPUT_PATH is the location where the job writes its data to
+JOB_OUTPUT_PATH="${OUTPUT_PATH}"
###
# Get all lines in part files and sort them numerically.
@@ -79,9 +42,6 @@
TEST_PROGRAM_JAR="${END_TO_END_DIR}/flink-file-sink-test/target/FileSinkProgram.
# sorted content of part files
###
function get_complete_result {
- if [ "${OUT_TYPE}" == "s3" ]; then
-s3_get_by_full_path_and_filename_prefix "$OUTPUT_PATH" "$S3_PREFIX"
"part-" true
- fi
find "${OUTPUT_PATH}" -type f \( -iname "part-*" \) -exec cat {} + | sort -g
}
@@ -89,20 +49,67 @@ function get_complete_result {
# Get total number of lines in part files.
#
# Globals:
-# S3_PREFIX
+# OUTPUT_PATH
# Arguments:
# None
# Returns:
# line number in part files
###
function get_total_number_of_valid_lines {
- if [ "${OUT_TYPE}" == "local" ]; then
-get_complete_result | wc -l | tr -d '[:space:]'
- elif [ "${OUT_TYPE}" == "s3" ]; then
-s3_get_number_of_lines_by_prefix "${S3_PREFIX}" "part-"
- fi
+ get_complete_result | wc -l | tr -d '[:space:]'
}
+if [ "${OUT_TYPE}" == "local" ]; then
+ echo "[INFO] Test run in local environment: No S3 environment is not loaded."
+elif [ "${OUT_TYPE}" == "s3" ]; then
+ # the s3 context requires additional
+ source "$(dirname "$0")"/common_s3.sh
+ s3_setup hadoop
+
+ # overwrites JOB_OUTPUT_PATH to point to S3
+ S3_DATA_PREFIX="${RANDOM_PREFIX}"
+ S3_CHECKPOINT_PREFIX="${RANDOM_PREFIX}-chk"
+ JOB_OUTPUT_PATH="s3://$IT_CASE_S3_BUCKET/${S3_DATA_PREFIX}"
+ set_config_key "state.checkpoints.dir"
"s3://$IT_CASE_S3_BUCKET/${S3_CHECKPOINT_PREFIX}"
+
+ # overwrites implementation for local runs
+ function get_complete_result {
+# copies the data from S3 to the local OUTPUT_PATH
+s3_get_by_full_path_and_filename_prefix "$OUTPUT_PATH"
"$FILE_SINK_TEST_TEMP_SUBFOLDER" "part-" true
+
+# and prints the sorted output
+find "${OUTP
