This is an automated email from the ASF dual-hosted git repository.
ferenc-csaky 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 249dc9c67 [FLINK-38795] Support Flink 2.0+ HA in standalone mode
249dc9c67 is described below
commit 249dc9c6717b3ad67a07aa3cc18549ffa5c9efb2
Author: Marc Demierre <[email protected]>
AuthorDate: Fri Jul 10 15:59:54 2026 +0200
[FLINK-38795] Support Flink 2.0+ HA in standalone mode
---
flink-kubernetes-standalone/pom.xml | 14 ++
.../CmdStandaloneJobManagerDecorator.java | 10 +-
.../CmdStandaloneJobManagerDecoratorTest.java | 142 ++++++++++++++++-----
3 files changed, 131 insertions(+), 35 deletions(-)
diff --git a/flink-kubernetes-standalone/pom.xml
b/flink-kubernetes-standalone/pom.xml
index 5d8da1de3..895187b87 100644
--- a/flink-kubernetes-standalone/pom.xml
+++ b/flink-kubernetes-standalone/pom.xml
@@ -47,12 +47,26 @@ under the License.
</dependency>
<!-- Test -->
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-container</artifactId>
+ <version>${flink.version}</version>
+ <scope>test</scope>
+ </dependency>
+
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter-api</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.assertj</groupId>
+ <artifactId>assertj-core</artifactId>
+ <version>${assertj.version}</version>
+ <scope>test</scope>
+ </dependency>
+
<dependency>
<groupId>org.hamcrest</groupId>
<artifactId>hamcrest-all</artifactId>
diff --git
a/flink-kubernetes-standalone/src/main/java/org/apache/flink/kubernetes/operator/kubeclient/decorators/CmdStandaloneJobManagerDecorator.java
b/flink-kubernetes-standalone/src/main/java/org/apache/flink/kubernetes/operator/kubeclient/decorators/CmdStandaloneJobManagerDecorator.java
index 2d063eaf7..ebff11c3b 100644
---
a/flink-kubernetes-standalone/src/main/java/org/apache/flink/kubernetes/operator/kubeclient/decorators/CmdStandaloneJobManagerDecorator.java
+++
b/flink-kubernetes-standalone/src/main/java/org/apache/flink/kubernetes/operator/kubeclient/decorators/CmdStandaloneJobManagerDecorator.java
@@ -17,6 +17,7 @@
package org.apache.flink.kubernetes.operator.kubeclient.decorators;
+import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.kubernetes.kubeclient.FlinkPod;
import
org.apache.flink.kubernetes.kubeclient.decorators.AbstractKubernetesStepDecorator;
import
org.apache.flink.kubernetes.operator.kubeclient.parameters.StandaloneKubernetesJobManagerParameters;
@@ -38,6 +39,9 @@ public class CmdStandaloneJobManagerDecorator extends
AbstractKubernetesStepDeco
public static final String APPLICATION_MODE_ARG = "standalone-job";
public static final String POD_IP_ARG =
String.format("$(%s)", Constants.ENV_FLINK_POD_IP_ADDRESS);
+ public static final String DYNAMIC_PROPERTY_FLAG = "-D";
+ public static final String JOBMANAGER_RPC_ADDRESS_ARG =
+ String.format("%s=%s", JobManagerOptions.ADDRESS.key(),
POD_IP_ARG);
private final StandaloneKubernetesJobManagerParameters
kubernetesJobManagerParameters;
@@ -64,7 +68,7 @@ public class CmdStandaloneJobManagerDecorator extends
AbstractKubernetesStepDeco
.addToArgs(JOBMANAGER_ENTRYPOINT_ARG);
if (kubernetesJobManagerParameters.isHAEnabled()) {
- containerBuilder.addToArgs(POD_IP_ARG);
+ containerBuilder.addToArgs(DYNAMIC_PROPERTY_FLAG,
JOBMANAGER_RPC_ADDRESS_ARG);
}
return containerBuilder.build();
@@ -99,8 +103,8 @@ public class CmdStandaloneJobManagerDecorator extends
AbstractKubernetesStepDeco
}
if (kubernetesJobManagerParameters.isHAEnabled()) {
- args.add("--host");
- args.add(POD_IP_ARG);
+ args.add(DYNAMIC_PROPERTY_FLAG);
+ args.add(JOBMANAGER_RPC_ADDRESS_ARG);
}
List<String> jobSpecArgs =
kubernetesJobManagerParameters.getJobSpecArgs();
diff --git
a/flink-kubernetes-standalone/src/test/java/org/apache/flink/kubernetes/operator/kubeclient/decorators/CmdStandaloneJobManagerDecoratorTest.java
b/flink-kubernetes-standalone/src/test/java/org/apache/flink/kubernetes/operator/kubeclient/decorators/CmdStandaloneJobManagerDecoratorTest.java
index d40f111c1..dea384566 100644
---
a/flink-kubernetes-standalone/src/test/java/org/apache/flink/kubernetes/operator/kubeclient/decorators/CmdStandaloneJobManagerDecoratorTest.java
+++
b/flink-kubernetes-standalone/src/test/java/org/apache/flink/kubernetes/operator/kubeclient/decorators/CmdStandaloneJobManagerDecoratorTest.java
@@ -20,21 +20,27 @@ package
org.apache.flink.kubernetes.operator.kubeclient.decorators;
import org.apache.flink.client.deployment.ClusterSpecification;
import org.apache.flink.client.deployment.application.ApplicationConfiguration;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.ConfigurationUtils;
import org.apache.flink.configuration.HighAvailabilityOptions;
+import org.apache.flink.configuration.JobManagerOptions;
+import
org.apache.flink.container.entrypoint.StandaloneApplicationClusterConfigurationParserFactory;
import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;
import org.apache.flink.kubernetes.kubeclient.FlinkPod;
import
org.apache.flink.kubernetes.operator.kubeclient.parameters.StandaloneKubernetesJobManagerParameters;
import
org.apache.flink.kubernetes.operator.standalone.StandaloneKubernetesConfigOptionsInternal;
+import org.apache.flink.runtime.entrypoint.ClusterConfiguration;
+import
org.apache.flink.runtime.entrypoint.EntrypointClusterConfigurationParserFactory;
+import org.apache.flink.runtime.entrypoint.parser.CommandLineParser;
+import org.apache.flink.runtime.entrypoint.parser.ParserResultFactory;
import org.apache.flink.runtime.jobgraph.SavepointConfigOptions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import java.util.ArrayList;
import java.util.List;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.contains;
-import static org.hamcrest.Matchers.containsInAnyOrder;
+import static org.assertj.core.api.Assertions.assertThat;
/**
* @link CmdStandaloneJobManagerDecorator unit tests
@@ -67,15 +73,15 @@ public class CmdStandaloneJobManagerDecoratorTest {
StandaloneKubernetesConfigOptionsInternal.ClusterMode.SESSION);
FlinkPod decoratedPod = decorator.decorateFlinkPod(new
FlinkPod.Builder().build());
- assertThat(
- decoratedPod.getMainContainer().getCommand(),
containsInAnyOrder(MOCK_ENTRYPATH));
- assertThat(
- decoratedPod.getMainContainer().getArgs(),
-
containsInAnyOrder(CmdStandaloneJobManagerDecorator.JOBMANAGER_ENTRYPOINT_ARG));
+ assertThat(decoratedPod.getMainContainer().getCommand())
+ .containsExactlyInAnyOrder(MOCK_ENTRYPATH);
+ assertThat(decoratedPod.getMainContainer().getArgs())
+ .containsExactlyInAnyOrder(
+
CmdStandaloneJobManagerDecorator.JOBMANAGER_ENTRYPOINT_ARG);
}
@Test
- public void testApplicationCommandAndArgsAdded() {
+ public void testApplicationCommandAndArgsAdded() throws Exception {
final String testMainClass = "org.main.class";
configuration.set(
StandaloneKubernetesConfigOptionsInternal.CLUSTER_MODE,
@@ -85,17 +91,20 @@ public class CmdStandaloneJobManagerDecoratorTest {
configuration.set(SavepointConfigOptions.SAVEPOINT_PATH,
"/tmp/savepoint/path");
FlinkPod decoratedPod = decorator.decorateFlinkPod(new
FlinkPod.Builder().build());
- assertThat(
- decoratedPod.getMainContainer().getCommand(),
containsInAnyOrder(MOCK_ENTRYPATH));
- assertThat(
- decoratedPod.getMainContainer().getArgs(),
- containsInAnyOrder(
+ assertThat(decoratedPod.getMainContainer().getCommand())
+ .containsExactlyInAnyOrder(MOCK_ENTRYPATH);
+ assertThat(decoratedPod.getMainContainer().getArgs())
+ .containsExactlyInAnyOrder(
CmdStandaloneJobManagerDecorator.APPLICATION_MODE_ARG,
"--allowNonRestoredState",
"--fromSavepoint",
"/tmp/savepoint/path",
"--job-classname",
- testMainClass));
+ testMainClass);
+
+ parseEntrypointConfig(
+ new StandaloneApplicationClusterConfigurationParserFactory(),
+ decoratedPod.getMainContainer().getArgs());
}
@Test
@@ -105,11 +114,10 @@ public class CmdStandaloneJobManagerDecoratorTest {
StandaloneKubernetesConfigOptionsInternal.ClusterMode.APPLICATION);
FlinkPod decoratedPod = decorator.decorateFlinkPod(new
FlinkPod.Builder().build());
- assertThat(
- decoratedPod.getMainContainer().getCommand(),
containsInAnyOrder(MOCK_ENTRYPATH));
- assertThat(
- decoratedPod.getMainContainer().getArgs(),
-
containsInAnyOrder(CmdStandaloneJobManagerDecorator.APPLICATION_MODE_ARG));
+ assertThat(decoratedPod.getMainContainer().getCommand())
+ .containsExactlyInAnyOrder(MOCK_ENTRYPATH);
+ assertThat(decoratedPod.getMainContainer().getArgs())
+
.containsExactlyInAnyOrder(CmdStandaloneJobManagerDecorator.APPLICATION_MODE_ARG);
}
@Test
@@ -124,13 +132,13 @@ public class CmdStandaloneJobManagerDecoratorTest {
FlinkPod decoratedPod = decorator.decorateFlinkPod(new
FlinkPod.Builder().build());
- assertThat(
- decoratedPod.getMainContainer().getCommand(),
containsInAnyOrder(MOCK_ENTRYPATH));
- assertThat(
- decoratedPod.getMainContainer().getArgs(),
- containsInAnyOrder(
+ assertThat(decoratedPod.getMainContainer().getCommand())
+ .containsExactlyInAnyOrder(MOCK_ENTRYPATH);
+ assertThat(decoratedPod.getMainContainer().getArgs())
+ .containsExactly(
CmdStandaloneJobManagerDecorator.JOBMANAGER_ENTRYPOINT_ARG,
- CmdStandaloneJobManagerDecorator.POD_IP_ARG));
+ CmdStandaloneJobManagerDecorator.DYNAMIC_PROPERTY_FLAG,
+
CmdStandaloneJobManagerDecorator.JOBMANAGER_RPC_ADDRESS_ARG);
}
@Test
@@ -144,13 +152,83 @@ public class CmdStandaloneJobManagerDecoratorTest {
FlinkPod decoratedPod = decorator.decorateFlinkPod(new
FlinkPod.Builder().build());
- assertThat(
- decoratedPod.getMainContainer().getArgs(),
- contains(
+ assertThat(decoratedPod.getMainContainer().getArgs())
+ .containsExactly(
CmdStandaloneJobManagerDecorator.APPLICATION_MODE_ARG,
- "--host",
- CmdStandaloneJobManagerDecorator.POD_IP_ARG,
+ CmdStandaloneJobManagerDecorator.DYNAMIC_PROPERTY_FLAG,
+
CmdStandaloneJobManagerDecorator.JOBMANAGER_RPC_ADDRESS_ARG,
"--test",
- "123"));
+ "123");
+ }
+
+ @Test
+ public void testApplicationHAArgsUnderstoodByFlinkRuntime() throws
Exception {
+ configuration.set(
+ StandaloneKubernetesConfigOptionsInternal.CLUSTER_MODE,
+
StandaloneKubernetesConfigOptionsInternal.ClusterMode.APPLICATION);
+ configuration.set(HighAvailabilityOptions.HA_MODE, "KUBERNETES");
+
+ FlinkPod decoratedPod = decorator.decorateFlinkPod(new
FlinkPod.Builder().build());
+
+ Configuration parsed =
+ parseEntrypointConfig(
+ new
StandaloneApplicationClusterConfigurationParserFactory(),
+ decoratedPod.getMainContainer().getArgs());
+
+ assertThat(parsed.get(JobManagerOptions.ADDRESS))
+ .isEqualTo(CmdStandaloneJobManagerDecorator.POD_IP_ARG);
+ }
+
+ @Test
+ public void testSessionHAArgsUnderstoodByFlinkRuntime() throws Exception {
+ configuration.set(
+ StandaloneKubernetesConfigOptionsInternal.CLUSTER_MODE,
+ StandaloneKubernetesConfigOptionsInternal.ClusterMode.SESSION);
+ configuration.set(
+ HighAvailabilityOptions.HA_MODE,
+
"org.apache.flink.kubernetes.highavailability.KubernetesHaServicesFactory");
+
+ FlinkPod decoratedPod = decorator.decorateFlinkPod(new
FlinkPod.Builder().build());
+
+ Configuration parsed =
+ parseEntrypointConfig(
+ new EntrypointClusterConfigurationParserFactory(),
+ decoratedPod.getMainContainer().getArgs());
+
+ assertThat(parsed.get(JobManagerOptions.ADDRESS))
+ .isEqualTo(CmdStandaloneJobManagerDecorator.POD_IP_ARG);
+ }
+
+ @Test
+ public void testApplicationWithoutHADoesNotSetRpcAddress() throws
Exception {
+ configuration.set(
+ StandaloneKubernetesConfigOptionsInternal.CLUSTER_MODE,
+
StandaloneKubernetesConfigOptionsInternal.ClusterMode.APPLICATION);
+
+ FlinkPod decoratedPod = decorator.decorateFlinkPod(new
FlinkPod.Builder().build());
+
+ Configuration parsed =
+ parseEntrypointConfig(
+ new
StandaloneApplicationClusterConfigurationParserFactory(),
+ decoratedPod.getMainContainer().getArgs());
+
+ assertThat(parsed.get(JobManagerOptions.ADDRESS)).isNull();
+ }
+
+ private static Configuration parseEntrypointConfig(
+ ParserResultFactory<? extends ClusterConfiguration> factory,
List<String> containerArgs)
+ throws Exception {
+ // Dropping the leading entrypoint token (standalone-job /
jobmanager); not seen by the JVM.
+ List<String> entrypointArgs = containerArgs.subList(1,
containerArgs.size());
+
+ // Adding configDir as it's required by the parser factory
+ List<String> argsWithConfigDir = new ArrayList<>();
+ argsWithConfigDir.add("--configDir");
+ argsWithConfigDir.add("unused");
+ argsWithConfigDir.addAll(entrypointArgs);
+
+ ClusterConfiguration parsed =
+ new
CommandLineParser<>(factory).parse(argsWithConfigDir.toArray(new String[0]));
+ return
ConfigurationUtils.createConfiguration(parsed.getDynamicProperties());
}
}
