diff --git a/flink-annotations/pom.xml b/flink-annotations/pom.xml new file mode 100644 index 0000000000000000000000000000000000000000..748d6f8dd4e60b1a5420d48ab30f32cd69c31dd7 --- /dev/null +++ b/flink-annotations/pom.xml @@ -0,0 +1,37 @@ + + + + + 4.0.0 + + + org.apache.flink + flink-parent + 1.0-SNAPSHOT + .. + + + flink-annotations + flink-annotations + + jar + + 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 0000000000000000000000000000000000000000..bf9a97c5204f2043dcc60bec4d62b8025db457a6 --- /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 { +} 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 0000000000000000000000000000000000000000..4df68478749cec8a79d9d1da3b8d1e9a2a37b133 --- /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 { +} 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 0000000000000000000000000000000000000000..28c4d8385a6cfbeae36f1c0c55560756fdc878bc --- /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 {} diff --git a/flink-core/pom.xml b/flink-core/pom.xml index f91608ea7fb4db81a3bbb8b7e81524bee14301b6..adc9a9bd98b136ec541b20f8332c4399c2c5adb3 100644 --- a/flink-core/pom.xml +++ b/flink-core/pom.xml @@ -35,6 +35,12 @@ under the License. jar + + org.apache.flink + flink-annotations + ${project.version} + + org.apache.flink ${shading-artifact.name} 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 e9d85417ecef792a88d63fc68bd81a1f78e4a2c9..a13b6534cc199ddb6a186838c09c2cd7853fca5d 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 { /** 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 a0d33631afb60a0bbcc6e32fa98d9d5e73caa21f..3e2e2fd2f504d279b121fd105946e547494bf74f 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. * */ +@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; } 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 b1463ccea720246a7a2b0d48bbee223ec2c8b1bb..f045bacd48e2504d9e6853209cdb4008b370aa5a 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 { /** 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 92d2b98bdcacc935bf39c338b499af26339ca528..1fff4daa427687eb26698e4681b8d7775c568774 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) { 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 13a1a32da23f3d2f9adda6fa92228e8748fb70d3..78cddfe2af6870129795aee564c30cd212e25049 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.

*/ +@Public public final class JobID extends AbstractID { private static final long serialVersionUID = 1L; 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 3a18eb443edca0a4764d8aa07f04dbe0578b81a8..91a838ba3509cbb57093bcd90744c1bc1ff807f4 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; 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 2ee5fa0ad37537ad23c3ef3ba9db829a1323301d..bf56cfe979f69c3c2b69f15d39dca119a716f992 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 extends Serializable, Cloneable { /** * @param value 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 0db1942bc669e7cd8743293627e263e72f69edf5..9c0f62f12a9012b15f1907f949c28b99b9ce1c33 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 { private static final long serialVersionUID = 3672555084179165255L; 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 b7a4bfcd6d73be9ffaa9d2f1676d769548dc8954..3e6a5ddc8e0422216bddcb72b1d5c2068ca6f2cd 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. */ 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 f587fee598d7aae80be13fab6ea59d34cabbdd2b..577795b0d81c084fbc84c0f5b46b677ba7b5f6ee 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> { private static final long serialVersionUID = 1L; 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 a97ba5124dcd46438936d800bbb1aa8a9c4fdfd7..133eb5c89bb3fdf1bc80d1fda63b6e9cb30cebff 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. */ 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 3af785cfde9141e1258cd6b752949a77f46a9219..917f4ba98afd24375503ee4bdba70e15e293b50f 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 The type of the accumulated objects */ +@Public public class ListAccumulator implements Accumulator> { private static final long serialVersionUID = 1L; 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 6614e137ea5bf4e441d6d3d0ee20d85cdb25e021..288f48473e3d7849a2a2b8416ee198e60f7d6807 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 extends Accumulator {} 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 a4261d012118d63b9d7147f9ad6b38eb6ee73687..35a82e86fefb6c937b97e2487d0d109589e613b6 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 { 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 fd9de67556b1d85b4e5efd3b196a0b8cb791fe84..ff2cbead73ea0c1752b1f2a52ca58b7149b147e6 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; 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 d38c1f83b8f4305cc4c4133dada29d01b4b2e00e..65c3c1fc795a4244612079f15a12bebdb9d44c97 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 The type of the elements in the list of the original untransformed broadcast variable. * @param The type of the transformed broadcast variable. */ +@Public public interface BroadcastVariableInitializer { /** 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 1098341125e771cad66d69610df8122574c94f4d..db40370c49bcb9d42815b536998125dba545909c 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 The data type of the second input data set. * @param The data type of the returned elements. */ +@Public public interface CoGroupFunction extends Function, Serializable { /** 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 af115b032b6a4f3cb1da8cce6137551e1b82626d..af05c0d1697eb2cb59619d78f3cff9c0bcbe8630 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 The data type processed by the combine function. * @param The data type emitted by the combine function. */ +@Public public interface CombineFunction extends Function, Serializable { /** 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 3660e637df35ad5fffdd33e49b11ceacc6510c2c..d264e025798d78844a289b02ceb2bf6f404903fc 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 The type of the elements in the second input. * @param The type of the result elements. */ +@Public public interface CrossFunction extends Function, Serializable { /** 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 1a80cb89909be7097e488ff23b6249de127f0c0a..e8e83b6f912ca259c57c1bd648cb2aef88569908 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 The type of the filtered elements. */ +@Public public interface FilterFunction extends Function, Serializable { /** 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 c79046b2a0b8945343ebba439dd45bad13951bb9..5d7c0ee9c0cebd10dbe1c00229f7fe9cf7e1202b 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 The type of the elements in the second input. * @param The type of the result elements. */ +@Public public interface FlatJoinFunction extends Function, Serializable { /** 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 5b955f758cd9b63d7632cc80f0127f72e63a2ac5..8b4a8f219f5fddee70c916738404cf0be6b58aad 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 Type of the input elements. * @param Type of the returned elements. */ +@Public public interface FlatMapFunction extends Function, Serializable { /** 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 c4be00b9e32eb04e2cd0312b6316795f1229ce04..8194663a62daba6bee12a94faad4b702f6814ca9 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 Type of the initial input and the returned element * @param Type of the elements that the group/list/stream contains */ +@Public public interface FoldFunction extends Function, Serializable { /** * The core method of FoldFunction, combining two values into one value of the same type. 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 8a1819c8cdae557c85924c7687248f3e520b46e2..86c42f835c44889e53a1e3cf4f2ba50f8042d5ee 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. * *

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.

*/ +@Public public interface Function extends java.io.Serializable { } 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 c0b153b3acaf6d9fbb194481aadc8c0bae6d7f77..53e7ade5ad04502785cd2009381f1be2898a4c47 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 The data type processed by the combine function. * @param The data type emitted by the combine function. */ +@Public public interface GroupCombineFunction extends Function, Serializable { /** 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 36500f62d0af6d2ed6705db8d19e402aaa595448..37490c5ffe6ef2e87f118afe33cd7ab846e2a500 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 Type of the elements that this function processes. * @param The type of the elements returned by the user-defined function. */ +@Public public interface GroupReduceFunction extends Function, Serializable { /** 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 611765e5424110f2f7e7c7479af0dd9df55ddf80..143180ffa273f3a97e78a3674aaed74094a22749 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; 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 82399215e2c035b4943188f868d321021d89d8f6..5dc4ec8f14b8d2c34ec5a46fa5f882c032929f3d 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 getIterationAggregator(String name); T getPreviousIterationAggregate(String name); 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 0032a0fcdc52419d39a1a3461bb383fbb7cf23eb..5897b472e580c8b613f34044a4f787d6661b5eaf 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 The type of the elements in the second input. * @param The type of the result elements. */ +@Public public interface JoinFunction extends Function, Serializable { /** 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 45f3d1f604e43f95cf96da0f1799f6d374d859f8..9a0a45adabaf593f9ffa58050ec659c6cc911982 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 Type of the input elements. * @param Type of the returned elements. */ +@Public public interface MapFunction extends Function, Serializable { /** 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 d2c00772baa40d8b255c7ae3bd5cdd6bb0bbc4a6..4caaaddec1edcabebd9316cae95958069f0910f3 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 Type of the input elements. * @param Type of the returned elements. */ +@Public public interface MapPartitionFunction extends Function, Serializable { /** 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 f686e94b603249981f8234d507be7f26f28b60b0..6c237ed57d5b311c7438cce59e235c35529b412a 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 The type of the key to be partitioned. */ +@Public public interface Partitioner extends java.io.Serializable { /** 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 1b34047e9a46fdb4901945ec5d8e25c624d519c0..357968075b8f8d026d4c7dc834e2d2000e3f8773 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 Type of the elements that this function processes. */ +@Public public interface ReduceFunction extends Function, Serializable { /** 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 2e1506edeec666a0966c14f7f180c4870a66a00a..517e5121751760d0ba28025e38cde136c4cb823a 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 The type of the elements in the second input. * @param The type of the result elements. */ +@Public public abstract class RichCoGroupFunction extends AbstractRichFunction implements CoGroupFunction { private static final long serialVersionUID = 1L; 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 c85ce932dbc9c1ca12ffd391494066f9db9ca4ac..dbeefd51c0a9239b77e3edfc6d1d5a30971bb69c 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 The type of the elements in the second input. * @param The type of the result elements. */ +@Public public abstract class RichCrossFunction extends AbstractRichFunction implements CrossFunction { private static final long serialVersionUID = 1L; 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 131baf323f20f1a4381bd92381e044a5c79ce641..44580387355ce0794536db879fae492f1e3da8f6 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 The type of the filtered elements. */ +@Public public abstract class RichFilterFunction extends AbstractRichFunction implements FilterFunction { private static final long serialVersionUID = 1L; 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 b8bc1b94eb44f36cc1031ca7c38dc3e4b2123f12..0e06d81270874de5df80b4a769006dac4174774f 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 The type of the elements in the second input. * @param The type of the result elements. */ +@Public public abstract class RichFlatJoinFunction extends AbstractRichFunction implements FlatJoinFunction { private static final long serialVersionUID = 1L; 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 ae61023da4b0cd5b4bd10719d484d5758049b14c..94443b8baed8f03af212971577fcc4a97217ba7d 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 Type of the input elements. * @param Type of the returned elements. */ +@Public public abstract class RichFlatMapFunction extends AbstractRichFunction implements FlatMapFunction { private static final long serialVersionUID = 1L; 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 f4c2942e848a83bd403490551962daa8e4b5040b..245550d8f47c6870e4f65354f315f378f86f69d5 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 Type of the initial input and the returned element * @param Type of the elements that the group/list/stream contains */ +@Public public abstract class RichFoldFunction extends AbstractRichFunction implements FoldFunction { private static final long serialVersionUID = 1L; 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 0cbde4a6b68bbf1605208f46a8a0a1fa0accf077..c0f0de2c29412292ffd9211d31b9897f0f6ce86a 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 { /** 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 55df232ab9b9c2ab8506505ad41c3b39031e8836..de340319fbe35f78daf8849ed8169857dceded3f 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 The data type of the elements to be combined. * @param The resulting data type of the elements to be combined. */ +@Public public abstract class RichGroupCombineFunction extends AbstractRichFunction implements GroupCombineFunction { private static final long serialVersionUID = 1L; 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 b5edc64ac76d08d532d429d2c6f9a5982598d4c6..94d60a8ee267f6692de0254aca51b7a3d79c640c 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 Type of the elements that this function processes. * @param The type of the elements returned by the user-defined function. */ +@Public public abstract class RichGroupReduceFunction extends AbstractRichFunction implements GroupReduceFunction, GroupCombineFunction { private static final long serialVersionUID = 1L; @@ -83,5 +85,6 @@ public abstract class RichGroupReduceFunction extends AbstractRichFunct */ @Retention(RetentionPolicy.RUNTIME) @Target(ElementType.TYPE) + @Public public static @interface Combinable {} } 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 1666a19ac1cf3d9d798a895a44bf88950214c4b8..ceeca798a6d3fad37b4a7149f2eea481a166cffe 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 The type of the elements in the second input. * @param The type of the result elements. */ +@Public public abstract class RichJoinFunction extends AbstractRichFunction implements JoinFunction { private static final long serialVersionUID = 1L; 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 7adb25ba5b8874c53819d78b700b831c6b177132..81fec5dc48ad976b2ff73eed952c90658b0551ea 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 Type of the input elements. * @param Type of the returned elements. */ +@Public public abstract class RichMapFunction extends AbstractRichFunction implements MapFunction { private static final long serialVersionUID = 1L; 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 d8b7e9b4444ada439d14eae97605c4f21fc79c72..b3b800af2eeedcc7ead0e1142729d667f0225e62 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 Type of the input elements. * @param Type of the returned elements. */ +@Public public abstract class RichMapPartitionFunction extends AbstractRichFunction implements MapPartitionFunction { private static final long serialVersionUID = 1L; 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 13e7e3f2d5e16dab539386e6ced9bae13d27439a..7b42ec7b16cfe70cb49e05a9a3dfef81127941ae 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 Type of the elements that this function processes. */ +@Public public abstract class RichReduceFunction extends AbstractRichFunction implements ReduceFunction { private static final long serialVersionUID = 1L; 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 da79a795a07ffa7b940d6c803985af1d561bd0b4..a1e9d7d0cfe356dc4657575f4b408617e1a1084e 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> 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); /** 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 46a5d5829e7d47afa0bf40306c61a248cdfb14b1..61e3a1ae05f00645c4258eb5dd0c01d3dc32a4e4 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 extends FileInputFormat { private static final long serialVersionUID = 1L; 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 9b4e1cf9242c908980f422ff9df94fcca399cfd6..a89e73e3e776b261b78cdec48748563b522636b9 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 extends FileOutputFormat { private static final long serialVersionUID = 1L; 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 71e244d8c7af1efa3ccbb92c55d02989d5bb43d0..0ac2e505b23514f12d0a8c2af4771c201efb0b2c 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; 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 4b912e1e521c6ff9cc67dc385c1538fbb6150f84..7f77677a7113d37caada5e83fe9b17f060865d5f 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 { /** 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 cb32fc34b0d3b679b199bd5c8dda2d4abb33b23e..f6b6d491de663398acef68e8cce7f34fb1ef0472 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; * *

The default delimiter is the newline character {@code '\n'}.

*/ +@Public public abstract class DelimitedInputFormat extends FileInputFormat { private static final long serialVersionUID = 1L; 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 37e8749c13b18d361762f79f5b55ff6f5af428a5..c4cd2b35ca551f8ec070b976bc43e6b0bb0ec88e 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; *

After the {@link #open(FileInputSplit)} method completed, the file input data is available * from the {@link #stream} field.

*/ +@Public public abstract class FileInputFormat extends RichInputFormat { // -------------------------------------- Constants ------------------------------------------- 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 a5515c8cefb6c3b6a6ced9044f0a67c540f300fa..082878590598d41384ad70578070c151ef2fbf6c 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 extends RichOutputFormat implements InitializeOnMaster, CleanupWhenUnsuccessful { private static final long serialVersionUID = 1L; 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 8b787eb50c06ea33092b974c56bba2526f8c38dd..879714f935630106de45f3244bc70f25a605df1d 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 { /** 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 89caf2fe5f9d93a3381c90f074286d4fdf11d197..ddbf1652277a2ffd8f2af54c1c47b49e4f19db6b 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 extends DelimitedInputFormat { private static final Logger LOG = LoggerFactory.getLogger(GenericCsvInputFormat.class); 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 0e3b1c21e75b13663ff146a33f4b41f1a364ae32..2be6f7ad5ac79a55cee6edb35973174cda22d170 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 extends RichInputFormat { private static final long serialVersionUID = 1L; 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 943fc8d574db79ea3b534fc7c836ae224c9cc6f7..0fb5157198e3b75c7b07be1d232030a785674992 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 { /** 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 7bb1598e26ebb47d38dd0666c837ffde61eadbad..0e978b913ea76037011b3ce9330cf6a34a64c34e 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 The type of the produced records. * @param The type of input split. */ +@Public public interface InputFormat extends InputSplitSource, Serializable { /** 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 9cd7dbf14c97b3f15e6e4239a50f95515a3ff8e1..cfd94bc8b7f021903cda61b4398550432fde2aea 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; *
* NB: {@link #seek(long)} and {@link #getPos()} are currently not supported. */ +@Public public class InputStreamFSInputWrapper extends FSDataInputStream { private final InputStream inStream; 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 c038da60602dff10c2fb65416f6e466ac0b5bc07..517f40af9418d2d5879b7ea90722a877119e4d27 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); 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 26e49a76077c7b9fa42c49d4b71c3b3e6e32fb48..78e9766c588b9bfd790d3dd3dc25f85c223863ba 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 { } 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 7594721d6218ca7ca1abeb445696634bc4239da3..eb7ef4040e2d45babe39c50a42439e80ffb6cc13 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 The type of the consumed records. */ +@Public public interface OutputFormat extends Serializable { /** diff --git a/flink-core/src/main/java/org/apache/flink/api/common/io/RichInputFormat.java b/flink-core/src/main/java/org/apache/flink/api/common/io/RichInputFormat.java index acc81bcb99d9ac49786f55e7ad5d12d733f7ab37..0c23e133ad6887926e09fe03f6fc291ccfccddec 100644 --- a/flink-core/src/main/java/org/apache/flink/api/common/io/RichInputFormat.java +++ b/flink-core/src/main/java/org/apache/flink/api/common/io/RichInputFormat.java @@ -18,6 +18,7 @@ c * Licensed to the Apache Software Foundation (ASF) under one package org.apache.flink.api.common.io; +import org.apache.flink.annotation.Public; import org.apache.flink.api.common.functions.RuntimeContext; import org.apache.flink.core.io.InputSplit; @@ -25,6 +26,7 @@ import org.apache.flink.core.io.InputSplit; * An abstract stub implementation for Rich input formats. * Rich formats have access to their runtime execution context via {@link #getRuntimeContext()}. */ +@Public public abstract class RichInputFormat implements InputFormat { private static final long serialVersionUID = 1L; diff --git a/flink-core/src/main/java/org/apache/flink/api/common/io/RichOutputFormat.java b/flink-core/src/main/java/org/apache/flink/api/common/io/RichOutputFormat.java index 927a97490284838d52bb66c85f9860d078c715a8..7e633dfbdc1da84218a044645a8382b31759bc92 100644 --- a/flink-core/src/main/java/org/apache/flink/api/common/io/RichOutputFormat.java +++ b/flink-core/src/main/java/org/apache/flink/api/common/io/RichOutputFormat.java @@ -18,12 +18,14 @@ c * Licensed to the Apache Software Foundation (ASF) under one package org.apache.flink.api.common.io; +import org.apache.flink.annotation.Public; import org.apache.flink.api.common.functions.RuntimeContext; /** * An abstract stub implementation for Rich output formats. * Rich formats have access to their runtime execution context via {@link #getRuntimeContext()}. */ +@Public public abstract class RichOutputFormat implements OutputFormat { private static final long serialVersionUID = 1L; diff --git a/flink-core/src/main/java/org/apache/flink/api/common/io/SerializedInputFormat.java b/flink-core/src/main/java/org/apache/flink/api/common/io/SerializedInputFormat.java index 957cba886cd17ef5a6f49fdbffdacdb2bb63b4f5..26efa486c274d4862f8f55db54dc458d53b8b0e0 100644 --- a/flink-core/src/main/java/org/apache/flink/api/common/io/SerializedInputFormat.java +++ b/flink-core/src/main/java/org/apache/flink/api/common/io/SerializedInputFormat.java @@ -20,6 +20,7 @@ package org.apache.flink.api.common.io; import java.io.IOException; +import org.apache.flink.annotation.Public; import org.apache.flink.core.io.IOReadableWritable; import org.apache.flink.core.memory.DataInputView; @@ -28,6 +29,7 @@ import org.apache.flink.core.memory.DataInputView; * * @see SerializedOutputFormat */ +@Public public class SerializedInputFormat extends BinaryInputFormat { private static final long serialVersionUID = 1L; diff --git a/flink-core/src/main/java/org/apache/flink/api/common/io/SerializedOutputFormat.java b/flink-core/src/main/java/org/apache/flink/api/common/io/SerializedOutputFormat.java index 24fc77d82a2355f581a80141d42ead8d2a3728c4..0ef160e857c96d67305b54ffcfec742f9ef6fb01 100644 --- a/flink-core/src/main/java/org/apache/flink/api/common/io/SerializedOutputFormat.java +++ b/flink-core/src/main/java/org/apache/flink/api/common/io/SerializedOutputFormat.java @@ -20,6 +20,7 @@ package org.apache.flink.api.common.io; import java.io.IOException; +import org.apache.flink.annotation.Public; import org.apache.flink.core.io.IOReadableWritable; import org.apache.flink.core.memory.DataOutputView; @@ -28,6 +29,7 @@ import org.apache.flink.core.memory.DataOutputView; * * @see SerializedInputFormat */ +@Public public class SerializedOutputFormat extends BinaryOutputFormat { private static final long serialVersionUID = 1L; diff --git a/flink-core/src/main/java/org/apache/flink/api/common/io/StrictlyLocalAssignment.java b/flink-core/src/main/java/org/apache/flink/api/common/io/StrictlyLocalAssignment.java index edf9bf9c0bafc484e2ec356a10ed689452fd6228..1869e603d31ed9b0529374d46c92c2f591f65c2c 100644 --- a/flink-core/src/main/java/org/apache/flink/api/common/io/StrictlyLocalAssignment.java +++ b/flink-core/src/main/java/org/apache/flink/api/common/io/StrictlyLocalAssignment.java @@ -18,4 +18,5 @@ package org.apache.flink.api.common.io; + public interface StrictlyLocalAssignment {} diff --git a/flink-core/src/main/java/org/apache/flink/api/common/io/statistics/BaseStatistics.java b/flink-core/src/main/java/org/apache/flink/api/common/io/statistics/BaseStatistics.java index 1dc89c105ccada462450e93b97fe887d7b7b384a..ef92f18360dcd98b39e98f6ecd70d0d25e439005 100644 --- a/flink-core/src/main/java/org/apache/flink/api/common/io/statistics/BaseStatistics.java +++ b/flink-core/src/main/java/org/apache/flink/api/common/io/statistics/BaseStatistics.java @@ -19,9 +19,12 @@ package org.apache.flink.api.common.io.statistics; +import org.apache.flink.annotation.Public; + /** * Interface describing the basic statistics that can be obtained from the input. */ +@Public public interface BaseStatistics { /** diff --git a/flink-core/src/main/java/org/apache/flink/api/common/operators/AbstractUdfOperator.java b/flink-core/src/main/java/org/apache/flink/api/common/operators/AbstractUdfOperator.java index d6451372aeb2aaab7dee5f8190ce860e682a2d5c..1ee6848e20e9f3a863ead66b687c81bf0619004a 100644 --- a/flink-core/src/main/java/org/apache/flink/api/common/operators/AbstractUdfOperator.java +++ b/flink-core/src/main/java/org/apache/flink/api/common/operators/AbstractUdfOperator.java @@ -73,7 +73,7 @@ public abstract class AbstractUdfOperator extends Oper return userFunction; } -// -------------------------------------------------------------------------------------------- + // -------------------------------------------------------------------------------------------- /** * Returns the input, or null, if none is set. diff --git a/flink-core/src/main/java/org/apache/flink/api/common/operators/Order.java b/flink-core/src/main/java/org/apache/flink/api/common/operators/Order.java index 65007fa28327772a4fbc7cb588784505e5f8d2f4..913125728fdf2cd273f7235135cdc7c0604302d1 100644 --- a/flink-core/src/main/java/org/apache/flink/api/common/operators/Order.java +++ b/flink-core/src/main/java/org/apache/flink/api/common/operators/Order.java @@ -19,9 +19,12 @@ package org.apache.flink.api.common.operators; +import org.apache.flink.annotation.Public; + /** * Enumeration representing order. May represent no order, an ascending order or a descending order. */ +@Public public enum Order { /** diff --git a/flink-core/src/main/java/org/apache/flink/api/common/operators/SingleInputSemanticProperties.java b/flink-core/src/main/java/org/apache/flink/api/common/operators/SingleInputSemanticProperties.java index 23bbc8be734064923a12a71f84e34f0e73e9e3e4..db27a9ed17f553e29c3c585678210dc3e93f9309 100644 --- a/flink-core/src/main/java/org/apache/flink/api/common/operators/SingleInputSemanticProperties.java +++ b/flink-core/src/main/java/org/apache/flink/api/common/operators/SingleInputSemanticProperties.java @@ -21,11 +21,13 @@ package org.apache.flink.api.common.operators; import java.util.HashMap; import java.util.Map; +import org.apache.flink.annotation.Public; import org.apache.flink.api.common.operators.util.FieldSet; /** * Container for the semantic properties associated to a single input operator. */ +@Public public class SingleInputSemanticProperties implements SemanticProperties { private static final long serialVersionUID = 1L; diff --git a/flink-core/src/main/java/org/apache/flink/api/common/state/OperatorState.java b/flink-core/src/main/java/org/apache/flink/api/common/state/OperatorState.java index 136b6f80557b107d98c7e06b6db3a0c724ccfa18..ec30f82d27a1a84351e9034c45d3a969ddc2c6b8 100644 --- a/flink-core/src/main/java/org/apache/flink/api/common/state/OperatorState.java +++ b/flink-core/src/main/java/org/apache/flink/api/common/state/OperatorState.java @@ -18,6 +18,8 @@ package org.apache.flink.api.common.state; +import org.apache.flink.annotation.Public; + import java.io.IOException; /** @@ -32,6 +34,7 @@ import java.io.IOException; * * @param Type of the value in the operator state */ +@Public public interface OperatorState { /** diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/TypeInformation.java b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/TypeInformation.java index 07d854482260dd900076dba1f0a63c4535581b1f..3f4b1962a68bc88c7ae92096697bfca1f4d4e680 100644 --- a/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/TypeInformation.java +++ b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/TypeInformation.java @@ -18,6 +18,7 @@ package org.apache.flink.api.common.typeinfo; +import org.apache.flink.annotation.Public; import org.apache.flink.api.common.ExecutionConfig; import org.apache.flink.api.common.typeutils.TypeSerializer; @@ -67,6 +68,7 @@ import java.util.List; * * @param The type represented by this type information. */ +@Public public abstract class TypeInformation implements Serializable { private static final long serialVersionUID = -7742311969684489493L; diff --git a/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java b/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java index 80727efe998e29adab2926a154bb610a586b7ec4..35376a6120e03a97448fd1c2f369853361e841d1 100644 --- a/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java +++ b/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java @@ -18,10 +18,13 @@ package org.apache.flink.configuration; +import org.apache.flink.annotation.Public; + /** * This class contains all constants for the configuration. That includes the configuration keys and * the default values. */ +@Public public final class ConfigConstants { // ------------------------------------------------------------------------ diff --git a/flink-core/src/main/java/org/apache/flink/configuration/Configuration.java b/flink-core/src/main/java/org/apache/flink/configuration/Configuration.java index da42958d40603f5c45f119460e7b2d2de64a2760..6b05053c82d89f5becc275ab45c1af3ca63b1d21 100644 --- a/flink-core/src/main/java/org/apache/flink/configuration/Configuration.java +++ b/flink-core/src/main/java/org/apache/flink/configuration/Configuration.java @@ -25,6 +25,7 @@ import java.util.HashSet; import java.util.Map; import java.util.Set; +import org.apache.flink.annotation.Public; import org.apache.flink.api.common.ExecutionConfig; import org.apache.flink.core.io.IOReadableWritable; import org.apache.flink.core.memory.DataInputView; @@ -36,6 +37,7 @@ import org.slf4j.LoggerFactory; /** * Lightweight configuration object which stores key/value pairs. */ +@Public public class Configuration extends ExecutionConfig.GlobalJobParameters implements IOReadableWritable, java.io.Serializable, Cloneable { diff --git a/flink-core/src/main/java/org/apache/flink/configuration/UnmodifiableConfiguration.java b/flink-core/src/main/java/org/apache/flink/configuration/UnmodifiableConfiguration.java index 5d92cf0f62a0fe97e9c2cf36e7e4a378cf124532..5514caf9e7cca7eafffc6563f77558d819d80cb9 100644 --- a/flink-core/src/main/java/org/apache/flink/configuration/UnmodifiableConfiguration.java +++ b/flink-core/src/main/java/org/apache/flink/configuration/UnmodifiableConfiguration.java @@ -18,9 +18,12 @@ package org.apache.flink.configuration; +import org.apache.flink.annotation.Public; + /** * Unmodifiable version of the Configuration class. */ +@Public public class UnmodifiableConfiguration extends Configuration { private static final long serialVersionUID = -8151292629158972280L; diff --git a/flink-core/src/main/java/org/apache/flink/core/fs/BlockLocation.java b/flink-core/src/main/java/org/apache/flink/core/fs/BlockLocation.java index b364ecf4e34b85d46018a29f868f52890cc08b4b..dff0c3ee8dbafbaf3e7da7116ce725ce3ac6fa6c 100644 --- a/flink-core/src/main/java/org/apache/flink/core/fs/BlockLocation.java +++ b/flink-core/src/main/java/org/apache/flink/core/fs/BlockLocation.java @@ -18,11 +18,14 @@ package org.apache.flink.core.fs; +import org.apache.flink.annotation.Public; + import java.io.IOException; /** * A BlockLocation lists hosts, offset and length of block. */ +@Public public interface BlockLocation extends Comparable { /** diff --git a/flink-core/src/main/java/org/apache/flink/core/fs/FSDataInputStream.java b/flink-core/src/main/java/org/apache/flink/core/fs/FSDataInputStream.java index 9ebc7f78e53c925df1b0966b82e949fccd5b95e2..c94a71df47c1da4ebb4f83c1e01380cf7b2393ed 100644 --- a/flink-core/src/main/java/org/apache/flink/core/fs/FSDataInputStream.java +++ b/flink-core/src/main/java/org/apache/flink/core/fs/FSDataInputStream.java @@ -25,6 +25,8 @@ package org.apache.flink.core.fs; +import org.apache.flink.annotation.Public; + import java.io.IOException; import java.io.InputStream; @@ -32,6 +34,7 @@ import java.io.InputStream; * Interface for a data input stream to a file on a {@link FileSystem}. * */ +@Public public abstract class FSDataInputStream extends InputStream { /** diff --git a/flink-core/src/main/java/org/apache/flink/core/fs/FSDataOutputStream.java b/flink-core/src/main/java/org/apache/flink/core/fs/FSDataOutputStream.java index 9974c9f9111b579d0e02bdab1f1385d77da720f3..a6becf7965b28ce4202569496ee6532ac209a9ae 100644 --- a/flink-core/src/main/java/org/apache/flink/core/fs/FSDataOutputStream.java +++ b/flink-core/src/main/java/org/apache/flink/core/fs/FSDataOutputStream.java @@ -18,12 +18,15 @@ package org.apache.flink.core.fs; +import org.apache.flink.annotation.Public; + import java.io.IOException; import java.io.OutputStream; /** * Interface for a data output stream to a file on a {@link FileSystem}. */ +@Public public abstract class FSDataOutputStream extends OutputStream { public abstract void flush() throws IOException; diff --git a/flink-core/src/main/java/org/apache/flink/core/fs/FileInputSplit.java b/flink-core/src/main/java/org/apache/flink/core/fs/FileInputSplit.java index 31ca3c2aea9344118361dba11db9142031058ef8..8af0a20a8cd0a2aebe670893e741df8612c3e212 100644 --- a/flink-core/src/main/java/org/apache/flink/core/fs/FileInputSplit.java +++ b/flink-core/src/main/java/org/apache/flink/core/fs/FileInputSplit.java @@ -18,12 +18,14 @@ package org.apache.flink.core.fs; +import org.apache.flink.annotation.Public; import org.apache.flink.core.io.LocatableInputSplit; /** * A file input split provides information on a particular part of a file, possibly * hosted on a distributed file system and replicated among several hosts. */ +@Public public class FileInputSplit extends LocatableInputSplit { private static final long serialVersionUID = 1L; diff --git a/flink-core/src/main/java/org/apache/flink/core/fs/FileStatus.java b/flink-core/src/main/java/org/apache/flink/core/fs/FileStatus.java index 8f380431a1bce9bd67cdc513698863dce639440d..8b62659c19dbc41507c0f4145113cef3b3cc19d1 100644 --- a/flink-core/src/main/java/org/apache/flink/core/fs/FileStatus.java +++ b/flink-core/src/main/java/org/apache/flink/core/fs/FileStatus.java @@ -25,11 +25,14 @@ package org.apache.flink.core.fs; +import org.apache.flink.annotation.Public; + /** * Interface that represents the client side information for a file * independent of the file system. * */ +@Public public interface FileStatus { /** diff --git a/flink-core/src/main/java/org/apache/flink/core/fs/FileSystem.java b/flink-core/src/main/java/org/apache/flink/core/fs/FileSystem.java index c2de956451df8f26149f737c818fdf78c7602cea..79ddd8a5d80aa8367fc0db3eb756b2a492f937dc 100644 --- a/flink-core/src/main/java/org/apache/flink/core/fs/FileSystem.java +++ b/flink-core/src/main/java/org/apache/flink/core/fs/FileSystem.java @@ -34,6 +34,7 @@ import java.net.URISyntaxException; import java.util.HashMap; import java.util.Map; +import org.apache.flink.annotation.Public; import org.apache.flink.util.OperatingSystem; /** @@ -41,6 +42,7 @@ import org.apache.flink.util.OperatingSystem; * may be implemented as a distributed file system, or as a local * one that reflects the locally-connected disk. */ +@Public public abstract class FileSystem { private static final String LOCAL_FILESYSTEM_CLASS = "org.apache.flink.core.fs.local.LocalFileSystem"; diff --git a/flink-core/src/main/java/org/apache/flink/core/fs/Path.java b/flink-core/src/main/java/org/apache/flink/core/fs/Path.java index c47bc0de36b018ba7404d83fd792aa9bc0b75c0b..4c77199c327a17840281859ac49d245ae5008f87 100644 --- a/flink-core/src/main/java/org/apache/flink/core/fs/Path.java +++ b/flink-core/src/main/java/org/apache/flink/core/fs/Path.java @@ -27,6 +27,7 @@ import java.io.Serializable; import java.net.URI; import java.net.URISyntaxException; +import org.apache.flink.annotation.Public; import org.apache.flink.core.io.IOReadableWritable; import org.apache.flink.core.memory.DataInputView; import org.apache.flink.core.memory.DataOutputView; @@ -38,6 +39,7 @@ import org.apache.flink.util.StringUtils; * * Tailing slashes are removed from the path. */ +@Public public class Path implements IOReadableWritable, Serializable { private static final long serialVersionUID = 1L; diff --git a/flink-core/src/main/java/org/apache/flink/core/io/GenericInputSplit.java b/flink-core/src/main/java/org/apache/flink/core/io/GenericInputSplit.java index d96d26459aba7747f422f4c51f02ceccde920dad..5c7bd0013088c1b4808b3b489dcc6a3bf186f583 100644 --- a/flink-core/src/main/java/org/apache/flink/core/io/GenericInputSplit.java +++ b/flink-core/src/main/java/org/apache/flink/core/io/GenericInputSplit.java @@ -18,9 +18,12 @@ package org.apache.flink.core.io; +import org.apache.flink.annotation.Public; + /** * A generic input split that has only a partition number. */ +@Public public class GenericInputSplit implements InputSplit, java.io.Serializable { private static final long serialVersionUID = 1L; diff --git a/flink-core/src/main/java/org/apache/flink/core/io/IOReadableWritable.java b/flink-core/src/main/java/org/apache/flink/core/io/IOReadableWritable.java index ee1f984b4414806ca91374d3b62f406e84798000..a192a21c78229a8fef5618c5bbde5c27e1a36790 100644 --- a/flink-core/src/main/java/org/apache/flink/core/io/IOReadableWritable.java +++ b/flink-core/src/main/java/org/apache/flink/core/io/IOReadableWritable.java @@ -20,6 +20,7 @@ package org.apache.flink.core.io; import java.io.IOException; +import org.apache.flink.annotation.Public; import org.apache.flink.core.memory.DataInputView; import org.apache.flink.core.memory.DataOutputView; @@ -30,6 +31,7 @@ import org.apache.flink.core.memory.DataOutputView; * to a binary representation. * When implementing this Interface make sure that the implementing class has a default (zero-argument) constructor! */ +@Public public interface IOReadableWritable { /** diff --git a/flink-core/src/main/java/org/apache/flink/core/io/InputSplit.java b/flink-core/src/main/java/org/apache/flink/core/io/InputSplit.java index 87dd0738598a6832e53d81e401bbd7c7b96eef43..90b63cc8eb0cfa288456ff908fa48f9270b54c20 100644 --- a/flink-core/src/main/java/org/apache/flink/core/io/InputSplit.java +++ b/flink-core/src/main/java/org/apache/flink/core/io/InputSplit.java @@ -18,6 +18,8 @@ package org.apache.flink.core.io; +import org.apache.flink.annotation.Public; + import java.io.Serializable; /** @@ -26,6 +28,7 @@ import java.io.Serializable; *

Input splits are transferred in serialized form via the messages, so they need to be serializable * as defined by {@link java.io.Serializable}.

*/ +@Public public interface InputSplit extends Serializable { /** diff --git a/flink-core/src/main/java/org/apache/flink/core/io/InputSplitAssigner.java b/flink-core/src/main/java/org/apache/flink/core/io/InputSplitAssigner.java index 3999c13bbb34d58c53fd5e8de489a6e577a8d438..cccb03ffa7084c51f68c69e5bbbb5a7a3232c6c9 100644 --- a/flink-core/src/main/java/org/apache/flink/core/io/InputSplitAssigner.java +++ b/flink-core/src/main/java/org/apache/flink/core/io/InputSplitAssigner.java @@ -18,6 +18,7 @@ package org.apache.flink.core.io; + /** * An input split assigner distributes the {@link InputSplit}s among the instances on which a * data source exists. diff --git a/flink-core/src/main/java/org/apache/flink/core/io/InputSplitSource.java b/flink-core/src/main/java/org/apache/flink/core/io/InputSplitSource.java index 3773c15d68465d23078e7c3e174f50b44692255c..77591109460b183204de7ffce08d17f70b1195bc 100644 --- a/flink-core/src/main/java/org/apache/flink/core/io/InputSplitSource.java +++ b/flink-core/src/main/java/org/apache/flink/core/io/InputSplitSource.java @@ -18,6 +18,8 @@ package org.apache.flink.core.io; +import org.apache.flink.annotation.Public; + import java.io.Serializable; /** @@ -26,6 +28,7 @@ import java.io.Serializable; * * @param The type of the input splits created by the source. */ +@Public public interface InputSplitSource extends Serializable { /** diff --git a/flink-core/src/main/java/org/apache/flink/core/io/LocatableInputSplit.java b/flink-core/src/main/java/org/apache/flink/core/io/LocatableInputSplit.java index 6544f1c1737edd6d3200bb22bf3afdbbf272e0f4..21a5093f84b053c12fd1db239d07e9d089dd2608 100644 --- a/flink-core/src/main/java/org/apache/flink/core/io/LocatableInputSplit.java +++ b/flink-core/src/main/java/org/apache/flink/core/io/LocatableInputSplit.java @@ -18,11 +18,14 @@ package org.apache.flink.core.io; +import org.apache.flink.annotation.Public; + import java.util.Arrays; /** * A locatable input split is an input split referring to input data which is located on one or more hosts. */ +@Public public class LocatableInputSplit implements InputSplit, java.io.Serializable { private static final long serialVersionUID = 1L; diff --git a/flink-core/src/main/java/org/apache/flink/core/memory/DataInputView.java b/flink-core/src/main/java/org/apache/flink/core/memory/DataInputView.java index c6c899e326dccb8727fb75d5689d869811cafd4e..88d54d9696ca27bd17da5f985ae864eaf663a93b 100644 --- a/flink-core/src/main/java/org/apache/flink/core/memory/DataInputView.java +++ b/flink-core/src/main/java/org/apache/flink/core/memory/DataInputView.java @@ -20,6 +20,8 @@ package org.apache.flink.core.memory; +import org.apache.flink.annotation.Public; + import java.io.DataInput; import java.io.IOException; @@ -28,6 +30,7 @@ import java.io.IOException; * This interface defines a view over some memory that can be used to sequentially read the contents of the memory. * The view is typically backed by one or more {@link org.apache.flink.core.memory.MemorySegment}. */ +@Public public interface DataInputView extends DataInput { /** diff --git a/flink-core/src/main/java/org/apache/flink/core/memory/DataOutputView.java b/flink-core/src/main/java/org/apache/flink/core/memory/DataOutputView.java index 2b17be63ae60f7074ebc1cb00540d8dde1b2508e..d9219b849e65ebc81220d3b23326f5185c1fa567 100644 --- a/flink-core/src/main/java/org/apache/flink/core/memory/DataOutputView.java +++ b/flink-core/src/main/java/org/apache/flink/core/memory/DataOutputView.java @@ -20,6 +20,8 @@ package org.apache.flink.core.memory; +import org.apache.flink.annotation.Public; + import java.io.DataOutput; import java.io.IOException; @@ -28,6 +30,7 @@ import java.io.IOException; * This interface defines a view over some memory that can be used to sequentially write contents to the memory. * The view is typically backed by one or more {@link org.apache.flink.core.memory.MemorySegment}. */ +@Public public interface DataOutputView extends DataOutput { /** diff --git a/flink-core/src/main/java/org/apache/flink/types/BooleanValue.java b/flink-core/src/main/java/org/apache/flink/types/BooleanValue.java index e0346487f587e86d141a436abd58443ee2d2144f..10ca069e71c6a889295b5f52078a505dbb9c909e 100644 --- a/flink-core/src/main/java/org/apache/flink/types/BooleanValue.java +++ b/flink-core/src/main/java/org/apache/flink/types/BooleanValue.java @@ -21,6 +21,7 @@ package org.apache.flink.types; import java.io.IOException; +import org.apache.flink.annotation.Public; import org.apache.flink.core.memory.DataInputView; import org.apache.flink.core.memory.DataOutputView; import org.apache.flink.core.memory.MemorySegment; @@ -31,6 +32,7 @@ import org.apache.flink.core.memory.MemorySegment; * * @see org.apache.flink.types.Key */ +@Public public class BooleanValue implements NormalizableKey, ResettableValue, CopyableValue { private static final long serialVersionUID = 1L; diff --git a/flink-core/src/main/java/org/apache/flink/types/ByteValue.java b/flink-core/src/main/java/org/apache/flink/types/ByteValue.java index 40ed1ad341ef7dfa0e77a3b474f03a5952a330b5..a1a9e325fe68dc7833c797f87c9c8cee695f3315 100644 --- a/flink-core/src/main/java/org/apache/flink/types/ByteValue.java +++ b/flink-core/src/main/java/org/apache/flink/types/ByteValue.java @@ -21,6 +21,7 @@ package org.apache.flink.types; import java.io.IOException; +import org.apache.flink.annotation.Public; import org.apache.flink.core.memory.DataInputView; import org.apache.flink.core.memory.DataOutputView; import org.apache.flink.core.memory.MemorySegment; @@ -31,6 +32,7 @@ import org.apache.flink.core.memory.MemorySegment; * * @see org.apache.flink.types.Key */ +@Public public class ByteValue implements NormalizableKey, ResettableValue, CopyableValue { private static final long serialVersionUID = 1L; diff --git a/flink-core/src/main/java/org/apache/flink/types/CharValue.java b/flink-core/src/main/java/org/apache/flink/types/CharValue.java index 06b67c7fd675e28299c4d383f216bfee78816807..8b2ab297566cbd902391066b6b52d025801be331 100644 --- a/flink-core/src/main/java/org/apache/flink/types/CharValue.java +++ b/flink-core/src/main/java/org/apache/flink/types/CharValue.java @@ -21,6 +21,7 @@ package org.apache.flink.types; import java.io.IOException; +import org.apache.flink.annotation.Public; import org.apache.flink.core.memory.DataInputView; import org.apache.flink.core.memory.DataOutputView; import org.apache.flink.core.memory.MemorySegment; @@ -31,6 +32,7 @@ import org.apache.flink.core.memory.MemorySegment; * * @see org.apache.flink.types.Key */ +@Public public class CharValue implements NormalizableKey, ResettableValue, CopyableValue { private static final long serialVersionUID = 1L; diff --git a/flink-core/src/main/java/org/apache/flink/types/CopyableValue.java b/flink-core/src/main/java/org/apache/flink/types/CopyableValue.java index 95498fe82f885ece7be53cb1952438e112cea40b..3732f15bf5e1fae28d8b15b3ddf0d726f216dac1 100644 --- a/flink-core/src/main/java/org/apache/flink/types/CopyableValue.java +++ b/flink-core/src/main/java/org/apache/flink/types/CopyableValue.java @@ -20,12 +20,14 @@ package org.apache.flink.types; import java.io.IOException; +import org.apache.flink.annotation.Public; import org.apache.flink.core.memory.DataInputView; import org.apache.flink.core.memory.DataOutputView; /** * Interface to be implemented by basic types that support to be copied efficiently. */ +@Public public interface CopyableValue extends Value { /** diff --git a/flink-core/src/main/java/org/apache/flink/types/DeserializationException.java b/flink-core/src/main/java/org/apache/flink/types/DeserializationException.java index 57fd652d9f690364ea478a317bf4e4292654ba4a..c3472ad313fc0d106bb54432ce70a472493f098c 100644 --- a/flink-core/src/main/java/org/apache/flink/types/DeserializationException.java +++ b/flink-core/src/main/java/org/apache/flink/types/DeserializationException.java @@ -20,9 +20,12 @@ package org.apache.flink.types; +import org.apache.flink.annotation.Public; + /** * An exception specifying that the deserialization caused an error. */ +@Public public class DeserializationException extends RuntimeException { /** diff --git a/flink-core/src/main/java/org/apache/flink/types/DoubleValue.java b/flink-core/src/main/java/org/apache/flink/types/DoubleValue.java index 3158e40c21321982497c091eeed476bc6ac30529..80fb77bd40bb10ceb6bcf2b7adddff01ebc25a95 100644 --- a/flink-core/src/main/java/org/apache/flink/types/DoubleValue.java +++ b/flink-core/src/main/java/org/apache/flink/types/DoubleValue.java @@ -21,6 +21,7 @@ package org.apache.flink.types; import java.io.IOException; +import org.apache.flink.annotation.Public; import org.apache.flink.core.memory.DataInputView; import org.apache.flink.core.memory.DataOutputView; @@ -30,6 +31,7 @@ import org.apache.flink.core.memory.DataOutputView; * * @see org.apache.flink.types.Key */ +@Public public class DoubleValue implements Key, ResettableValue, CopyableValue { private static final long serialVersionUID = 1L; diff --git a/flink-core/src/main/java/org/apache/flink/types/FloatValue.java b/flink-core/src/main/java/org/apache/flink/types/FloatValue.java index 5364203e053a647c2565e8b41ff3b6e2f230715a..371ce525e3d8df4df2b71b0dc2e842f1edf3a600 100644 --- a/flink-core/src/main/java/org/apache/flink/types/FloatValue.java +++ b/flink-core/src/main/java/org/apache/flink/types/FloatValue.java @@ -21,6 +21,7 @@ package org.apache.flink.types; import java.io.IOException; +import org.apache.flink.annotation.Public; import org.apache.flink.core.memory.DataInputView; import org.apache.flink.core.memory.DataOutputView; @@ -30,6 +31,7 @@ import org.apache.flink.core.memory.DataOutputView; * * @see org.apache.flink.types.Key */ +@Public public class FloatValue implements Key, ResettableValue, CopyableValue { private static final long serialVersionUID = 1L; diff --git a/flink-core/src/main/java/org/apache/flink/types/IntValue.java b/flink-core/src/main/java/org/apache/flink/types/IntValue.java index 1b893f06f9cb0d55d3a23df20e018e5e33b6e05d..0f6311710856b31214d0961246ce814e5a6cc6ee 100644 --- a/flink-core/src/main/java/org/apache/flink/types/IntValue.java +++ b/flink-core/src/main/java/org/apache/flink/types/IntValue.java @@ -21,6 +21,7 @@ package org.apache.flink.types; import java.io.IOException; +import org.apache.flink.annotation.Public; import org.apache.flink.core.memory.DataInputView; import org.apache.flink.core.memory.DataOutputView; import org.apache.flink.core.memory.MemorySegment; @@ -31,6 +32,7 @@ import org.apache.flink.core.memory.MemorySegment; * * @see org.apache.flink.types.Key */ +@Public public class IntValue implements NormalizableKey, ResettableValue, CopyableValue { private static final long serialVersionUID = 1L; diff --git a/flink-core/src/main/java/org/apache/flink/types/KeyFieldOutOfBoundsException.java b/flink-core/src/main/java/org/apache/flink/types/KeyFieldOutOfBoundsException.java index 3d04efafe980de20eab7ff33d2677d74f9513afb..ac7f1f2653ad68f0498b4ea49410b487e2a0f80c 100644 --- a/flink-core/src/main/java/org/apache/flink/types/KeyFieldOutOfBoundsException.java +++ b/flink-core/src/main/java/org/apache/flink/types/KeyFieldOutOfBoundsException.java @@ -20,9 +20,12 @@ package org.apache.flink.types; +import org.apache.flink.annotation.Public; + /** * An exception specifying that a required key field was not set in a record, i.e. was null. */ +@Public public class KeyFieldOutOfBoundsException extends RuntimeException { /** diff --git a/flink-core/src/main/java/org/apache/flink/types/ListValue.java b/flink-core/src/main/java/org/apache/flink/types/ListValue.java index a7cc9144a8fdc9229a03ff5c9d6ae5d608635509..717d9b81044287222c76d749570ebd58f2a58ef9 100644 --- a/flink-core/src/main/java/org/apache/flink/types/ListValue.java +++ b/flink-core/src/main/java/org/apache/flink/types/ListValue.java @@ -26,6 +26,7 @@ import java.util.Iterator; import java.util.List; import java.util.ListIterator; +import org.apache.flink.annotation.Public; import org.apache.flink.core.memory.DataInputView; import org.apache.flink.core.memory.DataOutputView; import org.apache.flink.util.ReflectionUtil; @@ -42,6 +43,7 @@ import org.apache.flink.util.ReflectionUtil; * * */ +@Public public abstract class ListValue implements Value, List { private static final long serialVersionUID = 1L; diff --git a/flink-core/src/main/java/org/apache/flink/types/LongValue.java b/flink-core/src/main/java/org/apache/flink/types/LongValue.java index 2b6cb1fec6c396924d0a527a13620027e4ba47a9..ef9386472c33272b33cb5bb6d962de8ece24dae3 100644 --- a/flink-core/src/main/java/org/apache/flink/types/LongValue.java +++ b/flink-core/src/main/java/org/apache/flink/types/LongValue.java @@ -21,6 +21,7 @@ package org.apache.flink.types; import java.io.IOException; +import org.apache.flink.annotation.Public; import org.apache.flink.core.memory.DataInputView; import org.apache.flink.core.memory.DataOutputView; import org.apache.flink.core.memory.MemorySegment; @@ -31,6 +32,7 @@ import org.apache.flink.core.memory.MemorySegment; * * @see org.apache.flink.types.Key */ +@Public public class LongValue implements NormalizableKey, ResettableValue, CopyableValue { private static final long serialVersionUID = 1L; diff --git a/flink-core/src/main/java/org/apache/flink/types/MapValue.java b/flink-core/src/main/java/org/apache/flink/types/MapValue.java index a6cafd2dd942b1a274d8ec453ff31b944688aa39..143a991edb89e663a38e8c3edd76e01cd1b265f5 100644 --- a/flink-core/src/main/java/org/apache/flink/types/MapValue.java +++ b/flink-core/src/main/java/org/apache/flink/types/MapValue.java @@ -25,6 +25,7 @@ import java.util.HashMap; import java.util.Map; import java.util.Set; +import org.apache.flink.annotation.Public; import org.apache.flink.core.memory.DataInputView; import org.apache.flink.core.memory.DataOutputView; import org.apache.flink.util.ReflectionUtil; @@ -42,6 +43,7 @@ import org.apache.flink.util.ReflectionUtil; * * */ +@Public public abstract class MapValue implements Value, Map { private static final long serialVersionUID = 1L; diff --git a/flink-core/src/main/java/org/apache/flink/types/NormalizableKey.java b/flink-core/src/main/java/org/apache/flink/types/NormalizableKey.java index ee3f7f5b2a2b05f1d2d33a9ec1f19a8fab889d54..6c5e20435d5cd476bcfd2d4856b970add958595d 100644 --- a/flink-core/src/main/java/org/apache/flink/types/NormalizableKey.java +++ b/flink-core/src/main/java/org/apache/flink/types/NormalizableKey.java @@ -19,6 +19,7 @@ package org.apache.flink.types; +import org.apache.flink.annotation.Public; import org.apache.flink.core.memory.MemorySegment; @@ -34,6 +35,7 @@ import org.apache.flink.core.memory.MemorySegment; * key. A normalized key is considered a prefix, if its length is less than the maximal normalized * key length. */ +@Public public interface NormalizableKey extends Key { /** diff --git a/flink-core/src/main/java/org/apache/flink/types/Nothing.java b/flink-core/src/main/java/org/apache/flink/types/Nothing.java index dbf6278b83fdb4b12a4a2a09a918afd33d96fe98..285ffcc4985a3bbed0d1609994f3029558531b18 100644 --- a/flink-core/src/main/java/org/apache/flink/types/Nothing.java +++ b/flink-core/src/main/java/org/apache/flink/types/Nothing.java @@ -19,7 +19,7 @@ package org.apache.flink.types; -/** + /** * A type for (synthetic) operators that do not output data. For example, data sinks. */ public class Nothing { diff --git a/flink-core/src/main/java/org/apache/flink/types/NullFieldException.java b/flink-core/src/main/java/org/apache/flink/types/NullFieldException.java index 5c48bf4792020b23c5b9b9c348fe587282c2df96..0e5ebdb2e8b7eccb204c0621676d5bdd6c6405ac 100644 --- a/flink-core/src/main/java/org/apache/flink/types/NullFieldException.java +++ b/flink-core/src/main/java/org/apache/flink/types/NullFieldException.java @@ -18,9 +18,12 @@ package org.apache.flink.types; +import org.apache.flink.annotation.Public; + /** * An exception specifying that a required field was not set in a record, i.e. was null. */ +@Public public class NullFieldException extends RuntimeException { /** diff --git a/flink-core/src/main/java/org/apache/flink/types/NullKeyFieldException.java b/flink-core/src/main/java/org/apache/flink/types/NullKeyFieldException.java index b02b91eb5b955d8b1fc88b880ed3c29c2bd70bd7..18a98a3db15724aa97b8e64fac9f6bf7d253641b 100644 --- a/flink-core/src/main/java/org/apache/flink/types/NullKeyFieldException.java +++ b/flink-core/src/main/java/org/apache/flink/types/NullKeyFieldException.java @@ -15,8 +15,6 @@ * See the License for the specific language governing permissions and * limitations under the License. */ - - package org.apache.flink.types; diff --git a/flink-core/src/main/java/org/apache/flink/types/NullValue.java b/flink-core/src/main/java/org/apache/flink/types/NullValue.java index aa56536c4ad3409b858e0f16c73bc315b5dbb5bd..9a3885d104a60a20b4c880e77a5fd5194810e75e 100644 --- a/flink-core/src/main/java/org/apache/flink/types/NullValue.java +++ b/flink-core/src/main/java/org/apache/flink/types/NullValue.java @@ -21,6 +21,7 @@ package org.apache.flink.types; import java.io.IOException; +import org.apache.flink.annotation.Public; import org.apache.flink.core.memory.DataInputView; import org.apache.flink.core.memory.DataOutputView; import org.apache.flink.core.memory.MemorySegment; @@ -30,6 +31,7 @@ import org.apache.flink.core.memory.MemorySegment; * * @see org.apache.flink.types.Key */ +@Public public final class NullValue implements NormalizableKey, CopyableValue { private static final long serialVersionUID = 1L; diff --git a/flink-core/src/main/java/org/apache/flink/types/Record.java b/flink-core/src/main/java/org/apache/flink/types/Record.java index 7cbbc44400a91a3eabc28bb151c8d5a51336b62a..9990ddfa3f6494839c60d31ab2afe5d7683e04a5 100644 --- a/flink-core/src/main/java/org/apache/flink/types/Record.java +++ b/flink-core/src/main/java/org/apache/flink/types/Record.java @@ -27,6 +27,7 @@ import java.io.Serializable; import java.io.UTFDataFormatException; import java.nio.ByteOrder; +import org.apache.flink.annotation.Public; import org.apache.flink.core.memory.DataInputView; import org.apache.flink.core.memory.DataOutputView; import org.apache.flink.core.memory.MemoryUtils; @@ -51,6 +52,7 @@ import org.apache.flink.util.InstantiationUtil; *

* This class is NOT thread-safe! */ +@Public public final class Record implements Value, CopyableValue { private static final long serialVersionUID = 1L; diff --git a/flink-core/src/main/java/org/apache/flink/types/ResettableValue.java b/flink-core/src/main/java/org/apache/flink/types/ResettableValue.java index 3a93e64eed1de830a736f089e98934fd60557bac..3cbe962b34a1fc53a3d4e9877c2fbabfee6517f8 100644 --- a/flink-core/src/main/java/org/apache/flink/types/ResettableValue.java +++ b/flink-core/src/main/java/org/apache/flink/types/ResettableValue.java @@ -19,6 +19,9 @@ package org.apache.flink.types; +import org.apache.flink.annotation.Public; + +@Public public interface ResettableValue extends Value { /** diff --git a/flink-core/src/main/java/org/apache/flink/types/ShortValue.java b/flink-core/src/main/java/org/apache/flink/types/ShortValue.java index f18ce7f745fa1b1ef78456eb57cbd404815c44ce..b91aaac58ca3fead07f9da75be0bb777adfda0cb 100644 --- a/flink-core/src/main/java/org/apache/flink/types/ShortValue.java +++ b/flink-core/src/main/java/org/apache/flink/types/ShortValue.java @@ -21,6 +21,7 @@ package org.apache.flink.types; import java.io.IOException; +import org.apache.flink.annotation.Public; import org.apache.flink.core.memory.DataInputView; import org.apache.flink.core.memory.DataOutputView; import org.apache.flink.core.memory.MemorySegment; @@ -31,6 +32,7 @@ import org.apache.flink.core.memory.MemorySegment; * * @see org.apache.flink.types.Key */ +@Public public class ShortValue implements NormalizableKey, ResettableValue, CopyableValue { private static final long serialVersionUID = 1L; diff --git a/flink-core/src/main/java/org/apache/flink/types/StringValue.java b/flink-core/src/main/java/org/apache/flink/types/StringValue.java index 873e1ddabb25d1f0a95231c2818ca48a8c9b4705..0f9105cff5ad92f033fb3f12f19643d2f907f110 100644 --- a/flink-core/src/main/java/org/apache/flink/types/StringValue.java +++ b/flink-core/src/main/java/org/apache/flink/types/StringValue.java @@ -23,6 +23,7 @@ import java.io.DataOutput; import java.io.IOException; import java.nio.CharBuffer; +import org.apache.flink.annotation.Public; import org.apache.flink.core.memory.DataInputView; import org.apache.flink.core.memory.DataOutputView; import org.apache.flink.core.memory.MemorySegment; @@ -42,6 +43,7 @@ import com.google.common.base.Preconditions; * @see java.lang.String * @see java.lang.CharSequence */ +@Public public class StringValue implements NormalizableKey, CharSequence, ResettableValue, CopyableValue, Appendable { diff --git a/flink-core/src/main/java/org/apache/flink/types/Value.java b/flink-core/src/main/java/org/apache/flink/types/Value.java index b071a4e483fcade06cdaed287ebc704bac2300ba..11bbd19ccecb09fe51468fdce42d756e1319a4d6 100644 --- a/flink-core/src/main/java/org/apache/flink/types/Value.java +++ b/flink-core/src/main/java/org/apache/flink/types/Value.java @@ -21,6 +21,7 @@ package org.apache.flink.types; import java.io.Serializable; +import org.apache.flink.annotation.Public; import org.apache.flink.core.io.IOReadableWritable; /** @@ -31,5 +32,6 @@ import org.apache.flink.core.io.IOReadableWritable; * * @see org.apache.flink.core.io.IOReadableWritable */ +@Public public interface Value extends IOReadableWritable, Serializable { } diff --git a/flink-core/src/main/java/org/apache/flink/util/Collector.java b/flink-core/src/main/java/org/apache/flink/util/Collector.java index 1ede1a812edb5a34b6a1f9a9fc394be0b366042b..09f5c690b902b5409cd2ada1e108e0f33ac0803f 100644 --- a/flink-core/src/main/java/org/apache/flink/util/Collector.java +++ b/flink-core/src/main/java/org/apache/flink/util/Collector.java @@ -19,10 +19,13 @@ package org.apache.flink.util; +import org.apache.flink.annotation.Public; + /** * Collects a record and forwards it. The collector is the "push" counterpart of the * {@link java.util.Iterator}, which "pulls" data in. */ +@Public public interface Collector { /** diff --git a/flink-core/src/main/java/org/apache/flink/util/NumberSequenceIterator.java b/flink-core/src/main/java/org/apache/flink/util/NumberSequenceIterator.java index c536f44a479dd4ad0f29135f401b91d417886df4..f9b02f32882c6e085c3f11b6ea4387ab175537f8 100644 --- a/flink-core/src/main/java/org/apache/flink/util/NumberSequenceIterator.java +++ b/flink-core/src/main/java/org/apache/flink/util/NumberSequenceIterator.java @@ -18,6 +18,8 @@ package org.apache.flink.util; +import org.apache.flink.annotation.Public; + import java.util.NoSuchElementException; /** @@ -25,6 +27,7 @@ import java.util.NoSuchElementException; * The iterator is splittable (as defined by {@link SplittableIterator}, i.e., it can be divided into multiple * iterators that each return a subsequence of the number sequence. */ +@Public public class NumberSequenceIterator extends SplittableIterator { private static final long serialVersionUID = 1L; diff --git a/flink-core/src/main/java/org/apache/flink/util/SplittableIterator.java b/flink-core/src/main/java/org/apache/flink/util/SplittableIterator.java index 3a585d704f5bdf7a0e0330c7d0c7abc24fbf6d13..bb7f9afe1c9da4c11846b72a1d63e02c4479e33a 100644 --- a/flink-core/src/main/java/org/apache/flink/util/SplittableIterator.java +++ b/flink-core/src/main/java/org/apache/flink/util/SplittableIterator.java @@ -20,6 +20,7 @@ package org.apache.flink.util; import java.io.Serializable; import java.util.Iterator; +import org.apache.flink.annotation.Public; /** * Abstract base class for iterators that can split themselves into multiple disjoint @@ -27,6 +28,7 @@ import java.util.Iterator; * * @param The type of elements returned by the iterator. */ +@Public public abstract class SplittableIterator implements Iterator, Serializable { private static final long serialVersionUID = 200377674313072307L; diff --git a/flink-core/src/main/java/org/apache/flink/util/TraversableOnceException.java b/flink-core/src/main/java/org/apache/flink/util/TraversableOnceException.java index c0dbefd7f3f1c23973c5734ead7b8bdc4af4b830..b4f271c94ebcbb94bb736636fab7b2a9e3348b22 100644 --- a/flink-core/src/main/java/org/apache/flink/util/TraversableOnceException.java +++ b/flink-core/src/main/java/org/apache/flink/util/TraversableOnceException.java @@ -18,10 +18,13 @@ package org.apache.flink.util; +import org.apache.flink.annotation.Public; + /** * An exception, indicating that an {@link java.lang.Iterable} can only be traversed once, but has been attempted * to traverse an additional time. */ +@Public public class TraversableOnceException extends RuntimeException { private static final long serialVersionUID = 7636881584773577290L; diff --git a/flink-core/src/main/java/org/apache/flink/util/XORShiftRandom.java b/flink-core/src/main/java/org/apache/flink/util/XORShiftRandom.java index fa6844253e0b7d68f8f8ca1abe8c0a30f1b59b44..47ed5615e8d45234fa5380c4d5b4a591d4199ec5 100644 --- a/flink-core/src/main/java/org/apache/flink/util/XORShiftRandom.java +++ b/flink-core/src/main/java/org/apache/flink/util/XORShiftRandom.java @@ -18,6 +18,7 @@ package org.apache.flink.util; import com.google.common.hash.Hashing; +import org.apache.flink.annotation.Public; import java.util.Random; @@ -29,8 +30,10 @@ import java.util.Random; * * @see XORShift Algorithm Paper */ +@Public public class XORShiftRandom extends Random { + private static final long serialVersionUID = -825722456120842841L; private long seed; public XORShiftRandom() { diff --git a/pom.xml b/pom.xml index 7e25d7bf6f5e0cf1e10bdf0b83179225dd7fd090..91e4ca1e3f3d83e0bd577f32cbb816cbd89618b5 100644 --- a/pom.xml +++ b/pom.xml @@ -50,6 +50,7 @@ under the License. + flink-annotations flink-shaded-hadoop flink-shaded-curator flink-core @@ -500,7 +501,6 @@ under the License. - @@ -701,6 +701,7 @@ under the License. + org.apache.maven.plugins maven-jar-plugin