[FLINK-24275][rest] Streamline JobCancellationMessageParameters

akalash · Oct 6, 2021 · bafbcef · bafbcef
1 parent b86b57a
commit bafbcef
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 6 deletions.
diff --git a/flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java b/flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java
@@ -436,10 +436,11 @@ public CompletableFuture<JobID> submitJob(@Nonnull JobGraph jobGraph) {
 
     @Override
     public CompletableFuture<Acknowledge> cancel(JobID jobID) {
-        JobCancellationMessageParameters params = new JobCancellationMessageParameters();
-        params.jobPathParameter.resolve(jobID);
-        params.terminationModeQueryParameter.resolve(
-                Collections.singletonList(TerminationModeQueryParameter.TerminationMode.CANCEL));
+        JobCancellationMessageParameters params =
+                new JobCancellationMessageParameters()
+                        .resolveJobId(jobID)
+                        .resolveTerminationMode(
+                                TerminationModeQueryParameter.TerminationMode.CANCEL);
         CompletableFuture<EmptyResponseBody> responseFuture =
                 sendRequest(JobCancellationHeaders.getInstance(), params);
         return responseFuture.thenApply(ignore -> Acknowledge.get());

diff --git a/...rc/main/java/org/apache/flink/runtime/rest/messages/JobCancellationMessageParameters.java b/...rc/main/java/org/apache/flink/runtime/rest/messages/JobCancellationMessageParameters.java
@@ -18,6 +18,8 @@
 
 package org.apache.flink.runtime.rest.messages;
 
+import org.apache.flink.api.common.JobID;
+
 import java.util.Collection;
 import java.util.Collections;
 
@@ -28,8 +30,8 @@
  */
 public class JobCancellationMessageParameters extends MessageParameters {
 
-    public final JobIDPathParameter jobPathParameter = new JobIDPathParameter();
-    public final TerminationModeQueryParameter terminationModeQueryParameter =
+    private final JobIDPathParameter jobPathParameter = new JobIDPathParameter();
+    private final TerminationModeQueryParameter terminationModeQueryParameter =
             new TerminationModeQueryParameter();
 
     @Override
@@ -41,4 +43,15 @@ public Collection<MessagePathParameter<?>> getPathParameters() {
     public Collection<MessageQueryParameter<?>> getQueryParameters() {
         return Collections.singleton(terminationModeQueryParameter);
     }
+
+    public JobCancellationMessageParameters resolveJobId(JobID jobId) {
+        jobPathParameter.resolve(jobId);
+        return this;
+    }
+
+    public JobCancellationMessageParameters resolveTerminationMode(
+            TerminationModeQueryParameter.TerminationMode terminationMode) {
+        terminationModeQueryParameter.resolve(Collections.singletonList(terminationMode));
+        return this;
+    }
 }