[FLINK-3093] Introduce annotations for interface stability in flink-core

This closes #1427

flink-annotations/pom.xml

+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied.  See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+
+	<modelVersion>4.0.0</modelVersion>
+
+	<parent>
+		<groupId>org.apache.flink</groupId>
+		<artifactId>flink-parent</artifactId>
+		<version>1.0-SNAPSHOT</version>
+		<relativePath>..</relativePath>
+	</parent>
+
+	<artifactId>flink-annotations</artifactId>
+	<name>flink-annotations</name>
+
+	<packaging>jar</packaging>
+
+</project>

flink-annotations/src/main/java/org/apache/flink/annotation/Experimental.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+package org.apache.flink.annotation;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Target;
+
+/**
+ * Interface to mark methods within stable, public APIs as experimental.
+ * It also allows to mark types explicitly as experimental
+ *
+ * An experimental API might change between minor releases.
+ */
+@Documented
+@Target({ ElementType.TYPE, ElementType.METHOD })
+public @interface Experimental {
+}

flink-annotations/src/main/java/org/apache/flink/annotation/Internal.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+package org.apache.flink.annotation;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Target;
+
+/**
+ * Interface to mark methods within stable, public APIs as an internal developer API.
+ *
+ * Developer APIs are stable but internal to Flink and might change across releases.
+ */
+@Documented
+@Target({ ElementType.TYPE, ElementType.METHOD })
+public @interface Internal {
+}

flink-annotations/src/main/java/org/apache/flink/annotation/Public.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+package org.apache.flink.annotation;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Target;
+
+/**
+ * Annotation for marking classes as public, stable interfaces.
+ *
+ * Classes, methods and fields with this annotation are stable across minor releases (1.0, 1.1, 1.2). In other words,
+ * applications using @PublicInterface annotated classes will compile against newer versions of the same major release.
+ *
+ * Only major releases (1.0, 2.0, 3.0) can break interfaces with this annotation.
+ */
+@Documented
+@Target(ElementType.TYPE)
+public @interface Public {}

flink-core/pom.xml

@@ -35,6 +35,12 @@ under the License.
 	<packaging>jar</packaging>
 
 	<dependencies>
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-annotations</artifactId>
+			<version>${project.version}</version>
+		</dependency>
+
 		<dependency>
 			<groupId>org.apache.flink</groupId>
 			<artifactId>${shading-artifact.name}</artifactId>

flink-core/src/main/java/org/apache/flink/api/common/CodeAnalysisMode.java

@@ -18,6 +18,8 @@
 
 package org.apache.flink.api.common;
 
+import org.apache.flink.annotation.Public;
+
 /**
  * Specifies to which extent user-defined functions are analyzed in order
  * to give the Flink optimizer an insight of UDF internals and inform
@@ -31,6 +33,7 @@ package org.apache.flink.api.common;
  *  - Warnings if a tuple access uses a wrong index
  *  - Information about the number of object creations (for manual optimization)
  */
+@Public
 public enum CodeAnalysisMode {
 
 	/**

flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java

@@ -19,6 +19,8 @@
 package org.apache.flink.api.common;
 
 import com.esotericsoftware.kryo.Serializer;
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.annotation.Public;
 
 import java.io.Serializable;
 import java.util.LinkedHashMap;
@@ -51,6 +53,7 @@ import java.util.Objects;
  *         automatically applied.</li>
  * </ul>
  */
+@Public
 public class ExecutionConfig implements Serializable {
 
 	private static final long serialVersionUID = 1L;
@@ -150,6 +153,7 @@ public class ExecutionConfig implements Serializable {
 	 *
 	 * @param interval The interval between watermarks in milliseconds.
 	 */
+	@Experimental
 	public ExecutionConfig setAutoWatermarkInterval(long interval) {
 		enableTimestamps();
 		this.autoWatermarkInterval = interval;
@@ -167,6 +171,7 @@ public class ExecutionConfig implements Serializable {
 	 *
 	 * @see #setAutoWatermarkInterval(long)
 	 */
+	@Experimental
 	public ExecutionConfig enableTimestamps() {
 		this.timestampsEnabled = true;
 		return this;
@@ -177,6 +182,7 @@ public class ExecutionConfig implements Serializable {
 	 *
 	 * @see #enableTimestamps()
 	 */
+	@Experimental
 	public ExecutionConfig disableTimestamps() {
 		this.timestampsEnabled = false;
 		return this;
@@ -187,6 +193,7 @@ public class ExecutionConfig implements Serializable {
 	 *
 	 * @see #enableTimestamps()
 	 */
+	@Experimental
 	public boolean areTimestampsEnabled() {
 		return timestampsEnabled;
 	}
@@ -196,6 +203,7 @@ public class ExecutionConfig implements Serializable {
 	 *
 	 * @see #setAutoWatermarkInterval(long)
 	 */
+	@Experimental
 	public long getAutoWatermarkInterval()  {
 		return this.autoWatermarkInterval;
 	}
@@ -377,6 +385,7 @@ public class ExecutionConfig implements Serializable {
 	 * 
 	 * @param codeAnalysisMode see {@link CodeAnalysisMode}
 	 */
+	@Experimental
 	public void setCodeAnalysisMode(CodeAnalysisMode codeAnalysisMode) {
 		this.codeAnalysisMode = codeAnalysisMode;
 	}
@@ -384,6 +393,7 @@ public class ExecutionConfig implements Serializable {
 	/**
 	 * Returns the {@link CodeAnalysisMode} of the program.
 	 */
+	@Experimental
 	public CodeAnalysisMode getCodeAnalysisMode() {
 		return codeAnalysisMode;
 	}

flink-core/src/main/java/org/apache/flink/api/common/ExecutionMode.java

@@ -18,10 +18,13 @@
 
 package org.apache.flink.api.common;
 
+import org.apache.flink.annotation.Public;
+
 /**
  * The execution mode specifies how a batch program is executed in terms
  * of data exchange: pipelining or batched.
  */
+@Public
 public enum ExecutionMode {
 
 	/**

flink-core/src/main/java/org/apache/flink/api/common/JobExecutionResult.java

@@ -18,6 +18,9 @@
 
 package org.apache.flink.api.common;
 
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.annotation.Public;
+
 import java.util.Collections;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
@@ -26,6 +29,7 @@ import java.util.concurrent.TimeUnit;
  * The result of a job execution. Gives access to the execution time of the job,
  * and to all accumulators created by this job.
  */
+@Public
 public class JobExecutionResult extends JobSubmissionResult {
 
 	private long netRuntime;
@@ -99,6 +103,8 @@ public class JobExecutionResult extends JobSubmissionResult {
 	 * @return Result of the counter, or null if the counter does not exist
 	 * @throws java.lang.ClassCastException Thrown, if the accumulator was not aggregating a {@link java.lang.Integer}
 	 */
+	@Deprecated
+	@Experimental
 	public Integer getIntCounterResult(String accumulatorName) {
 		Object result = this.accumulatorResults.get(accumulatorName);
 		if (result == null) {

flink-core/src/main/java/org/apache/flink/api/common/JobID.java

@@ -18,6 +18,7 @@
 
 package org.apache.flink.api.common;
 
+import org.apache.flink.annotation.Public;
 import org.apache.flink.util.AbstractID;
 import javax.xml.bind.DatatypeConverter;
 import java.nio.ByteBuffer;
@@ -30,6 +31,7 @@ import java.nio.ByteBuffer;
  * incrementally in different parts. Newer fragments of a graph can be attached to existing
  * graphs, thereby extending the current data flow graphs.</p>
  */
+@Public
 public final class JobID extends AbstractID {
 
 	private static final long serialVersionUID = 1L;

flink-core/src/main/java/org/apache/flink/api/common/JobSubmissionResult.java

@@ -18,9 +18,12 @@
 
 package org.apache.flink.api.common;
 
+import org.apache.flink.annotation.Public;
+
 /**
  * The result of submitting a job to a JobManager.
  */
+@Public
 public class JobSubmissionResult {
 	
 	private JobID jobID;

flink-core/src/main/java/org/apache/flink/api/common/accumulators/Accumulator.java

@@ -18,6 +18,8 @@
 
 package org.apache.flink.api.common.accumulators;
 
+import org.apache.flink.annotation.Public;
+
 import java.io.Serializable;
 
 /**
@@ -39,6 +41,7 @@ import java.io.Serializable;
  *            Type of the accumulator result as it will be reported to the
  *            client
  */
+@Public
 public interface Accumulator<V, R extends Serializable> extends Serializable, Cloneable {
 	/**
 	 * @param value

flink-core/src/main/java/org/apache/flink/api/common/accumulators/AverageAccumulator.java

@@ -18,10 +18,13 @@
 
 package org.apache.flink.api.common.accumulators;
 
+import org.apache.flink.annotation.Public;
+
 /**
  * An accumulator that computes the average value.
  * Input can be {@code long}, {@code integer}, or {@code double} and the result is {@code double}.
  */
+@Public
 public class AverageAccumulator implements SimpleAccumulator<Double> {
 
 	private static final long serialVersionUID = 3672555084179165255L;

flink-core/src/main/java/org/apache/flink/api/common/accumulators/DoubleCounter.java

@@ -18,6 +18,7 @@
 
 package org.apache.flink.api.common.accumulators;
 
+
 /**
  * An accumulator that sums up {@code double} values.
  */

flink-core/src/main/java/org/apache/flink/api/common/accumulators/Histogram.java

@@ -18,6 +18,8 @@
 
 package org.apache.flink.api.common.accumulators;
 
+import org.apache.flink.annotation.Public;
+
 import java.util.Map;
 import java.util.TreeMap;
 
@@ -29,6 +31,7 @@ import java.util.TreeMap;
  * This class does not extend to continuous values later, because it makes no
  * attempt to put the data in bins.
  */
+@Public
 public class Histogram implements Accumulator<Integer, TreeMap<Integer, Integer>> {
 
 	private static final long serialVersionUID = 1L;

flink-core/src/main/java/org/apache/flink/api/common/accumulators/IntCounter.java

@@ -19,6 +19,7 @@
 
 package org.apache.flink.api.common.accumulators;
 
+
 /**
  * An accumulator that sums up {@code Integer} values.
  */

flink-core/src/main/java/org/apache/flink/api/common/accumulators/ListAccumulator.java

@@ -18,6 +18,8 @@
 
 package org.apache.flink.api.common.accumulators;
 
+import org.apache.flink.annotation.Public;
+
 import java.util.ArrayList;
 
 /**
@@ -25,6 +27,7 @@ import java.util.ArrayList;
  *
  * @param <T> The type of the accumulated objects
  */
+@Public
 public class ListAccumulator<T> implements Accumulator<T, ArrayList<T>> {
 
 	private static final long serialVersionUID = 1L;

flink-core/src/main/java/org/apache/flink/api/common/accumulators/SimpleAccumulator.java

@@ -18,10 +18,13 @@
 
 package org.apache.flink.api.common.accumulators;
 
+import org.apache.flink.annotation.Public;
+
 import java.io.Serializable;
 
 /**
  * Similar to Accumulator, but the type of items to add and the result value
  * must be the same.
  */
+@Public
 public interface SimpleAccumulator<T extends Serializable> extends Accumulator<T,T> {}

flink-core/src/main/java/org/apache/flink/api/common/cache/DistributedCache.java

@@ -15,8 +15,6 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-
-
 package org.apache.flink.api.common.cache;
 
 
@@ -29,6 +27,7 @@ import java.util.Set;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 
+import org.apache.flink.annotation.Public;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.core.fs.Path;
 
@@ -36,6 +35,7 @@ import org.apache.flink.core.fs.Path;
  * DistributedCache provides static methods to write the registered cache files into job configuration or decode
  * them from job configuration. It also provides user access to the file locally.
  */
+@Public
 public class DistributedCache {
 	
 	public static class DistributedCacheEntry {

flink-core/src/main/java/org/apache/flink/api/common/functions/AbstractRichFunction.java

@@ -20,6 +20,7 @@ package org.apache.flink.api.common.functions;
 
 import java.io.Serializable;
 
+import org.apache.flink.annotation.Public;
 import org.apache.flink.configuration.Configuration;
 
 /**
@@ -28,6 +29,7 @@ import org.apache.flink.configuration.Configuration;
  * teardown ({@link #close()}), as well as access to their runtime execution context via
  * {@link #getRuntimeContext()}.
  */
+@Public
 public abstract class AbstractRichFunction implements RichFunction, Serializable {
 	
 	private static final long serialVersionUID = 1L;

flink-core/src/main/java/org/apache/flink/api/common/functions/BroadcastVariableInitializer.java

@@ -18,6 +18,8 @@
 
 package org.apache.flink.api.common.functions;
 
+import org.apache.flink.annotation.Public;
+
 /**
  * A broadcast variable initializer can be used to transform a broadcast variable
  * into another format during initialization. The transformed variable is shared
@@ -67,6 +69,7 @@ package org.apache.flink.api.common.functions;
  * @param <T> The type of the elements in the list of the original untransformed broadcast variable.
  * @param <O> The type of the transformed broadcast variable. 
  */
+@Public
 public interface BroadcastVariableInitializer<T, O> {
 
 	/**

flink-core/src/main/java/org/apache/flink/api/common/functions/CoGroupFunction.java

@@ -20,6 +20,7 @@ package org.apache.flink.api.common.functions;
 
 import java.io.Serializable;
 
+import org.apache.flink.annotation.Public;
 import org.apache.flink.util.Collector;
 
 /**
@@ -44,6 +45,7 @@ import org.apache.flink.util.Collector;
  * @param <IN2> The data type of the second input data set.
  * @param <O> The data type of the returned elements.
  */
+@Public
 public interface CoGroupFunction<IN1, IN2, O> extends Function, Serializable {
 	
 	/**

flink-core/src/main/java/org/apache/flink/api/common/functions/CombineFunction.java

@@ -18,6 +18,8 @@
 
 package org.apache.flink.api.common.functions;
 
+import org.apache.flink.annotation.Public;
+
 import java.io.Serializable;
 
 /**
@@ -34,6 +36,7 @@ import java.io.Serializable;
  * @param <IN> The data type processed by the combine function.
  * @param <OUT> The data type emitted by the combine function.
  */
+@Public
 public interface CombineFunction<IN, OUT> extends Function, Serializable {
 
 	/**

flink-core/src/main/java/org/apache/flink/api/common/functions/CrossFunction.java

@@ -18,6 +18,8 @@
 
 package org.apache.flink.api.common.functions;
 
+import org.apache.flink.annotation.Public;
+
 import java.io.Serializable;
 
 /**
@@ -41,6 +43,7 @@ import java.io.Serializable;
  * @param <IN2> The type of the elements in the second input.
  * @param <OUT> The type of the result elements.
  */
+@Public
 public interface CrossFunction<IN1, IN2, OUT> extends Function, Serializable {
 
 	/**

flink-core/src/main/java/org/apache/flink/api/common/functions/FilterFunction.java

@@ -18,6 +18,8 @@
 
 package org.apache.flink.api.common.functions;
 
+import org.apache.flink.annotation.Public;
+
 import java.io.Serializable;
 
 /**
@@ -37,6 +39,7 @@ import java.io.Serializable;
  * 
  * @param <T> The type of the filtered elements.
  */
+@Public
 public interface FilterFunction<T> extends Function, Serializable {
 	
 	/**

flink-core/src/main/java/org/apache/flink/api/common/functions/FlatJoinFunction.java

@@ -18,6 +18,7 @@
 
 package org.apache.flink.api.common.functions;
 
+import org.apache.flink.annotation.Public;
 import org.apache.flink.util.Collector;
 
 import java.io.Serializable;
@@ -53,6 +54,7 @@ import java.io.Serializable;
  * @param <IN2> The type of the elements in the second input.
  * @param <OUT> The type of the result elements.
  */
+@Public
 public interface FlatJoinFunction<IN1, IN2, OUT> extends Function, Serializable {
 
 	/**

flink-core/src/main/java/org/apache/flink/api/common/functions/FlatMapFunction.java

@@ -18,6 +18,7 @@
 
 package org.apache.flink.api.common.functions;
 
+import org.apache.flink.annotation.Public;
 import org.apache.flink.util.Collector;
 
 import java.io.Serializable;
@@ -38,6 +39,7 @@ import java.io.Serializable;
  * @param <T> Type of the input elements.
  * @param <O> Type of the returned elements.
  */
+@Public
 public interface FlatMapFunction<T, O> extends Function, Serializable {
 
 	/**

flink-core/src/main/java/org/apache/flink/api/common/functions/FoldFunction.java

@@ -18,6 +18,8 @@
 
 package org.apache.flink.api.common.functions;
 
+import org.apache.flink.annotation.Public;
+
 import java.io.Serializable;
 
 /**
@@ -37,6 +39,7 @@ import java.io.Serializable;
  * @param <T> Type of the initial input and the returned element
  * @param <O> Type of the elements that the group/list/stream contains
  */
+@Public
 public interface FoldFunction<O,T> extends Function, Serializable {
 	/**
 	 * The core method of FoldFunction, combining two values into one value of the same type.

flink-core/src/main/java/org/apache/flink/api/common/functions/Function.java

@@ -18,11 +18,14 @@
 
 package org.apache.flink.api.common.functions;
 
+import org.apache.flink.annotation.Public;
+
 /**
  * The base interface for all user-defined functions.
  * 
  * <p>This interface is empty in order to allow extending interfaces to
  * be SAM (single abstract method) interfaces that can be implemented via Java 8 lambdas.</p>
  */
+@Public
 public interface Function extends java.io.Serializable {
 }

flink-core/src/main/java/org/apache/flink/api/common/functions/GroupCombineFunction.java

@@ -20,6 +20,7 @@ package org.apache.flink.api.common.functions;
 
 import java.io.Serializable;
 
+import org.apache.flink.annotation.Public;
 import org.apache.flink.util.Collector;
 
 /**
@@ -36,6 +37,7 @@ import org.apache.flink.util.Collector;
  * @param <IN> The data type processed by the combine function.
  * @param <OUT> The data type emitted by the combine function.
  */
+@Public
 public interface GroupCombineFunction<IN, OUT> extends Function, Serializable {
 
 	/**

flink-core/src/main/java/org/apache/flink/api/common/functions/GroupReduceFunction.java

@@ -20,6 +20,7 @@ package org.apache.flink.api.common.functions;
 
 import java.io.Serializable;
 
+import org.apache.flink.annotation.Public;
 import org.apache.flink.util.Collector;
 
 /**
@@ -41,6 +42,7 @@ import org.apache.flink.util.Collector;
  * @param <T> Type of the elements that this function processes.
  * @param <O> The type of the elements returned by the user-defined function.
  */
+@Public
 public interface GroupReduceFunction<T, O> extends Function, Serializable {
 	
 	/**

flink-core/src/main/java/org/apache/flink/api/common/functions/InvalidTypesException.java

@@ -18,12 +18,14 @@
 
 package org.apache.flink.api.common.functions;
 
+import org.apache.flink.annotation.Public;
 import org.apache.flink.api.common.InvalidProgramException;
 
 /**
  * A special case of the {@link InvalidProgramException}, indicating that the types used in
  * an operation are invalid or inconsistent. 
  */
+@Public
 public class InvalidTypesException extends InvalidProgramException {
 
 	private static final long serialVersionUID = 1L;

flink-core/src/main/java/org/apache/flink/api/common/functions/IterationRuntimeContext.java

@@ -18,12 +18,12 @@
 
 package org.apache.flink.api.common.functions;
 
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.annotation.Public;
 import org.apache.flink.api.common.aggregators.Aggregator;
 import org.apache.flink.types.Value;
 
-/**
- * 
- */
+@Public
 public interface IterationRuntimeContext extends RuntimeContext {
 	
 	/**
@@ -32,7 +32,8 @@ public interface IterationRuntimeContext extends RuntimeContext {
 	 * @return The number of the current superstep.
 	 */
 	int getSuperstepNumber();
-	
+
+	@Experimental
 	<T extends Aggregator<?>> T getIterationAggregator(String name);
 	
 	<T extends Value> T getPreviousIterationAggregate(String name);

flink-core/src/main/java/org/apache/flink/api/common/functions/JoinFunction.java

@@ -18,6 +18,8 @@
 
 package org.apache.flink.api.common.functions;
 
+import org.apache.flink.annotation.Public;
+
 import java.io.Serializable;
 
 /**
@@ -48,6 +50,7 @@ import java.io.Serializable;
  * @param <IN2> The type of the elements in the second input.
  * @param <OUT> The type of the result elements.
  */
+@Public
 public interface JoinFunction<IN1, IN2, OUT> extends Function, Serializable {
 
 	/**

flink-core/src/main/java/org/apache/flink/api/common/functions/MapFunction.java

@@ -18,6 +18,8 @@
 
 package org.apache.flink.api.common.functions;
 
+import org.apache.flink.annotation.Public;
+
 import java.io.Serializable;
 
 /**
@@ -37,6 +39,7 @@ import java.io.Serializable;
  * @param <T> Type of the input elements.
  * @param <O> Type of the returned elements.
  */
+@Public
 public interface MapFunction<T, O> extends Function, Serializable {
 
 	/**

flink-core/src/main/java/org/apache/flink/api/common/functions/MapPartitionFunction.java

@@ -18,6 +18,7 @@
 
 package org.apache.flink.api.common.functions;
 
+import org.apache.flink.annotation.Public;
 import org.apache.flink.util.Collector;
 
 import java.io.Serializable;
@@ -40,6 +41,7 @@ import java.io.Serializable;
  * @param <T> Type of the input elements.
  * @param <O> Type of the returned elements.
  */
+@Public
 public interface MapPartitionFunction<T, O> extends Function, Serializable {
 	
 	/**

flink-core/src/main/java/org/apache/flink/api/common/functions/Partitioner.java

@@ -18,11 +18,14 @@
 
 package org.apache.flink.api.common.functions;
 
+import org.apache.flink.annotation.Public;
+
 /**
  * Function to implement a custom partition assignment for keys.
  * 
  * @param <K> The type of the key to be partitioned.
  */
+@Public
 public interface Partitioner<K> extends java.io.Serializable {
 
 	/**

flink-core/src/main/java/org/apache/flink/api/common/functions/ReduceFunction.java

@@ -18,6 +18,8 @@
 
 package org.apache.flink.api.common.functions;
 
+import org.apache.flink.annotation.Public;
+
 import java.io.Serializable;
 
 /**
@@ -42,6 +44,7 @@ import java.io.Serializable;
  * 
  * @param <T> Type of the elements that this function processes.
  */
+@Public
 public interface ReduceFunction<T> extends Function, Serializable {
 
 	/**

flink-core/src/main/java/org/apache/flink/api/common/functions/RichCoGroupFunction.java

@@ -18,9 +18,7 @@
 
 package org.apache.flink.api.common.functions;
 
-import org.apache.flink.api.common.functions.AbstractRichFunction;
-import org.apache.flink.api.common.functions.CoGroupFunction;
-import org.apache.flink.api.common.functions.RichFunction;
+import org.apache.flink.annotation.Public;
 import org.apache.flink.util.Collector;
 
 /**
@@ -33,6 +31,7 @@ import org.apache.flink.util.Collector;
  * @param <IN2> The type of the elements in the second input.
  * @param <OUT> The type of the result elements.
  */
+@Public
 public abstract class RichCoGroupFunction<IN1, IN2, OUT> extends AbstractRichFunction implements CoGroupFunction<IN1, IN2, OUT> {
 
 	private static final long serialVersionUID = 1L;

flink-core/src/main/java/org/apache/flink/api/common/functions/RichCrossFunction.java

@@ -18,9 +18,7 @@
 
 package org.apache.flink.api.common.functions;
 
-import org.apache.flink.api.common.functions.AbstractRichFunction;
-import org.apache.flink.api.common.functions.CrossFunction;
-import org.apache.flink.api.common.functions.RichFunction;
+import org.apache.flink.annotation.Public;
 
 /**
  * Rich variant of the {@link CrossFunction}. As a {@link RichFunction}, it gives access to the
@@ -32,6 +30,7 @@ import org.apache.flink.api.common.functions.RichFunction;
  * @param <IN2> The type of the elements in the second input.
  * @param <OUT> The type of the result elements.
  */
+@Public
 public abstract class RichCrossFunction<IN1, IN2, OUT> extends AbstractRichFunction implements CrossFunction<IN1, IN2, OUT> {
 	
 	private static final long serialVersionUID = 1L;

flink-core/src/main/java/org/apache/flink/api/common/functions/RichFilterFunction.java

@@ -18,9 +18,7 @@
 
 package org.apache.flink.api.common.functions;
 
-import org.apache.flink.api.common.functions.AbstractRichFunction;
-import org.apache.flink.api.common.functions.FilterFunction;
-import org.apache.flink.api.common.functions.RichFunction;
+import org.apache.flink.annotation.Public;
 
 /**
  * Rich variant of the {@link FilterFunction}. As a {@link RichFunction}, it gives access to the
@@ -30,6 +28,7 @@ import org.apache.flink.api.common.functions.RichFunction;
  * 
  * @param <T> The type of the filtered elements.
  */
+@Public
 public abstract class RichFilterFunction<T> extends AbstractRichFunction implements FilterFunction<T> {
 	
 	private static final long serialVersionUID = 1L;

flink-core/src/main/java/org/apache/flink/api/common/functions/RichFlatJoinFunction.java

@@ -18,9 +18,7 @@
 
 package org.apache.flink.api.common.functions;
 
-import org.apache.flink.api.common.functions.AbstractRichFunction;
-import org.apache.flink.api.common.functions.FlatJoinFunction;
-import org.apache.flink.api.common.functions.RichFunction;
+import org.apache.flink.annotation.Public;
 import org.apache.flink.util.Collector;
 
 /**
@@ -33,6 +31,7 @@ import org.apache.flink.util.Collector;
  * @param <IN2> The type of the elements in the second input.
  * @param <OUT> The type of the result elements.
  */
+@Public
 public abstract class RichFlatJoinFunction<IN1, IN2, OUT> extends AbstractRichFunction implements FlatJoinFunction<IN1, IN2, OUT> {
 
 	private static final long serialVersionUID = 1L;

flink-core/src/main/java/org/apache/flink/api/common/functions/RichFlatMapFunction.java

@@ -18,9 +18,7 @@
 
 package org.apache.flink.api.common.functions;
 
-import org.apache.flink.api.common.functions.AbstractRichFunction;
-import org.apache.flink.api.common.functions.FlatMapFunction;
-import org.apache.flink.api.common.functions.RichFunction;
+import org.apache.flink.annotation.Public;
 import org.apache.flink.util.Collector;
 
 /**
@@ -32,6 +30,7 @@ import org.apache.flink.util.Collector;
  * @param <IN> Type of the input elements.
  * @param <OUT> Type of the returned elements.
  */
+@Public
 public abstract class RichFlatMapFunction<IN, OUT> extends AbstractRichFunction implements FlatMapFunction<IN, OUT> {
 
 	private static final long serialVersionUID = 1L;

flink-core/src/main/java/org/apache/flink/api/common/functions/RichFoldFunction.java

@@ -18,9 +18,7 @@
 
 package org.apache.flink.api.common.functions;
 
-import org.apache.flink.api.common.functions.AbstractRichFunction;
-import org.apache.flink.api.common.functions.FoldFunction;
-import org.apache.flink.api.common.functions.RichFunction;
+import org.apache.flink.annotation.Public;
 
 /**
  * Rich variant of the {@link FoldFunction}. As a {@link RichFunction}, it gives access to the
@@ -31,6 +29,7 @@ import org.apache.flink.api.common.functions.RichFunction;
  * @param <T> Type of the initial input and the returned element
  * @param <O> Type of the elements that the group/list/stream contains
  */
+@Public
 public abstract class RichFoldFunction<O, T> extends AbstractRichFunction implements FoldFunction<O, T> {
 
 	private static final long serialVersionUID = 1L;

flink-core/src/main/java/org/apache/flink/api/common/functions/RichFunction.java

@@ -18,6 +18,7 @@
 
 package org.apache.flink.api.common.functions;
 
+import org.apache.flink.annotation.Public;
 import org.apache.flink.configuration.Configuration;
 
 /**
@@ -25,6 +26,7 @@ import org.apache.flink.configuration.Configuration;
  * the life cycle of the functions, as well as methods to access the context in which the functions
  * are executed.
  */
+@Public
 public interface RichFunction extends Function {
 	
 	/**

flink-core/src/main/java/org/apache/flink/api/common/functions/RichGroupCombineFunction.java

@@ -19,6 +19,7 @@
 package org.apache.flink.api.common.functions;
 
 
+import org.apache.flink.annotation.Public;
 import org.apache.flink.util.Collector;
 
 /**
@@ -30,6 +31,7 @@ import org.apache.flink.util.Collector;
  * @param <IN> The data type of the elements to be combined.
  * @param <OUT> The resulting data type of the elements to be combined.
  */
+@Public
 public abstract class RichGroupCombineFunction<IN, OUT> extends AbstractRichFunction implements GroupCombineFunction<IN, OUT> {
 
 	private static final long serialVersionUID = 1L;

flink-core/src/main/java/org/apache/flink/api/common/functions/RichGroupReduceFunction.java

@@ -23,6 +23,7 @@ import java.lang.annotation.Retention;
 import java.lang.annotation.RetentionPolicy;
 import java.lang.annotation.Target;
 
+import org.apache.flink.annotation.Public;
 import org.apache.flink.util.Collector;
 
 /**
@@ -34,6 +35,7 @@ import org.apache.flink.util.Collector;
  * @param <IN> Type of the elements that this function processes.
  * @param <OUT> The type of the elements returned by the user-defined function.
  */
+@Public
 public abstract class RichGroupReduceFunction<IN, OUT> extends AbstractRichFunction implements GroupReduceFunction<IN, OUT>, GroupCombineFunction<IN, IN> {
 	
 	private static final long serialVersionUID = 1L;
@@ -83,5 +85,6 @@ public abstract class RichGroupReduceFunction<IN, OUT> extends AbstractRichFunct
 	 */
 	@Retention(RetentionPolicy.RUNTIME)
 	@Target(ElementType.TYPE)
+	@Public
 	public static @interface Combinable {}
 }

flink-core/src/main/java/org/apache/flink/api/common/functions/RichJoinFunction.java

@@ -18,9 +18,7 @@
 
 package org.apache.flink.api.common.functions;
 
-import org.apache.flink.api.common.functions.AbstractRichFunction;
-import org.apache.flink.api.common.functions.JoinFunction;
-import org.apache.flink.api.common.functions.RichFunction;
+import org.apache.flink.annotation.Public;
 
 /**
  * Rich variant of the {@link JoinFunction}. As a {@link RichFunction}, it gives access to the
@@ -32,6 +30,7 @@ import org.apache.flink.api.common.functions.RichFunction;
  * @param <IN2> The type of the elements in the second input.
  * @param <OUT> The type of the result elements.
  */
+@Public
 public abstract class RichJoinFunction<IN1,IN2,OUT> extends AbstractRichFunction implements JoinFunction<IN1,IN2,OUT> {
 
 	private static final long serialVersionUID = 1L;

flink-core/src/main/java/org/apache/flink/api/common/functions/RichMapFunction.java

@@ -18,9 +18,7 @@
 
 package org.apache.flink.api.common.functions;
 
-import org.apache.flink.api.common.functions.AbstractRichFunction;
-import org.apache.flink.api.common.functions.MapFunction;
-import org.apache.flink.api.common.functions.RichFunction;
+import org.apache.flink.annotation.Public;
 
 /**
  * Rich variant of the {@link MapFunction}. As a {@link RichFunction}, it gives access to the
@@ -31,6 +29,7 @@ import org.apache.flink.api.common.functions.RichFunction;
  * @param <IN> Type of the input elements.
  * @param <OUT> Type of the returned elements.
  */
+@Public
 public abstract class RichMapFunction<IN, OUT> extends AbstractRichFunction implements MapFunction<IN, OUT> {
 
 	private static final long serialVersionUID = 1L;

flink-core/src/main/java/org/apache/flink/api/common/functions/RichMapPartitionFunction.java

@@ -18,6 +18,7 @@
 
 package org.apache.flink.api.common.functions;
 
+import org.apache.flink.annotation.Public;
 import org.apache.flink.util.Collector;
 
 /**
@@ -29,6 +30,7 @@ import org.apache.flink.util.Collector;
  * @param <I> Type of the input elements.
  * @param <O> Type of the returned elements.
  */
+@Public
 public abstract class RichMapPartitionFunction<I, O> extends AbstractRichFunction implements MapPartitionFunction<I, O> {
 
 	private static final long serialVersionUID = 1L;

flink-core/src/main/java/org/apache/flink/api/common/functions/RichReduceFunction.java

@@ -18,9 +18,7 @@
 
 package org.apache.flink.api.common.functions;
 
-import org.apache.flink.api.common.functions.AbstractRichFunction;
-import org.apache.flink.api.common.functions.ReduceFunction;
-import org.apache.flink.api.common.functions.RichFunction;
+import org.apache.flink.annotation.Public;
 
 /**
  * Rich variant of the {@link ReduceFunction}. As a {@link RichFunction}, it gives access to the
@@ -30,6 +28,7 @@ import org.apache.flink.api.common.functions.RichFunction;
  * 
  * @param <T> Type of the elements that this function processes.
  */
+@Public
 public abstract class RichReduceFunction<T> extends AbstractRichFunction implements ReduceFunction<T> {
 	
 	private static final long serialVersionUID = 1L;

flink-core/src/main/java/org/apache/flink/api/common/functions/RuntimeContext.java

@@ -22,6 +22,8 @@ import java.io.Serializable;
 import java.util.List;
 import java.util.Map;
 
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.annotation.Public;
 import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.common.accumulators.Accumulator;
 import org.apache.flink.api.common.accumulators.DoubleCounter;
@@ -40,6 +42,7 @@ import org.apache.flink.api.common.typeinfo.TypeInformation;
  * A function can, during runtime, obtain the RuntimeContext via a call to
  * {@link AbstractRichFunction#getRuntimeContext()}.
  */
+@Public
 public interface RuntimeContext {
 
 	/**
@@ -116,21 +119,25 @@ public interface RuntimeContext {
 	 * @deprecated Use getAccumulator(..) to obtain the value of an accumulator.
 	 */
 	@Deprecated
+	@Experimental
 	Map<String, Accumulator<?, ?>> getAllAccumulators();
 
 	/**
 	 * Convenience function to create a counter object for integers.
 	 */
+	@Experimental
 	IntCounter getIntCounter(String name);
 
 	/**
 	 * Convenience function to create a counter object for longs.
 	 */
+	@Experimental
 	LongCounter getLongCounter(String name);
 
 	/**
 	 * Convenience function to create a counter object for doubles.
 	 */
+	@Experimental
 	DoubleCounter getDoubleCounter(String name);
 
 	/**

flink-core/src/main/java/org/apache/flink/api/common/io/BinaryInputFormat.java

@@ -18,6 +18,7 @@
 
 package org.apache.flink.api.common.io;
 
+import org.apache.flink.annotation.Public;
 import org.apache.flink.api.common.io.statistics.BaseStatistics;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.core.fs.BlockLocation;
@@ -43,6 +44,7 @@ import java.util.List;
  * Base class for all input formats that use blocks of fixed size. The input splits are aligned to these blocks. Without
  * configuration, these block sizes equal the native block sizes of the HDFS.
  */
+@Public
 public abstract class BinaryInputFormat<T> extends FileInputFormat<T> {
 	private static final long serialVersionUID = 1L;
 

flink-core/src/main/java/org/apache/flink/api/common/io/BinaryOutputFormat.java

@@ -22,11 +22,13 @@ import java.io.FilterOutputStream;
 import java.io.IOException;
 import java.io.OutputStream;
 
+import org.apache.flink.annotation.Public;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.core.memory.DataOutputView;
 import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
 
 
+@Public
 public abstract class BinaryOutputFormat<T> extends FileOutputFormat<T> {
 	
 	private static final long serialVersionUID = 1L;

flink-core/src/main/java/org/apache/flink/api/common/io/BlockInfo.java

@@ -20,10 +20,12 @@ package org.apache.flink.api.common.io;
 
 import java.io.IOException;
 
+import org.apache.flink.annotation.Public;
 import org.apache.flink.core.io.IOReadableWritable;
 import org.apache.flink.core.memory.DataInputView;
 import org.apache.flink.core.memory.DataOutputView;
 
+@Public
 public class BlockInfo implements IOReadableWritable {
 
 	private long recordCount;

flink-core/src/main/java/org/apache/flink/api/common/io/CleanupWhenUnsuccessful.java

@@ -18,9 +18,12 @@
 
 package org.apache.flink.api.common.io;
 
+import org.apache.flink.annotation.Public;
+
 /**
  * {@link OutputFormat}s may implement this interface to run a cleanup hook when the execution is not successful.
  */
+@Public
 public interface CleanupWhenUnsuccessful {
 	
 	/**

flink-core/src/main/java/org/apache/flink/api/common/io/DelimitedInputFormat.java

@@ -21,6 +21,7 @@ package org.apache.flink.api.common.io;
 import java.io.IOException;
 import java.util.ArrayList;
 
+import org.apache.flink.annotation.Public;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.apache.flink.api.common.io.statistics.BaseStatistics;
@@ -41,6 +42,7 @@ import com.google.common.base.Charsets;
  * 
  * <p>The default delimiter is the newline character {@code '\n'}.</p>
  */
+@Public
 public abstract class DelimitedInputFormat<OT> extends FileInputFormat<OT> {
 	
 	private static final long serialVersionUID = 1L;

flink-core/src/main/java/org/apache/flink/api/common/io/FileInputFormat.java

@@ -28,6 +28,7 @@ import java.util.Map;
 import java.util.Set;
 
 import com.google.common.base.Preconditions;
+import org.apache.flink.annotation.Public;
 import org.apache.flink.api.common.io.compression.DeflateInflaterInputStreamFactory;
 import org.apache.flink.api.common.io.compression.GzipInflaterInputStreamFactory;
 import org.apache.flink.api.common.io.compression.InflaterInputStreamFactory;
@@ -53,6 +54,7 @@ import org.apache.flink.core.fs.Path;
  * <p>After the {@link #open(FileInputSplit)} method completed, the file input data is available
  * from the {@link #stream} field.</p>
  */
+@Public
 public abstract class FileInputFormat<OT> extends RichInputFormat<OT, FileInputSplit> {
 	
 	// -------------------------------------- Constants -------------------------------------------

flink-core/src/main/java/org/apache/flink/api/common/io/FileOutputFormat.java

@@ -21,6 +21,7 @@ package org.apache.flink.api.common.io;
 import java.io.FileNotFoundException;
 import java.io.IOException;
 
+import org.apache.flink.annotation.Public;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.apache.flink.configuration.ConfigConstants;
@@ -36,6 +37,7 @@ import org.apache.flink.core.fs.FileSystem.WriteMode;
  * open/close the target
  * file streams.
  */
+@Public
 public abstract class FileOutputFormat<IT> extends RichOutputFormat<IT> implements InitializeOnMaster, CleanupWhenUnsuccessful {
 	
 	private static final long serialVersionUID = 1L;

flink-core/src/main/java/org/apache/flink/api/common/io/FinalizeOnMaster.java

@@ -18,12 +18,15 @@
 
 package org.apache.flink.api.common.io;
 
+import org.apache.flink.annotation.Public;
+
 import java.io.IOException;
 
 /**
  * This interface may be implemented by {@link OutputFormat}s to have the master finalize them globally.
  * 
  */
+@Public
 public interface FinalizeOnMaster {
 
 	/**

flink-core/src/main/java/org/apache/flink/api/common/io/GenericCsvInputFormat.java

@@ -23,6 +23,7 @@ import com.google.common.base.Charsets;
 import com.google.common.base.Preconditions;
 import com.google.common.primitives.Ints;
 
+import org.apache.flink.annotation.Public;
 import org.apache.flink.core.fs.FileInputSplit;
 import org.apache.flink.core.fs.Path;
 import org.apache.flink.types.parser.FieldParser;
@@ -40,6 +41,7 @@ import java.util.ArrayList;
 import java.util.Map;
 import java.util.TreeMap;
 
+@Public
 public abstract class GenericCsvInputFormat<OT> extends DelimitedInputFormat<OT> {
 
 	private static final Logger LOG = LoggerFactory.getLogger(GenericCsvInputFormat.class);

flink-core/src/main/java/org/apache/flink/api/common/io/GenericInputFormat.java

@@ -21,6 +21,7 @@ package org.apache.flink.api.common.io;
 
 import java.io.IOException;
 
+import org.apache.flink.annotation.Public;
 import org.apache.flink.api.common.io.statistics.BaseStatistics;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.core.io.GenericInputSplit;
@@ -28,6 +29,7 @@ import org.apache.flink.core.io.GenericInputSplit;
 /**
  * Generic base class for all Rich inputs that are not based on files.
  */
+@Public
 public abstract class GenericInputFormat<OT> extends RichInputFormat<OT, GenericInputSplit> {
 
 	private static final long serialVersionUID = 1L;

flink-core/src/main/java/org/apache/flink/api/common/io/InitializeOnMaster.java

@@ -18,6 +18,8 @@
 
 package org.apache.flink.api.common.io;
 
+import org.apache.flink.annotation.Public;
+
 import java.io.IOException;
 
 /**
@@ -26,6 +28,7 @@ import java.io.IOException;
  * For example, the {@link FileOutputFormat} implements this behavior for distributed file systems and
  * creates/deletes target directories if necessary.
  */
+@Public
 public interface InitializeOnMaster {
 
 	/**

flink-core/src/main/java/org/apache/flink/api/common/io/InputFormat.java

@@ -21,6 +21,7 @@ package org.apache.flink.api.common.io;
 import java.io.IOException;
 import java.io.Serializable;
 
+import org.apache.flink.annotation.Public;
 import org.apache.flink.api.common.io.statistics.BaseStatistics;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.core.io.InputSplit;
@@ -60,6 +61,7 @@ import org.apache.flink.core.io.InputSplitSource;
  * @param <OT> The type of the produced records.
  * @param <T> The type of input split.
  */
+@Public
 public interface InputFormat<OT, T extends InputSplit> extends InputSplitSource<T>, Serializable {
 	
 	/**

flink-core/src/main/java/org/apache/flink/api/common/io/InputStreamFSInputWrapper.java

@@ -18,6 +18,7 @@
 
 package org.apache.flink.api.common.io;
 
+import org.apache.flink.annotation.Public;
 import org.apache.flink.core.fs.FSDataInputStream;
 
 import java.io.EOFException;
@@ -29,6 +30,7 @@ import java.io.InputStream;
  * <br>
  * <i>NB: {@link #seek(long)} and {@link #getPos()} are currently not supported.</i>
  */
+@Public
 public class InputStreamFSInputWrapper extends FSDataInputStream {
 
 	private final InputStream inStream;

flink-core/src/main/java/org/apache/flink/api/common/io/LocatableInputSplitAssigner.java

@@ -25,6 +25,7 @@ import java.util.Locale;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 
+import org.apache.flink.annotation.Public;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.apache.flink.core.io.InputSplitAssigner;
@@ -35,6 +36,7 @@ import org.apache.flink.util.NetUtils;
  * The locatable input split assigner assigns to each host splits that are local, before assigning
  * splits that are not local.
  */
+@Public
 public final class LocatableInputSplitAssigner implements InputSplitAssigner {
 
 	private static final Logger LOG = LoggerFactory.getLogger(LocatableInputSplitAssigner.class);

flink-core/src/main/java/org/apache/flink/api/common/io/NonParallelInput.java

@@ -18,6 +18,8 @@
 
 package org.apache.flink.api.common.io;
 
+import org.apache.flink.annotation.Public;
+
 /**
  * This interface acts as a marker for input formats for inputs which cannot be split.
  * Data sources with a non-parallel input formats are always executed with a parallelism
@@ -25,5 +27,6 @@ package org.apache.flink.api.common.io;
  * 
  * @see InputFormat
  */
+@Public
 public interface NonParallelInput {
 }

flink-core/src/main/java/org/apache/flink/api/common/io/OutputFormat.java

@@ -21,6 +21,7 @@ package org.apache.flink.api.common.io;
 import java.io.IOException;
 import java.io.Serializable;
 
+import org.apache.flink.annotation.Public;
 import org.apache.flink.configuration.Configuration;
 
 /**
@@ -38,6 +39,7 @@ import org.apache.flink.configuration.Configuration;
  * 
  * @param <IT> The type of the consumed records. 
  */
+@Public
 public interface OutputFormat<IT> extends Serializable {
 	
 	/**

flink-core/src/main/java/org/apache/flink/api/common/io/RichInputFormat.java

@@ -18,6 +18,7 @@ c * Licensed to the Apache Software Foundation (ASF) under one
 
 package org.apache.flink.api.common.io;
 
+import org.apache.flink.annotation.Public;
 import org.apache.flink.api.common.functions.RuntimeContext;
 import org.apache.flink.core.io.InputSplit;
 
@@ -25,6 +26,7 @@ import org.apache.flink.core.io.InputSplit;
  * An abstract stub implementation for Rich input formats.
  * Rich formats have access to their runtime execution context via {@link #getRuntimeContext()}.
  */
+@Public
 public abstract class RichInputFormat<OT, T extends InputSplit> implements InputFormat<OT, T> {
 	
 	private static final long serialVersionUID = 1L;

flink-core/src/main/java/org/apache/flink/api/common/io/RichOutputFormat.java

@@ -18,12 +18,14 @@ c * Licensed to the Apache Software Foundation (ASF) under one
 
 package org.apache.flink.api.common.io;
 
+import org.apache.flink.annotation.Public;
 import org.apache.flink.api.common.functions.RuntimeContext;
 
 /**
  * An abstract stub implementation for Rich output formats.
  * Rich formats have access to their runtime execution context via {@link #getRuntimeContext()}.
  */
+@Public
 public abstract class RichOutputFormat<IT> implements OutputFormat<IT> {
 	
 	private static final long serialVersionUID = 1L;

flink-core/src/main/java/org/apache/flink/api/common/io/SerializedInputFormat.java

@@ -20,6 +20,7 @@ package org.apache.flink.api.common.io;
 
 import java.io.IOException;
 
+import org.apache.flink.annotation.Public;
 import org.apache.flink.core.io.IOReadableWritable;
 import org.apache.flink.core.memory.DataInputView;
 
@@ -28,6 +29,7 @@ import org.apache.flink.core.memory.DataInputView;
  * 
  * @see SerializedOutputFormat
  */
+@Public
 public class SerializedInputFormat<T extends IOReadableWritable> extends BinaryInputFormat<T> {
 
 	private static final long serialVersionUID = 1L;

flink-core/src/main/java/org/apache/flink/api/common/io/SerializedOutputFormat.java

@@ -20,6 +20,7 @@ package org.apache.flink.api.common.io;
 
 import java.io.IOException;
 
+import org.apache.flink.annotation.Public;
 import org.apache.flink.core.io.IOReadableWritable;
 import org.apache.flink.core.memory.DataOutputView;
 
@@ -28,6 +29,7 @@ import org.apache.flink.core.memory.DataOutputView;
  * 
  * @see SerializedInputFormat
  */
+@Public
 public class SerializedOutputFormat<T extends IOReadableWritable> extends BinaryOutputFormat<T> {
 	
 	private static final long serialVersionUID = 1L;

flink-core/src/main/java/org/apache/flink/api/common/io/StrictlyLocalAssignment.java

@@ -18,4 +18,5 @@
 
 package org.apache.flink.api.common.io;
 
+
 public interface StrictlyLocalAssignment {}

flink-core/src/main/java/org/apache/flink/api/common/io/statistics/BaseStatistics.java

@@ -19,9 +19,12 @@
 
 package org.apache.flink.api.common.io.statistics;
 
+import org.apache.flink.annotation.Public;
+
 /**
  * Interface describing the basic statistics that can be obtained from the input.
  */
+@Public
 public interface BaseStatistics {
 	
 	/**

flink-core/src/main/java/org/apache/flink/api/common/operators/AbstractUdfOperator.java

@@ -73,7 +73,7 @@ public abstract class AbstractUdfOperator<OUT, FT extends Function> extends Oper
 		return userFunction;
 	}
 
-// --------------------------------------------------------------------------------------------
+	// --------------------------------------------------------------------------------------------
 
 	/**
 	 * Returns the input, or null, if none is set.

flink-core/src/main/java/org/apache/flink/api/common/operators/Order.java

@@ -19,9 +19,12 @@
 
 package org.apache.flink.api.common.operators;
 
+import org.apache.flink.annotation.Public;
+
 /**
  * Enumeration representing order. May represent no order, an ascending order or a descending order.
  */
+@Public
 public enum Order {
 	
 	/**

flink-core/src/main/java/org/apache/flink/api/common/operators/SingleInputSemanticProperties.java

@@ -21,11 +21,13 @@ package org.apache.flink.api.common.operators;
 import java.util.HashMap;
 import java.util.Map;
 
+import org.apache.flink.annotation.Public;
 import org.apache.flink.api.common.operators.util.FieldSet;
 
 /**
  * Container for the semantic properties associated to a single input operator.
  */
+@Public
 public class SingleInputSemanticProperties implements SemanticProperties {
 	private static final long serialVersionUID = 1L;
 

flink-core/src/main/java/org/apache/flink/api/common/state/OperatorState.java

@@ -18,6 +18,8 @@
 
 package org.apache.flink.api.common.state;
 
+import org.apache.flink.annotation.Public;
+
 import java.io.IOException;
 
 /**
@@ -32,6 +34,7 @@ import java.io.IOException;
  * 
  * @param <T> Type of the value in the operator state
  */
+@Public
 public interface OperatorState<T> {
 
 	/**

flink-core/src/main/java/org/apache/flink/api/common/typeinfo/TypeInformation.java

@@ -18,6 +18,7 @@
 
 package org.apache.flink.api.common.typeinfo;
 
+import org.apache.flink.annotation.Public;
 import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.common.typeutils.TypeSerializer;
 
@@ -67,6 +68,7 @@ import java.util.List;
  *
  * @param <T> The type represented by this type information.
  */
+@Public
 public abstract class TypeInformation<T> implements Serializable {
 	
 	private static final long serialVersionUID = -7742311969684489493L;

flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java

@@ -18,10 +18,13 @@
 
 package org.apache.flink.configuration;
 
+import org.apache.flink.annotation.Public;
+
 /**
  * This class contains all constants for the configuration. That includes the configuration keys and
  * the default values.
  */
+@Public
 public final class ConfigConstants {
 
 	// ------------------------------------------------------------------------

flink-core/src/main/java/org/apache/flink/configuration/Configuration.java

@@ -25,6 +25,7 @@ import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
 
+import org.apache.flink.annotation.Public;
 import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.core.io.IOReadableWritable;
 import org.apache.flink.core.memory.DataInputView;
@@ -36,6 +37,7 @@ import org.slf4j.LoggerFactory;
 /**
  * Lightweight configuration object which stores key/value pairs.
  */
+@Public
 public class Configuration extends ExecutionConfig.GlobalJobParameters 
 		implements IOReadableWritable, java.io.Serializable, Cloneable {
 

flink-core/src/main/java/org/apache/flink/configuration/UnmodifiableConfiguration.java

@@ -18,9 +18,12 @@
 
 package org.apache.flink.configuration;
 
+import org.apache.flink.annotation.Public;
+
 /**
  * Unmodifiable version of the Configuration class.
  */
+@Public
 public class UnmodifiableConfiguration extends Configuration {
 	
 	private static final long serialVersionUID = -8151292629158972280L;

flink-core/src/main/java/org/apache/flink/core/fs/BlockLocation.java

@@ -18,11 +18,14 @@
 
 package org.apache.flink.core.fs;
 
+import org.apache.flink.annotation.Public;
+
 import java.io.IOException;
 
 /**
  * A BlockLocation lists hosts, offset and length of block.
  */
+@Public
 public interface BlockLocation extends Comparable<BlockLocation> {
 
 	/**

flink-core/src/main/java/org/apache/flink/core/fs/FSDataInputStream.java

@@ -25,6 +25,8 @@
 
 package org.apache.flink.core.fs;
 
+import org.apache.flink.annotation.Public;
+
 import java.io.IOException;
 import java.io.InputStream;
 
@@ -32,6 +34,7 @@ import java.io.InputStream;
  * Interface for a data input stream to a file on a {@link FileSystem}.
  * 
  */
+@Public
 public abstract class FSDataInputStream extends InputStream {
 
 	/**

flink-core/src/main/java/org/apache/flink/core/fs/FSDataOutputStream.java

@@ -18,12 +18,15 @@
 
 package org.apache.flink.core.fs;
 
+import org.apache.flink.annotation.Public;
+
 import java.io.IOException;
 import java.io.OutputStream;
 
 /**
  * Interface for a data output stream to a file on a {@link FileSystem}.
  */
+@Public
 public abstract class FSDataOutputStream extends OutputStream {
 
 	public abstract void flush() throws IOException;

flink-core/src/main/java/org/apache/flink/core/fs/FileInputSplit.java

@@ -18,12 +18,14 @@
 
 package org.apache.flink.core.fs;
 
+import org.apache.flink.annotation.Public;
 import org.apache.flink.core.io.LocatableInputSplit;
 
 /**
  * A file input split provides information on a particular part of a file, possibly
  * hosted on a distributed file system and replicated among several hosts. 
  */
+@Public
 public class FileInputSplit extends LocatableInputSplit {
 
 	private static final long serialVersionUID = 1L;

flink-core/src/main/java/org/apache/flink/core/fs/FileStatus.java

@@ -25,11 +25,14 @@
 
 package org.apache.flink.core.fs;
 
+import org.apache.flink.annotation.Public;
+
 /**
  * Interface that represents the client side information for a file
  * independent of the file system.
  * 
  */
+@Public
 public interface FileStatus {
 
 	/**

flink-core/src/main/java/org/apache/flink/core/fs/FileSystem.java

@@ -34,6 +34,7 @@ import java.net.URISyntaxException;
 import java.util.HashMap;
 import java.util.Map;
 
+import org.apache.flink.annotation.Public;
 import org.apache.flink.util.OperatingSystem;
 
 /**
@@ -41,6 +42,7 @@ import org.apache.flink.util.OperatingSystem;
  * may be implemented as a distributed file system, or as a local
  * one that reflects the locally-connected disk.
  */
+@Public
 public abstract class FileSystem {
 
 	private static final String LOCAL_FILESYSTEM_CLASS = "org.apache.flink.core.fs.local.LocalFileSystem";

flink-core/src/main/java/org/apache/flink/core/fs/Path.java

@@ -27,6 +27,7 @@ import java.io.Serializable;
 import java.net.URI;
 import java.net.URISyntaxException;
 
+import org.apache.flink.annotation.Public;
 import org.apache.flink.core.io.IOReadableWritable;
 import org.apache.flink.core.memory.DataInputView;
 import org.apache.flink.core.memory.DataOutputView;
@@ -38,6 +39,7 @@ import org.apache.flink.util.StringUtils;
  *
  * Tailing slashes are removed from the path.
  */
+@Public
 public class Path implements IOReadableWritable, Serializable {
 	
 	private static final long serialVersionUID = 1L;

flink-core/src/main/java/org/apache/flink/core/io/GenericInputSplit.java

@@ -18,9 +18,12 @@
 
 package org.apache.flink.core.io;
 
+import org.apache.flink.annotation.Public;
+
 /**
  * A generic input split that has only a partition number.
  */
+@Public
 public class GenericInputSplit implements InputSplit, java.io.Serializable {
 
 	private static final long serialVersionUID = 1L;

flink-core/src/main/java/org/apache/flink/core/io/IOReadableWritable.java

@@ -20,6 +20,7 @@ package org.apache.flink.core.io;
 
 import java.io.IOException;
 
+import org.apache.flink.annotation.Public;
 import org.apache.flink.core.memory.DataInputView;
 import org.apache.flink.core.memory.DataOutputView;
 
@@ -30,6 +31,7 @@ import org.apache.flink.core.memory.DataOutputView;
  * to a binary representation.
  * When implementing this Interface make sure that the implementing class has a default (zero-argument) constructor!
  */
+@Public
 public interface IOReadableWritable {
 
 	/**

flink-core/src/main/java/org/apache/flink/core/io/InputSplit.java

@@ -18,6 +18,8 @@
 
 package org.apache.flink.core.io;
 
+import org.apache.flink.annotation.Public;
+
 import java.io.Serializable;
 
 /**
@@ -26,6 +28,7 @@ import java.io.Serializable;
  * <p>Input splits are transferred in serialized form via the messages, so they need to be serializable
  * as defined by {@link java.io.Serializable}.</p>
  */
+@Public
 public interface InputSplit extends Serializable {
 	
 	/**

flink-core/src/main/java/org/apache/flink/core/io/InputSplitAssigner.java

@@ -18,6 +18,7 @@
 
 package org.apache.flink.core.io;
 
+
 /**
  * An input split assigner distributes the {@link InputSplit}s among the instances on which a
  * data source exists.

flink-core/src/main/java/org/apache/flink/core/io/InputSplitSource.java

@@ -18,6 +18,8 @@
 
 package org.apache.flink.core.io;
 
+import org.apache.flink.annotation.Public;
+
 import java.io.Serializable;
 
 /**
@@ -26,6 +28,7 @@ import java.io.Serializable;
  *
  * @param <T> The type of the input splits created by the source.
  */
+@Public
 public interface InputSplitSource<T extends InputSplit> extends Serializable {
 
 	/**

flink-core/src/main/java/org/apache/flink/core/io/LocatableInputSplit.java

@@ -18,11 +18,14 @@
 
 package org.apache.flink.core.io;
 
+import org.apache.flink.annotation.Public;
+
 import java.util.Arrays;
 
 /**
  * A locatable input split is an input split referring to input data which is located on one or more hosts.
  */
+@Public
 public class LocatableInputSplit implements InputSplit, java.io.Serializable {
 	
 	private static final long serialVersionUID = 1L;

flink-core/src/main/java/org/apache/flink/core/memory/DataInputView.java

@@ -20,6 +20,8 @@
 package org.apache.flink.core.memory;
 
 
+import org.apache.flink.annotation.Public;
+
 import java.io.DataInput;
 import java.io.IOException;
 
@@ -28,6 +30,7 @@ import java.io.IOException;
  * This interface defines a view over some memory that can be used to sequentially read the contents of the memory.
  * The view is typically backed by one or more {@link org.apache.flink.core.memory.MemorySegment}.
  */
+@Public
 public interface DataInputView extends DataInput {
 	
 	/**

flink-core/src/main/java/org/apache/flink/core/memory/DataOutputView.java

@@ -20,6 +20,8 @@
 package org.apache.flink.core.memory;
 
 
+import org.apache.flink.annotation.Public;
+
 import java.io.DataOutput;
 import java.io.IOException;
 
@@ -28,6 +30,7 @@ import java.io.IOException;
  * This interface defines a view over some memory that can be used to sequentially write contents to the memory.
  * The view is typically backed by one or more {@link org.apache.flink.core.memory.MemorySegment}.
  */
+@Public
 public interface DataOutputView extends DataOutput {
 	
 	/**

flink-core/src/main/java/org/apache/flink/types/BooleanValue.java

@@ -21,6 +21,7 @@ package org.apache.flink.types;
 
 import java.io.IOException;
 
+import org.apache.flink.annotation.Public;
 import org.apache.flink.core.memory.DataInputView;
 import org.apache.flink.core.memory.DataOutputView;
 import org.apache.flink.core.memory.MemorySegment;
@@ -31,6 +32,7 @@ import org.apache.flink.core.memory.MemorySegment;
  * 
  * @see org.apache.flink.types.Key
  */
+@Public
 public class BooleanValue implements NormalizableKey<BooleanValue>, ResettableValue<BooleanValue>, CopyableValue<BooleanValue> {
 	
 	private static final long serialVersionUID = 1L;

flink-core/src/main/java/org/apache/flink/types/ByteValue.java

@@ -21,6 +21,7 @@ package org.apache.flink.types;
 
 import java.io.IOException;
 
+import org.apache.flink.annotation.Public;
 import org.apache.flink.core.memory.DataInputView;
 import org.apache.flink.core.memory.DataOutputView;
 import org.apache.flink.core.memory.MemorySegment;
@@ -31,6 +32,7 @@ import org.apache.flink.core.memory.MemorySegment;
  * 
  * @see org.apache.flink.types.Key
  */
+@Public
 public class ByteValue implements NormalizableKey<ByteValue>, ResettableValue<ByteValue>, CopyableValue<ByteValue> {
 	private static final long serialVersionUID = 1L;
 	

flink-core/src/main/java/org/apache/flink/types/CharValue.java

@@ -21,6 +21,7 @@ package org.apache.flink.types;
 
 import java.io.IOException;
 
+import org.apache.flink.annotation.Public;
 import org.apache.flink.core.memory.DataInputView;
 import org.apache.flink.core.memory.DataOutputView;
 import org.apache.flink.core.memory.MemorySegment;
@@ -31,6 +32,7 @@ import org.apache.flink.core.memory.MemorySegment;
  * 
  * @see org.apache.flink.types.Key
  */
+@Public
 public class CharValue implements NormalizableKey<CharValue>, ResettableValue<CharValue>, CopyableValue<CharValue> {
 	private static final long serialVersionUID = 1L;