You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by al...@apache.org on 2017/04/26 10:07:41 UTC
[19/23] flink git commit: [FLINK-6107] Enable trailing whitespace
check in streaming checkstyle
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa93a27/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/RemoteStreamEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/RemoteStreamEnvironment.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/RemoteStreamEnvironment.java
index 28a0675..ff5af87 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/RemoteStreamEnvironment.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/RemoteStreamEnvironment.java
@@ -46,7 +46,7 @@ import org.slf4j.LoggerFactory;
*/
@Public
public class RemoteStreamEnvironment extends StreamExecutionEnvironment {
-
+
private static final Logger LOG = LoggerFactory.getLogger(RemoteStreamEnvironment.class);
/** The hostname of the JobManager. */
@@ -60,14 +60,14 @@ public class RemoteStreamEnvironment extends StreamExecutionEnvironment {
/** The jar files that need to be attached to each job. */
private final List<URL> jarFiles;
-
+
/** The classpaths that need to be attached to each job. */
private final List<URL> globalClasspaths;
/**
* Creates a new RemoteStreamEnvironment that points to the master
* (JobManager) described by the given host name and port.
- *
+ *
* @param host
* The host name or address of the master (JobManager), where the
* program should be executed.
@@ -125,9 +125,9 @@ public class RemoteStreamEnvironment extends StreamExecutionEnvironment {
* cluster. If the program uses user-defined functions,
* user-defined input formats, or any libraries, those must be
* provided in the JAR files.
- * @param globalClasspaths
- * The paths of directories and JAR files that are added to each user code
- * classloader on all nodes in the cluster. Note that the paths must specify a
+ * @param globalClasspaths
+ * The paths of directories and JAR files that are added to each user code
+ * classloader on all nodes in the cluster. Note that the paths must specify a
* protocol (e.g. file://) and be accessible on all nodes (e.g. by means of a NFS share).
* The protocol must be supported by the {@link java.net.URLClassLoader}.
*/
@@ -137,7 +137,7 @@ public class RemoteStreamEnvironment extends StreamExecutionEnvironment {
"The RemoteEnvironment cannot be used when submitting a program through a client, " +
"or running in a TestEnvironment context.");
}
-
+
if (host == null) {
throw new NullPointerException("Host must not be null.");
}
@@ -178,7 +178,7 @@ public class RemoteStreamEnvironment extends StreamExecutionEnvironment {
/**
* Executes the remote job.
- *
+ *
* @param streamGraph
* Stream Graph to execute
* @param jarFiles
@@ -192,10 +192,10 @@ public class RemoteStreamEnvironment extends StreamExecutionEnvironment {
ClassLoader usercodeClassLoader = JobWithJars.buildUserCodeClassLoader(jarFiles, globalClasspaths,
getClass().getClassLoader());
-
+
Configuration configuration = new Configuration();
configuration.addAll(this.clientConfiguration);
-
+
configuration.setString(ConfigConstants.JOB_MANAGER_IPC_ADDRESS_KEY, host);
configuration.setInteger(ConfigConstants.JOB_MANAGER_IPC_PORT_KEY, port);
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa93a27/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 354eb4a..3f73eb2 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
@@ -1639,11 +1639,11 @@ public abstract class StreamExecutionEnvironment {
/**
* Creates a {@link LocalStreamEnvironment} for local program execution that also starts the
* web monitoring UI.
- *
+ *
* <p>The local execution environment will run the program in a multi-threaded fashion in
* the same JVM as the environment was created in. It will use the parallelism specified in the
* parameter.
- *
+ *
* <p>If the configuration key 'jobmanager.web.port' was set in the configuration, that particular
* port will be used for the web UI. Otherwise, the default port (8081) will be used.
*/
@@ -1773,7 +1773,7 @@ public abstract class StreamExecutionEnvironment {
protected static void initializeContextEnvironment(StreamExecutionEnvironmentFactory ctx) {
contextEnvironmentFactory = ctx;
}
-
+
protected static void resetContextEnvironment() {
contextEnvironmentFactory = null;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa93a27/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/AssignerWithPeriodicWatermarks.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/AssignerWithPeriodicWatermarks.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/AssignerWithPeriodicWatermarks.java
index 4388c9d..823a5c3 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/AssignerWithPeriodicWatermarks.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/AssignerWithPeriodicWatermarks.java
@@ -28,7 +28,7 @@ import javax.annotation.Nullable;
* and generates low watermarks that signal event time progress within the stream.
* These timestamps and watermarks are used by functions and operators that operate
* on event time, for example event time windows.
- *
+ *
* <p>Use this class to generate watermarks in a periodical interval.
* At most every {@code i} milliseconds (configured via
* {@link ExecutionConfig#getAutoWatermarkInterval()}), the system will call the
@@ -45,9 +45,9 @@ import javax.annotation.Nullable;
* milliseconds since the Epoch (midnight, January 1, 1970 UTC).
* A watermark with a certain value {@code t} indicates that no elements with event
* timestamps {@code x}, where {@code x} is lower or equal to {@code t}, will occur any more.
- *
+ *
* @param <T> The type of the elements to which this assigner assigns timestamps.
- *
+ *
* @see org.apache.flink.streaming.api.watermark.Watermark
*/
public interface AssignerWithPeriodicWatermarks<T> extends TimestampAssigner<T> {
@@ -56,7 +56,7 @@ public interface AssignerWithPeriodicWatermarks<T> extends TimestampAssigner<T>
* Returns the current watermark. This method is periodically called by the
* system to retrieve the current watermark. The method may return {@code null} to
* indicate that no new Watermark is available.
- *
+ *
* <p>The returned watermark will be emitted only if it is non-null and its timestamp
* is larger than that of the previously emitted watermark (to preserve the contract of
* ascending watermarks). If the current watermark is still
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa93a27/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/AssignerWithPunctuatedWatermarks.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/AssignerWithPunctuatedWatermarks.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/AssignerWithPunctuatedWatermarks.java
index 5b5694c..a69a839 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/AssignerWithPunctuatedWatermarks.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/AssignerWithPunctuatedWatermarks.java
@@ -33,7 +33,7 @@ import javax.annotation.Nullable;
* The system will generate a new watermark, if the probed value is non-null
* and has a timestamp larger than that of the previous watermark (to preserve
* the contract of ascending watermarks).
- *
+ *
* <p>For use cases that should periodically emit watermarks based on element timestamps,
* use the {@link AssignerWithPeriodicWatermarks} instead.
*
@@ -41,7 +41,7 @@ import javax.annotation.Nullable;
* generator. It assumes elements carry a timestamp that describes when they were created,
* and that some elements carry a flag, marking them as the end of a sequence such that no
* elements with smaller timestamps can come anymore.
- *
+ *
* <pre>{@code
* public class WatermarkOnFlagAssigner implements AssignerWithPunctuatedWatermarks<MyElement> {
*
@@ -54,18 +54,18 @@ import javax.annotation.Nullable;
* }
* }
* }</pre>
- *
+ *
* <p>Timestamps and watermarks are defined as {@code longs} that represent the
* milliseconds since the Epoch (midnight, January 1, 1970 UTC).
* A watermark with a certain value {@code t} indicates that no elements with event
* timestamps {@code x}, where {@code x} is lower or equal to {@code t}, will occur any more.
- *
+ *
* @param <T> The type of the elements to which this assigner assigns timestamps.
- *
+ *
* @see org.apache.flink.streaming.api.watermark.Watermark
*/
public interface AssignerWithPunctuatedWatermarks<T> extends TimestampAssigner<T> {
-
+
/**
* Asks this implementation if it wants to emit a watermark. This method is called right after
* the {@link #extractTimestamp(Object, long)} method.
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa93a27/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/IngestionTimeExtractor.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/IngestionTimeExtractor.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/IngestionTimeExtractor.java
index c7c60d5..0b18c1c 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/IngestionTimeExtractor.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/IngestionTimeExtractor.java
@@ -22,16 +22,16 @@ import org.apache.flink.streaming.api.watermark.Watermark;
/**
* A timestamp assigner that assigns timestamps based on the machine's wall clock.
- *
+ *
* <p>If this assigner is used after a stream source, it realizes "ingestion time" semantics.
- *
+ *
* @param <T> The elements that get timestamps assigned.
*/
public class IngestionTimeExtractor<T> implements AssignerWithPeriodicWatermarks<T> {
private static final long serialVersionUID = -4072216356049069301L;
-
+
private long maxTimestamp;
-
+
@Override
public long extractTimestamp(T element, long previousElementTimestamp) {
// make sure timestamps are monotonously increasing, even when the system clock re-syncs
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa93a27/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/ProcessFunction.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/ProcessFunction.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/ProcessFunction.java
index 43683ef..c2c130e 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/ProcessFunction.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/ProcessFunction.java
@@ -105,7 +105,7 @@ public abstract class ProcessFunction<I, O> extends AbstractRichFunction {
/**
* Emits a record to the side output identified by the {@link OutputTag}.
- *
+ *
* @param outputTag the {@code OutputTag} that identifies the side output to emit to.
* @param value The record to emit.
*/
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa93a27/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/TimestampAssigner.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/TimestampAssigner.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/TimestampAssigner.java
index 5167fc8..60debb9 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/TimestampAssigner.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/TimestampAssigner.java
@@ -24,25 +24,25 @@ import org.apache.flink.api.common.functions.Function;
* A {@code TimestampAssigner} assigns event time timestamps to elements.
* These timestamps are used by all functions that operate on event time,
* for example event time windows.
- *
+ *
* <p>Timestamps are represented in milliseconds since the Epoch
* (midnight, January 1, 1970 UTC).
- *
+ *
* @param <T> The type of the elements to which this assigner assigns timestamps.
*/
public interface TimestampAssigner<T> extends Function {
/**
* Assigns a timestamp to an element, in milliseconds since the Epoch.
- *
+ *
* <p>The method is passed the previously assigned timestamp of the element.
* That previous timestamp may have been assigned from a previous assigner,
* by ingestion time. If the element did not carry a timestamp before, this value is
- * {@code Long.MIN_VALUE}.
+ * {@code Long.MIN_VALUE}.
*
* @param element The element that the timestamp is wil be assigned to.
* @param previousElementTimestamp The previous internal timestamp of the element,
- * or a negative value, if no timestamp has been assigned, yet.
+ * or a negative value, if no timestamp has been assigned, yet.
* @return The new timestamp.
*/
long extractTimestamp(T element, long previousElementTimestamp);
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa93a27/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/TimestampExtractor.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/TimestampExtractor.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/TimestampExtractor.java
index 2dc9091..c71e10b 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/TimestampExtractor.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/TimestampExtractor.java
@@ -35,7 +35,7 @@ import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExt
*
* @deprecated This class has been replaced by {@link AssignerWithPeriodicWatermarks} and
* {@link AssignerWithPunctuatedWatermarks}
- *
+ *
* @see AssignerWithPeriodicWatermarks
* @see AssignerWithPunctuatedWatermarks
*/
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa93a27/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/aggregation/ComparableAggregator.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/aggregation/ComparableAggregator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/aggregation/ComparableAggregator.java
index 3216e3a..a2a84fe 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/aggregation/ComparableAggregator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/aggregation/ComparableAggregator.java
@@ -36,7 +36,7 @@ public class ComparableAggregator<T> extends AggregationFunction<T> {
private boolean byAggregate;
private boolean first;
private final FieldAccessor<T, Object> fieldAccessor;
-
+
private ComparableAggregator(AggregationType aggregationType, FieldAccessor<T, Object> fieldAccessor, boolean first) {
this.comparator = Comparator.getForAggregation(aggregationType);
this.byAggregate = (aggregationType == AggregationType.MAXBY) || (aggregationType == AggregationType.MINBY);
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa93a27/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 419c3ee..5bb4459 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,21 +26,21 @@ 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,
* 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.
* An error can also be propagate to the async IO operator by
* {@link AsyncCollector#collect(Throwable)}.
*
* <p>Callback example usage:
- *
+ *
* <pre>{@code
* public class HBaseAsyncFunc implements AsyncFunction<String, String> {
- *
+ *
* public void asyncInvoke(String row, AsyncCollector<String> collector) throws Exception {
* HBaseCallback cb = new HBaseCallback(collector);
* Get get = new Get(Bytes.toBytes(row));
@@ -50,10 +50,10 @@ import java.io.Serializable;
* }</pre>
*
* <p>Future example usage:
- *
+ *
* <pre>{@code
* public class HBaseAsyncFunc implements AsyncFunction<String, String> {
- *
+ *
* public void asyncInvoke(String row, final AsyncCollector<String> collector) throws Exception {
* Get get = new Get(Bytes.toBytes(row));
* ListenableFuture<Result> future = hbase.asyncGet(get);
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa93a27/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/co/CoFlatMapFunction.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/co/CoFlatMapFunction.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/co/CoFlatMapFunction.java
index 3f996f5..03bfcbd 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/co/CoFlatMapFunction.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/co/CoFlatMapFunction.java
@@ -27,14 +27,14 @@ import org.apache.flink.util.Collector;
/**
* A CoFlatMapFunction implements a flat-map transformation over two
* connected streams.
- *
+ *
* <p>The same instance of the transformation function is used to transform
* both of the connected streams. That way, the stream transformations can
* share state.
- *
+ *
* <p>An example for the use of connected streams would be to apply rules that change over time
* onto elements of a stream. One of the connected streams has the rules, the other stream the
- * elements to apply the rules to. The operation on the connected stream maintains the
+ * elements to apply the rules to. The operation on the connected stream maintains the
* current set of rules in the state. It may receive either a rule update (from the first stream)
* and update the state, or a data element (from the second stream) and apply the rules in the
* state to the element. The result of applying the rules would be emitted.
@@ -48,7 +48,7 @@ public interface CoFlatMapFunction<IN1, IN2, OUT> extends Function, Serializable
/**
* This method is called for each element in the first of the connected streams.
- *
+ *
* @param value The stream element
* @param out The collector to emit resulting elements to
* @throws Exception The function may throw exceptions which cause the streaming program
@@ -58,7 +58,7 @@ public interface CoFlatMapFunction<IN1, IN2, OUT> extends Function, Serializable
/**
* This method is called for each element in the second of the connected streams.
- *
+ *
* @param value The stream element
* @param out The collector to emit resulting elements to
* @throws Exception The function may throw exceptions which cause the streaming program
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa93a27/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/co/CoProcessFunction.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/co/CoProcessFunction.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/co/CoProcessFunction.java
index 4285cc3..89a52d2 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/co/CoProcessFunction.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/co/CoProcessFunction.java
@@ -53,7 +53,7 @@ public abstract class CoProcessFunction<IN1, IN2, OUT> extends AbstractRichFunct
*
* <p>This function can output zero or more elements using the {@link Collector} parameter
* and also update internal state or set timers using the {@link Context} parameter.
- *
+ *
* @param value The stream element
* @param ctx A {@link Context} that allows querying the timestamp of the element,
* querying the {@link TimeDomain} of the firing timer and getting a
@@ -70,7 +70,7 @@ public abstract class CoProcessFunction<IN1, IN2, OUT> extends AbstractRichFunct
*
* <p>This function can output zero or more elements using the {@link Collector} parameter
* and also update internal state or set timers using the {@link Context} parameter.
- *
+ *
* @param value The stream element
* @param ctx A {@link Context} that allows querying the timestamp of the element,
* querying the {@link TimeDomain} of the firing timer and getting a
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa93a27/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/DiscardingSink.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/DiscardingSink.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/DiscardingSink.java
index c79b0dd..d116e74 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/DiscardingSink.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/DiscardingSink.java
@@ -22,12 +22,12 @@ import org.apache.flink.annotation.Public;
/**
* A stream sink that ignores all elements.
- *
+ *
* @param <T> The type of elements received by the sink.
*/
@Public
public class DiscardingSink<T> implements SinkFunction<T> {
-
+
private static final long serialVersionUID = 1L;
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa93a27/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/OutputFormatSinkFunction.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/OutputFormatSinkFunction.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/OutputFormatSinkFunction.java
index 3f40095..ed308c5 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/OutputFormatSinkFunction.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/OutputFormatSinkFunction.java
@@ -35,16 +35,16 @@ import org.slf4j.LoggerFactory;
/**
* Simple implementation of the SinkFunction writing tuples in the specified
* OutputFormat format.
- *
+ *
* @param <IN> Input type
*/
@PublicEvolving
public class OutputFormatSinkFunction<IN> extends RichSinkFunction<IN> implements InputTypeConfigurable {
-
+
private static final long serialVersionUID = 1L;
-
+
private static final Logger LOG = LoggerFactory.getLogger(OutputFormatSinkFunction.class);
-
+
private OutputFormat<IN> format;
private boolean cleanupCalled = false;
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa93a27/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/PrintSinkFunction.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/PrintSinkFunction.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/PrintSinkFunction.java
index bbdd7a7..bc47e9f 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/PrintSinkFunction.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/PrintSinkFunction.java
@@ -26,7 +26,7 @@ import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
/**
* Implementation of the SinkFunction writing every tuple to the standard
* output or standard error stream.
- *
+ *
* @param <IN>
* Input record type
*/
@@ -36,19 +36,19 @@ public class PrintSinkFunction<IN> extends RichSinkFunction<IN> {
private static final boolean STD_OUT = false;
private static final boolean STD_ERR = true;
-
- private boolean target;
+
+ private boolean target;
private transient PrintStream stream;
private transient String prefix;
-
+
/**
* Instantiates a print sink function that prints to standard out.
*/
public PrintSinkFunction() {}
-
+
/**
* Instantiates a print sink function that prints to standard out.
- *
+ *
* @param stdErr True, if the format should print to standard error instead of standard out.
*/
public PrintSinkFunction(boolean stdErr) {
@@ -58,20 +58,20 @@ public class PrintSinkFunction<IN> extends RichSinkFunction<IN> {
public void setTargetToStandardOut() {
target = STD_OUT;
}
-
+
public void setTargetToStandardErr() {
target = STD_ERR;
}
-
+
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
StreamingRuntimeContext context = (StreamingRuntimeContext) getRuntimeContext();
// get the target stream
stream = target == STD_OUT ? System.out : System.err;
-
+
// set the prefix if we have a >1 parallelism
- prefix = (context.getNumberOfParallelSubtasks() > 1) ?
+ prefix = (context.getNumberOfParallelSubtasks() > 1) ?
((context.getIndexOfThisSubtask() + 1) + "> ") : null;
}
@@ -84,13 +84,13 @@ public class PrintSinkFunction<IN> extends RichSinkFunction<IN> {
stream.println(record.toString());
}
}
-
+
@Override
public void close() {
this.stream = null;
this.prefix = null;
}
-
+
@Override
public String toString() {
return "Print to " + (target == STD_OUT ? "System.out" : "System.err");
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa93a27/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/SocketClientSink.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/SocketClientSink.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/SocketClientSink.java
index 0d0669a..21c35a3 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/SocketClientSink.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/SocketClientSink.java
@@ -46,26 +46,26 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
public class SocketClientSink<IN> extends RichSinkFunction<IN> {
private static final long serialVersionUID = 1L;
-
+
private static final Logger LOG = LoggerFactory.getLogger(SocketClientSink.class);
private static final int CONNECTION_RETRY_DELAY = 500;
-
-
+
+
private final SerializableObject lock = new SerializableObject();
private final SerializationSchema<IN> schema;
private final String hostName;
private final int port;
private final int maxNumRetries;
private final boolean autoFlush;
-
+
private transient Socket client;
private transient OutputStream outputStream;
-
+
private int retries;
private volatile boolean isRunning = true;
-
+
/**
* Creates a new SocketClientSink. The sink will not attempt to retry connections upon failure
* and will not auto-flush the stream.
@@ -118,7 +118,7 @@ public class SocketClientSink<IN> extends RichSinkFunction<IN> {
// ------------------------------------------------------------------------
// Life cycle
// ------------------------------------------------------------------------
-
+
/**
* Initialize the connection with the Socket in the server.
* @param parameters Configuration.
@@ -134,8 +134,8 @@ public class SocketClientSink<IN> extends RichSinkFunction<IN> {
throw new IOException("Cannot connect to socket server at " + hostName + ":" + port, e);
}
}
-
-
+
+
/**
* Called when new data arrives to the sink, and forwards it to Socket.
*
@@ -157,19 +157,19 @@ public class SocketClientSink<IN> extends RichSinkFunction<IN> {
throw new IOException("Failed to send message '" + value + "' to socket server at "
+ hostName + ":" + port + ". Connection re-tries are not enabled.", e);
}
-
- LOG.error("Failed to send message '" + value + "' to socket server at " + hostName + ":" + port +
+
+ LOG.error("Failed to send message '" + value + "' to socket server at " + hostName + ":" + port +
". Trying to reconnect..." , e);
-
+
// do the retries in locked scope, to guard against concurrent close() calls
// note that the first re-try comes immediately, without a wait!
-
+
synchronized (lock) {
IOException lastException = null;
retries = 0;
-
+
while (isRunning && (maxNumRetries < 0 || retries < maxNumRetries)) {
-
+
// first, clean up the old resources
try {
if (outputStream != null) {
@@ -187,17 +187,17 @@ public class SocketClientSink<IN> extends RichSinkFunction<IN> {
catch (IOException ee) {
LOG.error("Could not close socket from failed write attempt", ee);
}
-
+
// try again
retries++;
-
+
try {
// initialize a new connection
createConnection();
-
+
// re-try the write
outputStream.write(msg);
-
+
// success!
return;
}
@@ -209,7 +209,7 @@ public class SocketClientSink<IN> extends RichSinkFunction<IN> {
// wait before re-attempting to connect
lock.wait(CONNECTION_RETRY_DELAY);
}
-
+
// throw an exception if the task is still running, otherwise simply leave the method
if (isRunning) {
throw new IOException("Failed to send message '" + value + "' to socket server at "
@@ -218,7 +218,7 @@ public class SocketClientSink<IN> extends RichSinkFunction<IN> {
}
}
}
-
+
/**
* Closes the connection with the Socket server.
*/
@@ -226,13 +226,13 @@ public class SocketClientSink<IN> extends RichSinkFunction<IN> {
public void close() throws Exception {
// flag this as not running any more
isRunning = false;
-
+
// clean up in locked scope, so there is no concurrent change to the stream and client
synchronized (lock) {
// we notify first (this statement cannot fail). The notified thread will not continue
// anyways before it can re-acquire the lock
lock.notifyAll();
-
+
try {
if (outputStream != null) {
outputStream.close();
@@ -249,19 +249,19 @@ public class SocketClientSink<IN> extends RichSinkFunction<IN> {
// ------------------------------------------------------------------------
// Utilities
// ------------------------------------------------------------------------
-
+
private void createConnection() throws IOException {
client = new Socket(hostName, port);
client.setKeepAlive(true);
client.setTcpNoDelay(true);
-
+
outputStream = client.getOutputStream();
}
-
+
// ------------------------------------------------------------------------
// For testing
// ------------------------------------------------------------------------
-
+
int getCurrentNumberOfRetries() {
synchronized (lock) {
return retries;
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa93a27/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/WriteFormatAsCsv.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/WriteFormatAsCsv.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/WriteFormatAsCsv.java
index da03859..234dd39 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/WriteFormatAsCsv.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/WriteFormatAsCsv.java
@@ -27,7 +27,7 @@ import java.util.ArrayList;
/**
* Writes tuples in csv format.
- *
+ *
* @param <IN>
* Input tuple type
*/
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa93a27/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/WriteSinkFunction.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/WriteSinkFunction.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/WriteSinkFunction.java
index a9c724f..fa2bd8d 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/WriteSinkFunction.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/WriteSinkFunction.java
@@ -28,7 +28,7 @@ import java.util.ArrayList;
* the file specified by path. Tuples are collected to a list and written to the
* file periodically. The file specified by path is created if it does not
* exist, cleared if it exists before the writing.
- *
+ *
* @param <IN>
* Input tuple type
*/
@@ -48,7 +48,7 @@ public abstract class WriteSinkFunction<IN> implements SinkFunction<IN> {
/**
* Creates target file if it does not exist, cleans it if it exists.
- *
+ *
* @param path
* is the path to the location where the tuples are written
*/
@@ -65,7 +65,7 @@ public abstract class WriteSinkFunction<IN> implements SinkFunction<IN> {
/**
* Condition for writing the contents of tupleList and clearing it.
- *
+ *
* @return value of the updating condition
*/
protected abstract boolean updateCondition();
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa93a27/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/WriteSinkFunctionByMillis.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/WriteSinkFunctionByMillis.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/WriteSinkFunctionByMillis.java
index 62ad8c3..4753c5c 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/WriteSinkFunctionByMillis.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/WriteSinkFunctionByMillis.java
@@ -22,7 +22,7 @@ import org.apache.flink.annotation.PublicEvolving;
/**
* Implementation of WriteSinkFunction. Writes tuples to file in every millis
* milliseconds.
- *
+ *
* @param <IN>
* Input tuple type
*/
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa93a27/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/FromElementsFunction.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/FromElementsFunction.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/FromElementsFunction.java
index 68a1750..691d3d6 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/FromElementsFunction.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/FromElementsFunction.java
@@ -40,7 +40,7 @@ import java.util.List;
/**
* A stream source function that returns a sequence of elements.
- *
+ *
* <p>Upon construction, this source function serializes the elements using Flink's type information.
* That way, any object transport using Java serialization will not be affected by the serializability
* of the elements.</p>
@@ -51,15 +51,15 @@ import java.util.List;
*/
@PublicEvolving
public class FromElementsFunction<T> implements SourceFunction<T>, CheckpointedFunction {
-
+
private static final long serialVersionUID = 1L;
/** The (de)serializer to be used for the data elements. */
private final TypeSerializer<T> serializer;
-
+
/** The actual data elements, in serialized form. */
private final byte[] elementsSerialized;
-
+
/** The number of serialized elements. */
private final int numElements;
@@ -68,7 +68,7 @@ public class FromElementsFunction<T> implements SourceFunction<T>, CheckpointedF
/** The number of elements to skip initially. */
private volatile int numElementsToSkip;
-
+
/** Flag to make the source cancelable. */
private volatile boolean isRunning = true;
@@ -77,7 +77,7 @@ public class FromElementsFunction<T> implements SourceFunction<T>, CheckpointedF
public FromElementsFunction(TypeSerializer<T> serializer, T... elements) throws IOException {
this(serializer, Arrays.asList(elements));
}
-
+
public FromElementsFunction(TypeSerializer<T> serializer, Iterable<T> elements) throws IOException {
ByteArrayOutputStream baos = new ByteArrayOutputStream();
DataOutputViewStreamWrapper wrapper = new DataOutputViewStreamWrapper(baos);
@@ -128,7 +128,7 @@ public class FromElementsFunction<T> implements SourceFunction<T>, CheckpointedF
public void run(SourceContext<T> ctx) throws Exception {
ByteArrayInputStream bais = new ByteArrayInputStream(elementsSerialized);
final DataInputView input = new DataInputViewStreamWrapper(bais);
-
+
// if we are restored from a checkpoint and need to skip elements, skip them now.
int toSkip = numElementsToSkip;
if (toSkip > 0) {
@@ -143,12 +143,12 @@ public class FromElementsFunction<T> implements SourceFunction<T>, CheckpointedF
"If you are using user-defined serialization (Value and Writable types), check the " +
"serialization functions.\nSerializer is " + serializer);
}
-
+
this.numElementsEmitted = this.numElementsToSkip;
}
-
+
final Object lock = ctx.getCheckpointLock();
-
+
while (isRunning && numElementsEmitted < numElements) {
T next;
try {
@@ -159,7 +159,7 @@ public class FromElementsFunction<T> implements SourceFunction<T>, CheckpointedF
"If you are using user-defined serialization (Value and Writable types), check the " +
"serialization functions.\nSerializer is " + serializer);
}
-
+
synchronized (lock) {
ctx.collect(next);
numElementsEmitted++;
@@ -175,7 +175,7 @@ public class FromElementsFunction<T> implements SourceFunction<T>, CheckpointedF
/**
* Gets the number of elements produced in total by this function.
- *
+ *
* @return The number of elements produced in total.
*/
public int getNumElements() {
@@ -184,7 +184,7 @@ public class FromElementsFunction<T> implements SourceFunction<T>, CheckpointedF
/**
* Gets the number of elements emitted so far.
- *
+ *
* @return The number of elements emitted so far.
*/
public int getNumElementsEmitted() {
@@ -211,10 +211,10 @@ public class FromElementsFunction<T> implements SourceFunction<T>, CheckpointedF
/**
* Verifies that all elements in the collection are non-null, and are of the given class, or
* a subclass thereof.
- *
+ *
* @param elements The collection to check.
* @param viewedAs The class to which the elements must be assignable to.
- *
+ *
* @param <OUT> The generic type of the collection to be checked.
*/
public static <OUT> void checkCollection(Collection<OUT> elements, Class<OUT> viewedAs) {
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa93a27/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/InputFormatSourceFunction.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/InputFormatSourceFunction.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/InputFormatSourceFunction.java
index 9592c88..3fdedf1 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/InputFormatSourceFunction.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/InputFormatSourceFunction.java
@@ -86,7 +86,7 @@ public class InputFormatSourceFunction<OUT> extends RichParallelSourceFunction<O
// for each element we also check if cancel
// was called by checking the isRunning flag
-
+
while (isRunning && !format.reachedEnd()) {
nextElement = format.nextRecord(nextElement);
if (nextElement != null) {
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa93a27/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/RichParallelSourceFunction.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/RichParallelSourceFunction.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/RichParallelSourceFunction.java
index 1a0437e..94b85b6 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/RichParallelSourceFunction.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/RichParallelSourceFunction.java
@@ -24,7 +24,7 @@ import org.apache.flink.api.common.functions.AbstractRichFunction;
* Base class for implementing a parallel data source. Upon execution, the runtime will
* execute as many parallel instances of this function function as configured parallelism
* of the source.
- *
+ *
* <p>The data source has access to context information (such as the number of parallel
* instances of the source, and which parallel instance the current instance is)
* via {@link #getRuntimeContext()}. It also provides additional life-cycle methods
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa93a27/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/SocketTextStreamFunction.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/SocketTextStreamFunction.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/SocketTextStreamFunction.java
index 8d77b15..dd852a4 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/SocketTextStreamFunction.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/SocketTextStreamFunction.java
@@ -46,7 +46,7 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
public class SocketTextStreamFunction implements SourceFunction<String> {
private static final long serialVersionUID = 1L;
-
+
private static final Logger LOG = LoggerFactory.getLogger(SocketTextStreamFunction.class);
/** Default delay between successive connection attempts. */
@@ -54,28 +54,28 @@ public class SocketTextStreamFunction implements SourceFunction<String> {
/** Default connection timeout when connecting to the server socket (infinite). */
private static final int CONNECTION_TIMEOUT_TIME = 0;
-
-
+
+
private final String hostname;
private final int port;
private final String delimiter;
private final long maxNumRetries;
private final long delayBetweenRetries;
-
+
private transient Socket currentSocket;
-
+
private volatile boolean isRunning = true;
-
+
public SocketTextStreamFunction(String hostname, int port, String delimiter, long maxNumRetries) {
this(hostname, port, delimiter, maxNumRetries, DEFAULT_CONNECTION_RETRY_SLEEP);
}
-
+
public SocketTextStreamFunction(String hostname, int port, String delimiter, long maxNumRetries, long delayBetweenRetries) {
checkArgument(port > 0 && port < 65536, "port is out of range");
checkArgument(maxNumRetries >= -1, "maxNumRetries must be zero or larger (num retries), or -1 (infinite retries)");
checkArgument(delayBetweenRetries >= 0, "delayBetweenRetries must be zero or positive");
-
+
this.hostname = checkNotNull(hostname, "hostname must not be null");
this.port = port;
this.delimiter = delimiter;
@@ -87,12 +87,12 @@ public class SocketTextStreamFunction implements SourceFunction<String> {
public void run(SourceContext<String> ctx) throws Exception {
final StringBuilder buffer = new StringBuilder();
long attempt = 0;
-
+
while (isRunning) {
-
+
try (Socket socket = new Socket()) {
currentSocket = socket;
-
+
LOG.info("Connecting to server socket " + hostname + ':' + port);
socket.connect(new InetSocketAddress(hostname, port), CONNECTION_TIMEOUT_TIME);
BufferedReader reader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
@@ -138,7 +138,7 @@ public class SocketTextStreamFunction implements SourceFunction<String> {
@Override
public void cancel() {
isRunning = false;
-
+
// we need to close the socket as well, because the Thread.interrupt() function will
// not wake the thread in the socketStream.read() method when blocked.
Socket theSocket = this.currentSocket;
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa93a27/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/SourceFunction.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/SourceFunction.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/SourceFunction.java
index ba18df4..df5f9b6 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/SourceFunction.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/SourceFunction.java
@@ -70,8 +70,8 @@ import java.io.Serializable;
*
* <h3>Timestamps and watermarks:</h3>
* Sources may assign timestamps to elements and may manually emit watermarks.
- * However, these are only interpreted if the streaming program runs on
- * {@link TimeCharacteristic#EventTime}. On other time characteristics
+ * However, these are only interpreted if the streaming program runs on
+ * {@link TimeCharacteristic#EventTime}. On other time characteristics
* ({@link TimeCharacteristic#IngestionTime} and {@link TimeCharacteristic#ProcessingTime}),
* the watermarks from the source function are ignored.
*
@@ -79,13 +79,13 @@ import java.io.Serializable;
* Functions may additionally implement the {@link org.apache.flink.api.common.functions.StoppableFunction}
* interface. "Stopping" a function, in contrast to "canceling" means a graceful exit that leaves the
* state and the emitted elements in a consistent state.
- *
+ *
* <p>When a source is stopped, the executing thread is not interrupted, but expected to leave the
* {@link #run(SourceContext)} method in reasonable time on its own, preserving the atomicity
- * of state updates and element emission.
- *
+ * of state updates and element emission.
+ *
* @param <T> The type of the elements produced by this source.
- *
+ *
* @see org.apache.flink.api.common.functions.StoppableFunction
* @see org.apache.flink.streaming.api.TimeCharacteristic
*/
@@ -95,7 +95,7 @@ public interface SourceFunction<T> extends Function, Serializable {
/**
* Starts the source. Implementations can use the {@link SourceContext} emit
* elements.
- *
+ *
* <p>Sources that implement {@link org.apache.flink.streaming.api.checkpoint.Checkpointed}
* must lock on the checkpoint lock (using a synchronized block) before updating internal
* state and emitting elements, to make both an atomic operation:
@@ -132,10 +132,10 @@ public interface SourceFunction<T> extends Function, Serializable {
* Cancels the source. Most sources will have a while loop inside the
* {@link #run(SourceContext)} method. The implementation needs to ensure that the
* source will break out of that loop after this method is called.
- *
+ *
* <p>A typical pattern is to have an {@code "volatile boolean isRunning"} flag that is set to
* {@code false} in this method. That flag is checked in the loop condition.
- *
+ *
* <p>When a source is canceled, the executing thread will also be interrupted
* (via {@link Thread#interrupt()}). The interruption happens strictly after this
* method has been called, so any interruption handler can rely on the fact that
@@ -148,7 +148,7 @@ public interface SourceFunction<T> extends Function, Serializable {
// ------------------------------------------------------------------------
// source context
// ------------------------------------------------------------------------
-
+
/**
* Interface that source functions use to emit elements, and possibly watermarks.
*
@@ -160,7 +160,7 @@ public interface SourceFunction<T> extends Function, Serializable {
/**
* Emits one element from the source, without attaching a timestamp. In most cases,
* this is the default way of emitting elements.
- *
+ *
* <p>The timestamp that the element will get assigned depends on the time characteristic of
* the streaming program:
* <ul>
@@ -203,7 +203,7 @@ public interface SourceFunction<T> extends Function, Serializable {
* Emits the given {@link Watermark}. A Watermark of value {@code t} declares that no
* elements with a timestamp {@code t' <= t} will occur any more. If further such
* elements will be emitted, those elements are considered <i>late</i>.
- *
+ *
* <p>This method is only relevant when running on {@link TimeCharacteristic#EventTime}.
* On {@link TimeCharacteristic#ProcessingTime},Watermarks will be ignored. On
* {@link TimeCharacteristic#IngestionTime}, the Watermarks will be replaced by the
@@ -233,8 +233,8 @@ public interface SourceFunction<T> extends Function, Serializable {
* Returns the checkpoint lock. Please refer to the class-level comment in
* {@link SourceFunction} for details about how to write a consistent checkpointed
* source.
- *
- * @return The object to use as the lock
+ *
+ * @return The object to use as the lock
*/
Object getCheckpointLock();
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa93a27/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/StatefulSequenceSource.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/StatefulSequenceSource.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/StatefulSequenceSource.java
index fb1c94d..b6a0c3b 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/StatefulSequenceSource.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/StatefulSequenceSource.java
@@ -42,7 +42,7 @@ import java.util.Deque;
*/
@PublicEvolving
public class StatefulSequenceSource extends RichParallelSourceFunction<Long> implements CheckpointedFunction {
-
+
private static final long serialVersionUID = 1L;
private final long start;
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa93a27/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/timestamps/AscendingTimestampExtractor.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/timestamps/AscendingTimestampExtractor.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/timestamps/AscendingTimestampExtractor.java
index 253608f..ceed996 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/timestamps/AscendingTimestampExtractor.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/timestamps/AscendingTimestampExtractor.java
@@ -36,16 +36,16 @@ import static java.util.Objects.requireNonNull;
*/
@PublicEvolving
public abstract class AscendingTimestampExtractor<T> implements AssignerWithPeriodicWatermarks<T> {
-
+
private static final long serialVersionUID = 1L;
-
+
/** The current timestamp. */
private long currentTimestamp = Long.MIN_VALUE;
/** Handler that is called when timestamp monotony is violated. */
private MonotonyViolationHandler violationHandler = new LoggingHandler();
-
-
+
+
/**
* Extracts the timestamp from the given element. The timestamp must be monotonically increasing.
*
@@ -56,7 +56,7 @@ public abstract class AscendingTimestampExtractor<T> implements AssignerWithPeri
/**
* Sets the handler for violations to the ascending timestamp order.
- *
+ *
* @param handler The violation handler to use.
* @return This extractor.
*/
@@ -64,9 +64,9 @@ public abstract class AscendingTimestampExtractor<T> implements AssignerWithPeri
this.violationHandler = requireNonNull(handler);
return this;
}
-
+
// ------------------------------------------------------------------------
-
+
@Override
public final long extractTimestamp(T element, long elementPrevTimestamp) {
final long newTimestamp = extractAscendingTimestamp(element);
@@ -97,7 +97,7 @@ public abstract class AscendingTimestampExtractor<T> implements AssignerWithPeri
/**
* Called when the property of monotonously ascending timestamps is violated, i.e.,
* when {@code elementTimestamp < lastTimestamp}.
- *
+ *
* @param elementTimestamp The timestamp of the current element.
* @param lastTimestamp The last timestamp.
*/
@@ -119,7 +119,7 @@ public abstract class AscendingTimestampExtractor<T> implements AssignerWithPeri
*/
public static final class FailingHandler implements MonotonyViolationHandler {
private static final long serialVersionUID = 1L;
-
+
@Override
public void handleViolation(long elementTimestamp, long lastTimestamp) {
throw new RuntimeException("Ascending timestamps condition violated. Element timestamp "
@@ -134,7 +134,7 @@ public abstract class AscendingTimestampExtractor<T> implements AssignerWithPeri
private static final long serialVersionUID = 1L;
private static final Logger LOG = LoggerFactory.getLogger(AscendingTimestampExtractor.class);
-
+
@Override
public void handleViolation(long elementTimestamp, long lastTimestamp) {
LOG.warn("Timestamp monotony violated: {} < {}", elementTimestamp, lastTimestamp);
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa93a27/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/AggregateApplyWindowFunction.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/AggregateApplyWindowFunction.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/AggregateApplyWindowFunction.java
index 912a5c0..0341084 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/AggregateApplyWindowFunction.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/AggregateApplyWindowFunction.java
@@ -29,7 +29,7 @@ import java.util.Collections;
* A {@link WindowFunction} that composes an {@link AggregateFunction} and {@link WindowFunction}.
* Upon invocation, this first applies {@code AggregateFunction} to the input, and then
* finally the {@code WindowFunction} to the single result element.
- *
+ *
* @param <K> The key type
* @param <W> The window type
* @param <T> The type of the input to the AggregateFunction
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa93a27/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/InternalProcessApplyWindowContext.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/InternalProcessApplyWindowContext.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/InternalProcessApplyWindowContext.java
index f547adc..0cf0a8d 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/InternalProcessApplyWindowContext.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/InternalProcessApplyWindowContext.java
@@ -26,7 +26,7 @@ import org.apache.flink.streaming.api.windowing.windows.Window;
*
* @param <IN> The type of the input value.
* @param <OUT> The type of the output value.
- * @param <KEY> The type of the window key.
+ * @param <KEY> The type of the window key.
* @param <W> The type of the window.
*/
@Internal
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa93a27/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/WindowFunction.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/WindowFunction.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/WindowFunction.java
index 154fe88..030e973 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/WindowFunction.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/WindowFunction.java
@@ -43,8 +43,8 @@ public interface WindowFunction<IN, OUT, KEY, W extends Window> extends Function
* @param window The window that is being evaluated.
* @param input The elements in the window being evaluated.
* @param out A collector for emitting elements.
- *
- * @throws Exception The function may throw exceptions to fail the program and trigger recovery.
+ *
+ * @throws Exception The function may throw exceptions to fail the program and trigger recovery.
*/
void apply(KEY key, W window, Iterable<IN> input, Collector<OUT> out) throws Exception;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa93a27/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/CosineDistance.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/CosineDistance.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/CosineDistance.java
index a8fc600..2bf15a0 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/CosineDistance.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/CosineDistance.java
@@ -23,9 +23,9 @@ import org.apache.flink.streaming.api.functions.windowing.delta.extractor.Extrac
/**
* This delta function calculates the cosine distance between two given vectors.
* The cosine distance is defined as: cosineDistance=1-cosineSimilarity
- *
+ *
* <p>Cosine similarity: http://en.wikipedia.org/wiki/Cosine_similarity
- *
+ *
* @param <DATA>
* The input data type. This delta function works with a double[],
* but can extract/convert to it from any other given object in case
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa93a27/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/DeltaFunction.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/DeltaFunction.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/DeltaFunction.java
index d9295c4..ed1f830 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/DeltaFunction.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/DeltaFunction.java
@@ -35,7 +35,7 @@ public interface DeltaFunction<DATA> extends Serializable {
/**
* Calculates the delta between two given data points.
- *
+ *
* @param oldDataPoint
* the old data point.
* @param newDataPoint
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa93a27/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/EuclideanDistance.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/EuclideanDistance.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/EuclideanDistance.java
index c4b416c..28f9c24 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/EuclideanDistance.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/EuclideanDistance.java
@@ -23,9 +23,9 @@ import org.apache.flink.streaming.api.functions.windowing.delta.extractor.Extrac
/**
* This delta function calculates the euclidean distance between two given
* points.
- *
+ *
* <p>Euclidean distance: http://en.wikipedia.org/wiki/Euclidean_distance
- *
+ *
* @param <DATA>
* The input data type. This delta function works with a double[],
* but can extract/convert to it from any other given object in case
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa93a27/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/ExtractionAwareDeltaFunction.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/ExtractionAwareDeltaFunction.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/ExtractionAwareDeltaFunction.java
index 535d513..dc5b6be 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/ExtractionAwareDeltaFunction.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/ExtractionAwareDeltaFunction.java
@@ -25,7 +25,7 @@ import org.apache.flink.streaming.api.functions.windowing.delta.extractor.Extrac
* extracting the data on which the delta is calculated from a more complex data
* structure. For example in case you want to be able to run a delta only on one
* field of a Tuple type or only on some fields from an array.
- *
+ *
* @param <DATA>
* The input data type. The input of this type will be passed to the
* extractor which will transform into a TO-object. The delta
@@ -38,7 +38,7 @@ import org.apache.flink.streaming.api.functions.windowing.delta.extractor.Extrac
public abstract class ExtractionAwareDeltaFunction<DATA, TO> implements DeltaFunction<DATA> {
private static final long serialVersionUID = 6927486219702689554L;
-
+
private Extractor<DATA, TO> converter;
public ExtractionAwareDeltaFunction(Extractor<DATA, TO> converter) {
@@ -51,7 +51,7 @@ public abstract class ExtractionAwareDeltaFunction<DATA, TO> implements DeltaFun
* with the extracted data. In case no extractor is set the input data gets
* passes to {@link #getNestedDelta} as-is. The return value is just
* forwarded from {@link #getNestedDelta}.
- *
+ *
* @param oldDataPoint
* the older data point as raw data (before extraction).
* @param newDataPoint
@@ -78,7 +78,7 @@ public abstract class ExtractionAwareDeltaFunction<DATA, TO> implements DeltaFun
* result of the previously done extractions as input. Therefore, this
* method only does the actual calculation of the delta but no data
* extraction or conversion.
- *
+ *
* @param oldDataPoint
* the older data point.
* @param newDataPoint
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa93a27/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/extractor/ArrayFromTuple.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/extractor/ArrayFromTuple.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/extractor/ArrayFromTuple.java
index 16d0935..fd892b7 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/extractor/ArrayFromTuple.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/extractor/ArrayFromTuple.java
@@ -41,7 +41,7 @@ public class ArrayFromTuple implements Extractor<Tuple, Object[]> {
/**
* Using this constructor the extractor will combine the fields as specified
* in the indexes parameter in an object array.
- *
+ *
* @param indexes
* the field ids (enumerated from 0)
*/
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa93a27/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/extractor/ConcatenatedExtract.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/extractor/ConcatenatedExtract.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/extractor/ConcatenatedExtract.java
index 6fd9a67..71233cf 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/extractor/ConcatenatedExtract.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/extractor/ConcatenatedExtract.java
@@ -41,7 +41,7 @@ public class ConcatenatedExtract<FROM, OVER, TO> implements Extractor<FROM, TO>
/**
* Combines two extractors which will be executed one after each other.
- *
+ *
* @param e1
* First extractor: This extractor gets applied to the input data
* first. Its output as then passed as input to the second
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa93a27/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/extractor/Extractor.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/extractor/Extractor.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/extractor/Extractor.java
index 76d95b9..4263285 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/extractor/Extractor.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/extractor/Extractor.java
@@ -25,7 +25,7 @@ import java.io.Serializable;
* Extractors allow to extract/convert one type to another. They are mostly used
* to extract some fields out of a more complex structure (Tuple/Array) to run
* further calculation on the extraction result.
- *
+ *
* @param <FROM>
* The input data type.
* @param <TO>
@@ -36,7 +36,7 @@ public interface Extractor<FROM, TO> extends Serializable {
/**
* Extracts/Converts the given input to an object of the output type.
- *
+ *
* @param in
* the input data
* @return the extracted/converted data
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa93a27/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/extractor/FieldFromArray.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/extractor/FieldFromArray.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/extractor/FieldFromArray.java
index c44f32b..9fc6354 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/extractor/FieldFromArray.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/extractor/FieldFromArray.java
@@ -23,7 +23,7 @@ import java.lang.reflect.Array;
/**
* Extracts a single field out of an array.
- *
+ *
* @param <OUT>
* The type of the extracted field.
*/
@@ -31,7 +31,7 @@ import java.lang.reflect.Array;
public class FieldFromArray<OUT> implements Extractor<Object, OUT> {
private static final long serialVersionUID = -5161386546695574359L;
-
+
private int fieldId = 0;
/**
@@ -43,7 +43,7 @@ public class FieldFromArray<OUT> implements Extractor<Object, OUT> {
/**
* Extracts the field with the given id from the array.
- *
+ *
* @param fieldId
* The id of the field which will be extracted from the array.
*/
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa93a27/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/extractor/FieldFromTuple.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/extractor/FieldFromTuple.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/extractor/FieldFromTuple.java
index 468b878..a456a22 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/extractor/FieldFromTuple.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/extractor/FieldFromTuple.java
@@ -22,7 +22,7 @@ import org.apache.flink.api.java.tuple.Tuple;
/**
* Extracts a single field out of a tuple.
- *
+ *
* @param <OUT>
* The type of the extracted field.
*/
@@ -35,7 +35,7 @@ public class FieldFromTuple<OUT> implements Extractor<Tuple, OUT> {
/**
* Extracts the field with the given id from the tuple.
- *
+ *
* @param fieldId
* The id of the field which will be extracted from the tuple.
*/
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa93a27/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/extractor/FieldsFromArray.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/extractor/FieldsFromArray.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/extractor/FieldsFromArray.java
index afc2b4d..1fa9eb8 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/extractor/FieldsFromArray.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/extractor/FieldsFromArray.java
@@ -34,14 +34,14 @@ import java.lang.reflect.Array;
public class FieldsFromArray<OUT> implements Extractor<Object, OUT[]> {
private static final long serialVersionUID = 8075055384516397670L;
-
+
private int[] order;
private Class<OUT> clazz;
/**
* Extracts multiple fields from an array and puts them in the given order
* into a new array of the specified type.
- *
+ *
* @param clazz
* the Class object representing the component type of the new
* array
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa93a27/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/extractor/FieldsFromTuple.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/extractor/FieldsFromTuple.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/extractor/FieldsFromTuple.java
index 2321878..e2ba284 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/extractor/FieldsFromTuple.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/windowing/delta/extractor/FieldsFromTuple.java
@@ -33,7 +33,7 @@ public class FieldsFromTuple implements Extractor<Tuple, double[]> {
/**
* Extracts one or more fields of the the type Double from a tuple and puts
* them into a new double[] (in the specified order).
- *
+ *
* @param indexes
* The indexes of the fields to be extracted.
*/
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa93a27/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamConfig.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamConfig.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamConfig.java
index be591f7..b053bff 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamConfig.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamConfig.java
@@ -49,11 +49,11 @@ import org.apache.flink.util.Preconditions;
public class StreamConfig implements Serializable {
private static final long serialVersionUID = 1L;
-
+
// ------------------------------------------------------------------------
// Config Keys
// ------------------------------------------------------------------------
-
+
private static final String NUMBER_OF_OUTPUTS = "numberOfOutputs";
private static final String NUMBER_OF_INPUTS = "numberOfInputs";
private static final String CHAINED_OUTPUTS = "chainedOutputs";
@@ -80,22 +80,22 @@ public class StreamConfig implements Serializable {
private static final String CHECKPOINTING_ENABLED = "checkpointing";
private static final String CHECKPOINT_MODE = "checkpointMode";
-
+
private static final String STATE_BACKEND = "statebackend";
private static final String STATE_PARTITIONER = "statePartitioner";
private static final String STATE_KEY_SERIALIZER = "statekeyser";
-
+
private static final String TIME_CHARACTERISTIC = "timechar";
-
+
// ------------------------------------------------------------------------
// Default Values
// ------------------------------------------------------------------------
-
+
private static final long DEFAULT_TIMEOUT = 100;
private static final CheckpointingMode DEFAULT_CHECKPOINTING_MODE = CheckpointingMode.EXACTLY_ONCE;
-
-
+
+
// ------------------------------------------------------------------------
// Config
// ------------------------------------------------------------------------
@@ -121,7 +121,7 @@ public class StreamConfig implements Serializable {
public Integer getVertexID() {
return config.getInteger(VERTEX_NAME, -1);
}
-
+
public void setTimeCharacteristic(TimeCharacteristic characteristic) {
config.setInteger(TIME_CHARACTERISTIC, characteristic.ordinal());
}
@@ -134,7 +134,7 @@ public class StreamConfig implements Serializable {
throw new CorruptConfigurationException("time characteristic is not set");
}
}
-
+
public void setTypeSerializerIn1(TypeSerializer<?> serializer) {
setTypeSerializer(TYPE_SERIALIZER_IN_1, serializer);
}
@@ -150,7 +150,7 @@ public class StreamConfig implements Serializable {
public void setTypeSerializerSideOut(OutputTag<?> outputTag, TypeSerializer<?> serializer) {
setTypeSerializer(TYPE_SERIALIZER_SIDEOUT_PREFIX + outputTag.getId(), serializer);
}
-
+
public <T> TypeSerializer<T> getTypeSerializerIn1(ClassLoader cl) {
try {
return InstantiationUtil.readObjectFromConfig(this.config, TYPE_SERIALIZER_IN_1, cl);
@@ -158,7 +158,7 @@ public class StreamConfig implements Serializable {
throw new StreamTaskException("Could not instantiate serializer.", e);
}
}
-
+
public <T> TypeSerializer<T> getTypeSerializerIn2(ClassLoader cl) {
try {
return InstantiationUtil.readObjectFromConfig(this.config, TYPE_SERIALIZER_IN_2, cl);
@@ -212,7 +212,7 @@ public class StreamConfig implements Serializable {
}
}
}
-
+
public <T> T getStreamOperator(ClassLoader cl) {
try {
return InstantiationUtil.readObjectFromConfig(this.config, SERIALIZEDUDF, cl);
@@ -220,13 +220,13 @@ public class StreamConfig implements Serializable {
catch (ClassNotFoundException e) {
String classLoaderInfo = ClassLoaderUtil.getUserCodeClassLoaderInfo(cl);
boolean loadableDoubleCheck = ClassLoaderUtil.validateClassLoadable(e, cl);
-
+
String exceptionMessage = "Cannot load user class: " + e.getMessage()
- + "\nClassLoader info: " + classLoaderInfo +
- (loadableDoubleCheck ?
+ + "\nClassLoader info: " + classLoaderInfo +
+ (loadableDoubleCheck ?
"\nClass was actually found in classloader - deserialization issue." :
"\nClass not resolvable through given classloader.");
-
+
throw new StreamTaskException(exceptionMessage);
}
catch (Exception e) {
@@ -241,13 +241,13 @@ public class StreamConfig implements Serializable {
throw new StreamTaskException("Could not serialize output selectors", e);
}
}
-
+
public <T> List<OutputSelector<T>> getOutputSelectors(ClassLoader userCodeClassloader) {
try {
- List<OutputSelector<T>> selectors =
+ List<OutputSelector<T>> selectors =
InstantiationUtil.readObjectFromConfig(this.config, OUTPUT_SELECTOR_WRAPPER, userCodeClassloader);
return selectors == null ? Collections.<OutputSelector<T>>emptyList() : selectors;
-
+
} catch (Exception e) {
throw new StreamTaskException("Could not read output selectors", e);
}
@@ -292,7 +292,7 @@ public class StreamConfig implements Serializable {
throw new StreamTaskException("Cannot serialize non chained outputs.", e);
}
}
-
+
public List<StreamEdge> getNonChainedOutputs(ClassLoader cl) {
try {
List<StreamEdge> nonChainedOutputs = InstantiationUtil.readObjectFromConfig(this.config, NONCHAINED_OUTPUTS, cl);
@@ -309,7 +309,7 @@ public class StreamConfig implements Serializable {
throw new StreamTaskException("Cannot serialize chained outputs.", e);
}
}
-
+
public List<StreamEdge> getChainedOutputs(ClassLoader cl) {
try {
List<StreamEdge> chainedOutputs = InstantiationUtil.readObjectFromConfig(this.config, CHAINED_OUTPUTS, cl);
@@ -326,7 +326,7 @@ public class StreamConfig implements Serializable {
throw new StreamTaskException("Cannot serialize outward edges.", e);
}
}
-
+
public List<StreamEdge> getOutEdges(ClassLoader cl) {
try {
List<StreamEdge> outEdges = InstantiationUtil.readObjectFromConfig(this.config, OUT_STREAM_EDGES, cl);
@@ -343,7 +343,7 @@ public class StreamConfig implements Serializable {
throw new StreamTaskException("Cannot serialize inward edges.", e);
}
}
-
+
public List<StreamEdge> getInPhysicalEdges(ClassLoader cl) {
try {
List<StreamEdge> inEdges = InstantiationUtil.readObjectFromConfig(this.config, IN_STREAM_EDGES, cl);
@@ -354,7 +354,7 @@ public class StreamConfig implements Serializable {
}
// --------------------- checkpointing -----------------------
-
+
public void setCheckpointingEnabled(boolean enabled) {
config.setBoolean(CHECKPOINTING_ENABLED, enabled);
}
@@ -362,7 +362,7 @@ public class StreamConfig implements Serializable {
public boolean isCheckpointingEnabled() {
return config.getBoolean(CHECKPOINTING_ENABLED, false);
}
-
+
public void setCheckpointMode(CheckpointingMode mode) {
config.setInteger(CHECKPOINT_MODE, mode.ordinal());
}
@@ -372,7 +372,7 @@ public class StreamConfig implements Serializable {
if (ordinal >= 0) {
return CheckpointingMode.values()[ordinal];
} else {
- return DEFAULT_CHECKPOINTING_MODE;
+ return DEFAULT_CHECKPOINTING_MODE;
}
}
@@ -383,7 +383,7 @@ public class StreamConfig implements Serializable {
throw new StreamTaskException("Could not serialize outputs in order.", e);
}
}
-
+
public List<StreamEdge> getOutEdgesInOrder(ClassLoader cl) {
try {
List<StreamEdge> outEdgesInOrder = InstantiationUtil.readObjectFromConfig(this.config, EDGES_IN_ORDER, cl);
@@ -401,7 +401,7 @@ public class StreamConfig implements Serializable {
throw new StreamTaskException("Could not serialize configuration.", e);
}
}
-
+
public Map<Integer, StreamConfig> getTransitiveChainedTaskConfigs(ClassLoader cl) {
try {
Map<Integer, StreamConfig> confs = InstantiationUtil.readObjectFromConfig(this.config, CHAINED_TASK_CONFIG, cl);
@@ -410,11 +410,11 @@ public class StreamConfig implements Serializable {
throw new StreamTaskException("Could not instantiate configuration.", e);
}
}
-
+
public void setOperatorName(String name) {
this.config.setString(OPERATOR_NAME,name);
}
-
+
public String getOperatorName() {
return this.config.getString(OPERATOR_NAME, null);
}
@@ -426,11 +426,11 @@ public class StreamConfig implements Serializable {
public int getChainIndex() {
return this.config.getInteger(CHAIN_INDEX, 0);
}
-
+
// ------------------------------------------------------------------------
// State backend
// ------------------------------------------------------------------------
-
+
public void setStateBackend(AbstractStateBackend backend) {
if (backend != null) {
try {
@@ -440,7 +440,7 @@ public class StreamConfig implements Serializable {
}
}
}
-
+
public AbstractStateBackend getStateBackend(ClassLoader cl) {
try {
return InstantiationUtil.readObjectFromConfig(this.config, STATE_BACKEND, cl);
@@ -452,7 +452,7 @@ public class StreamConfig implements Serializable {
public byte[] getSerializedStateBackend() {
return this.config.getBytes(STATE_BACKEND, null);
}
-
+
public void setStatePartitioner(int input, KeySelector<?, ?> partitioner) {
try {
InstantiationUtil.writeObjectToConfig(partitioner, this.config, STATE_PARTITIONER + input);
@@ -460,7 +460,7 @@ public class StreamConfig implements Serializable {
throw new StreamTaskException("Could not serialize state partitioner.", e);
}
}
-
+
public KeySelector<?, Serializable> getStatePartitioner(int input, ClassLoader cl) {
try {
return InstantiationUtil.readObjectFromConfig(this.config, STATE_PARTITIONER + input, cl);
@@ -486,11 +486,11 @@ public class StreamConfig implements Serializable {
}
-
+
// ------------------------------------------------------------------------
// Miscellansous
// ------------------------------------------------------------------------
-
+
public void setChainStart() {
config.setBoolean(IS_CHAINED_VERTEX, true);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa93a27/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamEdge.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamEdge.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamEdge.java
index 94e624f..555d67b 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamEdge.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamEdge.java
@@ -102,7 +102,7 @@ public class StreamEdge implements Serializable {
public StreamPartitioner<?> getPartitioner() {
return outputPartitioner;
}
-
+
public void setPartitioner(StreamPartitioner<?> partitioner) {
this.outputPartitioner = partitioner;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa93a27/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java
index 157023e..b132416 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java
@@ -67,11 +67,11 @@ import org.slf4j.LoggerFactory;
/**
* Class representing the streaming topology. It contains all the information
* necessary to build the jobgraph for the execution.
- *
+ *
*/
@Internal
public class StreamGraph extends StreamingPlan {
-
+
private static final Logger LOG = LoggerFactory.getLogger(StreamGraph.class);
private String jobName = StreamExecutionEnvironment.DEFAULT_JOB_NAME;
@@ -79,7 +79,7 @@ public class StreamGraph extends StreamingPlan {
private final StreamExecutionEnvironment environment;
private final ExecutionConfig executionConfig;
private final CheckpointConfig checkpointConfig;
-
+
private boolean chaining;
private Map<Integer, StreamNode> streamNodes;
@@ -118,8 +118,8 @@ public class StreamGraph extends StreamingPlan {
sources = new HashSet<>();
sinks = new HashSet<>();
}
-
-
+
+
public StreamExecutionEnvironment getEnvironment() {
return environment;
}
@@ -127,7 +127,7 @@ public class StreamGraph extends StreamingPlan {
public ExecutionConfig getExecutionConfig() {
return executionConfig;
}
-
+
public CheckpointConfig getCheckpointConfig() {
return checkpointConfig;
}
@@ -153,11 +153,11 @@ public class StreamGraph extends StreamingPlan {
}
// Checkpointing
-
+
public boolean isChainingEnabled() {
return chaining;
}
-
+
public boolean isIterative() {
return!vertexIDtoLoopTimeout.isEmpty();
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa93a27/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphGenerator.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphGenerator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphGenerator.java
index be80371..4a141eb 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphGenerator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphGenerator.java
@@ -273,7 +273,7 @@ public class StreamGraphGenerator {
* Transforms a {@code SelectTransformation}.
*
* <p>For this we create a virtual node in the {@code StreamGraph} holds the selected names.
- *
+ *
* @see org.apache.flink.streaming.api.graph.StreamGraphGenerator
*/
private <T> Collection<Integer> transformSelect(SelectTransformation<T> select) {