flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From rmetz...@apache.org
Subject [2/2] flink git commit: [FLINK-3093] Introduce annotations for interface stability in flink-core
Date Thu, 07 Jan 2016 15:03:18 GMT
[FLINK-3093] Introduce annotations for interface stability in flink-core

This closes #1427


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/c674a655
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/c674a655
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/c674a655

Branch: refs/heads/master
Commit: c674a6558e07f1d918961a21ddaad4a6148698a0
Parents: dcf86c2
Author: Robert Metzger <rmetzger@apache.org>
Authored: Tue Dec 1 19:32:00 2015 +0100
Committer: Robert Metzger <rmetzger@apache.org>
Committed: Thu Jan 7 16:01:23 2016 +0100

----------------------------------------------------------------------
 flink-annotations/pom.xml                       | 37 ++++++++++++++++++++
 .../apache/flink/annotation/Experimental.java   | 34 ++++++++++++++++++
 .../org/apache/flink/annotation/Internal.java   | 33 +++++++++++++++++
 .../org/apache/flink/annotation/Public.java     | 35 ++++++++++++++++++
 flink-core/pom.xml                              |  6 ++++
 .../flink/api/common/CodeAnalysisMode.java      |  3 ++
 .../flink/api/common/ExecutionConfig.java       | 10 ++++++
 .../apache/flink/api/common/ExecutionMode.java  |  3 ++
 .../flink/api/common/JobExecutionResult.java    |  6 ++++
 .../java/org/apache/flink/api/common/JobID.java |  2 ++
 .../flink/api/common/JobSubmissionResult.java   |  3 ++
 .../api/common/accumulators/Accumulator.java    |  3 ++
 .../common/accumulators/AverageAccumulator.java |  3 ++
 .../api/common/accumulators/DoubleCounter.java  |  1 +
 .../api/common/accumulators/Histogram.java      |  3 ++
 .../api/common/accumulators/IntCounter.java     |  1 +
 .../common/accumulators/ListAccumulator.java    |  3 ++
 .../common/accumulators/SimpleAccumulator.java  |  3 ++
 .../api/common/cache/DistributedCache.java      |  4 +--
 .../common/functions/AbstractRichFunction.java  |  2 ++
 .../functions/BroadcastVariableInitializer.java |  3 ++
 .../api/common/functions/CoGroupFunction.java   |  2 ++
 .../api/common/functions/CombineFunction.java   |  3 ++
 .../api/common/functions/CrossFunction.java     |  3 ++
 .../api/common/functions/FilterFunction.java    |  3 ++
 .../api/common/functions/FlatJoinFunction.java  |  2 ++
 .../api/common/functions/FlatMapFunction.java   |  2 ++
 .../api/common/functions/FoldFunction.java      |  3 ++
 .../flink/api/common/functions/Function.java    |  3 ++
 .../common/functions/GroupCombineFunction.java  |  2 ++
 .../common/functions/GroupReduceFunction.java   |  2 ++
 .../common/functions/InvalidTypesException.java |  2 ++
 .../functions/IterationRuntimeContext.java      |  9 ++---
 .../api/common/functions/JoinFunction.java      |  3 ++
 .../flink/api/common/functions/MapFunction.java |  3 ++
 .../common/functions/MapPartitionFunction.java  |  2 ++
 .../flink/api/common/functions/Partitioner.java |  3 ++
 .../api/common/functions/ReduceFunction.java    |  3 ++
 .../common/functions/RichCoGroupFunction.java   |  5 ++-
 .../api/common/functions/RichCrossFunction.java |  5 ++-
 .../common/functions/RichFilterFunction.java    |  5 ++-
 .../common/functions/RichFlatJoinFunction.java  |  5 ++-
 .../common/functions/RichFlatMapFunction.java   |  5 ++-
 .../api/common/functions/RichFoldFunction.java  |  5 ++-
 .../api/common/functions/RichFunction.java      |  2 ++
 .../functions/RichGroupCombineFunction.java     |  2 ++
 .../functions/RichGroupReduceFunction.java      |  3 ++
 .../api/common/functions/RichJoinFunction.java  |  5 ++-
 .../api/common/functions/RichMapFunction.java   |  5 ++-
 .../functions/RichMapPartitionFunction.java     |  2 ++
 .../common/functions/RichReduceFunction.java    |  5 ++-
 .../api/common/functions/RuntimeContext.java    |  7 ++++
 .../flink/api/common/io/BinaryInputFormat.java  |  2 ++
 .../flink/api/common/io/BinaryOutputFormat.java |  2 ++
 .../apache/flink/api/common/io/BlockInfo.java   |  2 ++
 .../api/common/io/CleanupWhenUnsuccessful.java  |  3 ++
 .../api/common/io/DelimitedInputFormat.java     |  2 ++
 .../flink/api/common/io/FileInputFormat.java    |  2 ++
 .../flink/api/common/io/FileOutputFormat.java   |  2 ++
 .../flink/api/common/io/FinalizeOnMaster.java   |  3 ++
 .../api/common/io/GenericCsvInputFormat.java    |  2 ++
 .../flink/api/common/io/GenericInputFormat.java |  2 ++
 .../flink/api/common/io/InitializeOnMaster.java |  3 ++
 .../apache/flink/api/common/io/InputFormat.java |  2 ++
 .../common/io/InputStreamFSInputWrapper.java    |  2 ++
 .../common/io/LocatableInputSplitAssigner.java  |  2 ++
 .../flink/api/common/io/NonParallelInput.java   |  3 ++
 .../flink/api/common/io/OutputFormat.java       |  2 ++
 .../flink/api/common/io/RichInputFormat.java    |  2 ++
 .../flink/api/common/io/RichOutputFormat.java   |  2 ++
 .../api/common/io/SerializedInputFormat.java    |  2 ++
 .../api/common/io/SerializedOutputFormat.java   |  2 ++
 .../api/common/io/StrictlyLocalAssignment.java  |  1 +
 .../common/io/statistics/BaseStatistics.java    |  3 ++
 .../common/operators/AbstractUdfOperator.java   |  2 +-
 .../flink/api/common/operators/Order.java       |  3 ++
 .../SingleInputSemanticProperties.java          |  2 ++
 .../flink/api/common/state/OperatorState.java   |  3 ++
 .../api/common/typeinfo/TypeInformation.java    |  2 ++
 .../flink/configuration/ConfigConstants.java    |  3 ++
 .../flink/configuration/Configuration.java      |  2 ++
 .../UnmodifiableConfiguration.java              |  3 ++
 .../org/apache/flink/core/fs/BlockLocation.java |  3 ++
 .../apache/flink/core/fs/FSDataInputStream.java |  3 ++
 .../flink/core/fs/FSDataOutputStream.java       |  3 ++
 .../apache/flink/core/fs/FileInputSplit.java    |  2 ++
 .../org/apache/flink/core/fs/FileStatus.java    |  3 ++
 .../org/apache/flink/core/fs/FileSystem.java    |  2 ++
 .../java/org/apache/flink/core/fs/Path.java     |  2 ++
 .../apache/flink/core/io/GenericInputSplit.java |  3 ++
 .../flink/core/io/IOReadableWritable.java       |  2 ++
 .../org/apache/flink/core/io/InputSplit.java    |  3 ++
 .../flink/core/io/InputSplitAssigner.java       |  1 +
 .../apache/flink/core/io/InputSplitSource.java  |  3 ++
 .../flink/core/io/LocatableInputSplit.java      |  3 ++
 .../apache/flink/core/memory/DataInputView.java |  3 ++
 .../flink/core/memory/DataOutputView.java       |  3 ++
 .../org/apache/flink/types/BooleanValue.java    |  2 ++
 .../java/org/apache/flink/types/ByteValue.java  |  2 ++
 .../java/org/apache/flink/types/CharValue.java  |  2 ++
 .../org/apache/flink/types/CopyableValue.java   |  2 ++
 .../flink/types/DeserializationException.java   |  3 ++
 .../org/apache/flink/types/DoubleValue.java     |  2 ++
 .../java/org/apache/flink/types/FloatValue.java |  2 ++
 .../java/org/apache/flink/types/IntValue.java   |  2 ++
 .../types/KeyFieldOutOfBoundsException.java     |  3 ++
 .../java/org/apache/flink/types/ListValue.java  |  2 ++
 .../java/org/apache/flink/types/LongValue.java  |  2 ++
 .../java/org/apache/flink/types/MapValue.java   |  2 ++
 .../org/apache/flink/types/NormalizableKey.java |  2 ++
 .../java/org/apache/flink/types/Nothing.java    |  2 +-
 .../apache/flink/types/NullFieldException.java  |  3 ++
 .../flink/types/NullKeyFieldException.java      |  2 --
 .../java/org/apache/flink/types/NullValue.java  |  2 ++
 .../java/org/apache/flink/types/Record.java     |  2 ++
 .../org/apache/flink/types/ResettableValue.java |  3 ++
 .../java/org/apache/flink/types/ShortValue.java |  2 ++
 .../org/apache/flink/types/StringValue.java     |  2 ++
 .../main/java/org/apache/flink/types/Value.java |  2 ++
 .../java/org/apache/flink/util/Collector.java   |  3 ++
 .../flink/util/NumberSequenceIterator.java      |  3 ++
 .../apache/flink/util/SplittableIterator.java   |  2 ++
 .../flink/util/TraversableOnceException.java    |  3 ++
 .../org/apache/flink/util/XORShiftRandom.java   |  3 ++
 pom.xml                                         |  3 +-
 125 files changed, 443 insertions(+), 38 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-annotations/pom.xml
----------------------------------------------------------------------
diff --git a/flink-annotations/pom.xml b/flink-annotations/pom.xml
new file mode 100644
index 0000000..748d6f8
--- /dev/null
+++ b/flink-annotations/pom.xml
@@ -0,0 +1,37 @@
+<?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>

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-annotations/src/main/java/org/apache/flink/annotation/Experimental.java
----------------------------------------------------------------------
diff --git a/flink-annotations/src/main/java/org/apache/flink/annotation/Experimental.java b/flink-annotations/src/main/java/org/apache/flink/annotation/Experimental.java
new file mode 100644
index 0000000..bf9a97c
--- /dev/null
+++ b/flink-annotations/src/main/java/org/apache/flink/annotation/Experimental.java
@@ -0,0 +1,34 @@
+/*
+ * 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 {
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-annotations/src/main/java/org/apache/flink/annotation/Internal.java
----------------------------------------------------------------------
diff --git a/flink-annotations/src/main/java/org/apache/flink/annotation/Internal.java b/flink-annotations/src/main/java/org/apache/flink/annotation/Internal.java
new file mode 100644
index 0000000..4df6847
--- /dev/null
+++ b/flink-annotations/src/main/java/org/apache/flink/annotation/Internal.java
@@ -0,0 +1,33 @@
+/*
+ * 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 {
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-annotations/src/main/java/org/apache/flink/annotation/Public.java
----------------------------------------------------------------------
diff --git a/flink-annotations/src/main/java/org/apache/flink/annotation/Public.java b/flink-annotations/src/main/java/org/apache/flink/annotation/Public.java
new file mode 100644
index 0000000..28c4d83
--- /dev/null
+++ b/flink-annotations/src/main/java/org/apache/flink/annotation/Public.java
@@ -0,0 +1,35 @@
+/*
+ * 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 {}

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/pom.xml
----------------------------------------------------------------------
diff --git a/flink-core/pom.xml b/flink-core/pom.xml
index f91608e..adc9a9b 100644
--- a/flink-core/pom.xml
+++ b/flink-core/pom.xml
@@ -37,6 +37,12 @@ under the License.
 	<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>
 			<version>${project.version}</version>
 		</dependency>

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/CodeAnalysisMode.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/CodeAnalysisMode.java b/flink-core/src/main/java/org/apache/flink/api/common/CodeAnalysisMode.java
index e9d8541..a13b653 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/CodeAnalysisMode.java
+++ b/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 {
 
 	/**

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java b/flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java
index a0d3363..3e2e2fd 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java
+++ b/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;
 	}

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/ExecutionMode.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/ExecutionMode.java b/flink-core/src/main/java/org/apache/flink/api/common/ExecutionMode.java
index b1463cc..f045bac 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/ExecutionMode.java
+++ b/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 {
 
 	/**

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/JobExecutionResult.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/JobExecutionResult.java b/flink-core/src/main/java/org/apache/flink/api/common/JobExecutionResult.java
index 92d2b98..1fff4da 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/JobExecutionResult.java
+++ b/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) {

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/JobID.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/JobID.java b/flink-core/src/main/java/org/apache/flink/api/common/JobID.java
index 13a1a32..78cddfe 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/JobID.java
+++ b/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;

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/JobSubmissionResult.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/JobSubmissionResult.java b/flink-core/src/main/java/org/apache/flink/api/common/JobSubmissionResult.java
index 3a18eb4..91a838b 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/JobSubmissionResult.java
+++ b/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;

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/accumulators/Accumulator.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/accumulators/Accumulator.java b/flink-core/src/main/java/org/apache/flink/api/common/accumulators/Accumulator.java
index 2ee5fa0..bf56cfe 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/accumulators/Accumulator.java
+++ b/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

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/accumulators/AverageAccumulator.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/accumulators/AverageAccumulator.java b/flink-core/src/main/java/org/apache/flink/api/common/accumulators/AverageAccumulator.java
index 0db1942..9c0f62f 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/accumulators/AverageAccumulator.java
+++ b/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;

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/accumulators/DoubleCounter.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/accumulators/DoubleCounter.java b/flink-core/src/main/java/org/apache/flink/api/common/accumulators/DoubleCounter.java
index b7a4bfc..3e6a5dd 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/accumulators/DoubleCounter.java
+++ b/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.
  */

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/accumulators/Histogram.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/accumulators/Histogram.java b/flink-core/src/main/java/org/apache/flink/api/common/accumulators/Histogram.java
index f587fee..577795b 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/accumulators/Histogram.java
+++ b/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;

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/accumulators/IntCounter.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/accumulators/IntCounter.java b/flink-core/src/main/java/org/apache/flink/api/common/accumulators/IntCounter.java
index a97ba51..133eb5c 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/accumulators/IntCounter.java
+++ b/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.
  */

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/accumulators/ListAccumulator.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/accumulators/ListAccumulator.java b/flink-core/src/main/java/org/apache/flink/api/common/accumulators/ListAccumulator.java
index 3af785c..917f4ba 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/accumulators/ListAccumulator.java
+++ b/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;

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/accumulators/SimpleAccumulator.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/accumulators/SimpleAccumulator.java b/flink-core/src/main/java/org/apache/flink/api/common/accumulators/SimpleAccumulator.java
index 6614e13..288f484 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/accumulators/SimpleAccumulator.java
+++ b/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> {}

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/cache/DistributedCache.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/cache/DistributedCache.java b/flink-core/src/main/java/org/apache/flink/api/common/cache/DistributedCache.java
index a4261d0..35a82e8 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/cache/DistributedCache.java
+++ b/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 {

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/functions/AbstractRichFunction.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/AbstractRichFunction.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/AbstractRichFunction.java
index fd9de67..ff2cbea 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/functions/AbstractRichFunction.java
+++ b/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;

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/functions/BroadcastVariableInitializer.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/BroadcastVariableInitializer.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/BroadcastVariableInitializer.java
index d38c1f8..65c3c1f 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/functions/BroadcastVariableInitializer.java
+++ b/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> {
 
 	/**

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/functions/CoGroupFunction.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/CoGroupFunction.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/CoGroupFunction.java
index 1098341..db40370 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/functions/CoGroupFunction.java
+++ b/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 {
 	
 	/**

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/functions/CombineFunction.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/CombineFunction.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/CombineFunction.java
index af115b0..af05c0d 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/functions/CombineFunction.java
+++ b/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 {
 
 	/**

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/functions/CrossFunction.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/CrossFunction.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/CrossFunction.java
index 3660e63..d264e02 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/functions/CrossFunction.java
+++ b/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 {
 
 	/**

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/functions/FilterFunction.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/FilterFunction.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/FilterFunction.java
index 1a80cb8..e8e83b6 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/functions/FilterFunction.java
+++ b/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 {
 	
 	/**

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/functions/FlatJoinFunction.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/FlatJoinFunction.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/FlatJoinFunction.java
index c79046b..5d7c0ee 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/functions/FlatJoinFunction.java
+++ b/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 {
 
 	/**

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/functions/FlatMapFunction.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/FlatMapFunction.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/FlatMapFunction.java
index 5b955f7..8b4a8f2 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/functions/FlatMapFunction.java
+++ b/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 {
 
 	/**

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/functions/FoldFunction.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/FoldFunction.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/FoldFunction.java
index c4be00b..8194663 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/functions/FoldFunction.java
+++ b/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.

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/functions/Function.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/Function.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/Function.java
index 8a1819c..86c42f8 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/functions/Function.java
+++ b/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 {
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/functions/GroupCombineFunction.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/GroupCombineFunction.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/GroupCombineFunction.java
index c0b153b..53e7ade 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/functions/GroupCombineFunction.java
+++ b/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 {
 
 	/**

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/functions/GroupReduceFunction.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/GroupReduceFunction.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/GroupReduceFunction.java
index 36500f6..37490c5 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/functions/GroupReduceFunction.java
+++ b/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 {
 	
 	/**

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/functions/InvalidTypesException.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/InvalidTypesException.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/InvalidTypesException.java
index 611765e..143180f 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/functions/InvalidTypesException.java
+++ b/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;

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/functions/IterationRuntimeContext.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/IterationRuntimeContext.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/IterationRuntimeContext.java
index 8239921..5dc4ec8 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/functions/IterationRuntimeContext.java
+++ b/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);

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/functions/JoinFunction.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/JoinFunction.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/JoinFunction.java
index 0032a0f..5897b47 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/functions/JoinFunction.java
+++ b/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 {
 
 	/**

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/functions/MapFunction.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/MapFunction.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/MapFunction.java
index 45f3d1f..9a0a45a 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/functions/MapFunction.java
+++ b/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 {
 
 	/**

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/functions/MapPartitionFunction.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/MapPartitionFunction.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/MapPartitionFunction.java
index d2c0077..4caaadd 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/functions/MapPartitionFunction.java
+++ b/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 {
 	
 	/**

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/functions/Partitioner.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/Partitioner.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/Partitioner.java
index f686e94..6c237ed 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/functions/Partitioner.java
+++ b/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 {
 
 	/**

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/functions/ReduceFunction.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/ReduceFunction.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/ReduceFunction.java
index 1b34047..3579680 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/functions/ReduceFunction.java
+++ b/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 {
 
 	/**

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/functions/RichCoGroupFunction.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/RichCoGroupFunction.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/RichCoGroupFunction.java
index 2e1506e..517e512 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/functions/RichCoGroupFunction.java
+++ b/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;

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/functions/RichCrossFunction.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/RichCrossFunction.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/RichCrossFunction.java
index c85ce93..dbeefd5 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/functions/RichCrossFunction.java
+++ b/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;

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/functions/RichFilterFunction.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/RichFilterFunction.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/RichFilterFunction.java
index 131baf3..4458038 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/functions/RichFilterFunction.java
+++ b/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;

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/functions/RichFlatJoinFunction.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/RichFlatJoinFunction.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/RichFlatJoinFunction.java
index b8bc1b9..0e06d81 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/functions/RichFlatJoinFunction.java
+++ b/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;

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/functions/RichFlatMapFunction.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/RichFlatMapFunction.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/RichFlatMapFunction.java
index ae61023..94443b8 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/functions/RichFlatMapFunction.java
+++ b/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;

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/functions/RichFoldFunction.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/RichFoldFunction.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/RichFoldFunction.java
index f4c2942..245550d 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/functions/RichFoldFunction.java
+++ b/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;

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/functions/RichFunction.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/RichFunction.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/RichFunction.java
index 0cbde4a..c0f0de2 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/functions/RichFunction.java
+++ b/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 {
 	
 	/**

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/functions/RichGroupCombineFunction.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/RichGroupCombineFunction.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/RichGroupCombineFunction.java
index 55df232..de34031 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/functions/RichGroupCombineFunction.java
+++ b/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;

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/functions/RichGroupReduceFunction.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/RichGroupReduceFunction.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/RichGroupReduceFunction.java
index b5edc64..94d60a8 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/functions/RichGroupReduceFunction.java
+++ b/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 {}
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/functions/RichJoinFunction.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/RichJoinFunction.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/RichJoinFunction.java
index 1666a19..ceeca79 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/functions/RichJoinFunction.java
+++ b/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;

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/functions/RichMapFunction.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/RichMapFunction.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/RichMapFunction.java
index 7adb25b..81fec5d 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/functions/RichMapFunction.java
+++ b/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;

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/functions/RichMapPartitionFunction.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/RichMapPartitionFunction.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/RichMapPartitionFunction.java
index d8b7e9b..b3b800a 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/functions/RichMapPartitionFunction.java
+++ b/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;

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/functions/RichReduceFunction.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/RichReduceFunction.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/RichReduceFunction.java
index 13e7e3f..7b42ec7 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/functions/RichReduceFunction.java
+++ b/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;

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/functions/RuntimeContext.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/RuntimeContext.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/RuntimeContext.java
index da79a79..a1e9d7d 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/functions/RuntimeContext.java
+++ b/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);
 
 	/**

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/io/BinaryInputFormat.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/io/BinaryInputFormat.java b/flink-core/src/main/java/org/apache/flink/api/common/io/BinaryInputFormat.java
index 46a5d58..61e3a1a 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/io/BinaryInputFormat.java
+++ b/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;
 

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/io/BinaryOutputFormat.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/io/BinaryOutputFormat.java b/flink-core/src/main/java/org/apache/flink/api/common/io/BinaryOutputFormat.java
index 9b4e1cf..a89e73e 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/io/BinaryOutputFormat.java
+++ b/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;

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/io/BlockInfo.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/io/BlockInfo.java b/flink-core/src/main/java/org/apache/flink/api/common/io/BlockInfo.java
index 71e244d..0ac2e50 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/io/BlockInfo.java
+++ b/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;

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/io/CleanupWhenUnsuccessful.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/io/CleanupWhenUnsuccessful.java b/flink-core/src/main/java/org/apache/flink/api/common/io/CleanupWhenUnsuccessful.java
index 4b912e1..7f77677 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/io/CleanupWhenUnsuccessful.java
+++ b/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 {
 	
 	/**

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/io/DelimitedInputFormat.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/io/DelimitedInputFormat.java b/flink-core/src/main/java/org/apache/flink/api/common/io/DelimitedInputFormat.java
index cb32fc3..f6b6d49 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/io/DelimitedInputFormat.java
+++ b/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;

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/io/FileInputFormat.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/io/FileInputFormat.java b/flink-core/src/main/java/org/apache/flink/api/common/io/FileInputFormat.java
index 37e8749..c4cd2b3 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/io/FileInputFormat.java
+++ b/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 -------------------------------------------

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/io/FileOutputFormat.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/io/FileOutputFormat.java b/flink-core/src/main/java/org/apache/flink/api/common/io/FileOutputFormat.java
index a5515c8..0828785 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/io/FileOutputFormat.java
+++ b/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;

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/io/FinalizeOnMaster.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/io/FinalizeOnMaster.java b/flink-core/src/main/java/org/apache/flink/api/common/io/FinalizeOnMaster.java
index 8b787eb..879714f 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/io/FinalizeOnMaster.java
+++ b/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 {
 
 	/**

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/io/GenericCsvInputFormat.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/io/GenericCsvInputFormat.java b/flink-core/src/main/java/org/apache/flink/api/common/io/GenericCsvInputFormat.java
index 89caf2f..ddbf165 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/io/GenericCsvInputFormat.java
+++ b/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);

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/io/GenericInputFormat.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/io/GenericInputFormat.java b/flink-core/src/main/java/org/apache/flink/api/common/io/GenericInputFormat.java
index 0e3b1c2..2be6f7a 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/io/GenericInputFormat.java
+++ b/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;

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/io/InitializeOnMaster.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/io/InitializeOnMaster.java b/flink-core/src/main/java/org/apache/flink/api/common/io/InitializeOnMaster.java
index 943fc8d..0fb5157 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/io/InitializeOnMaster.java
+++ b/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 {
 
 	/**

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/io/InputFormat.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/io/InputFormat.java b/flink-core/src/main/java/org/apache/flink/api/common/io/InputFormat.java
index 7bb1598..0e978b9 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/io/InputFormat.java
+++ b/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 {
 	
 	/**

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/io/InputStreamFSInputWrapper.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/io/InputStreamFSInputWrapper.java b/flink-core/src/main/java/org/apache/flink/api/common/io/InputStreamFSInputWrapper.java
index 9cd7dbf..cfd94bc 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/io/InputStreamFSInputWrapper.java
+++ b/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;

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/io/LocatableInputSplitAssigner.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/io/LocatableInputSplitAssigner.java b/flink-core/src/main/java/org/apache/flink/api/common/io/LocatableInputSplitAssigner.java
index c038da6..517f40a 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/io/LocatableInputSplitAssigner.java
+++ b/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);

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/io/NonParallelInput.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/io/NonParallelInput.java b/flink-core/src/main/java/org/apache/flink/api/common/io/NonParallelInput.java
index 26e49a7..78e9766 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/io/NonParallelInput.java
+++ b/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 {
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/c674a655/flink-core/src/main/java/org/apache/flink/api/common/io/OutputFormat.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/io/OutputFormat.java b/flink-core/src/main/java/org/apache/flink/api/common/io/OutputFormat.java
index 7594721..eb7ef40 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/io/OutputFormat.java
+++ b/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 {
 	
 	/**


Mime
View raw message