This is an automated email from the ASF dual-hosted git repository.
gyfora 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 56ddd0578 [FLINK-40467] Randomise application result store paths
56ddd0578 is described below
commit 56ddd057828193404b424d5495ebc1fcc92f31d6
Author: Dale Lane <[email protected]>
AuthorDate: Mon Sep 14 20:27:20 2026 +0100
[FLINK-40467] Randomise application result store paths
---
docs/content.zh/docs/deployment/leader-election.md | 2 +
docs/content/docs/deployment/leader-election.md | 2 +
.../deployment/ApplicationReconciler.java | 47 +++++++++++++
.../deployment/ApplicationReconcilerTest.java | 76 ++++++++++++++++++++++
4 files changed, 127 insertions(+)
diff --git a/docs/content.zh/docs/deployment/leader-election.md
b/docs/content.zh/docs/deployment/leader-election.md
index 3e66501a7..2e02824db 100644
--- a/docs/content.zh/docs/deployment/leader-election.md
+++ b/docs/content.zh/docs/deployment/leader-election.md
@@ -97,6 +97,8 @@ drwxr-xr-x 2 9999 9999 60 May 12 09:46
a6031ec7-ab3e-4b30-ba77-6498e58e6b7f
drwxr-xr-x 2 9999 9999 60 May 11 15:11 b6fb2a9c-d1cd-4e65-a9a1-e825c4b47543
```
+From Flink 2.3 the operator applies the same two settings to the
ApplicationResultStore
([FLINK-40467](https://issues.apache.org/jira/browse/FLINK-40467)):
`application-result-store.delete-on-commit` is turned off unless set
explicitly, and `application-result-store.storage-path` is pointed at a unique
directory on every cluster launch. This means a terminal application result
left by an earlier deployment can no longer stop the replacement cluster from
submitting its new job, and a JobM [...]
+
### Limitations
The following are known limitations of the operator's Flink Job HA integration:
diff --git a/docs/content/docs/deployment/leader-election.md
b/docs/content/docs/deployment/leader-election.md
index 3e66501a7..2e02824db 100644
--- a/docs/content/docs/deployment/leader-election.md
+++ b/docs/content/docs/deployment/leader-election.md
@@ -97,6 +97,8 @@ drwxr-xr-x 2 9999 9999 60 May 12 09:46
a6031ec7-ab3e-4b30-ba77-6498e58e6b7f
drwxr-xr-x 2 9999 9999 60 May 11 15:11 b6fb2a9c-d1cd-4e65-a9a1-e825c4b47543
```
+From Flink 2.3 the operator applies the same two settings to the
ApplicationResultStore
([FLINK-40467](https://issues.apache.org/jira/browse/FLINK-40467)):
`application-result-store.delete-on-commit` is turned off unless set
explicitly, and `application-result-store.storage-path` is pointed at a unique
directory on every cluster launch. This means a terminal application result
left by an earlier deployment can no longer stop the replacement cluster from
submitting its new job, and a JobM [...]
+
### Limitations
The following are known limitations of the operator's Flink Job HA integration:
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconciler.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconciler.java
index 99ff5f789..38e8df1e6 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconciler.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconciler.java
@@ -20,6 +20,8 @@ package
org.apache.flink.kubernetes.operator.reconciler.deployment;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobStatus;
import org.apache.flink.autoscaler.JobAutoScaler;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.HighAvailabilityOptions;
import org.apache.flink.configuration.PipelineOptionsInternal;
@@ -72,6 +74,24 @@ public class ApplicationReconciler
static final String MSG_RECOVERY = "Recovering lost deployment";
static final String MSG_RESTART_UNHEALTHY = "Restarting unhealthy job";
+ /**
+ * Mirrors {@code ApplicationResultStoreOptions#STORAGE_PATH}, introduced
in Flink 2.3. The
+ * option is unknown and ignored in earlier versions.
+ */
+ static final ConfigOption<String> APPLICATION_RESULT_STORE_STORAGE_PATH =
+ ConfigOptions.key("application-result-store.storage-path")
+ .stringType()
+ .noDefaultValue();
+
+ /**
+ * Mirrors {@code ApplicationResultStoreOptions#DELETE_ON_COMMIT},
introduced in Flink 2.3. The
+ * option is unknown and ignored in earlier versions.
+ */
+ static final ConfigOption<Boolean>
APPLICATION_RESULT_STORE_DELETE_ON_COMMIT =
+ ConfigOptions.key("application-result-store.delete-on-commit")
+ .booleanType()
+ .defaultValue(true);
+
public ApplicationReconciler(
EventRecorder eventRecorder,
StatusRecorder<FlinkDeployment, FlinkDeploymentStatus>
statusRecorder,
@@ -173,6 +193,7 @@ public class ApplicationReconciler
setOwnerReference(relatedResource, deployConfig);
setRandomJobResultStorePath(deployConfig);
+ setRandomApplicationResultStorePath(deployConfig);
if (status.getJobManagerDeploymentStatus() !=
JobManagerDeploymentStatus.MISSING) {
Preconditions.checkArgument(ReconciliationUtils.isJobInTerminalState(status));
@@ -273,6 +294,32 @@ public class ApplicationReconciler
}
}
+ private static void setRandomApplicationResultStorePath(Configuration
effectiveConfig) {
+ if (effectiveConfig.contains(HighAvailabilityOptions.HA_STORAGE_PATH))
{
+
+ // With shutdown-on-application-finish disabled, a JobManager that
restarts
+ // after an application has terminated would find no result and
run the
+ // application's main() again. Retaining the entry avoids that.
+ if
(!effectiveConfig.contains(APPLICATION_RESULT_STORE_DELETE_ON_COMMIT)) {
+ effectiveConfig.set(APPLICATION_RESULT_STORE_DELETE_ON_COMMIT,
false);
+ }
+
+ // The application result store shares the HA storage path, which
outlives
+ // the HA metadata deleted on upgrade. Dirty entries are recovered
unkeyed,
+ // so a terminal entry left behind by a previous deployment makes
the
+ // replacement cluster skip submitting the new job.
+ // Giving each deployment a unique path makes sure that nothing
stale is
+ // recovered, including entries retained above.
+ effectiveConfig.set(
+ APPLICATION_RESULT_STORE_STORAGE_PATH,
+
effectiveConfig.getString(HighAvailabilityOptions.HA_STORAGE_PATH)
+ + "/application-result-store/"
+ +
effectiveConfig.getString(KubernetesConfigOptions.CLUSTER_ID)
+ + "/"
+ + UUID.randomUUID());
+ }
+ }
+
@Override
public boolean reconcileOtherChanges(FlinkResourceContext<FlinkDeployment>
ctx)
throws Exception {
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java
index d1188c7ff..5b6b36422 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java
@@ -131,6 +131,8 @@ import static
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConf
import static
org.apache.flink.kubernetes.operator.reconciler.SnapshotType.CHECKPOINT;
import static
org.apache.flink.kubernetes.operator.reconciler.SnapshotType.SAVEPOINT;
import static
org.apache.flink.kubernetes.operator.reconciler.deployment.AbstractFlinkResourceReconciler.MSG_SUBMIT;
+import static
org.apache.flink.kubernetes.operator.reconciler.deployment.ApplicationReconciler.APPLICATION_RESULT_STORE_DELETE_ON_COMMIT;
+import static
org.apache.flink.kubernetes.operator.reconciler.deployment.ApplicationReconciler.APPLICATION_RESULT_STORE_STORAGE_PATH;
import static
org.apache.flink.kubernetes.operator.reconciler.deployment.ApplicationReconciler.MSG_RECOVERY;
import static
org.apache.flink.kubernetes.operator.utils.SnapshotUtils.getLastSnapshotStatus;
import static org.assertj.core.api.Assertions.assertThat;
@@ -802,6 +804,80 @@ public class ApplicationReconcilerTest extends
OperatorTestBase {
assertNotEquals(path1, path2);
}
+ /**
+ * The application result store lives under the HA storage path and
survives the HA metadata
+ * deletion performed on upgrade, so a terminal entry left by the previous
deployment would
+ * otherwise be recovered by the replacement cluster, which then skips
submitting the new job.
+ */
+ @Test
+ public void testRandomApplicationResultStorePath() throws Exception {
+ FlinkDeployment flinkApp = TestUtils.buildApplicationCluster();
+ final String haStoragePath = "file:///flink-data/ha";
+ flinkApp.getSpec()
+ .getFlinkConfiguration()
+ .put(HighAvailabilityOptions.HA_STORAGE_PATH.key(),
haStoragePath);
+
+ ObjectMeta deployMeta = flinkApp.getMetadata();
+ FlinkDeploymentStatus status = flinkApp.getStatus();
+ FlinkDeploymentSpec spec = flinkApp.getSpec();
+ Configuration deployConfig = configManager.getDeployConfig(deployMeta,
spec);
+
+
status.getJobStatus().setState(org.apache.flink.api.common.JobStatus.FINISHED);
+ status.setJobManagerDeploymentStatus(JobManagerDeploymentStatus.READY);
+ reconciler
+ .getReconciler()
+ .deploy(getResourceContext(flinkApp), spec, deployConfig,
Optional.empty(), false);
+ String path1 = deployConfig.get(APPLICATION_RESULT_STORE_STORAGE_PATH);
+ assertThat(path1).startsWith(haStoragePath);
+
+
status.getJobStatus().setState(org.apache.flink.api.common.JobStatus.FINISHED);
+ status.setJobManagerDeploymentStatus(JobManagerDeploymentStatus.READY);
+ reconciler
+ .getReconciler()
+ .deploy(getResourceContext(flinkApp), spec, deployConfig,
Optional.empty(), false);
+ String path2 = deployConfig.get(APPLICATION_RESULT_STORE_STORAGE_PATH);
+ assertThat(path2).startsWith(haStoragePath).isNotEqualTo(path1);
+ }
+
+ /**
+ * A terminal application result must outlive its cleanup, otherwise a
JobManager restart before
+ * the cluster is shut down finds no record of the application and runs it
again.
+ */
+ @Test
+ public void testApplicationResultStoreRetainsResults() throws Exception {
+ var deployConfig = deployWithHaStorage(Map.of());
+
assertThat(deployConfig.get(APPLICATION_RESULT_STORE_DELETE_ON_COMMIT)).isFalse();
+ }
+
+ @Test
+ public void testApplicationResultStoreDeleteOnCommitNotOverridden() throws
Exception {
+ var deployConfig =
+ deployWithHaStorage(
+
Map.of(APPLICATION_RESULT_STORE_DELETE_ON_COMMIT.key(), "true"));
+
assertThat(deployConfig.get(APPLICATION_RESULT_STORE_DELETE_ON_COMMIT)).isTrue();
+ }
+
+ /** Deploys an application cluster with HA storage configured, plus any
extra config. */
+ private Configuration deployWithHaStorage(Map<String, String> extraConfig)
throws Exception {
+ final FlinkDeployment flinkApp = TestUtils.buildApplicationCluster();
+ flinkApp.getSpec()
+ .getFlinkConfiguration()
+ .put(HighAvailabilityOptions.HA_STORAGE_PATH.key(),
"file:///flink-data/ha");
+ flinkApp.getSpec().getFlinkConfiguration().putAllFrom(extraConfig);
+
+ final Configuration deployConfig =
+ configManager.getDeployConfig(flinkApp.getMetadata(),
flinkApp.getSpec());
+ reconciler
+ .getReconciler()
+ .deploy(
+ getResourceContext(flinkApp),
+ flinkApp.getSpec(),
+ deployConfig,
+ Optional.empty(),
+ false);
+ return deployConfig;
+ }
+
@Test
public void testAlwaysSavepointOnFlinkVersionChange() throws Exception {
var deployment = TestUtils.buildApplicationCluster(FlinkVersion.v1_18);
