Repository: flink Updated Branches: refs/heads/master 30c9e2b68 -> 417597fbf
[hotfix] [docs] Fix JavaDoc errors in 'flink-streaming-java' Project: http://git-wip-us.apache.org/repos/asf/flink/repo Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/1f9f38bf Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/1f9f38bf Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/1f9f38bf Branch: refs/heads/master Commit: 1f9f38bf6312529af3ac527bf2f80f2ecee4d62b Parents: 30c9e2b Author: Stephan Ewen <[email protected]> Authored: Wed Feb 22 12:50:16 2017 +0100 Committer: Stephan Ewen <[email protected]> Committed: Thu Feb 23 17:44:25 2017 +0100 ---------------------------------------------------------------------- .../api/datastream/AllWindowedStream.java | 4 +-- .../api/datastream/WindowedStream.java | 4 +-- .../environment/StreamExecutionEnvironment.java | 7 +++--- .../api/functions/async/AsyncFunction.java | 26 ++++++++++---------- .../api/operators/StreamSourceContexts.java | 6 ++--- .../api/operators/async/AsyncWaitOperator.java | 11 +++++---- .../operators/windowing/MergingWindowSet.java | 2 +- 7 files changed, 30 insertions(+), 30 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/flink/blob/1f9f38bf/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/AllWindowedStream.java ---------------------------------------------------------------------- diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/AllWindowedStream.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/AllWindowedStream.java index c3c7424..4f4546e 100644 --- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/AllWindowedStream.java +++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/AllWindowedStream.java @@ -826,7 +826,7 @@ public class AllWindowedStream<T, W extends Window> { * @param function The window function. * @return The data stream that is the result of applying the window function to the window. * - * @deprecated Use {@link #fold(R, FoldFunction, AllWindowFunction)} instead. + * @deprecated Use {@link #fold(Object, FoldFunction, AllWindowFunction)} instead. */ @Deprecated public <R> SingleOutputStreamOperator<R> apply(R initialValue, FoldFunction<T, R> foldFunction, AllWindowFunction<R, R, W> function) { @@ -851,7 +851,7 @@ public class AllWindowedStream<T, W extends Window> { * @param resultType Type information for the result type of the window function * @return The data stream that is the result of applying the window function to the window. * - * @deprecated Use {@link #fold(R, FoldFunction, AllWindowFunction, TypeInformation, TypeInformation)} instead. + * @deprecated Use {@link #fold(Object, FoldFunction, AllWindowFunction, TypeInformation, TypeInformation)} instead. */ @Deprecated public <R> SingleOutputStreamOperator<R> apply(R initialValue, FoldFunction<T, R> foldFunction, AllWindowFunction<R, R, W> function, TypeInformation<R> resultType) { http://git-wip-us.apache.org/repos/asf/flink/blob/1f9f38bf/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/WindowedStream.java ---------------------------------------------------------------------- diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/WindowedStream.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/WindowedStream.java index 6809df0..b28434c 100644 --- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/WindowedStream.java +++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/WindowedStream.java @@ -1284,7 +1284,7 @@ public class WindowedStream<T, K, W extends Window> { * @param function The window function. * @return The data stream that is the result of applying the window function to the window. * - * @deprecated Use {@link #fold(R, FoldFunction, WindowFunction)} instead. + * @deprecated Use {@link #fold(Object, FoldFunction, WindowFunction)} instead. */ @Deprecated public <R> SingleOutputStreamOperator<R> apply(R initialValue, FoldFunction<T, R> foldFunction, WindowFunction<R, R, K, W> function) { @@ -1309,7 +1309,7 @@ public class WindowedStream<T, K, W extends Window> { * @param resultType Type information for the result type of the window function * @return The data stream that is the result of applying the window function to the window. * - * @deprecated Use {@link #fold(R, FoldFunction, WindowFunction, TypeInformation, TypeInformation)} instead. + * @deprecated Use {@link #fold(Object, FoldFunction, WindowFunction, TypeInformation, TypeInformation)} instead. */ @Deprecated public <R> SingleOutputStreamOperator<R> apply(R initialValue, FoldFunction<T, R> foldFunction, WindowFunction<R, R, K, W> function, TypeInformation<R> resultType) { http://git-wip-us.apache.org/repos/asf/flink/blob/1f9f38bf/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java ---------------------------------------------------------------------- diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java index dab0a06..e299e84 100644 --- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java +++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java @@ -31,6 +31,7 @@ import org.apache.flink.api.common.io.FilePathFilter; import org.apache.flink.api.common.io.InputFormat; import org.apache.flink.api.common.restartstrategy.RestartStrategies; import org.apache.flink.api.common.state.ValueState; +import org.apache.flink.api.common.time.Time; import org.apache.flink.api.common.typeinfo.BasicTypeInfo; import org.apache.flink.api.common.typeinfo.TypeInformation; import org.apache.flink.api.java.ClosureCleaner; @@ -484,7 +485,7 @@ public abstract class StreamExecutionEnvironment { * The number of times the system will try to re-execute failed tasks. * * @deprecated This method will be replaced by {@link #setRestartStrategy}. The - * {@link RestartStrategies.FixedDelayRestartStrategyConfiguration} contains the number of + * {@link RestartStrategies#fixedDelayRestart(int, Time)} contains the number of * execution retries. */ @Deprecated @@ -500,9 +501,7 @@ public abstract class StreamExecutionEnvironment { * * @return The number of times the system will try to re-execute failed tasks. * - * @deprecated This method will be replaced by {@link #getRestartStrategy}. The - * {@link RestartStrategies.FixedDelayRestartStrategyConfiguration} contains the number of - * execution retries. + * @deprecated This method will be replaced by {@link #getRestartStrategy}. */ @Deprecated @PublicEvolving http://git-wip-us.apache.org/repos/asf/flink/blob/1f9f38bf/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/async/AsyncFunction.java ---------------------------------------------------------------------- diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/async/AsyncFunction.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/async/AsyncFunction.java index 4de2db1..419c3ee 100644 --- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/async/AsyncFunction.java +++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/async/AsyncFunction.java @@ -26,34 +26,34 @@ import java.io.Serializable; /** * A function to trigger Async I/O operation. - * <p> - * For each #asyncInvoke, an async io operation can be triggered, and once it has been done, + * + * <p>For each #asyncInvoke, an async io operation can be triggered, and once it has been done, * the result can be collected by calling {@link AsyncCollector#collect}. For each async * operation, its context is stored in the operator immediately after invoking * #asyncInvoke, avoiding blocking for each stream input as long as the internal buffer is not full. - * <p> - * {@link AsyncCollector} can be passed into callbacks or futures to collect the result data. + * + * <p>{@link AsyncCollector} can be passed into callbacks or futures to collect the result data. * An error can also be propagate to the async IO operator by * {@link AsyncCollector#collect(Throwable)}. * - * <p> - * Callback example usage: + * <p>Callback example usage: + * * <pre>{@code * public class HBaseAsyncFunc implements AsyncFunction<String, String> { - * @Override + * * public void asyncInvoke(String row, AsyncCollector<String> collector) throws Exception { * HBaseCallback cb = new HBaseCallback(collector); * Get get = new Get(Bytes.toBytes(row)); * hbase.asyncGet(get, cb); * } * } - * </pre> + * }</pre> * - * <p> - * Future example usage: + * <p>Future example usage: + * * <pre>{@code * public class HBaseAsyncFunc implements AsyncFunction<String, String> { - * @Override + * * public void asyncInvoke(String row, final AsyncCollector<String> collector) throws Exception { * Get get = new Get(Bytes.toBytes(row)); * ListenableFuture<Result> future = hbase.asyncGet(get); @@ -68,14 +68,14 @@ import java.io.Serializable; * }); * } * } - * </pre> + * }</pre> * * @param <IN> The type of the input elements. * @param <OUT> The type of the returned elements. */ - @PublicEvolving public interface AsyncFunction<IN, OUT> extends Function, Serializable { + /** * Trigger async operation for each stream input. * http://git-wip-us.apache.org/repos/asf/flink/blob/1f9f38bf/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamSourceContexts.java ---------------------------------------------------------------------- diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamSourceContexts.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamSourceContexts.java index 98281c4..e4d051c 100644 --- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamSourceContexts.java +++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamSourceContexts.java @@ -38,9 +38,9 @@ public class StreamSourceContexts { * Depending on the {@link TimeCharacteristic}, this method will return the adequate * {@link org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext}. That is: * <ul> - * <li>{@link TimeCharacteristic#IngestionTime} = {@link AutomaticWatermarkContext}</li> - * <li>{@link TimeCharacteristic#ProcessingTime} = {@link NonTimestampContext}</li> - * <li>{@link TimeCharacteristic#EventTime} = {@link ManualWatermarkContext}</li> + * <li>{@link TimeCharacteristic#IngestionTime} = {@code AutomaticWatermarkContext}</li> + * <li>{@link TimeCharacteristic#ProcessingTime} = {@code NonTimestampContext}</li> + * <li>{@link TimeCharacteristic#EventTime} = {@code ManualWatermarkContext}</li> * </ul> * */ public static <OUT> SourceFunction.SourceContext<OUT> getSourceContext( http://git-wip-us.apache.org/repos/asf/flink/blob/1f9f38bf/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperator.java ---------------------------------------------------------------------- diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperator.java index a70d825..4cf79b1 100644 --- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperator.java +++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperator.java @@ -26,6 +26,7 @@ import org.apache.flink.runtime.concurrent.AcceptFunction; import org.apache.flink.runtime.state.StateInitializationContext; import org.apache.flink.runtime.state.StateSnapshotContext; import org.apache.flink.streaming.api.datastream.AsyncDataStream; +import org.apache.flink.streaming.api.datastream.AsyncDataStream.OutputMode; import org.apache.flink.streaming.api.functions.async.AsyncFunction; import org.apache.flink.streaming.api.functions.async.collector.AsyncCollector; import org.apache.flink.streaming.api.graph.StreamConfig; @@ -61,13 +62,13 @@ import java.util.concurrent.TimeoutException; * Within the async function, the user can complete the async collector arbitrarily. Once the async * collector has been completed, the result is emitted by the operator's emitter to downstream * operators. - * <p> - * The operator offers different output modes depending on the chosen - * {@link AsyncDataStream.OutputMode}. In order to give exactly once processing guarantees, the + * + * <p>The operator offers different output modes depending on the chosen + * {@link OutputMode}. In order to give exactly once processing guarantees, the * operator stores all currently in-flight {@link StreamElement} in it's operator state. Upon * recovery the recorded set of stream elements is replayed. - * <p> - * In case of chaining of this operator, it has to be made sure that the operators in the chain are + * + * <p>In case of chaining of this operator, it has to be made sure that the operators in the chain are * opened tail to head. The reason for this is that an opened {@link AsyncWaitOperator} starts * already emitting recovered {@link StreamElement} to downstream operators. * http://git-wip-us.apache.org/repos/asf/flink/blob/1f9f38bf/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/MergingWindowSet.java ---------------------------------------------------------------------- diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/MergingWindowSet.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/MergingWindowSet.java index b79a3fa..de37cb6 100644 --- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/MergingWindowSet.java +++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/MergingWindowSet.java @@ -42,7 +42,7 @@ import java.util.Map; * * <p>A new window can be added to the set of in-flight windows using * {@link #addWindow(Window, MergeFunction)}. This might merge other windows and the caller - * must react accordingly in the {@link MergeFunction#merge(Object, Collection, Object, Collection) + * must react accordingly in the {@link MergeFunction#merge(Object, Collection, Object, Collection)} * and adjust the outside view of windows and state. * * <p>Windows can be removed from the set of windows using {@link #retireWindow(Window)}.
