We are running same job all the time. And that error is happening from time
to time.
Here is job submittion code:
private JobSubmissionResult submitProgramToCluster(PackagedProgram
packagedProgram) throws JobSubmitterException,
ProgramMissingJobException, ProgramInvocationException {
ClusterClient clusterClient =
clusterClientUtil.getPrototypeClusterClient();
int parallelism =
Integer.parseInt(serverConfiguration.envParallelism);
return clusterClient.run(packagedProgram, parallelism);
}
And here our util for retrieving ClusterClient.
public class ClusterClientUtil {
...
public ClusterClient getPrototypeClusterClient() throws
JobSubmitterException {
return createClusterClientInstance();
}
private synchronized ClusterClient createClusterClientInstance() throws
JobSubmitterException {
try {
LOG.info("Creating new ClusterClient instance.");
Configuration configuration =
flinkConfigurator.getFlinkConfiguration();
ApplicationId applicationId =
ConverterUtils.toApplicationId(configuration.getString(FlinkConfigurator.PROPERTY_FLINK_APP_ID,
""));
logger.debug("Retrieved Flink applicationId: {}",
applicationId.toString());
YarnClusterDescriptor yarnClusterDescriptor = new
YarnClusterDescriptor(configuration,
serverConfiguration.getFlinkConfigurationDirectory);
ApplicationReport applicationReport =
yarnClient.getApplicationReport(applicationId);
final int numberTaskManagers =
configuration.getInteger(FlinkConfigurator.PROPERTY_FLINK_NUMBER_TASK_MANAGERS,
0);
final int slotsPerTaskManager =
configuration.getInteger(FlinkConfigurator.PROPERTY_FLINK_SLOTS_PER_TASK_MANAGER,
-1);
return new YarnClusterClient(yarnClusterDescriptor,
numberTaskManagers, slotsPerTaskManager, yarnClient, applicationReport,
configuration, false);
} catch (Exception e) {
throw new JobSubmitterException("Unable to create
YarnClusterClient.", e);
}
}
What Yarn settings to you need?
--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/
