[FLINK-3093] Introduce annotations for interface stability in remaini…

…ng modules

This closes apache#1428

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

@@ -29,6 +29,7 @@
  * An experimental API might change between minor releases.
  */
 @Documented
-@Target({ ElementType.TYPE, ElementType.METHOD })
+@Target({ ElementType.TYPE, ElementType.METHOD, ElementType.FIELD, ElementType.CONSTRUCTOR })
+@Public
 public @interface Experimental {
 }

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

@@ -28,6 +28,7 @@
  * Developer APIs are stable but internal to Flink and might change across releases.
  */
 @Documented
-@Target({ ElementType.TYPE, ElementType.METHOD })
+@Target({ ElementType.TYPE, ElementType.METHOD, ElementType.CONSTRUCTOR })
+@Public
 public @interface Internal {
 }

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

@@ -32,4 +32,5 @@
  */
 @Documented
 @Target(ElementType.TYPE)
+@Public
 public @interface Public {}

flink-batch-connectors/flink-avro/src/main/java/org/apache/flink/api/java/io/AvroInputFormat.java

@@ -38,7 +38,6 @@
 import org.apache.flink.core.fs.Path;
 import org.apache.flink.util.InstantiationUtil;
 
-
 public class AvroInputFormat<E> extends FileInputFormat<E> implements ResultTypeQueryable<E> {
 
  private static final long serialVersionUID = 1L;

flink-batch-connectors/flink-avro/src/main/java/org/apache/flink/api/java/io/AvroOutputFormat.java

@@ -15,11 +15,8 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-
-
 package org.apache.flink.api.java.io;
 
-
 import org.apache.avro.Schema;
 import org.apache.avro.file.DataFileWriter;
 import org.apache.avro.io.DatumWriter;

flink-batch-connectors/flink-hadoop-compatibility/src/main/java/org/apache/flink/hadoopcompatibility/mapred/HadoopMapFunction.java

@@ -23,6 +23,7 @@
 import java.io.ObjectOutputStream;
 import java.io.Serializable;
 
+import org.apache.flink.annotation.Public;
 import org.apache.flink.api.common.functions.RichFlatMapFunction;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.tuple.Tuple2;
@@ -42,6 +43,7 @@
  * This wrapper maps a Hadoop Mapper (mapred API) to a Flink FlatMapFunction. 
  */
 @SuppressWarnings("rawtypes")
+@Public
 public final class HadoopMapFunction<KEYIN, VALUEIN, KEYOUT, VALUEOUT> 
  extends RichFlatMapFunction<Tuple2<KEYIN,VALUEIN>, Tuple2<KEYOUT,VALUEOUT>> 
  implements ResultTypeQueryable<Tuple2<KEYOUT,VALUEOUT>>, Serializable {

flink-batch-connectors/flink-hadoop-compatibility/src/main/java/org/apache/flink/hadoopcompatibility/mapred/HadoopReduceCombineFunction.java

@@ -23,6 +23,7 @@
 import java.io.ObjectOutputStream;
 import java.io.Serializable;
 
+import org.apache.flink.annotation.Public;
 import org.apache.flink.api.common.functions.GroupCombineFunction;
 import org.apache.flink.api.common.functions.RichGroupReduceFunction;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
@@ -45,6 +46,7 @@
  * This wrapper maps a Hadoop Reducer and Combiner (mapred API) to a combinable Flink GroupReduceFunction.
  */
 @SuppressWarnings("rawtypes")
+@Public
 public final class HadoopReduceCombineFunction<KEYIN, VALUEIN, KEYOUT, VALUEOUT> 
  extends RichGroupReduceFunction<Tuple2<KEYIN,VALUEIN>,Tuple2<KEYOUT,VALUEOUT>>
  implements GroupCombineFunction<Tuple2<KEYIN,VALUEIN>, Tuple2<KEYIN,VALUEIN>>,

flink-batch-connectors/flink-hadoop-compatibility/src/main/java/org/apache/flink/hadoopcompatibility/mapred/HadoopReduceFunction.java

@@ -23,6 +23,7 @@
 import java.io.ObjectOutputStream;
 import java.io.Serializable;
 
+import org.apache.flink.annotation.Public;
 import org.apache.flink.api.common.functions.RichGroupReduceFunction;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.common.typeutils.TypeSerializer;
@@ -44,6 +45,7 @@
  * This wrapper maps a Hadoop Reducer (mapred API) to a non-combinable Flink GroupReduceFunction. 
  */
 @SuppressWarnings("rawtypes")
+@Public
 public final class HadoopReduceFunction<KEYIN, VALUEIN, KEYOUT, VALUEOUT> 
  extends RichGroupReduceFunction<Tuple2<KEYIN,VALUEIN>,Tuple2<KEYOUT,VALUEOUT>> 
  implements ResultTypeQueryable<Tuple2<KEYOUT,VALUEOUT>>, Serializable {

flink-batch-connectors/flink-hbase/src/main/java/org/apache/flink/addons/hbase/TableInputFormat.java

@@ -15,8 +15,6 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-
-
 package org.apache.flink.addons.hbase;
 
 import java.io.IOException;

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

@@ -239,6 +239,7 @@ public interface RuntimeContext {
  * @throws UnsupportedOperationException Thrown, if no partitioned state is available for the
  * function (function is not part of a KeyedStream).
  */
+ @Experimental
  <T> ValueState<T> getState(ValueStateDescriptor<T> stateProperties);
 
  /**
@@ -282,6 +283,7 @@ public interface RuntimeContext {
  * @throws UnsupportedOperationException Thrown, if no partitioned state is available for the
  * function (function is not part os a KeyedStream).
  */
+ @Experimental
  <T> ListState<T> getListState(ListStateDescriptor<T> stateProperties);
 
  /**
@@ -321,6 +323,7 @@ public interface RuntimeContext {
  * @throws UnsupportedOperationException Thrown, if no partitioned state is available for the
  * function (function is not part of a KeyedStream).
  */
+ @Experimental
  <T> ReducingState<T> getReducingState(ReducingStateDescriptor<T> stateProperties);
 
  /**
@@ -380,6 +383,7 @@ public interface RuntimeContext {
  * @deprecated Use the more expressive {@link #getState(ValueStateDescriptor)} instead.
  */
  @Deprecated
+ @Experimental
  <S> OperatorState<S> getKeyValueState(String name, Class<S> stateType, S defaultState);
 
  /**
@@ -432,5 +436,6 @@ public interface RuntimeContext {
  * @deprecated Use the more expressive {@link #getState(ValueStateDescriptor)} instead.
  */
  @Deprecated
+ @Experimental
  <S> OperatorState<S> getKeyValueState(String name, TypeInformation<S> stateType, S defaultState);
 }

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

@@ -24,6 +24,7 @@
 import java.util.concurrent.Future;
 
 import com.google.common.base.Preconditions;
+import org.apache.flink.annotation.Experimental;
 import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.common.TaskInfo;
 import org.apache.flink.api.common.accumulators.Accumulator;
@@ -175,32 +176,37 @@ private <V, A extends Serializable> Accumulator<V, A> getAccumulator(String name
  }
 
  @Override
+ @Experimental
  public <T> ValueState<T> getState(ValueStateDescriptor<T> stateProperties) {
  throw new UnsupportedOperationException(
  "This state is only accessible by functions executed on a KeyedStream");
  }
 
  @Override
+ @Experimental
  public <T> ListState<T> getListState(ListStateDescriptor<T> stateProperties) {
  throw new UnsupportedOperationException(
  "This state is only accessible by functions executed on a KeyedStream");
  }
 
  @Override
+ @Experimental
  public <T> ReducingState<T> getReducingState(ReducingStateDescriptor<T> stateProperties) {
  throw new UnsupportedOperationException(
  "This state is only accessible by functions executed on a KeyedStream");
  }
 
  @Override
  @Deprecated
+ @Experimental
  public <S> OperatorState<S> getKeyValueState(String name, Class<S> stateType, S defaultState) {
  throw new UnsupportedOperationException(
  "This state is only accessible by functions executed on a KeyedStream");
  }
 
  @Override
  @Deprecated
+ @Experimental
  public <S> OperatorState<S> getKeyValueState(String name, TypeInformation<S> stateType, S defaultState) {
  throw new UnsupportedOperationException(
  "This state is only accessible by functions executed on a KeyedStream");

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

@@ -23,7 +23,6 @@
 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;
@@ -41,7 +40,6 @@
 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/statistics/BaseStatistics.java

@@ -19,6 +19,7 @@
 
 package org.apache.flink.api.common.io.statistics;
 
+import org.apache.flink.annotation.Experimental;
 import org.apache.flink.annotation.Public;
 
 /**
@@ -30,16 +31,19 @@ public interface BaseStatistics {
  /**
  * Constant indicating that the input size is unknown.
  */
+ @Experimental
  public static final long SIZE_UNKNOWN = -1;
 
  /**
  * Constant indicating that the number of records is unknown;
  */
+ @Experimental
  public static final long NUM_RECORDS_UNKNOWN = -1;
 
  /**
  * Constant indicating that average record width is unknown.
  */
+ @Experimental
  public static final float AVG_RECORD_BYTES_UNKNOWN = -1.0f;
 
  // --------------------------------------------------------------------------------------------
@@ -49,19 +53,22 @@ public interface BaseStatistics {
  * 
  * @return The total size of the input, in bytes.
  */
+ @Experimental
  public long getTotalInputSize();
 
  /**
  * Gets the number of records in the input (= base cardinality).
  * 
  * @return The number of records in the input.
  */
+ @Experimental
  public long getNumberOfRecords();
 
  /**
  * Gets the average width of a record, in bytes.
  * 
  * @return The average width of a record in bytes.
  */
+ @Experimental
  public float getAverageRecordWidth();
 }

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

@@ -21,13 +21,11 @@
 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/typeinfo/AtomicType.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.TypeComparator;
 
@@ -29,6 +30,7 @@
  * In contrast to atomic types are composite types, where the type information is aware of the individual
  * fields and individual fields may be used as a key.
  */
+@Public
 public interface AtomicType<T> {
 
  /**

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

@@ -23,12 +23,15 @@
 import java.util.Objects;
 
 import com.google.common.base.Preconditions;
+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.typeutils.TypeSerializer;
 import org.apache.flink.api.common.typeutils.base.array.StringArraySerializer;
 import org.apache.flink.api.common.functions.InvalidTypesException;
 import org.apache.flink.api.common.typeutils.base.GenericArraySerializer;
 
+@Public
 public final class BasicArrayTypeInfo<T, C> extends TypeInformation<T> {
 
  private static final long serialVersionUID = 1L;
@@ -58,45 +61,54 @@ private BasicArrayTypeInfo(Class<T> arrayClass, BasicTypeInfo<C> componentInfo)
  // --------------------------------------------------------------------------------------------
 
  @Override
+ @Experimental
  public boolean isBasicType() {
  return false;
  }
 
  @Override
+ @Experimental
  public boolean isTupleType() {
  return false;
  }
 
  @Override
+ @Experimental
  public int getArity() {
  return 1;
  }
 
  @Override
+ @Experimental
  public int getTotalFields() {
  return 1;
  }
 
  @Override
+ @Experimental
  public Class<T> getTypeClass() {
  return this.arrayClass;
  }
 
+ @Experimental
  public Class<C> getComponentTypeClass() {
  return this.componentInfo.getTypeClass();
  }
-
+
+ @Experimental
  public TypeInformation<C> getComponentInfo() {
  return componentInfo;
  }
 
  @Override
+ @Experimental
  public boolean isKeyType() {
  return false;
  }
 
  @Override
  @SuppressWarnings("unchecked")
+ @Experimental
  public TypeSerializer<T> createSerializer(ExecutionConfig executionConfig) {
  // special case the string array
  if (componentInfo.getTypeClass().equals(String.class)) {
@@ -140,6 +152,7 @@ public String toString() {
  // --------------------------------------------------------------------------------------------
 
  @SuppressWarnings("unchecked")
+ @Experimental
  public static <X, C> BasicArrayTypeInfo<X, C> getInfoFor(Class<X> type) {
  if (!type.isArray()) {
  throw new InvalidTypesException("The given class is no array.");