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
commit 8f95987a15530f109361d2416e204a7e422e8d7c Author: Gyula Fora <[email protected]> AuthorDate: Thu Sep 17 17:24:54 2026 +0200 Revert "[FLINK-40465] Add version-independent classes to parse job config (#1193)" This reverts commit 3ed982e5cf2e4259aa0836a0b3fb8d902ac03632. --- .../flink/runtime/rest/messages/JobConfigInfo.java | 260 --------------------- .../operator/service/AbstractFlinkServiceTest.java | 121 +++++++--- .../runtime/rest/messages/JobConfigInfoTest.java | 136 ----------- 3 files changed, 89 insertions(+), 428 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 deleted file mode 100644 index b492a3210..000000000 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/runtime/rest/messages/JobConfigInfo.java +++ /dev/null @@ -1,260 +0,0 @@ -/* - * 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 a212f2658..8735ca7f6 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,7 +64,6 @@ 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; @@ -159,7 +158,6 @@ 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; @@ -1597,6 +1595,95 @@ 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(); @@ -1620,36 +1707,6 @@ 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 deleted file mode 100644 index f56bc5c9d..000000000 --- a/flink-kubernetes-operator/src/test/java/org/apache/flink/runtime/rest/messages/JobConfigInfoTest.java +++ /dev/null @@ -1,136 +0,0 @@ -/* - * 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); - } -}
