You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by wa...@apache.org on 2022/04/24 08:44:52 UTC
[flink-kubernetes-operator] branch main updated: [FLINK-27360] Rename clusterId field of FlinkSessionJobSpec to deploymentName
This is an automated email from the ASF dual-hosted git repository.
wangyang0918 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 4559495 [FLINK-27360] Rename clusterId field of FlinkSessionJobSpec to deploymentName
4559495 is described below
commit 4559495fc554595c03f4569fa5fd52ff214967e3
Author: Aitozi <yu...@alibaba-inc.com>
AuthorDate: Sun Apr 24 14:36:52 2022 +0800
[FLINK-27360] Rename clusterId field of FlinkSessionJobSpec to deploymentName
---
docs/content/docs/custom-resource/reference.md | 2 +-
.../apache/flink/kubernetes/operator/artifact/ArtifactManager.java | 2 +-
.../kubernetes/operator/controller/FlinkSessionJobController.java | 5 +++--
.../flink/kubernetes/operator/crd/spec/FlinkSessionJobSpec.java | 7 ++-----
.../org/apache/flink/kubernetes/operator/service/FlinkService.java | 2 +-
.../flink/kubernetes/operator/validation/DefaultValidator.java | 2 +-
.../test/java/org/apache/flink/kubernetes/operator/TestUtils.java | 2 +-
.../flink/kubernetes/operator/validation/DefaultValidatorTest.java | 2 +-
.../crds/flinksessionjobs.flink.apache.org-v1.yml | 2 +-
9 files changed, 12 insertions(+), 14 deletions(-)
diff --git a/docs/content/docs/custom-resource/reference.md b/docs/content/docs/custom-resource/reference.md
index adcf44c..b56d8f9 100644
--- a/docs/content/docs/custom-resource/reference.md
+++ b/docs/content/docs/custom-resource/reference.md
@@ -73,7 +73,7 @@ This page serves as a full reference for FlinkDeployment custom resource definit
| Parameter | Type | Docs |
| ----------| ---- | ---- |
-| clusterId | java.lang.String | The cluster id of the target session cluster. When deployed using the operator the cluster id is the name of the deployment. |
+| deploymentName | java.lang.String | The name of the target session cluster deployment. |
### FlinkVersion
**Class**: org.apache.flink.kubernetes.operator.crd.spec.FlinkVersion
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/artifact/ArtifactManager.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/artifact/ArtifactManager.java
index 2ed8114..bdbfed8 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/artifact/ArtifactManager.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/artifact/ArtifactManager.java
@@ -67,7 +67,7 @@ public class ArtifactManager {
new String[] {
baseDir.getAbsolutePath(),
sessionJob.getMetadata().getNamespace(),
- sessionJob.getSpec().getClusterId(),
+ sessionJob.getSpec().getDeploymentName(),
sessionJob.getMetadata().getName()
});
}
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkSessionJobController.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkSessionJobController.java
index d140cd7..608c7ba 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkSessionJobController.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkSessionJobController.java
@@ -172,7 +172,7 @@ public class FlinkSessionJobController
primaryResourceRetriever(),
sessionJob ->
new ResourceID(
- sessionJob.getSpec().getClusterId(),
+ sessionJob.getSpec().getDeploymentName(),
sessionJob.getMetadata().getNamespace()),
false) {
@Override
@@ -245,7 +245,8 @@ public class FlinkSessionJobController
}
private Map<String, Function<FlinkSessionJob, List<String>>> clusterToSessionJobIndexer() {
- return Map.of(CLUSTER_ID_INDEX, sessionJob -> List.of(sessionJob.getSpec().getClusterId()));
+ return Map.of(
+ CLUSTER_ID_INDEX, sessionJob -> List.of(sessionJob.getSpec().getDeploymentName()));
}
private Optional<String> validateSessionJob(FlinkSessionJob sessionJob, Context context) {
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/FlinkSessionJobSpec.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/FlinkSessionJobSpec.java
index 82714a1..80399f2 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/FlinkSessionJobSpec.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/FlinkSessionJobSpec.java
@@ -36,9 +36,6 @@ import lombok.experimental.SuperBuilder;
@SuperBuilder
public class FlinkSessionJobSpec extends AbstractFlinkSpec {
- /**
- * The cluster id of the target session cluster. When deployed using the operator the cluster id
- * is the name of the deployment.
- */
- private String clusterId;
+ /** The name of the target session cluster deployment. */
+ private String deploymentName;
}
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 4efeff8..a928180 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
@@ -213,7 +213,7 @@ public class FlinkService {
jarFile.exists(),
String.format("The jar file %s not exists", jarFile.getAbsolutePath()));
JarUploadHeaders headers = JarUploadHeaders.getInstance();
- String clusterId = sessionJob.getSpec().getClusterId();
+ String clusterId = sessionJob.getSpec().getDeploymentName();
String namespace = sessionJob.getMetadata().getNamespace();
int port = conf.getInteger(RestOptions.PORT);
String host =
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java
index 2dfe5ff..8f79f09 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java
@@ -313,7 +313,7 @@ public class DefaultValidator implements FlinkResourceValidator {
if (!deployment
.getMetadata()
.getName()
- .equals(sessionJob.getSpec().getClusterId())) {
+ .equals(sessionJob.getSpec().getDeploymentName())) {
return Optional.of(
"The session job's cluster id is not match with the session cluster");
} else {
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 044d063..8f72965 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
@@ -115,7 +115,7 @@ public class TestUtils {
.build());
sessionJob.setSpec(
FlinkSessionJobSpec.builder()
- .clusterId(TEST_DEPLOYMENT_NAME)
+ .deploymentName(TEST_DEPLOYMENT_NAME)
.job(
JobSpec.builder()
.jarURI(SAMPLE_JAR)
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/validation/DefaultValidatorTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/validation/DefaultValidatorTest.java
index 7b7e56f..0357e9a 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/validation/DefaultValidatorTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/validation/DefaultValidatorTest.java
@@ -301,7 +301,7 @@ public class DefaultValidatorTest {
testSessionJobValidateSuccess(job -> {}, session -> {});
testSessionJobValidateError(
- sessionJob -> sessionJob.getSpec().setClusterId("not-match"),
+ sessionJob -> sessionJob.getSpec().setDeploymentName("not-match"),
deployment -> {},
"The session job's cluster id is not match with the session cluster");
diff --git a/helm/flink-kubernetes-operator/crds/flinksessionjobs.flink.apache.org-v1.yml b/helm/flink-kubernetes-operator/crds/flinksessionjobs.flink.apache.org-v1.yml
index 64fe0e7..6d20c4c 100644
--- a/helm/flink-kubernetes-operator/crds/flinksessionjobs.flink.apache.org-v1.yml
+++ b/helm/flink-kubernetes-operator/crds/flinksessionjobs.flink.apache.org-v1.yml
@@ -19,7 +19,7 @@ spec:
properties:
spec:
properties:
- clusterId:
+ deploymentName:
type: string
job:
properties: