@@ -0,0 +1,3435 @@
+/*%0A * Licensed to the Apache Software Foundation (ASF) under one%0A * or more contributor license agreements.  See the NOTICE file%0A * distributed with this work for additional information%0A * regarding copyright ownership.  The ASF licenses this file%0A * to you under the Apache License, Version 2.0 (the%0A * %22License%22); you may not use this file except in compliance%0A * with the License.  You may obtain a copy of the License at%0A *%0A *     http://www.apache.org/licenses/LICENSE-2.0%0A *%0A * Unless required by applicable law or agreed to in writing, software%0A * distributed under the License is distributed on an %22AS IS%22 BASIS,%0A * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.%0A * See the License for the specific language governing permissions and%0A * limitations under the License.%0A */%0Apackage org.apache.beam.runners.flink;%0A%0Aimport org.apache.flink.configuration.Configuration;%0Aimport org.apache.flink.configuration.RestOptions;%0Aimport org.apache.flink.runtime.minicluster.MiniCluster;%0Aimport org.apache.flink.runtime.minicluster.MiniClusterConfiguration;%0Aimport org.kohsuke.args4j.CmdLineException;%0Aimport org.kohsuke.args4j.CmdLineParser;%0Aimport org.kohsuke.args4j.Option;%0Aimport org.slf4j.Logger;%0Aimport org.slf4j.LoggerFactory;%0A%0A/** Entry point for starting an embedded Flink cluster. */%0Apublic class FlinkMiniClusterEntryPoint %7B%0A%0A  private static final Logger LOG = LoggerFactory.getLogger(FlinkMiniClusterEntryPoint.class);%0A%0A  static class MiniClusterArgs %7B%0A    @Option(name = %22--rest-port%22)%0A    int restPort = 0;%0A%0A    @Option(name = %22--num-task-managers%22)%0A    int numTaskManagers = 1;%0A%0A    @Option(name = %22--num-task-slots-per-taskmanager%22)%0A    int numSlotsPerTaskManager = 1;%0A  %7D%0A%0A  public static void main(String%5B%5D args) throws Exception %7B%0A    MiniClusterArgs miniClusterArgs = parseArgs(args);%0A%0A    Configuration flinkConfig = new Configuration();%0A    flinkConfig.setInteger(RestOptions.PORT, miniClusterArgs.restPort);%0A%0A    MiniClusterConfiguration clusterConfig =%0A        new MiniClusterConfiguration.Builder()%0A            .setConfiguration(flinkConfig)%0A            .setNumTaskManagers(miniClusterArgs.numTaskManagers)%0A            .setNumSlotsPerTaskManager(miniClusterArgs.numSlotsPerTaskManager)%0A            .build();%0A%0A    try (MiniCluster miniCluster = new MiniCluster(clusterConfig)) %7B%0A      miniCluster.start();%0A      System.out.println(%0A          String.format(%0A              %22Started Flink mini cluster (%25s TaskManagers with %25s task slots) with Rest API at %25s%22,%0A              miniClusterArgs.numTaskManagers,%0A              miniClusterArgs.numSlotsPerTaskManager,%0A              miniCluster.getRestAddress()));%0A      Thread.sleep(Long.MAX_VALUE);%0A    %7D%0A  %7D%0A%0A  private static MiniClusterArgs parseArgs(String%5B%5D args) %7B%0A    MiniClusterArgs configuration = new MiniClusterArgs();%0A    CmdLineParser parser = new CmdLineParser(configuration);%0A    try %7B%0A      parser.parseArgument(args);%0A    %7D catch (CmdLineException e) %7B%0A      LOG.error(%22Unable to parse command line arguments.%22, e);%0A      printUsage(parser);%0A      throw new IllegalArgumentException(%22Unable to parse command line arguments.%22, e);%0A    %7D%0A    return configuration;%0A  %7D%0A%0A  private static void printUsage(CmdLineParser parser) %7B%0A    System.err.println(%0A        String.format(%0A            %22Usage: java %25s arguments...%22, FlinkMiniClusterEntryPoint.class.getSimpleName()));%0A    parser.printUsage(System.err);%0A    System.err.println();%0A  %7D%0A%7D%0A

Add FlinkMiniClusterEntryPoint for testing the uber jar submission