docs: protected ClusterClient<ApplicationId> deployInternal(
  			ClusterSpecification clusterSpecification,
  			String applicationName,
  			String yarnClusterEntrypoint,
  			@Nullable JobGraph jobGraph,
  			boolean detached) throws Exception {
  
  		// ------------------ Check if configuration is valid --------------------
  		validateClusterSpecification(clusterSpecification);
  
  		if (UserGroupInformation.isSecurityEnabled()) {
  			// note: UGI::hasKerberosCredentials inaccurately reports false
  			// for logins based on a keytab (fixed in Hadoop 2.6.1, see HADOOP-10786),
  			// so we check only in ticket cache scenario.
  			boolean useTicketCache = flinkConfiguration.getBoolean(SecurityOptions.KERBEROS_LOGIN_USETICKETCACHE);
  
  			UserGroupInformation loginUser = UserGroupInformation.getCurrentUser();
  			if (loginUser.getAuthenticationMethod() == UserGroupInformation.AuthenticationMethod.KERBEROS
  				&& useTicketCache && !loginUser.hasKerberosCredentials()) {
  				LOG.error("Hadoop security with Kerberos is enabled but the login user does not have Kerberos credentials");
  				throw new RuntimeException("Hadoop security with Kerberos is enabled but the login user " +
  					"does not have Kerberos credentials");
  			}
  		}
  
  		isReadyForDeployment(clusterSpecification);
  
  		// ------------------ Check if the specified queue exists --------------------
  
  		checkYarnQueues(yarnClient);
  
  		// ------------------ Add dynamic properties to local flinkConfiguraton ------
  		Map<String, String> dynProperties = getDynamicProperties(dynamicPropertiesEncoded);
  		for (Map.Entry<String, String> dynProperty : dynProperties.entrySet()) {
  			flinkConfiguration.setString(dynProperty.getKey(), dynProperty.getValue());
  		}
  
  		// ------------------ Check if the YARN ClusterClient has the requested resources --------------
  
  		// Create application via yarnClient
  		final YarnClientApplication yarnApplication = yarnClient.createApplication();
  		final GetNewApplicationResponse appResponse = yarnApplication.getNewApplicationResponse();
  
  		Resource maxRes = appResponse.getMaximumResourceCapability();
  
  		final ClusterResourceDescription freeClusterMem;
  		try {
  			freeClusterMem = getCurrentFreeClusterResources(yarnClient);
  		} catch (YarnException | IOException e) {
  			failSessionDuringDeployment(yarnClient, yarnApplication);
  			throw new YarnDeploymentException("Could not retrieve information about free cluster resources.", e);
  		}
  
  		final int yarnMinAllocationMB = yarnConfiguration.getInt(YarnConfiguration.RM_SCHEDULER_MINIMUM_ALLOCATION_MB, 0);
  
  		final ClusterSpecification validClusterSpecification;
  		try {
  			validClusterSpecification = validateClusterResources(
  				clusterSpecification,
  				yarnMinAllocationMB,
  				maxRes,
  				freeClusterMem);
  		} catch (YarnDeploymentException yde) {
  			failSessionDuringDeployment(yarnClient, yarnApplication);
  			throw yde;
  		}
  
  		LOG.info("Cluster specification: {}", validClusterSpecification);
  
  		final ClusterEntrypoint.ExecutionMode executionMode = detached ?
  			ClusterEntrypoint.ExecutionMode.DETACHED
  			: ClusterEntrypoint.ExecutionMode.NORMAL;
  
  		flinkConfiguration.setString(ClusterEntrypoint.EXECUTION_MODE, executionMode.toString());
  
  		ApplicationReport report = startAppMaster(
  			flinkConfiguration,
  			applicationName,
  			yarnClusterEntrypoint,
  			jobGraph,
  			yarnClient,
  			yarnApplication,
  			validClusterSpecification);
  
  		String host = report.getHost();
  		int port = report.getRpcPort();
  
  		// Correctly initialize the Flink config
  		flinkConfiguration.setString(JobManagerOptions.ADDRESS, host);
  		flinkConfiguration.setInteger(JobManagerOptions.PORT, port);
  
  		flinkConfiguration.setString(RestOptions.ADDRESS, host);
  		flinkConfiguration.setInteger(RestOptions.PORT, port);
  
  		// the Flink cluster is deployed in YARN. Represent cluster
  		return createYarnClusterClient(
  			this,
  			validClusterSpecification.getNumberTaskManagers(),
  			validClusterSpecification.getSlotsPerTaskManager(),
  			report,
  			flinkConfiguration,
  			true);
  	}