docs: @Internal
  	public Plan createProgramPlan(String jobName, boolean clearSinks) {
  		if (this.sinks.isEmpty()) {
  			if (wasExecuted) {
  				throw new RuntimeException("No new data sinks have been defined since the " +
  						"last execution. The last execution refers to the latest call to " +
  						"'execute()', 'count()', 'collect()', or 'print()'.");
  			} else {
  				throw new RuntimeException("No data sinks have been created yet. " +
  						"A program needs at least one sink that consumes data. " +
  						"Examples are writing the data set or printing it.");
  			}
  		}
  
  		if (jobName == null) {
  			jobName = getDefaultName();
  		}
  
  		OperatorTranslation translator = new OperatorTranslation();
  		Plan plan = translator.translateToPlan(this.sinks, jobName);
  
  		if (getParallelism() > 0) {
  			plan.setDefaultParallelism(getParallelism());
  		}
  		plan.setExecutionConfig(getConfig());
  
  		// Check plan for GenericTypeInfo's and register the types at the serializers.
  		if (!config.isAutoTypeRegistrationDisabled()) {
  			plan.accept(new Visitor<org.apache.flink.api.common.operators.Operator<?>>() {
  
  				private final Set<Class<?>> registeredTypes = new HashSet<>();
  				private final Set<org.apache.flink.api.common.operators.Operator<?>> visitedOperators = new HashSet<>();
  
  				@Override
  				public boolean preVisit(org.apache.flink.api.common.operators.Operator<?> visitable) {
  					if (!visitedOperators.add(visitable)) {
  						return false;
  					}
  					OperatorInformation<?> opInfo = visitable.getOperatorInfo();
  					Serializers.recursivelyRegisterType(opInfo.getOutputType(), config, registeredTypes);
  					return true;
  				}
  
  				@Override
  				public void postVisit(org.apache.flink.api.common.operators.Operator<?> visitable) {}
  			});
  		}
  
  		try {
  			registerCachedFilesWithPlan(plan);
  		} catch (Exception e) {
  			throw new RuntimeException("Error while registering cached files: " + e.getMessage(), e);
  		}
  
  		// clear all the sinks such that the next execution does not redo everything
  		if (clearSinks) {
  			this.sinks.clear();
  			wasExecuted = true;
  		}
  
  		// All types are registered now. Print information.
  		int registeredTypes = config.getRegisteredKryoTypes().size() +
  				config.getRegisteredPojoTypes().size() +
  				config.getRegisteredTypesWithKryoSerializerClasses().size() +
  				config.getRegisteredTypesWithKryoSerializers().size();
  		int defaultKryoSerializers = config.getDefaultKryoSerializers().size() +
  				config.getDefaultKryoSerializerClasses().size();
  		LOG.info("The job has {} registered types and {} default Kryo serializers", registeredTypes, defaultKryoSerializers);
  
  		if (config.isForceKryoEnabled() && config.isForceAvroEnabled()) {
  			LOG.warn("In the ExecutionConfig, both Avro and Kryo are enforced. Using Kryo serializer");
  		}
  		if (config.isForceKryoEnabled()) {
  			LOG.info("Using KryoSerializer for serializing POJOs");
  		}
  		if (config.isForceAvroEnabled()) {
  			LOG.info("Using AvroSerializer for serializing POJOs");
  		}
  
  		if (LOG.isDebugEnabled()) {
  			LOG.debug("Registered Kryo types: {}", config.getRegisteredKryoTypes().toString());
  			LOG.debug("Registered Kryo with Serializers types: {}", config.getRegisteredTypesWithKryoSerializers().entrySet().toString());
  			LOG.debug("Registered Kryo with Serializer Classes types: {}", config.getRegisteredTypesWithKryoSerializerClasses().entrySet().toString());
  			LOG.debug("Registered Kryo default Serializers: {}", config.getDefaultKryoSerializers().entrySet().toString());
  			LOG.debug("Registered Kryo default Serializers Classes {}", config.getDefaultKryoSerializerClasses().entrySet().toString());
  			LOG.debug("Registered POJO types: {}", config.getRegisteredPojoTypes().toString());
  
  			// print information about static code analysis
  			LOG.debug("Static code analysis mode: {}", config.getCodeAnalysisMode());
  		}
  
  		return plan;
  	}