[FLINK-15121] Add public constructors for execution environments that…

… take Configuration

This makes useful constructors on ExecutionEnvironment public and adds
Javadoc.

This makes the useless zero-argument constructor of
StreamExecutionEnvironment protected and adds Javadoc to
StreamExecutionEnvironment constructors.

This allows creating execution environments with arbitrary executor
configurations, i.e. one can now create an execution environment that
uses a YARN executor.

flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java

@@ -137,17 +137,23 @@ public class ExecutionEnvironment {
  private final ClassLoader userClassloader;
 
  /**
- * Creates a new Execution Environment.
+ * Creates a new {@link ExecutionEnvironment} that will use the given {@link Configuration} to
+ * configure the {@link org.apache.flink.core.execution.Executor}.
  */
- protected ExecutionEnvironment() {
- this(new Configuration());
- }
-
- protected ExecutionEnvironment(final Configuration configuration) {
+ @PublicEvolving
+ public ExecutionEnvironment(final Configuration configuration) {
  this(DefaultExecutorServiceLoader.INSTANCE, configuration, null);
  }
 
- protected ExecutionEnvironment(
+ /**
+ * Creates a new {@link ExecutionEnvironment} that will use the given {@link
+ * Configuration} to configure the {@link org.apache.flink.core.execution.Executor}.
+ *
+ * <p>In addition, this constructor allows specifying the {@link ExecutorServiceLoader} and
+ * user code {@link ClassLoader}.
+ */
+ @PublicEvolving
+ public ExecutionEnvironment(
  final ExecutorServiceLoader executorServiceLoader,
  final Configuration configuration,
  final ClassLoader userClassloader) {
@@ -156,6 +162,13 @@ protected ExecutionEnvironment(
  this.userClassloader = userClassloader == null ? getClass().getClassLoader() : userClassloader;
  }
 
+ /**
+ * Creates a new Execution Environment.
+ */
+ protected ExecutionEnvironment() {
+ this(new Configuration());
+ }
+
  @Internal
  public ClassLoader getUserCodeClassLoader() {
  return userClassloader;

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java

@@ -173,12 +173,28 @@ public class StreamExecutionEnvironment {
 
  public StreamExecutionEnvironment() {
  this(new Configuration());
+ // unfortunately, StreamExecutionEnvironment always (implicitly) had a public constructor.
+ // This constructor is not useful because the execution environment cannot be used for
+ // execution. We're keeping this to appease the binary compatibiliy checks.
  }
 
+ /**
+ * Creates a new {@link StreamExecutionEnvironment} that will use the given {@link
+ * Configuration} to configure the {@link org.apache.flink.core.execution.Executor}.
+ */
+ @PublicEvolving
  public StreamExecutionEnvironment(final Configuration configuration) {
  this(DefaultExecutorServiceLoader.INSTANCE, configuration, null);
  }
 
+ /**
+ * Creates a new {@link StreamExecutionEnvironment} that will use the given {@link
+ * Configuration} to configure the {@link org.apache.flink.core.execution.Executor}.
+ *
+ * <p>In addition, this constructor allows specifying the {@link ExecutorServiceLoader} and
+ * user code {@link ClassLoader}.
+ */
+ @PublicEvolving
  public StreamExecutionEnvironment(
  final ExecutorServiceLoader executorServiceLoader,
  final Configuration configuration,