This is an automated email from the ASF dual-hosted git repository. gyfora pushed a commit to branch release-1.16 in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git
commit 1cd4f1d2a3e70ae17a45238e55f658ce8daaedc6 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); + } +}
