code: private <T> void runClusterAction(CustomCommandLine<T> activeCommandLine, CommandLine commandLine, ClusterAction<T> clusterAction) throws FlinkException {
  		final ClusterDescriptor<T> clusterDescriptor = activeCommandLine.createClusterDescriptor(commandLine);
  
  		final T clusterId = activeCommandLine.getClusterId(commandLine);
  
  		if (clusterId == null) {
  			throw new FlinkException("No cluster id was specified. Please specify a cluster to which " +
  				"you would like to connect.");
  		} else {
  			try {
  				final ClusterClient<T> clusterClient = clusterDescriptor.retrieve(clusterId);
  
  				try {
  					clusterAction.runAction(clusterClient);
  				} finally {
  					try {
  						clusterClient.shutdown();
  					} catch (Exception e) {
  						LOG.info("Could not properly shut down the cluster client.", e);
  					}
  				}
  			} finally {
  				try {
  					clusterDescriptor.close();
  				} catch (Exception e) {
  					LOG.info("Could not properly close the cluster descriptor.", e);
  				}
  			}
  		}
  	}
docs: Retrieves the {@link ClusterClient} from the given {@link CustomCommandLine} and runs the given
  {@link ClusterAction} against it.
  
  @param activeCommandLine to create the {@link ClusterDescriptor} from
  @param commandLine containing the parsed command line options
  @param clusterAction the cluster action to run against the retrieved {@link ClusterClient}.
  @param <T> type of the cluster id
  @throws FlinkException if something goes wrong