You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gy...@apache.org on 2022/05/12 08:12:03 UTC
[flink-kubernetes-operator] branch main updated: [FLINK-27499] Bump base Flink version to 1.15.0
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 fb51b6a [FLINK-27499] Bump base Flink version to 1.15.0
fb51b6a is described below
commit fb51b6afef489a306efc23cef950d26bbb3ef260
Author: SteNicholas <pr...@163.com>
AuthorDate: Tue May 10 22:11:16 2022 +0800
[FLINK-27499] Bump base Flink version to 1.15.0
---
.github/workflows/ci.yml | 4 +-
README.md | 2 +-
docs/README.md | 2 +-
.../content/docs/custom-resource/job-management.md | 4 +-
docs/content/docs/custom-resource/overview.md | 6 +-
docs/content/docs/custom-resource/pod-template.md | 6 +-
docs/content/docs/operations/ingress.md | 4 +-
e2e-tests/data/flinkdep-cr.yaml | 4 +-
e2e-tests/data/sessionjob-cr.yaml | 4 +-
examples/advanced-ingress.yaml | 4 +-
examples/basic-checkpoint-ha.yaml | 4 +-
examples/basic-ingress.yaml | 4 +-
examples/basic-session-job.yaml | 8 +-
examples/basic-session.yaml | 4 +-
examples/basic.yaml | 4 +-
examples/custom-logging.yaml | 4 +-
examples/pod-template.yaml | 4 +-
flink-kubernetes-operator/pom.xml | 12 +--
.../operator/config/FlinkConfigBuilder.java | 14 +---
.../operator/reconciler/ReconciliationUtils.java | 4 +-
.../kubernetes/operator/service/FlinkService.java | 36 ++++++++-
.../savepoints/SavepointTriggerRequestBody.java | 90 +++++++++++++++++++++
.../stop/StopWithSavepointRequestBody.java | 91 ++++++++++++++++++++++
.../src/main/resources/META-INF/NOTICE | 38 ++++-----
.../kubernetes/operator/FlinkOperatorITCase.java | 4 +-
.../flink/kubernetes/operator/TestUtils.java | 4 +-
.../kubernetes/operator/TestingClusterClient.java | 12 ++-
.../operator/service/FlinkServiceTest.java | 6 +-
flink-kubernetes-shaded/pom.xml | 2 +-
flink-kubernetes-webhook/pom.xml | 2 +-
pom.xml | 3 +-
31 files changed, 294 insertions(+), 96 deletions(-)
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index f2c7056..7b69f82 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -41,7 +41,7 @@ jobs:
- name: Build with Maven
run: |
set -o pipefail; mvn clean install -Pgenerate-docs | tee ./mvn.log; set +o pipefail
- if [[ $(grep -c "overlapping classes" ./mvn.log) -gt 0 ]];then
+ if [[ $(cat ./mvn.log | grep -E -v 'flink-runtime-.*.jar, flink-kubernetes-operator-.*.jar define 2 overlapping classes' | grep -c "overlapping classes" -) -gt 0 ]];then
echo "Found overlapping classes, please fix"
exit 1
fi
@@ -123,7 +123,7 @@ jobs:
- name: Build with Maven
run: |
set -o pipefail; mvn clean install -DskipTests | tee ./mvn.log; set +o pipefail
- if [[ $(grep -c "overlapping classes" ./mvn.log) -gt 0 ]];then
+ if [[ $(cat ./mvn.log | grep -E -v 'flink-runtime-.*.jar, flink-kubernetes-operator-.*.jar define 2 overlapping classes' | grep -c "overlapping classes" -) -gt 0 ]];then
echo "Found overlapping classes, please fix"
exit 1
fi
diff --git a/README.md b/README.md
index 5514682..5d40ab1 100644
--- a/README.md
+++ b/README.md
@@ -30,7 +30,7 @@ If you are currently using the v1alpha1 version of the APIs in your manifests, p
| Operator Version | API Version | Flink Release | Operator Image Tag | Helm Chart Version |
| ------------- | ------------- | ------------- | ------------- | ------------- |
-| `latest` (main HEAD) | `v1beta1` | 1.14 | | |
+| `latest` (main HEAD) | `v1beta1` | 1.15 | | |
| `release-0.1.0` | `v1alpha1` | 1.14 | `2c166e3` | 0.1.0 |
## Support
diff --git a/docs/README.md b/docs/README.md
index c5419e9..618dcc4 100644
--- a/docs/README.md
+++ b/docs/README.md
@@ -98,7 +98,7 @@ should copy into their pom.xml file. It will render out to:
```xml
<dependency>
<groupdId>org.apache.flink</groupId>
- <artifactId>flink-streaming-java_2.12</artifactId>
+ <artifactId>flink-streaming-java</artifactId>
<version><!-- current flink version --></version>
</dependency>
```
diff --git a/docs/content/docs/custom-resource/job-management.md b/docs/content/docs/custom-resource/job-management.md
index af8494f..a095887 100644
--- a/docs/content/docs/custom-resource/job-management.md
+++ b/docs/content/docs/custom-resource/job-management.md
@@ -102,8 +102,8 @@ metadata:
namespace: default
name: basic-checkpoint-ha-example
spec:
- image: flink:1.14
- flinkVersion: v1_14
+ image: flink:1.15
+ flinkVersion: v1_15
flinkConfiguration:
taskmanager.numberOfTaskSlots: "2"
state.savepoints.dir: file:///flink-data/savepoints
diff --git a/docs/content/docs/custom-resource/overview.md b/docs/content/docs/custom-resource/overview.md
index 8d95dca..a920b54 100644
--- a/docs/content/docs/custom-resource/overview.md
+++ b/docs/content/docs/custom-resource/overview.md
@@ -82,7 +82,7 @@ The spec contains all the information the operator need to deploy and manage you
Most deployments will define at least the following fields:
- `image` : Docker used to run Flink job and task manager processes
- - `flinkVersion` : Flink version used in the image (`v1_14`, `v1_15`...)
+ - `flinkVersion` : Flink version used in the image (`v1_13`, `v1_14`, `v1_15`...)
- `serviceAccount` : Kubernetes service account used by the Flink pods
- `taskManager, jobManager` : Job and Task manager pod resource specs (cpu, memory, etc.)
- `flinkConfiguration` : Map of Flink configuration overrides such as HA and checkpointing configs
@@ -111,8 +111,8 @@ metadata:
namespace: default
name: basic-example
spec:
- image: flink:1.14
- flinkVersion: v1_14
+ image: flink:1.15
+ flinkVersion: v1_15
flinkConfiguration:
taskmanager.numberOfTaskSlots: "2"
serviceAccount: flink
diff --git a/docs/content/docs/custom-resource/pod-template.md b/docs/content/docs/custom-resource/pod-template.md
index 05f5bd8..4d49908 100644
--- a/docs/content/docs/custom-resource/pod-template.md
+++ b/docs/content/docs/custom-resource/pod-template.md
@@ -49,8 +49,8 @@ metadata:
namespace: default
name: pod-template-example
spec:
- image: flink:1.14
- flinkVersion: v1_14
+ image: flink:1.15
+ flinkVersion: v1_15
flinkConfiguration:
taskmanager.numberOfTaskSlots: "2"
podTemplate:
@@ -102,5 +102,5 @@ spec:
{{< hint info >}}
When using the operator with Flink native Kubernetes integration, please refer to [pod template field precedence](
-https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/deployment/resource-providers/native_kubernetes/#fields-overwritten-by-flink).
+https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/deployment/resource-providers/native_kubernetes/#fields-overwritten-by-flink).
{{< /hint >}}
diff --git a/docs/content/docs/operations/ingress.md b/docs/content/docs/operations/ingress.md
index 38ff2b8..2b981d7 100644
--- a/docs/content/docs/operations/ingress.md
+++ b/docs/content/docs/operations/ingress.md
@@ -35,8 +35,8 @@ metadata:
namespace: default
name: advanced-ingress
spec:
- image: flink:1.14
- flinkVersion: v1_14
+ image: flink:1.15
+ flinkVersion: v1_15
ingress:
template: "flink.k8s.io/{{namespace}}/{{name}}(/|$)(.*)"
className: "nginx"
diff --git a/e2e-tests/data/flinkdep-cr.yaml b/e2e-tests/data/flinkdep-cr.yaml
index 76930cd..d48af2b 100644
--- a/e2e-tests/data/flinkdep-cr.yaml
+++ b/e2e-tests/data/flinkdep-cr.yaml
@@ -22,8 +22,8 @@ metadata:
namespace: default
name: flink-example-statemachine
spec:
- image: flink:1.14
- flinkVersion: v1_14
+ image: flink:1.15
+ flinkVersion: v1_15
ingress:
template: "/{{namespace}}/{{name}}(/|$)(.*)"
className: "nginx"
diff --git a/e2e-tests/data/sessionjob-cr.yaml b/e2e-tests/data/sessionjob-cr.yaml
index 8b821d7..53b49f0 100644
--- a/e2e-tests/data/sessionjob-cr.yaml
+++ b/e2e-tests/data/sessionjob-cr.yaml
@@ -22,8 +22,8 @@ metadata:
namespace: default
name: session-cluster-1
spec:
- image: flink:1.14
- flinkVersion: v1_14
+ image: flink:1.15
+ flinkVersion: v1_15
ingress:
template: "/{{namespace}}/{{name}}(/|$)(.*)"
className: "nginx"
diff --git a/examples/advanced-ingress.yaml b/examples/advanced-ingress.yaml
index 02e0ad7..62062d6 100644
--- a/examples/advanced-ingress.yaml
+++ b/examples/advanced-ingress.yaml
@@ -21,8 +21,8 @@ kind: FlinkDeployment
metadata:
name: advanced-ingress
spec:
- image: flink:1.14
- flinkVersion: v1_14
+ image: flink:1.15
+ flinkVersion: v1_15
ingress:
template: "/{{namespace}}/{{name}}(/|$)(.*)"
className: "nginx"
diff --git a/examples/basic-checkpoint-ha.yaml b/examples/basic-checkpoint-ha.yaml
index 60bf5d7..7454998 100644
--- a/examples/basic-checkpoint-ha.yaml
+++ b/examples/basic-checkpoint-ha.yaml
@@ -21,8 +21,8 @@ kind: FlinkDeployment
metadata:
name: basic-checkpoint-ha-example
spec:
- image: flink:1.14
- flinkVersion: v1_14
+ image: flink:1.15
+ flinkVersion: v1_15
flinkConfiguration:
taskmanager.numberOfTaskSlots: "2"
state.savepoints.dir: file:///flink-data/savepoints
diff --git a/examples/basic-ingress.yaml b/examples/basic-ingress.yaml
index 223c5da..52e3632 100644
--- a/examples/basic-ingress.yaml
+++ b/examples/basic-ingress.yaml
@@ -21,8 +21,8 @@ kind: FlinkDeployment
metadata:
name: basic-ingress
spec:
- image: flink:1.14
- flinkVersion: v1_14
+ image: flink:1.15
+ flinkVersion: v1_15
ingress:
template: "{{name}}.{{namespace}}.flink.k8s.io"
flinkConfiguration:
diff --git a/examples/basic-session-job.yaml b/examples/basic-session-job.yaml
index 8feeddb..f4a56a0 100644
--- a/examples/basic-session-job.yaml
+++ b/examples/basic-session-job.yaml
@@ -21,8 +21,8 @@ kind: FlinkDeployment
metadata:
name: basic-session-cluster
spec:
- image: flink:1.14
- flinkVersion: v1_14
+ image: flink:1.15
+ flinkVersion: v1_15
jobManager:
resource:
memory: "2048m"
@@ -40,7 +40,7 @@ metadata:
spec:
deploymentName: basic-session-cluster
job:
- jarURI: https://repo1.maven.org/maven2/org/apache/flink/flink-examples-streaming_2.12/1.14.4/flink-examples-streaming_2.12-1.14.4-TopSpeedWindowing.jar
+ jarURI: https://repo1.maven.org/maven2/org/apache/flink/flink-examples-streaming_2.12/1.15.0/flink-examples-streaming_2.12-1.15.0-TopSpeedWindowing.jar
parallelism: 4
upgradeMode: stateless
@@ -52,7 +52,7 @@ metadata:
spec:
deploymentName: basic-session-cluster
job:
- jarURI: https://repo1.maven.org/maven2/org/apache/flink/flink-examples-streaming_2.12/1.14.4/flink-examples-streaming_2.12-1.14.4.jar
+ jarURI: https://repo1.maven.org/maven2/org/apache/flink/flink-examples-streaming_2.12/1.15.0/flink-examples-streaming_2.12-1.15.0.jar
parallelism: 2
upgradeMode: stateless
entryClass: org.apache.flink.streaming.examples.statemachine.StateMachineExample
diff --git a/examples/basic-session.yaml b/examples/basic-session.yaml
index 8ef2281..8bb55d5 100644
--- a/examples/basic-session.yaml
+++ b/examples/basic-session.yaml
@@ -21,8 +21,8 @@ kind: FlinkDeployment
metadata:
name: basic-session-example
spec:
- image: flink:1.14
- flinkVersion: v1_14
+ image: flink:1.15
+ flinkVersion: v1_15
flinkConfiguration:
taskmanager.numberOfTaskSlots: "2"
serviceAccount: flink
diff --git a/examples/basic.yaml b/examples/basic.yaml
index c08b703..d63861d 100644
--- a/examples/basic.yaml
+++ b/examples/basic.yaml
@@ -21,8 +21,8 @@ kind: FlinkDeployment
metadata:
name: basic-example
spec:
- image: flink:1.14
- flinkVersion: v1_14
+ image: flink:1.15
+ flinkVersion: v1_15
flinkConfiguration:
taskmanager.numberOfTaskSlots: "2"
serviceAccount: flink
diff --git a/examples/custom-logging.yaml b/examples/custom-logging.yaml
index a88a282..5122049 100644
--- a/examples/custom-logging.yaml
+++ b/examples/custom-logging.yaml
@@ -21,8 +21,8 @@ kind: FlinkDeployment
metadata:
name: custom-logging-example
spec:
- image: flink:1.14
- flinkVersion: v1_14
+ image: flink:1.15
+ flinkVersion: v1_15
flinkConfiguration:
taskmanager.numberOfTaskSlots: "2"
serviceAccount: flink
diff --git a/examples/pod-template.yaml b/examples/pod-template.yaml
index e1620fa..f6ec08c 100644
--- a/examples/pod-template.yaml
+++ b/examples/pod-template.yaml
@@ -21,8 +21,8 @@ kind: FlinkDeployment
metadata:
name: pod-template-example
spec:
- image: flink:1.14
- flinkVersion: v1_14
+ image: flink:1.15
+ flinkVersion: v1_15
flinkConfiguration:
taskmanager.numberOfTaskSlots: "2"
podTemplate:
diff --git a/flink-kubernetes-operator/pom.xml b/flink-kubernetes-operator/pom.xml
index 639d772..2147304 100644
--- a/flink-kubernetes-operator/pom.xml
+++ b/flink-kubernetes-operator/pom.xml
@@ -59,19 +59,13 @@ under the License.
<!-- Flink -->
<dependency>
<groupId>org.apache.flink</groupId>
- <artifactId>flink-clients_${scala.version}</artifactId>
+ <artifactId>flink-clients</artifactId>
<version>${flink.version}</version>
- <exclusions>
- <exclusion>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-scala_${scala.version}</artifactId>
- </exclusion>
- </exclusions>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
- <artifactId>flink-runtime-web_${scala.version}</artifactId>
+ <artifactId>flink-runtime-web</artifactId>
<version>${flink.version}</version>
</dependency>
@@ -138,7 +132,7 @@ under the License.
<dependency>
<groupId>org.apache.flink</groupId>
- <artifactId>flink-test-utils_${scala.version}</artifactId>
+ <artifactId>flink-test-utils</artifactId>
<version>${flink.version}</version>
<scope>test</scope>
</dependency>
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigBuilder.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigBuilder.java
index c8053f6..2a768e4 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigBuilder.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigBuilder.java
@@ -53,6 +53,8 @@ import java.nio.file.Files;
import java.time.Duration;
import java.util.Collections;
+import static org.apache.flink.configuration.DeploymentOptions.SHUTDOWN_ON_APPLICATION_FINISH;
+import static org.apache.flink.configuration.DeploymentOptions.SUBMIT_FAILED_JOB_ON_APPLICATION_ERROR;
import static org.apache.flink.configuration.DeploymentOptionsInternal.CONF_DIR;
import static org.apache.flink.configuration.WebOptions.CANCEL_ENABLE;
import static org.apache.flink.kubernetes.configuration.KubernetesConfigOptions.REST_SERVICE_EXPOSED_TYPE;
@@ -65,18 +67,6 @@ public class FlinkConfigBuilder {
private static final Logger LOG = LoggerFactory.getLogger(FlinkConfigBuilder.class);
- // Flink 1.15+ specific config options, should be removed once the Flink base dependency is
- // bumped
- private static final ConfigOption<Boolean> SHUTDOWN_ON_APPLICATION_FINISH =
- ConfigOptions.key("execution.shutdown-on-application-finish")
- .booleanType()
- .noDefaultValue();
-
- public static final ConfigOption<Boolean> SUBMIT_FAILED_JOB_ON_APPLICATION_ERROR =
- ConfigOptions.key("execution.submit-failed-job-on-application-error")
- .booleanType()
- .noDefaultValue();
-
public static final ConfigOption<FlinkVersion> FLINK_VERSION =
ConfigOptions.key("$internal.flink.version")
.enumType(FlinkVersion.class)
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/ReconciliationUtils.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/ReconciliationUtils.java
index 6440776..315eee5 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/ReconciliationUtils.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/ReconciliationUtils.java
@@ -274,9 +274,7 @@ public class ReconciliationUtils {
KubernetesOperatorConfigOptions.OPERATOR_RECOVER_JM_DEPLOYMENT_ENABLED)
.orElse(
conf.get(FlinkConfigBuilder.FLINK_VERSION)
- .isNewerVersionThan(FlinkVersion.v1_14)
- ? true
- : false);
+ .isNewerVersionThan(FlinkVersion.v1_14));
}
public static boolean jmMissingForRunningDeployment(FlinkDeploymentStatus status) {
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/FlinkService.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/FlinkService.java
index 2b61e09..8367e34 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/FlinkService.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/FlinkService.java
@@ -32,12 +32,14 @@ import org.apache.flink.client.program.rest.RestClusterClient;
import org.apache.flink.configuration.CheckpointingOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
+import org.apache.flink.core.execution.SavepointFormatType;
import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;
import org.apache.flink.kubernetes.kubeclient.decorators.ExternalServiceDecorator;
import org.apache.flink.kubernetes.operator.artifact.ArtifactManager;
import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
import org.apache.flink.kubernetes.operator.crd.FlinkSessionJob;
+import org.apache.flink.kubernetes.operator.crd.spec.FlinkVersion;
import org.apache.flink.kubernetes.operator.crd.spec.JobSpec;
import org.apache.flink.kubernetes.operator.crd.spec.UpgradeMode;
import org.apache.flink.kubernetes.operator.crd.status.FlinkDeploymentStatus;
@@ -48,6 +50,7 @@ import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
import org.apache.flink.kubernetes.operator.utils.FlinkUtils;
import org.apache.flink.runtime.client.JobStatusMessage;
import org.apache.flink.runtime.highavailability.nonha.standalone.StandaloneClientHAServices;
+import org.apache.flink.runtime.jobgraph.RestoreMode;
import org.apache.flink.runtime.rest.FileUpload;
import org.apache.flink.runtime.rest.RestClient;
import org.apache.flink.runtime.rest.handler.async.AsynchronousOperationResult;
@@ -102,6 +105,8 @@ import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import static org.apache.flink.kubernetes.operator.config.FlinkConfigBuilder.FLINK_VERSION;
+
/** Service for submitting and interacting with Flink clusters and jobs. */
public class FlinkService {
@@ -192,7 +197,10 @@ public class FlinkService {
job.getParallelism() > 0 ? job.getParallelism() : null,
jobID,
job.getAllowNonRestoredState(),
- savepoint);
+ savepoint,
+ conf.get(FLINK_VERSION).isNewerVersionThan(FlinkVersion.v1_14)
+ ? RestoreMode.DEFAULT
+ : null);
LOG.info("Submitting job: {} to session cluster.", jobID.toHexString());
return clusterClient
.sendRequest(headers, parameters, runRequestBody)
@@ -349,7 +357,12 @@ public class FlinkService {
.stopWithSavepoint(
Preconditions.checkNotNull(jobId),
false,
- savepointDirectory)
+ savepointDirectory,
+ conf.get(FLINK_VERSION)
+ .isNewerVersionThan(
+ FlinkVersion.v1_14)
+ ? SavepointFormatType.DEFAULT
+ : null)
.get(timeout, TimeUnit.SECONDS);
savepointOpt = Optional.of(savepoint);
} else if (ReconciliationUtils.isJobInTerminalState(deploymentStatus)) {
@@ -447,7 +460,15 @@ public class FlinkService {
try {
String savepoint =
clusterClient
- .stopWithSavepoint(jobID, false, savepointDirectory)
+ .stopWithSavepoint(
+ jobID,
+ false,
+ savepointDirectory,
+ conf.get(FLINK_VERSION)
+ .isNewerVersionThan(
+ FlinkVersion.v1_14)
+ ? SavepointFormatType.DEFAULT
+ : null)
.get(timeout, TimeUnit.SECONDS);
savepointOpt = Optional.of(savepoint);
} catch (TimeoutException exception) {
@@ -491,7 +512,14 @@ public class FlinkService {
.sendRequest(
savepointTriggerHeaders,
savepointTriggerMessageParameters,
- new SavepointTriggerRequestBody(savepointDirectory, false))
+ new SavepointTriggerRequestBody(
+ savepointDirectory,
+ false,
+ conf.get(FLINK_VERSION)
+ .isNewerVersionThan(FlinkVersion.v1_14)
+ ? SavepointFormatType.DEFAULT
+ : null,
+ null))
.get(timeout, TimeUnit.SECONDS);
LOG.info("Savepoint successfully triggered: " + response.getTriggerId().toHexString());
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/runtime/rest/messages/job/savepoints/SavepointTriggerRequestBody.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/runtime/rest/messages/job/savepoints/SavepointTriggerRequestBody.java
new file mode 100644
index 0000000..edb1973
--- /dev/null
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/runtime/rest/messages/job/savepoints/SavepointTriggerRequestBody.java
@@ -0,0 +1,90 @@
+/*
+ * 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.job.savepoints;
+
+import org.apache.flink.core.execution.SavepointFormatType;
+import org.apache.flink.runtime.rest.messages.RequestBody;
+import org.apache.flink.runtime.rest.messages.TriggerId;
+
+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.JsonInclude;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import javax.annotation.Nullable;
+
+import java.util.Optional;
+
+/** {@link RequestBody} to trigger savepoints. */
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class SavepointTriggerRequestBody implements RequestBody {
+
+ public static final String FIELD_NAME_TARGET_DIRECTORY = "target-directory";
+
+ private static final String FIELD_NAME_CANCEL_JOB = "cancel-job";
+
+ private static final String FIELD_NAME_TRIGGER_ID = "triggerId";
+ private static final String FIELD_NAME_FORMAT_TYPE = "formatType";
+
+ @JsonProperty(FIELD_NAME_TARGET_DIRECTORY)
+ @Nullable
+ private final String targetDirectory;
+
+ @JsonProperty(FIELD_NAME_CANCEL_JOB)
+ private final boolean cancelJob;
+
+ @JsonProperty(FIELD_NAME_TRIGGER_ID)
+ @Nullable
+ private final TriggerId triggerId;
+
+ @JsonProperty(FIELD_NAME_FORMAT_TYPE)
+ @Nullable
+ private final SavepointFormatType formatType;
+
+ @JsonCreator
+ public SavepointTriggerRequestBody(
+ @Nullable @JsonProperty(FIELD_NAME_TARGET_DIRECTORY) final String targetDirectory,
+ @Nullable @JsonProperty(FIELD_NAME_CANCEL_JOB) final Boolean cancelJob,
+ @Nullable @JsonProperty(FIELD_NAME_FORMAT_TYPE) final SavepointFormatType formatType,
+ @Nullable @JsonProperty(FIELD_NAME_TRIGGER_ID) TriggerId triggerId) {
+ this.targetDirectory = targetDirectory;
+ this.cancelJob = cancelJob != null ? cancelJob : false;
+ this.triggerId = triggerId;
+ this.formatType = formatType;
+ }
+
+ @JsonIgnore
+ public Optional<String> getTargetDirectory() {
+ return Optional.ofNullable(targetDirectory);
+ }
+
+ @JsonIgnore
+ public Optional<TriggerId> getTriggerId() {
+ return Optional.ofNullable(triggerId);
+ }
+
+ public boolean isCancelJob() {
+ return cancelJob;
+ }
+
+ @JsonIgnore
+ public SavepointFormatType getFormatType() {
+ return formatType;
+ }
+}
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/runtime/rest/messages/job/savepoints/stop/StopWithSavepointRequestBody.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/runtime/rest/messages/job/savepoints/stop/StopWithSavepointRequestBody.java
new file mode 100644
index 0000000..45df5e1
--- /dev/null
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/runtime/rest/messages/job/savepoints/stop/StopWithSavepointRequestBody.java
@@ -0,0 +1,91 @@
+/*
+ * 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.job.savepoints.stop;
+
+import org.apache.flink.core.execution.SavepointFormatType;
+import org.apache.flink.runtime.rest.messages.RequestBody;
+import org.apache.flink.runtime.rest.messages.TriggerId;
+
+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.JsonInclude;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import javax.annotation.Nullable;
+
+import java.util.Optional;
+
+/** {@link RequestBody} for stopping a job with a savepoint. */
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class StopWithSavepointRequestBody implements RequestBody {
+
+ public static final String FIELD_NAME_TARGET_DIRECTORY = "targetDirectory";
+
+ public static final String FIELD_NAME_FORMAT_TYPE = "formatType";
+
+ private static final String FIELD_NAME_DRAIN = "drain";
+
+ private static final String FIELD_NAME_TRIGGER_ID = "triggerId";
+
+ @JsonProperty(FIELD_NAME_TARGET_DIRECTORY)
+ @Nullable
+ private final String targetDirectory;
+
+ @JsonProperty(FIELD_NAME_DRAIN)
+ private final boolean drain;
+
+ @JsonProperty(FIELD_NAME_TRIGGER_ID)
+ @Nullable
+ private final TriggerId triggerId;
+
+ @JsonProperty(FIELD_NAME_FORMAT_TYPE)
+ @Nullable
+ private final SavepointFormatType formatType;
+
+ @JsonCreator
+ public StopWithSavepointRequestBody(
+ @Nullable @JsonProperty(FIELD_NAME_TARGET_DIRECTORY) final String targetDirectory,
+ @Nullable @JsonProperty(FIELD_NAME_DRAIN) final Boolean drain,
+ @Nullable @JsonProperty(FIELD_NAME_FORMAT_TYPE) final SavepointFormatType formatType,
+ @Nullable @JsonProperty(FIELD_NAME_TRIGGER_ID) TriggerId triggerId) {
+ this.targetDirectory = targetDirectory;
+ this.drain = drain != null ? drain : false;
+ this.triggerId = triggerId;
+ this.formatType = formatType;
+ }
+
+ @JsonIgnore
+ public Optional<String> getTargetDirectory() {
+ return Optional.ofNullable(targetDirectory);
+ }
+
+ public boolean shouldDrain() {
+ return drain;
+ }
+
+ @JsonIgnore
+ public Optional<TriggerId> getTriggerId() {
+ return Optional.ofNullable(triggerId);
+ }
+
+ @JsonIgnore
+ public SavepointFormatType getFormatType() {
+ return formatType;
+ }
+}
diff --git a/flink-kubernetes-operator/src/main/resources/META-INF/NOTICE b/flink-kubernetes-operator/src/main/resources/META-INF/NOTICE
index 4072483..2663458 100644
--- a/flink-kubernetes-operator/src/main/resources/META-INF/NOTICE
+++ b/flink-kubernetes-operator/src/main/resources/META-INF/NOTICE
@@ -70,25 +70,25 @@ This project bundles the following dependencies under the Apache Software Licens
- org.apache.commons:commons-compress:1.21
- org.apache.commons:commons-lang3:3.12.0
- org.apache.commons:commons-math3:3.5
-- org.apache.flink:flink-annotations:1.14.4
-- org.apache.flink:flink-clients_2.12:1.14.4
-- org.apache.flink:flink-core:1.14.4
-- org.apache.flink:flink-file-sink-common:1.14.4
-- org.apache.flink:flink-hadoop-fs:1.14.4
-- org.apache.flink:flink-java:1.14.4
-- org.apache.flink:flink-metrics-core:1.14.4
-- org.apache.flink:flink-optimizer:1.14.4
-- org.apache.flink:flink-queryable-state-client-java:1.14.4
-- org.apache.flink:flink-rpc-akka-loader:1.14.4
-- org.apache.flink:flink-rpc-core:1.14.4
-- org.apache.flink:flink-runtime:1.14.4
-- org.apache.flink:flink-shaded-asm-7:7.1-14.0
-- org.apache.flink:flink-shaded-force-shading:14.0
-- org.apache.flink:flink-shaded-guava:30.1.1-jre-14.0
-- org.apache.flink:flink-shaded-jackson:2.12.4-14.0
-- org.apache.flink:flink-shaded-netty:4.1.65.Final-14.0
-- org.apache.flink:flink-shaded-zookeeper-3:3.4.14-14.0
-- org.apache.flink:flink-streaming-java_2.12:1.14.4
+- org.apache.flink:flink-annotations:1.15.0
+- org.apache.flink:flink-clients:1.15.0
+- org.apache.flink:flink-core:1.15.0
+- org.apache.flink:flink-file-sink-common:1.15.0
+- org.apache.flink:flink-hadoop-fs:1.15.0
+- org.apache.flink:flink-java:1.15.0
+- org.apache.flink:flink-metrics-core:1.15.0
+- org.apache.flink:flink-optimizer:1.15.0
+- org.apache.flink:flink-queryable-state-client-java:1.15.0
+- org.apache.flink:flink-rpc-akka-loader:1.15.0
+- org.apache.flink:flink-rpc-core:1.15.0
+- org.apache.flink:flink-runtime:1.15.0
+- org.apache.flink:flink-shaded-asm-9:9.2-15.0
+- org.apache.flink:flink-shaded-force-shading:15.0
+- org.apache.flink:flink-shaded-guava:30.1.1-jre-15.0
+- org.apache.flink:flink-shaded-jackson:2.12.4-15.0
+- org.apache.flink:flink-shaded-netty:4.1.70.Final-15.0
+- org.apache.flink:flink-shaded-zookeeper-3:3.5.9-15.0
+- org.apache.flink:flink-streaming-java:1.15.0
- org.apache.logging.log4j:log4j-1.2-api:2.17.1
- org.apache.logging.log4j:log4j-api:2.17.1
- org.apache.logging.log4j:log4j-core:2.17.1
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/FlinkOperatorITCase.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/FlinkOperatorITCase.java
index e94c7b6..d96847d 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/FlinkOperatorITCase.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/FlinkOperatorITCase.java
@@ -50,7 +50,7 @@ public class FlinkOperatorITCase {
private static final String TEST_NAMESPACE = "flink-operator-test";
private static final String SERVICE_ACCOUNT = "flink-operator";
private static final String CLUSTER_ROLE_BINDING = "flink-operator-cluster-role-binding";
- private static final String FLINK_VERSION = "1.14.4";
+ private static final String FLINK_VERSION = "1.15.0";
private static final String IMAGE = String.format("flink:%s", FLINK_VERSION);
private static final Logger LOG = LoggerFactory.getLogger(FlinkOperatorITCase.class);
private KubernetesClient client;
@@ -107,7 +107,7 @@ public class FlinkOperatorITCase {
.build());
FlinkDeploymentSpec spec = new FlinkDeploymentSpec();
spec.setImage(IMAGE);
- spec.setFlinkVersion(FlinkVersion.v1_14);
+ spec.setFlinkVersion(FlinkVersion.v1_15);
spec.setServiceAccount(SERVICE_ACCOUNT);
Resource resource = new Resource();
resource.setMemory("2048m");
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 6efbd5d..ef763b9 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
@@ -89,7 +89,7 @@ public class TestUtils {
public static final String SAMPLE_JAR = "local:///tmp/sample.jar";
public static FlinkDeployment buildSessionCluster() {
- return buildSessionCluster(FlinkVersion.v1_14);
+ return buildSessionCluster(FlinkVersion.v1_15);
}
public static FlinkDeployment buildSessionCluster(FlinkVersion version) {
@@ -119,7 +119,7 @@ public class TestUtils {
}
public static FlinkDeployment buildApplicationCluster() {
- return buildApplicationCluster(FlinkVersion.v1_14);
+ return buildApplicationCluster(FlinkVersion.v1_15);
}
public static FlinkSessionJob buildSessionJob() {
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingClusterClient.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingClusterClient.java
index bc83d59..b5942e8 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingClusterClient.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingClusterClient.java
@@ -22,6 +22,7 @@ import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobStatus;
import org.apache.flink.client.program.rest.RestClusterClient;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.execution.SavepointFormatType;
import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;
import org.apache.flink.runtime.client.JobStatusMessage;
import org.apache.flink.runtime.highavailability.nonha.standalone.StandaloneClientHAServices;
@@ -165,19 +166,22 @@ public class TestingClusterClient<T> extends RestClusterClient<T> {
@Override
public CompletableFuture<String> cancelWithSavepoint(
- JobID jobId, @Nullable String savepointDirectory) {
+ JobID jobId, @Nullable String savepointDirectory, SavepointFormatType formatType) {
throw new UnsupportedOperationException();
}
@Override
public CompletableFuture<String> stopWithSavepoint(
- JobID jobId, boolean advanceToEndOfEventTime, @Nullable String savepointDirectory) {
- return stopWithSavepointFunction.apply(jobId, advanceToEndOfEventTime, savepointDirectory);
+ JobID jobId,
+ boolean advanceToEndOfTime,
+ @Nullable String savepointDirectory,
+ SavepointFormatType formatType) {
+ return stopWithSavepointFunction.apply(jobId, advanceToEndOfTime, savepointDirectory);
}
@Override
public CompletableFuture<String> triggerSavepoint(
- JobID jobId, @Nullable String savepointDirectory) {
+ JobID jobId, @Nullable String savepointDirectory, SavepointFormatType formatType) {
throw new UnsupportedOperationException();
}
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/FlinkServiceTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/FlinkServiceTest.java
index 0118fe1..d944614 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/FlinkServiceTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/FlinkServiceTest.java
@@ -29,6 +29,7 @@ import org.apache.flink.kubernetes.operator.TestUtils;
import org.apache.flink.kubernetes.operator.TestingClusterClient;
import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
+import org.apache.flink.kubernetes.operator.crd.spec.FlinkVersion;
import org.apache.flink.kubernetes.operator.crd.spec.UpgradeMode;
import org.apache.flink.kubernetes.operator.crd.status.JobManagerDeploymentStatus;
import org.apache.flink.kubernetes.operator.crd.status.JobStatus;
@@ -47,6 +48,7 @@ import org.junit.jupiter.api.Test;
import java.util.concurrent.CompletableFuture;
+import static org.apache.flink.kubernetes.operator.config.FlinkConfigBuilder.FLINK_VERSION;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -63,6 +65,7 @@ public class FlinkServiceTest {
public void setup() {
configuration.set(KubernetesConfigOptions.CLUSTER_ID, TestUtils.TEST_DEPLOYMENT_NAME);
configuration.set(KubernetesConfigOptions.NAMESPACE, TestUtils.TEST_NAMESPACE);
+ configuration.set(FLINK_VERSION, FlinkVersion.v1_15);
}
@Test
@@ -176,7 +179,8 @@ public class FlinkServiceTest {
((SavepointTriggerMessageParameters) parameters)
.jobID.getValue(),
((SavepointTriggerRequestBody) requestBody)
- .getTargetDirectory(),
+ .getTargetDirectory()
+ .get(),
((SavepointTriggerRequestBody) requestBody).isCancelJob()));
return CompletableFuture.completedFuture(new TriggerResponse(new TriggerId()));
});
diff --git a/flink-kubernetes-shaded/pom.xml b/flink-kubernetes-shaded/pom.xml
index 9a3002a..459c973 100644
--- a/flink-kubernetes-shaded/pom.xml
+++ b/flink-kubernetes-shaded/pom.xml
@@ -34,7 +34,7 @@ under the License.
<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
- <artifactId>flink-kubernetes_${scala.version}</artifactId>
+ <artifactId>flink-kubernetes</artifactId>
<version>${flink.version}</version>
<exclusions>
<exclusion>
diff --git a/flink-kubernetes-webhook/pom.xml b/flink-kubernetes-webhook/pom.xml
index fe27466..931ca78 100644
--- a/flink-kubernetes-webhook/pom.xml
+++ b/flink-kubernetes-webhook/pom.xml
@@ -48,7 +48,7 @@ under the License.
<dependency>
<groupId>org.apache.flink</groupId>
- <artifactId>flink-kubernetes_${scala.version}</artifactId>
+ <artifactId>flink-kubernetes</artifactId>
<version>${flink.version}</version>
<scope>test</scope>
</dependency>
diff --git a/pom.xml b/pom.xml
index e2a6cb7..ad2ae58 100644
--- a/pom.xml
+++ b/pom.xml
@@ -70,8 +70,7 @@ under the License.
<fabric8.version>5.12.1</fabric8.version>
<lombok.version>1.18.22</lombok.version>
- <scala.version>2.12</scala.version>
- <flink.version>1.14.4</flink.version>
+ <flink.version>1.15.0</flink.version>
<flink.shaded.version>15.0</flink.shaded.version>
<slf4j.version>1.7.36</slf4j.version>