[FLINK-16658] Make StandaloneJobClusterEntryPoint extend directly Clu…

…sterEntrypoint

The StandaloneJobClusterEntryPoint is actually a
StandaloneApplicationClusterEntryPoint now. To illustrate that, we break
the previous hierarchy which would be misleading, but we keep the same
name for backwards compatibility reasons.

flink-container/src/main/java/org/apache/flink/container/entrypoint/StandaloneJobClusterEntryPoint.java

@@ -29,6 +29,9 @@
 import org.apache.flink.configuration.DeploymentOptions;
 import org.apache.flink.configuration.PipelineOptions;
 import org.apache.flink.configuration.PipelineOptionsInternal;
+import org.apache.flink.runtime.concurrent.ScheduledExecutor;
+import org.apache.flink.runtime.dispatcher.ArchivedExecutionGraphStore;
+import org.apache.flink.runtime.dispatcher.MemoryArchivedExecutionGraphStore;
 import org.apache.flink.runtime.dispatcher.SessionDispatcherFactory;
 import org.apache.flink.runtime.dispatcher.runner.DefaultDispatcherRunnerFactory;
 import org.apache.flink.runtime.entrypoint.ClusterEntrypoint;
@@ -58,7 +61,7 @@
  * {@link JobClusterEntrypoint} which is started with a job in a predefined
  * location.
  */
-public final class StandaloneJobClusterEntryPoint extends JobClusterEntrypoint {
+public final class StandaloneJobClusterEntryPoint extends ClusterEntrypoint {
 
  public static final JobID ZERO_JOB_ID = new JobID(0, 0);
 
@@ -81,6 +84,13 @@ protected DispatcherResourceManagerComponentFactory createDispatcherResourceMana
  JobRestEndpointFactory.INSTANCE);
  }
 
+ @Override
+ protected ArchivedExecutionGraphStore createSerializableExecutionGraphStore(
+ Configuration configuration,
+ ScheduledExecutor scheduledExecutor) {
+ return new MemoryArchivedExecutionGraphStore();
+ }
+
  public static void main(String[] args) {
  // startup checks and logging
  EnvironmentInformation.logEnvironmentInfo(LOG, StandaloneJobClusterEntryPoint.class.getSimpleName(), args);