docs: @Override
  	public ClusterClient<ApplicationId> retrieve(ApplicationId applicationId) throws ClusterRetrieveException {
  
  		try {
  			// check if required Hadoop environment variables are set. If not, warn user
  			if (System.getenv("HADOOP_CONF_DIR") == null &&
  				System.getenv("YARN_CONF_DIR") == null) {
  				LOG.warn("Neither the HADOOP_CONF_DIR nor the YARN_CONF_DIR environment variable is set." +
  					"The Flink YARN Client needs one of these to be set to properly load the Hadoop " +
  					"configuration for accessing YARN.");
  			}
  
  			final ApplicationReport appReport = yarnClient.getApplicationReport(applicationId);
  
  			if (appReport.getFinalApplicationStatus() != FinalApplicationStatus.UNDEFINED) {
  				// Flink cluster is not running anymore
  				LOG.error("The application {} doesn't run anymore. It has previously completed with final status: {}",
  					applicationId, appReport.getFinalApplicationStatus());
  				throw new RuntimeException("The Yarn application " + applicationId + " doesn't run anymore.");
  			}
  
  			final String host = appReport.getHost();
  			final int rpcPort = appReport.getRpcPort();
  
  			LOG.info("Found application JobManager host name '{}' and port '{}' from supplied application id '{}'",
  				host, rpcPort, applicationId);
  
  			flinkConfiguration.setString(JobManagerOptions.ADDRESS, host);
  			flinkConfiguration.setInteger(JobManagerOptions.PORT, rpcPort);
  
  			flinkConfiguration.setString(RestOptions.ADDRESS, host);
  			flinkConfiguration.setInteger(RestOptions.PORT, rpcPort);
  
  			return createYarnClusterClient(
  				this,
  				-1, // we don't know the number of task managers of a started Flink cluster
  				-1, // we don't know how many slots each task manager has for a started Flink cluster
  				appReport,
  				flinkConfiguration,
  				false);
  		} catch (Exception e) {
  			throw new ClusterRetrieveException("Couldn't retrieve Yarn cluster", e);
  		}
  	}