[FLINK-11545] [container] Add job ID to StandaloneJobClusterConfigura…

…tion [FLINK-11545] [container] Add null checks and reduce visibility in StandaloneJobClusterConfiguration Fix line breaks in StandaloneJobClusterConfiguration
spotify · Feb 15, 2019 · 4c1eeca · 4c1eeca
1 parent 6f023c8
commit 4c1eeca
Show file tree

Hide file tree

Showing 2 changed files with 27 additions and 5 deletions.
diff --git a/...rc/main/java/org/apache/flink/container/entrypoint/StandaloneJobClusterConfiguration.java b/...rc/main/java/org/apache/flink/container/entrypoint/StandaloneJobClusterConfiguration.java
@@ -18,6 +18,7 @@
 
 package org.apache.flink.container.entrypoint;
 
+import org.apache.flink.api.common.JobID;
 import org.apache.flink.runtime.entrypoint.EntrypointClusterConfiguration;
 import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
 
@@ -26,6 +27,8 @@
 
 import java.util.Properties;
 
+import static java.util.Objects.requireNonNull;
+
 /**
  * Configuration for the {@link StandaloneJobClusterEntryPoint}.
  */
@@ -37,10 +40,22 @@ final class StandaloneJobClusterConfiguration extends EntrypointClusterConfigura
  @Nonnull
  private final SavepointRestoreSettings savepointRestoreSettings;
 
- public StandaloneJobClusterConfiguration(@Nonnull String configDir, @Nonnull Properties dynamicProperties, @Nonnull String[] args, @Nullable String hostname, int restPort, @Nonnull String jobClassName, @Nonnull SavepointRestoreSettings savepointRestoreSettings) {
+ @Nonnull
+ private final JobID jobId;
+
+ StandaloneJobClusterConfiguration(
+ @Nonnull String configDir,
+ @Nonnull Properties dynamicProperties,
+ @Nonnull String[] args,
+ @Nullable String hostname,
+ int restPort,
+ @Nonnull String jobClassName,
+ @Nonnull SavepointRestoreSettings savepointRestoreSettings,
+ @Nonnull JobID jobId) {
  super(configDir, dynamicProperties, args, hostname, restPort);
- this.jobClassName = jobClassName;
- this.savepointRestoreSettings = savepointRestoreSettings;
+ this.jobClassName = requireNonNull(jobClassName, "jobClassName");
+ this.savepointRestoreSettings = requireNonNull(savepointRestoreSettings, "savepointRestoreSettings");
+ this.jobId = requireNonNull(jobId, "jobId");
  }
 
  @Nonnull
@@ -49,7 +64,12 @@ String getJobClassName() {
  }
 
  @Nonnull
- public SavepointRestoreSettings getSavepointRestoreSettings() {
+ SavepointRestoreSettings getSavepointRestoreSettings() {
  return savepointRestoreSettings;
  }
+
+ @Nonnull
+ JobID getJobId() {
+ return jobId;
+ }
 }
diff --git a/...org/apache/flink/container/entrypoint/StandaloneJobClusterConfigurationParserFactory.java b/...org/apache/flink/container/entrypoint/StandaloneJobClusterConfigurationParserFactory.java
@@ -18,6 +18,7 @@
 
 package org.apache.flink.container.entrypoint;
 
+import org.apache.flink.api.common.JobID;
 import org.apache.flink.client.cli.CliFrontendParser;
 import org.apache.flink.runtime.entrypoint.parser.ParserResultFactory;
 import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
@@ -79,6 +80,7 @@ public StandaloneJobClusterConfiguration createResult(@Nonnull CommandLine comma
  hostname,
  restPort,
  jobClassName,
- savepointRestoreSettings);
+ savepointRestoreSettings,
+ new JobID());
  }
 }