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 3ed982e5 [FLINK-40465] Add version-independent classes to parse job
config (#1193)
3ed982e5 is described below
commit 3ed982e5cf2e4259aa0836a0b3fb8d902ac03632
Author: Mate Czagany <[email protected]>
AuthorDate: Thu Aug 27 12:28:14 2026 +0200
[FLINK-40465] Add version-independent classes to parse job config (#1193)
---
.../flink/runtime/rest/messages/JobConfigInfo.java | 260 +++++++++++++++++++++
.../operator/service/AbstractFlinkServiceTest.java | 121 +++-------
.../runtime/rest/messages/JobConfigInfoTest.java | 136 +++++++++++
3 files changed, 428 insertions(+), 89 deletions(-)
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/runtime/rest/messages/JobConfigInfo.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/runtime/rest/messages/JobConfigInfo.java
new file mode 100644
index 00000000..b492a321
--- /dev/null
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/runtime/rest/messages/JobConfigInfo.java
@@ -0,0 +1,260 @@
+/*
+ * 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.runtime.rest.messages;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.rest.util.RestMapperUtils;
+import org.apache.flink.util.Preconditions;
+
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnore;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonParser;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.DeserializationContext;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.SerializerProvider;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.deser.std.StdDeserializer;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ser.std.StdSerializer;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * Copied from Flink 1.20 to parse job config responses from all supported
Flink versions, with
+ * small modifications: {@code executionMode} is nullable because the
deprecated {@code
+ * execution-mode} field was removed in Flink 2.0, and the nested {@link
ExecutionConfigInfo} is
+ * parsed with the flexible object mapper so fields added by future Flink
versions are tolerated.
+ */
+@JsonSerialize(using = JobConfigInfo.Serializer.class)
+@JsonDeserialize(using = JobConfigInfo.Deserializer.class)
+public class JobConfigInfo implements ResponseBody {
+
+ public static final String FIELD_NAME_JOB_ID = "jid";
+ public static final String FIELD_NAME_JOB_NAME = "name";
+ public static final String FIELD_NAME_EXECUTION_CONFIG =
"execution-config";
+
+ private final JobID jobId;
+
+ private final String jobName;
+
+ @Nullable private final ExecutionConfigInfo executionConfigInfo;
+
+ public JobConfigInfo(
+ JobID jobId, String jobName, @Nullable ExecutionConfigInfo
executionConfigInfo) {
+ this.jobId = Preconditions.checkNotNull(jobId);
+ this.jobName = Preconditions.checkNotNull(jobName);
+ this.executionConfigInfo = executionConfigInfo;
+ }
+
+ public JobID getJobId() {
+ return jobId;
+ }
+
+ public String getJobName() {
+ return jobName;
+ }
+
+ @Nullable
+ public ExecutionConfigInfo getExecutionConfigInfo() {
+ return executionConfigInfo;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ JobConfigInfo that = (JobConfigInfo) o;
+ return Objects.equals(jobId, that.jobId)
+ && Objects.equals(jobName, that.jobName)
+ && Objects.equals(executionConfigInfo,
that.executionConfigInfo);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(jobId, jobName, executionConfigInfo);
+ }
+
+ //
---------------------------------------------------------------------------------
+ // Static helper classes
+ //
---------------------------------------------------------------------------------
+
+ /** Json serializer for the {@link JobConfigInfo}. */
+ public static final class Serializer extends StdSerializer<JobConfigInfo> {
+
+ private static final long serialVersionUID = -1551666039618928811L;
+
+ public Serializer() {
+ super(JobConfigInfo.class);
+ }
+
+ @Override
+ public void serialize(
+ JobConfigInfo jobConfigInfo,
+ JsonGenerator jsonGenerator,
+ SerializerProvider serializerProvider)
+ throws IOException {
+ jsonGenerator.writeStartObject();
+
+ jsonGenerator.writeStringField(FIELD_NAME_JOB_ID,
jobConfigInfo.getJobId().toString());
+ jsonGenerator.writeStringField(FIELD_NAME_JOB_NAME,
jobConfigInfo.getJobName());
+
+ if (jobConfigInfo.getExecutionConfigInfo() != null) {
+ jsonGenerator.writeObjectField(
+ FIELD_NAME_EXECUTION_CONFIG,
jobConfigInfo.getExecutionConfigInfo());
+ }
+
+ jsonGenerator.writeEndObject();
+ }
+ }
+
+ /** Json deserializer for the {@link JobConfigInfo}. */
+ public static final class Deserializer extends
StdDeserializer<JobConfigInfo> {
+
+ private static final long serialVersionUID = -3580088509877177213L;
+
+ public Deserializer() {
+ super(JobConfigInfo.class);
+ }
+
+ @Override
+ public JobConfigInfo deserialize(
+ JsonParser jsonParser, DeserializationContext
deserializationContext)
+ throws IOException {
+ JsonNode rootNode = jsonParser.readValueAsTree();
+
+ final JobID jobId =
JobID.fromHexString(rootNode.get(FIELD_NAME_JOB_ID).asText());
+ final String jobName = rootNode.get(FIELD_NAME_JOB_NAME).asText();
+
+ final ExecutionConfigInfo executionConfigInfo;
+
+ if (rootNode.has(FIELD_NAME_EXECUTION_CONFIG)) {
+ executionConfigInfo =
+ RestMapperUtils.getFlexibleObjectMapper()
+ .treeToValue(
+
rootNode.get(FIELD_NAME_EXECUTION_CONFIG),
+ ExecutionConfigInfo.class);
+ } else {
+ executionConfigInfo = null;
+ }
+
+ return new JobConfigInfo(jobId, jobName, executionConfigInfo);
+ }
+ }
+
+ /** Nested class to encapsulate the execution configuration. */
+ public static final class ExecutionConfigInfo {
+
+ @Deprecated public static final String FIELD_NAME_EXECUTION_MODE =
"execution-mode";
+ public static final String FIELD_NAME_RESTART_STRATEGY =
"restart-strategy";
+ public static final String FIELD_NAME_PARALLELISM = "job-parallelism";
+ public static final String FIELD_NAME_OBJECT_REUSE_MODE =
"object-reuse-mode";
+ public static final String FIELD_NAME_GLOBAL_JOB_PARAMETERS =
"user-config";
+
+ /** Removed in Flink 2.0, only returned by Flink 1.x. */
+ @Deprecated
+ @Nullable
+ @JsonProperty(FIELD_NAME_EXECUTION_MODE)
+ private final String executionMode;
+
+ @JsonProperty(FIELD_NAME_RESTART_STRATEGY)
+ private final String restartStrategy;
+
+ @JsonProperty(FIELD_NAME_PARALLELISM)
+ private final int parallelism;
+
+ @JsonProperty(FIELD_NAME_OBJECT_REUSE_MODE)
+ private final boolean isObjectReuse;
+
+ @JsonProperty(FIELD_NAME_GLOBAL_JOB_PARAMETERS)
+ private final Map<String, String> globalJobParameters;
+
+ @JsonCreator
+ public ExecutionConfigInfo(
+ @JsonProperty(FIELD_NAME_EXECUTION_MODE) @Nullable String
executionMode,
+ @JsonProperty(FIELD_NAME_RESTART_STRATEGY) String
restartStrategy,
+ @JsonProperty(FIELD_NAME_PARALLELISM) int parallelism,
+ @JsonProperty(FIELD_NAME_OBJECT_REUSE_MODE) boolean
isObjectReuse,
+ @JsonProperty(FIELD_NAME_GLOBAL_JOB_PARAMETERS)
+ Map<String, String> globalJobParameters) {
+ this.executionMode = executionMode;
+ this.restartStrategy = Preconditions.checkNotNull(restartStrategy);
+ this.parallelism = parallelism;
+ this.isObjectReuse = isObjectReuse;
+ this.globalJobParameters =
Preconditions.checkNotNull(globalJobParameters);
+ }
+
+ @Nullable
+ public String getExecutionMode() {
+ return executionMode;
+ }
+
+ public String getRestartStrategy() {
+ return restartStrategy;
+ }
+
+ public int getParallelism() {
+ return parallelism;
+ }
+
+ @JsonIgnore
+ public boolean isObjectReuse() {
+ return isObjectReuse;
+ }
+
+ public Map<String, String> getGlobalJobParameters() {
+ return globalJobParameters;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ ExecutionConfigInfo that = (ExecutionConfigInfo) o;
+ return parallelism == that.parallelism
+ && isObjectReuse == that.isObjectReuse
+ && Objects.equals(executionMode, that.executionMode)
+ && Objects.equals(restartStrategy, that.restartStrategy)
+ && Objects.equals(globalJobParameters,
that.globalJobParameters);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(
+ executionMode,
+ restartStrategy,
+ parallelism,
+ isObjectReuse,
+ globalJobParameters);
+ }
+ }
+}
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkServiceTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkServiceTest.java
index 8735ca7f..a212f265 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkServiceTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkServiceTest.java
@@ -64,6 +64,7 @@ import
org.apache.flink.runtime.rest.handler.async.TriggerResponse;
import org.apache.flink.runtime.rest.messages.ConfigurationInfo;
import org.apache.flink.runtime.rest.messages.ConfigurationInfoEntry;
import org.apache.flink.runtime.rest.messages.DashboardConfiguration;
+import org.apache.flink.runtime.rest.messages.JobConfigHeaders;
import org.apache.flink.runtime.rest.messages.JobConfigInfo;
import org.apache.flink.runtime.rest.messages.JobExceptionsInfoWithHistory;
import org.apache.flink.runtime.rest.messages.MessageHeaders;
@@ -158,6 +159,7 @@ import static
org.apache.flink.api.common.JobStatus.FINISHED;
import static org.apache.flink.api.common.JobStatus.RUNNING;
import static
org.apache.flink.kubernetes.operator.config.FlinkConfigBuilder.FLINK_VERSION;
import static
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.OPERATOR_SAVEPOINT_FORMAT_TYPE;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -1595,95 +1597,6 @@ public class AbstractFlinkServiceTest {
+ declared);
}
- @Test
- void testAllExecutionConfigInfoFieldNamesCoveredByMapping() throws
Exception {
- Set<String> declared =
discoverFieldNameConstants(JobConfigInfo.ExecutionConfigInfo.class);
- assertFalse(declared.isEmpty(), "Should discover FIELD_NAME_*
constants");
-
- Set<String> handled =
- new HashSet<>(
- Set.of(
-
JobConfigInfo.ExecutionConfigInfo.FIELD_NAME_PARALLELISM,
-
JobConfigInfo.ExecutionConfigInfo.FIELD_NAME_OBJECT_REUSE_MODE,
-
JobConfigInfo.ExecutionConfigInfo.FIELD_NAME_GLOBAL_JOB_PARAMETERS,
- // Informational only, not runtime config
overrides
-
JobConfigInfo.ExecutionConfigInfo.FIELD_NAME_EXECUTION_MODE,
-
JobConfigInfo.ExecutionConfigInfo.FIELD_NAME_RESTART_STRATEGY));
-
- declared.removeAll(handled);
- assertTrue(
- declared.isEmpty(),
- "ExecutionConfigInfo FIELD_NAME constants not covered: "
- + declared
- + ". Add mapping or exclude explicitly.");
- }
-
- @Test
- void testMapJobConfigurationMapsAllExpectedFields() {
- JobConfigInfo configInfo =
- new JobConfigInfo(
- new JobID(),
- "test-job",
- new JobConfigInfo.ExecutionConfigInfo(
- "PIPELINED",
- "fixedDelay",
- 4,
- true,
- Map.of("user.param", "value1")));
-
- Map<String, String> result =
FlinkRuntimeConfigurationUtils.mapJobConfiguration(configInfo);
-
- assertEquals("4", result.get("parallelism.default"));
- assertEquals("true", result.get("pipeline.object-reuse"));
- assertEquals("value1", result.get("user.param"));
- assertEquals(3, result.size());
- }
-
- @Test
- void testMapJobConfigurationDropsOperatorControlledGlobalParameters() {
- JobConfigInfo configInfo =
- new JobConfigInfo(
- new JobID(),
- "test-job",
- new JobConfigInfo.ExecutionConfigInfo(
- "PIPELINED",
- "fixedDelay",
- 4,
- true,
- Map.of(
- "user.param",
- "value1",
-
"kubernetes.operator.job.upgrade.last-state-fallback.enabled",
- "false",
- "job.autoscaler.enabled",
- "false")));
-
- Map<String, String> result =
FlinkRuntimeConfigurationUtils.mapJobConfiguration(configInfo);
-
- // A job must not be able to change how the operator manages it.
- assertNull(
-
result.get("kubernetes.operator.job.upgrade.last-state-fallback.enabled"),
- "operator keys must not be taken from global job parameters");
- assertNull(
- result.get("job.autoscaler.enabled"),
- "autoscaler keys must not be taken from global job
parameters");
-
- // Unrelated parameters and the mapped execution fields are still
present.
- assertEquals("value1", result.get("user.param"));
- assertEquals("4", result.get("parallelism.default"));
- assertEquals("true", result.get("pipeline.object-reuse"));
- assertEquals(3, result.size());
- }
-
- @Test
- void testMapJobConfigurationHandlesNullGracefully() {
-
assertTrue(FlinkRuntimeConfigurationUtils.mapJobConfiguration(null).isEmpty());
- assertTrue(
- FlinkRuntimeConfigurationUtils.mapJobConfiguration(
- new JobConfigInfo(new JobID(), "test-job",
null))
- .isEmpty());
- }
-
@Test
void testGetRuntimeConfigurationIncludesJmConfig() throws Exception {
var jmConfig = new ConfigurationInfo();
@@ -1707,6 +1620,36 @@ public class AbstractFlinkServiceTest {
assertEquals("ZOOKEEPER", result.get("high-availability"));
}
+ @Test
+ void testGetRuntimeConfigurationIncludesVersionIndependentJobConfig()
throws Exception {
+ var executionConfig =
+ new JobConfigInfo.ExecutionConfigInfo(
+ null, "fixedDelay", 4, true, Map.of("user.param",
"value1"));
+ var jobConfigInfo = new JobConfigInfo(new JobID(), "test-job",
executionConfig);
+ var jobConfigHeadersUsed = new AtomicBoolean();
+
+ var service =
+ getTestingService(
+ (headers, params, body) -> {
+ if (headers instanceof JobConfigHeaders) {
+ jobConfigHeadersUsed.set(true);
+ return
CompletableFuture.completedFuture(jobConfigInfo);
+ }
+ return CompletableFuture.failedFuture(
+ new UnsupportedOperationException());
+ });
+
+ var result = service.getRuntimeConfiguration(configuration,
JobID.generate());
+
+ assertThat(jobConfigHeadersUsed).isTrue();
+ assertThat(result)
+ .containsExactlyInAnyOrderEntriesOf(
+ Map.of(
+ "parallelism.default", "4",
+ "pipeline.object-reuse", "true",
+ "user.param", "value1"));
+ }
+
@Test
void testGetRuntimeConfigurationThrowsWhenAllFetchesFail() throws
Exception {
var service =
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/runtime/rest/messages/JobConfigInfoTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/runtime/rest/messages/JobConfigInfoTest.java
new file mode 100644
index 00000000..f56bc5c9
--- /dev/null
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/runtime/rest/messages/JobConfigInfoTest.java
@@ -0,0 +1,136 @@
+/*
+ * 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.runtime.rest.messages;
+
+import org.apache.flink.api.common.JobID;
+import
org.apache.flink.kubernetes.operator.utils.FlinkRuntimeConfigurationUtils;
+import org.apache.flink.runtime.rest.util.RestMapperUtils;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.util.Map;
+import java.util.stream.Stream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for the shadowed multi-version compatible {@link JobConfigInfo}. */
+class JobConfigInfoTest {
+
+ @ParameterizedTest
+ @MethodSource("flinkJobConfigResponses")
+ void testJobConfigurationCompatibilityAcrossFlinkVersions(String json)
throws Exception {
+ var configInfo = parseJobConfig(json);
+
+ assertThat(configInfo.getJobId())
+
.isEqualTo(JobID.fromHexString("bc8cc01941f17a4fb5f8873b45512e19"));
+ assertThat(configInfo.getJobName()).isEqualTo("test-job");
+
assertThat(FlinkRuntimeConfigurationUtils.mapJobConfiguration(configInfo))
+ .containsExactlyInAnyOrderEntriesOf(
+ Map.of(
+ "parallelism.default", "4",
+ "pipeline.object-reuse", "true",
+ "user.param", "value1"));
+ }
+
+ @Test
+ void testMapJobConfigurationDropsOperatorControlledGlobalParameters() {
+ var executionConfig =
+ new JobConfigInfo.ExecutionConfigInfo(
+ null,
+ "fixedDelay",
+ 4,
+ true,
+ Map.of(
+ "user.param",
+ "value1",
+
"kubernetes.operator.job.upgrade.last-state-fallback.enabled",
+ "false",
+ "job.autoscaler.enabled",
+ "false"));
+ var configInfo = new JobConfigInfo(new JobID(), "test-job",
executionConfig);
+
+
assertThat(FlinkRuntimeConfigurationUtils.mapJobConfiguration(configInfo))
+ .containsExactlyInAnyOrderEntriesOf(
+ Map.of(
+ "parallelism.default", "4",
+ "pipeline.object-reuse", "true",
+ "user.param", "value1"));
+ }
+
+ @Test
+ void testMapJobConfigurationHandlesMissingExecutionConfig() throws
Exception {
+ var withoutExecutionConfig =
+ parseJobConfig(
+ """
+ {
+ "jid": "bc8cc01941f17a4fb5f8873b45512e19",
+ "name": "test-job"
+ }
+ """);
+
+ assertThat(withoutExecutionConfig.getExecutionConfigInfo()).isNull();
+
assertThat(FlinkRuntimeConfigurationUtils.mapJobConfiguration(withoutExecutionConfig))
+ .isEmpty();
+
assertThat(FlinkRuntimeConfigurationUtils.mapJobConfiguration(null)).isEmpty();
+ }
+
+ private static Stream<String> flinkJobConfigResponses() {
+ return Stream.of(
+ // Flink 1.x includes the deprecated execution-mode field.
+ """
+ {
+ "jid": "bc8cc01941f17a4fb5f8873b45512e19",
+ "name": "test-job",
+ "execution-config": {
+ "execution-mode": "PIPELINED",
+ "restart-strategy": "fixedDelay",
+ "job-parallelism": 4,
+ "object-reuse-mode": true,
+ "user-config": {
+ "user.param": "value1"
+ }
+ }
+ }
+ """,
+ // Flink 2.x removed execution-mode. Unknown future fields
must also be tolerated.
+ """
+ {
+ "jid": "bc8cc01941f17a4fb5f8873b45512e19",
+ "name": "test-job",
+ "future-root-field": "ignored",
+ "execution-config": {
+ "restart-strategy": "Cluster level default restart
strategy",
+ "job-parallelism": 4,
+ "object-reuse-mode": true,
+ "user-config": {
+ "user.param": "value1"
+ },
+ "future-execution-field": "ignored"
+ }
+ }
+ """);
+ }
+
+ /** Parses the response the same way {@code RestClient} does. */
+ private static JobConfigInfo parseJobConfig(String json) throws Exception {
+ return RestMapperUtils.getFlexibleObjectMapper().readValue(json,
JobConfigInfo.class);
+ }
+}
