This is an automated email from the ASF dual-hosted git repository. mbalassi pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git
commit 29a4aa5adf101920cbe1a3a9a178ff16f52c746d Author: Marton Balassi <[email protected]> AuthorDate: Tue Apr 16 13:56:51 2024 +0200 [FLINK-35116] Bump operator sdk version to 4.8.3 --- .../generated/kubernetes_operator_config_configuration.html | 2 +- docs/layouts/shortcodes/generated/system_section.html | 2 +- .../operator/api/validation/CrdCompatibilityChecker.java | 6 +++++- .../org/apache/flink/kubernetes/operator/FlinkOperator.java | 2 +- .../operator/config/FlinkOperatorConfiguration.java | 3 +-- .../kubernetes/operator/metrics/OperatorJosdkMetrics.java | 4 ++-- .../src/main/resources/META-INF/NOTICE | 6 +++--- .../apache/flink/kubernetes/operator/FlinkOperatorTest.java | 13 +------------ .../org/apache/flink/kubernetes/operator/TestUtils.java | 6 ++++++ .../operator/service/AbstractFlinkServiceTest.java | 2 +- pom.xml | 2 +- 11 files changed, 23 insertions(+), 25 deletions(-) diff --git a/docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html b/docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html index d0680627..918b4074 100644 --- a/docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html +++ b/docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html @@ -304,7 +304,7 @@ </tr> <tr> <td><h5>kubernetes.operator.reconcile.parallelism</h5></td> - <td style="word-wrap: break-word;">200</td> + <td style="word-wrap: break-word;">50</td> <td>Integer</td> <td>The maximum number of threads running the reconciliation loop. Use -1 for infinite.</td> </tr> diff --git a/docs/layouts/shortcodes/generated/system_section.html b/docs/layouts/shortcodes/generated/system_section.html index aa053c2f..ec303011 100644 --- a/docs/layouts/shortcodes/generated/system_section.html +++ b/docs/layouts/shortcodes/generated/system_section.html @@ -100,7 +100,7 @@ </tr> <tr> <td><h5>kubernetes.operator.reconcile.parallelism</h5></td> - <td style="word-wrap: break-word;">200</td> + <td style="word-wrap: break-word;">50</td> <td>Integer</td> <td>The maximum number of threads running the reconciliation loop. Use -1 for infinite.</td> </tr> diff --git a/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/validation/CrdCompatibilityChecker.java b/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/validation/CrdCompatibilityChecker.java index 658e25aa..faf6024a 100644 --- a/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/validation/CrdCompatibilityChecker.java +++ b/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/validation/CrdCompatibilityChecker.java @@ -93,7 +93,11 @@ public class CrdCompatibilityChecker { // This field was removed from Kubernetes ObjectMeta v1 in 1.25 as it was unused // for a long time. If set for any reason (very unlikely as it does nothing), // the property will be dropped / ignored by the api server. - if (!fieldPath.endsWith(".metadata.clusterName")) { + if (!fieldPath.endsWith(".metadata.clusterName") + // This claims field was removed in Kubernetes 1.28 as it was mistakenly + // added in the first place. For more context please refer to + // https://github.com/kubernetes/api/commit/8b14183 + && !fieldPath.contains(".volumeClaimTemplate.spec.resources.claims")) { err(fieldPath + " has been removed"); } } else { diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java index a5846f59..0ecd7c83 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java +++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java @@ -137,7 +137,7 @@ public class FlinkOperator { overrider.withExecutorService(Executors.newCachedThreadPool()); } else { LOG.info("Configuring operator with {} reconciliation threads.", parallelism); - overrider.withExecutorService(Executors.newFixedThreadPool(parallelism)); + overrider.withConcurrentReconciliationThreads(parallelism); } if (operatorConf.isJosdkMetricsEnabled()) { diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkOperatorConfiguration.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkOperatorConfiguration.java index 9fcdb6a1..225bebd5 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkOperatorConfiguration.java +++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkOperatorConfiguration.java @@ -266,8 +266,7 @@ public class FlinkOperatorConfiguration { null, conf.get(KubernetesOperatorConfigOptions.OPERATOR_LEADER_ELECTION_LEASE_DURATION), conf.get(KubernetesOperatorConfigOptions.OPERATOR_LEADER_ELECTION_RENEW_DEADLINE), - conf.get(KubernetesOperatorConfigOptions.OPERATOR_LEADER_ELECTION_RETRY_PERIOD), - null); + conf.get(KubernetesOperatorConfigOptions.OPERATOR_LEADER_ELECTION_RETRY_PERIOD)); } private static Optional<String> getEnv(String key) { diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/OperatorJosdkMetrics.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/OperatorJosdkMetrics.java index 679aa57c..47165f78 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/OperatorJosdkMetrics.java +++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/OperatorJosdkMetrics.java @@ -205,9 +205,9 @@ public class OperatorJosdkMetrics implements Metrics { Class<? extends AbstractFlinkResource<?, ?>> resourceClass; - if (resourceGvk.kind.equals(FlinkDeployment.class.getSimpleName())) { + if (resourceGvk.getKind().equals(FlinkDeployment.class.getSimpleName())) { resourceClass = FlinkDeployment.class; - } else if (resourceGvk.kind.equals(FlinkSessionJob.class.getSimpleName())) { + } else if (resourceGvk.getKind().equals(FlinkSessionJob.class.getSimpleName())) { resourceClass = FlinkSessionJob.class; } else { return Optional.empty(); diff --git a/flink-kubernetes-operator/src/main/resources/META-INF/NOTICE b/flink-kubernetes-operator/src/main/resources/META-INF/NOTICE index 1b7d2325..bd46972e 100644 --- a/flink-kubernetes-operator/src/main/resources/META-INF/NOTICE +++ b/flink-kubernetes-operator/src/main/resources/META-INF/NOTICE @@ -47,10 +47,10 @@ This project bundles the following dependencies under the Apache Software Licens - io.fabric8:kubernetes-model-scheduling:jar:6.8.1 - io.fabric8:kubernetes-model-storageclass:jar:6.8.1 - io.fabric8:zjsonpatch:jar:0.3.0 -- io.javaoperatorsdk:operator-framework-core:jar:4.4.4 -- io.javaoperatorsdk:operator-framework:jar:4.4.4 +- io.javaoperatorsdk:operator-framework-core:jar:4.8.3 +- io.javaoperatorsdk:operator-framework:jar:4.8.3 - org.apache.commons:commons-compress:1.21 -- org.apache.commons:commons-lang3:3.13.0 +- org.apache.commons:commons-lang3:3.14.0 - org.apache.commons:commons-math3:3.6.1 - org.apache.commons:commons-text:jar:1.10.0 - org.apache.logging.log4j:log4j-1.2-api:2.17.1 diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/FlinkOperatorTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/FlinkOperatorTest.java index 64550964..b60551a9 100644 --- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/FlinkOperatorTest.java +++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/FlinkOperatorTest.java @@ -70,21 +70,10 @@ public class FlinkOperatorTest { var configService = testOperator.getOperator().getConfigurationService(); - // Test parallelism being passed expectedly + // Test parallelism being passed var executorService = configService.getExecutorService(); Assertions.assertInstanceOf(ThreadPoolExecutor.class, executorService); ThreadPoolExecutor threadPoolExecutor = (ThreadPoolExecutor) executorService; - for (int i = 0; i < testParallelism * 2; i++) { - threadPoolExecutor.execute( - () -> { - try { - Thread.sleep(1000); - } catch (InterruptedException e) { - e.printStackTrace(); - } - }); - } - Assertions.assertEquals(threadPoolExecutor.getPoolSize(), testParallelism); Assertions.assertEquals(threadPoolExecutor.getMaximumPoolSize(), testParallelism); // Test label selector being passed diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestUtils.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestUtils.java index 2b2e23a7..62e94988 100644 --- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestUtils.java +++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestUtils.java @@ -51,6 +51,7 @@ import io.fabric8.kubernetes.client.KubernetesClient; import io.fabric8.mockwebserver.utils.ResponseProvider; import io.javaoperatorsdk.operator.api.config.ControllerConfiguration; import io.javaoperatorsdk.operator.api.reconciler.Context; +import io.javaoperatorsdk.operator.api.reconciler.IndexedResourceCache; import io.javaoperatorsdk.operator.api.reconciler.ResourceDiscriminator; import io.javaoperatorsdk.operator.api.reconciler.RetryInfo; import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.ManagedDependentResourceContext; @@ -497,5 +498,10 @@ public class TestUtils extends BaseTestUtils { public ExecutorService getWorkflowExecutorService() { throw new UnsupportedOperationException("Not implemented"); } + + @Override + public IndexedResourceCache<T> getPrimaryCache() { + return null; + } } } 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 8e9683d0..c4f24ee9 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 @@ -1013,7 +1013,7 @@ public class AbstractFlinkServiceTest { namespace, deploymentName); String watchUrl = String.format( - "/apis/apps/v1/namespaces/%s/deployments?fieldSelector=metadata.name%%3D%s&timeoutSeconds=600&allowWatchBookmarks=true&watch=true", + "/apis/apps/v1/namespaces/%s/deployments?allowWatchBookmarks=true&fieldSelector=metadata.name%%3D%s&timeoutSeconds=600&watch=true", namespace, deploymentName); var flinkService = new TestingService(null); diff --git a/pom.xml b/pom.xml index 823dd033..a5a698d5 100644 --- a/pom.xml +++ b/pom.xml @@ -75,7 +75,7 @@ under the License. <maven-javadoc-plugin.version>3.3.2</maven-javadoc-plugin.version> <git-commit-id-maven-plugin.version>5.0.0</git-commit-id-maven-plugin.version> - <operator.sdk.version>4.4.4</operator.sdk.version> + <operator.sdk.version>4.8.3</operator.sdk.version> <operator.sdk.webhook-framework.version>1.1.1</operator.sdk.webhook-framework.version> <fabric8.version>6.8.1</fabric8.version>
