ferenc-csaky commented on code in PR #1150:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/1150#discussion_r3551524553
##########
flink-kubernetes-standalone/src/test/java/org/apache/flink/kubernetes/operator/kubeclient/decorators/CmdStandaloneJobManagerDecoratorTest.java:
##########
@@ -148,9 +165,97 @@ public void testApplicationKubernetesHAArgsAdded() {
decoratedPod.getMainContainer().getArgs(),
contains(
CmdStandaloneJobManagerDecorator.APPLICATION_MODE_ARG,
- "--host",
- CmdStandaloneJobManagerDecorator.POD_IP_ARG,
+
CmdStandaloneJobManagerDecorator.JOBMANAGER_RPC_ADDRESS_ARG,
"--test",
"123"));
}
+
+ /**
+ * Verifies that the HA args produced for an application cluster are
actually understood by the
+ * real Flink standalone application entrypoint parser and resolve to
{@code
+ * jobmanager.rpc.address}. This guards against regressions where the
generated args are removed
+ * or renamed in a future Flink version (as {@code --host} was in Flink
2.0).
+ */
Review Comment:
Let's remove the javadoc from all test cases. It's just noise.
##########
flink-kubernetes-standalone/src/test/java/org/apache/flink/kubernetes/operator/kubeclient/decorators/CmdStandaloneJobManagerDecoratorTest.java:
##########
@@ -148,9 +165,97 @@ public void testApplicationKubernetesHAArgsAdded() {
decoratedPod.getMainContainer().getArgs(),
contains(
CmdStandaloneJobManagerDecorator.APPLICATION_MODE_ARG,
- "--host",
- CmdStandaloneJobManagerDecorator.POD_IP_ARG,
+
CmdStandaloneJobManagerDecorator.JOBMANAGER_RPC_ADDRESS_ARG,
"--test",
"123"));
}
+
+ /**
+ * Verifies that the HA args produced for an application cluster are
actually understood by the
+ * real Flink standalone application entrypoint parser and resolve to
{@code
+ * jobmanager.rpc.address}. This guards against regressions where the
generated args are removed
+ * or renamed in a future Flink version (as {@code --host} was in Flink
2.0).
+ */
+ @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());
+
+ // The leading APPLICATION_MODE_ARG ("standalone-job") is consumed by
the Flink
+ // docker-entrypoint before the JVM, so it is not part of the
entrypoint's own args.
+ Configuration parsed =
+ parseEntrypointConfig(
+ new
StandaloneApplicationClusterConfigurationParserFactory(),
+
dropEntrypointToken(decoratedPod.getMainContainer().getArgs()));
+
+ assertEquals(
+ CmdStandaloneJobManagerDecorator.POD_IP_ARG,
parsed.get(JobManagerOptions.ADDRESS));
+ }
+
+ /**
+ * Verifies that the HA args produced for a session cluster are actually
understood by the real
+ * Flink standalone session entrypoint parser and resolve to {@code
jobmanager.rpc.address}.
+ */
+ @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());
+
+ // The leading JOBMANAGER_ENTRYPOINT_ARG ("jobmanager") is consumed by
the Flink
+ // docker-entrypoint before the JVM, so it is not part of the
entrypoint's own args.
+ Configuration parsed =
+ parseEntrypointConfig(
+ new EntrypointClusterConfigurationParserFactory(),
+
dropEntrypointToken(decoratedPod.getMainContainer().getArgs()));
+
+ assertEquals(
+ CmdStandaloneJobManagerDecorator.POD_IP_ARG,
parsed.get(JobManagerOptions.ADDRESS));
+ }
+
+ /** Without HA the entrypoint args must not set {@code
jobmanager.rpc.address}. */
+ @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(),
+
dropEntrypointToken(decoratedPod.getMainContainer().getArgs()));
+
+ assertNull(parsed.get(JobManagerOptions.ADDRESS));
+ }
+
+ private static List<String> dropEntrypointToken(List<String> args) {
+ return args.subList(1, args.size());
+ }
+
+ private static Configuration parseEntrypointConfig(
+ ParserResultFactory<? extends ClusterConfiguration> factory,
List<String> args)
+ throws Exception {
+ // The standalone entrypoints require --configDir; the Flink
docker-entrypoint always
+ // supplies it. Its value is irrelevant here since we assert on the
parsed dynamic
+ // properties rather than the loaded flink-conf.yaml.
Review Comment:
Let's shorten this to smth like `Adding configDir as it's required by the
parser factory`. The rest is just noise.
##########
flink-kubernetes-standalone/src/test/java/org/apache/flink/kubernetes/operator/kubeclient/decorators/CmdStandaloneJobManagerDecoratorTest.java:
##########
@@ -148,9 +165,97 @@ public void testApplicationKubernetesHAArgsAdded() {
decoratedPod.getMainContainer().getArgs(),
contains(
CmdStandaloneJobManagerDecorator.APPLICATION_MODE_ARG,
- "--host",
- CmdStandaloneJobManagerDecorator.POD_IP_ARG,
+
CmdStandaloneJobManagerDecorator.JOBMANAGER_RPC_ADDRESS_ARG,
"--test",
"123"));
}
+
+ /**
+ * Verifies that the HA args produced for an application cluster are
actually understood by the
+ * real Flink standalone application entrypoint parser and resolve to
{@code
+ * jobmanager.rpc.address}. This guards against regressions where the
generated args are removed
+ * or renamed in a future Flink version (as {@code --host} was in Flink
2.0).
+ */
+ @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());
+
+ // The leading APPLICATION_MODE_ARG ("standalone-job") is consumed by
the Flink
+ // docker-entrypoint before the JVM, so it is not part of the
entrypoint's own args.
+ Configuration parsed =
+ parseEntrypointConfig(
+ new
StandaloneApplicationClusterConfigurationParserFactory(),
+
dropEntrypointToken(decoratedPod.getMainContainer().getArgs()));
+
+ assertEquals(
+ CmdStandaloneJobManagerDecorator.POD_IP_ARG,
parsed.get(JobManagerOptions.ADDRESS));
+ }
+
+ /**
+ * Verifies that the HA args produced for a session cluster are actually
understood by the real
+ * Flink standalone session entrypoint parser and resolve to {@code
jobmanager.rpc.address}.
+ */
+ @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());
+
+ // The leading JOBMANAGER_ENTRYPOINT_ARG ("jobmanager") is consumed by
the Flink
+ // docker-entrypoint before the JVM, so it is not part of the
entrypoint's own args.
+ Configuration parsed =
+ parseEntrypointConfig(
+ new EntrypointClusterConfigurationParserFactory(),
+
dropEntrypointToken(decoratedPod.getMainContainer().getArgs()));
+
+ assertEquals(
+ CmdStandaloneJobManagerDecorator.POD_IP_ARG,
parsed.get(JobManagerOptions.ADDRESS));
+ }
+
+ /** Without HA the entrypoint args must not set {@code
jobmanager.rpc.address}. */
+ @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(),
+
dropEntrypointToken(decoratedPod.getMainContainer().getArgs()));
+
+ assertNull(parsed.get(JobManagerOptions.ADDRESS));
+ }
+
+ private static List<String> dropEntrypointToken(List<String> args) {
+ return args.subList(1, args.size());
+ }
Review Comment:
I think we should just integrate this to `parseEntrypointConfig`, it's only
relevant there and would make the call side more concise. And would also move
the explanation (in a shortened format) why the leading argument is removed
from the list), and remove those comments from the test cases.
##########
flink-kubernetes-standalone/src/test/java/org/apache/flink/kubernetes/operator/kubeclient/decorators/CmdStandaloneJobManagerDecoratorTest.java:
##########
@@ -96,6 +106,13 @@ public void testApplicationCommandAndArgsAdded() {
"/tmp/savepoint/path",
"--job-classname",
testMainClass));
+
+ // The full application command must be accepted by the real Flink
standalone application
+ // entrypoint parser, i.e. --job-classname / --fromSavepoint /
--allowNonRestoredState are
+ // valid, correctly-spelled options (parsing throws on an unrecognized
option).
Review Comment:
Pls. remove this. it's just noise.
##########
flink-kubernetes-standalone/src/test/java/org/apache/flink/kubernetes/operator/kubeclient/decorators/CmdStandaloneJobManagerDecoratorTest.java:
##########
@@ -148,9 +165,97 @@ public void testApplicationKubernetesHAArgsAdded() {
decoratedPod.getMainContainer().getArgs(),
contains(
CmdStandaloneJobManagerDecorator.APPLICATION_MODE_ARG,
- "--host",
- CmdStandaloneJobManagerDecorator.POD_IP_ARG,
+
CmdStandaloneJobManagerDecorator.JOBMANAGER_RPC_ADDRESS_ARG,
"--test",
"123"));
}
+
+ /**
+ * Verifies that the HA args produced for an application cluster are
actually understood by the
+ * real Flink standalone application entrypoint parser and resolve to
{@code
+ * jobmanager.rpc.address}. This guards against regressions where the
generated args are removed
+ * or renamed in a future Flink version (as {@code --host} was in Flink
2.0).
+ */
+ @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());
+
+ // The leading APPLICATION_MODE_ARG ("standalone-job") is consumed by
the Flink
+ // docker-entrypoint before the JVM, so it is not part of the
entrypoint's own args.
+ Configuration parsed =
+ parseEntrypointConfig(
+ new
StandaloneApplicationClusterConfigurationParserFactory(),
+
dropEntrypointToken(decoratedPod.getMainContainer().getArgs()));
+
+ assertEquals(
+ CmdStandaloneJobManagerDecorator.POD_IP_ARG,
parsed.get(JobManagerOptions.ADDRESS));
+ }
+
+ /**
+ * Verifies that the HA args produced for a session cluster are actually
understood by the real
+ * Flink standalone session entrypoint parser and resolve to {@code
jobmanager.rpc.address}.
+ */
+ @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());
+
+ // The leading JOBMANAGER_ENTRYPOINT_ARG ("jobmanager") is consumed by
the Flink
+ // docker-entrypoint before the JVM, so it is not part of the
entrypoint's own args.
+ Configuration parsed =
+ parseEntrypointConfig(
+ new EntrypointClusterConfigurationParserFactory(),
+
dropEntrypointToken(decoratedPod.getMainContainer().getArgs()));
+
+ assertEquals(
+ CmdStandaloneJobManagerDecorator.POD_IP_ARG,
parsed.get(JobManagerOptions.ADDRESS));
+ }
+
+ /** Without HA the entrypoint args must not set {@code
jobmanager.rpc.address}. */
+ @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(),
+
dropEntrypointToken(decoratedPod.getMainContainer().getArgs()));
+
+ assertNull(parsed.get(JobManagerOptions.ADDRESS));
+ }
+
+ private static List<String> dropEntrypointToken(List<String> args) {
+ return args.subList(1, args.size());
+ }
+
+ private static Configuration parseEntrypointConfig(
+ ParserResultFactory<? extends ClusterConfiguration> factory,
List<String> args)
+ throws Exception {
+ // The standalone entrypoints require --configDir; the Flink
docker-entrypoint always
+ // supplies it. Its value is irrelevant here since we assert on the
parsed dynamic
+ // properties rather than the loaded flink-conf.yaml.
+ List<String> argsWithConfigDir = new ArrayList<>();
+ argsWithConfigDir.add("--configDir");
+ argsWithConfigDir.add("/opt/flink/conf");
Review Comment:
Since this won't be resolved let's give it a string like `dummy` to make it
more explicit this path won't be resolved at all.
##########
flink-kubernetes-standalone/src/test/java/org/apache/flink/kubernetes/operator/kubeclient/decorators/CmdStandaloneJobManagerDecoratorTest.java:
##########
@@ -148,9 +165,97 @@ public void testApplicationKubernetesHAArgsAdded() {
decoratedPod.getMainContainer().getArgs(),
contains(
CmdStandaloneJobManagerDecorator.APPLICATION_MODE_ARG,
- "--host",
- CmdStandaloneJobManagerDecorator.POD_IP_ARG,
+
CmdStandaloneJobManagerDecorator.JOBMANAGER_RPC_ADDRESS_ARG,
"--test",
"123"));
}
+
+ /**
+ * Verifies that the HA args produced for an application cluster are
actually understood by the
+ * real Flink standalone application entrypoint parser and resolve to
{@code
+ * jobmanager.rpc.address}. This guards against regressions where the
generated args are removed
+ * or renamed in a future Flink version (as {@code --host} was in Flink
2.0).
+ */
+ @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());
+
+ // The leading APPLICATION_MODE_ARG ("standalone-job") is consumed by
the Flink
+ // docker-entrypoint before the JVM, so it is not part of the
entrypoint's own args.
+ Configuration parsed =
+ parseEntrypointConfig(
+ new
StandaloneApplicationClusterConfigurationParserFactory(),
+
dropEntrypointToken(decoratedPod.getMainContainer().getArgs()));
+
+ assertEquals(
Review Comment:
Pls use AssertJ for any kind of assertions instead of JUnit asserts.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
