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 2015/06/04 18:47:47 UTC
[2/3] flink git commit: [FLINK-2098] Ensure checkpoints and element
emission are in order
[FLINK-2098] Ensure checkpoints and element emission are in order
Before, it could happen that a streaming source would keep emitting
elements while a checkpoint is being performed. This can lead to
inconsistencies in the checkpoints with other operators.
This also adds a test that checks whether only one checkpoint is
executed at a time and the serial behaviour of checkpointing and
emission.
This changes the SourceFunction interface to have run()/cancel() methods
where the run() method takes a lock object on which it needs to
synchronize updates to state and emission of elements.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/6685b0ba
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/6685b0ba
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/6685b0ba
Branch: refs/heads/master
Commit: 6685b0bac54b95705426fd78bb819171271004e6
Parents: 39010dc
Author: Aljoscha Krettek <al...@gmail.com>
Authored: Thu May 28 10:24:37 2015 +0200
Committer: Aljoscha Krettek <al...@gmail.com>
Committed: Thu Jun 4 17:51:38 2015 +0200
----------------------------------------------------------------------
docs/apis/streaming_guide.md | 19 +-
.../jobgraph/tasks/CheckpointedOperator.java | 6 +-
.../hbase/example/HBaseWriteStreamExample.java | 15 +-
.../connectors/kafka/KafkaProducerExample.java | 25 +-
.../connectors/kafka/api/KafkaSource.java | 67 ++---
.../streaming/connectors/kafka/KafkaITCase.java | 251 ++++++++++-------
.../connectors/rabbitmq/RMQSource.java | 50 +---
.../connectors/twitter/TwitterSource.java | 48 ++--
.../connectors/twitter/TwitterStreaming.java | 32 +--
.../connectors/twitter/TwitterTopology.java | 42 +--
.../source/FileMonitoringFunction.java | 86 +++---
.../functions/source/FileSourceFunction.java | 45 ++-
.../functions/source/FromElementsFunction.java | 26 +-
.../functions/source/FromIteratorFunction.java | 15 +-
.../source/FromSplittableIteratorFunction.java | 21 +-
.../source/SocketTextStreamFunction.java | 150 +++++-----
.../api/functions/source/SourceFunction.java | 107 ++++---
.../api/operators/AbstractStreamOperator.java | 2 +-
.../operators/AbstractUdfStreamOperator.java | 9 +-
.../streaming/api/operators/StreamSource.java | 22 +-
.../runtime/tasks/SourceStreamTask.java | 20 +-
.../streaming/runtime/tasks/StreamTask.java | 12 +-
.../api/ChainedRuntimeContextTest.java | 7 +-
.../flink/streaming/api/TypeFillTest.java | 11 +-
.../api/complex/ComplexIntegrationTest.java | 82 +++---
.../windowing/WindowIntegrationTest.java | 28 +-
.../api/streamtask/StreamVertexTest.java | 16 +-
.../runtime/tasks/SourceStreamTaskTest.java | 248 +++++++++++++++++
.../runtime/tasks/StreamMockEnvironment.java | 277 +++++++++++++++++++
.../runtime/tasks/StreamTaskTestBase.java | 109 ++++++++
.../apache/flink/streaming/util/MockSource.java | 4 +-
.../examples/iteration/IterateExample.java | 23 +-
.../streaming/examples/join/WindowJoin.java | 44 +--
.../ml/IncrementalLearningSkeleton.java | 77 ++++--
.../examples/windowing/SessionWindowing.java | 57 ++--
.../windowing/TopSpeedWindowingExample.java | 88 +++---
.../api/scala/StreamExecutionEnvironment.scala | 16 +-
.../StreamCheckpointingITCase.java | 146 +++++++---
.../ProcessFailureStreamingRecoveryITCase.java | 180 +++++-------
39 files changed, 1616 insertions(+), 867 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/6685b0ba/docs/apis/streaming_guide.md
----------------------------------------------------------------------
diff --git a/docs/apis/streaming_guide.md b/docs/apis/streaming_guide.md
index 49dc068..4c7178e 100644
--- a/docs/apis/streaming_guide.md
+++ b/docs/apis/streaming_guide.md
@@ -309,18 +309,13 @@ parallelism of 1. To create parallel sources the users source function needs to
the parallelism of the environment. The parallelism for ParallelSourceFunctions can be changed
after creation by using `source.setParallelism(parallelism)`.
-The `SourceFunction` interface has two methods: `reachedEnd()` and `next()`. The former is used
-by the system to determine whether more input data is available. This method can block if there
-is no data available right now but there might come more data in the future. The `next()` method
-is called to get next data element. This method will only be called if `reachedEnd()` returns
-false. This method can also block if no data is currently available but more will arrive in the
-future.
-
-The methods must react to thread interrupt calls and break out of blocking calls with
-`InterruptedException`. The method may ignore interrupt calls and/or swallow InterruptedExceptions,
-if it is guaranteed that the method returns quasi immediately irrespectively of the input.
-This is true for example for file streams, where the call is guaranteed to return after a very
-short I/O delay in the order of milliseconds.
+The `SourceFunction` interface has two methods: `run(SourceContext)` and `cancel()`. The `run()`
+method is not expected to return until the source has either finished by itself or received
+a cancel request. The source can communicate with the outside world using the source context. For
+example, the `emit(element)` method is used to emit one element from the source. Most sources will
+have an infinite while loop inside the `run()` method to read from the input and emit elements.
+Upon invocation of the `cancel()` method the source is required to break out of its internal
+loop and return from the `run()` method.
In addition to the bounded data sources (with similar method signatures as the
[batch API](programming_guide.html#data-sources)) there are several predefined stream sources
http://git-wip-us.apache.org/repos/asf/flink/blob/6685b0ba/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/tasks/CheckpointedOperator.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/tasks/CheckpointedOperator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/tasks/CheckpointedOperator.java
index 17ba947..84072e1 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/tasks/CheckpointedOperator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/tasks/CheckpointedOperator.java
@@ -19,6 +19,10 @@
package org.apache.flink.runtime.jobgraph.tasks;
public interface CheckpointedOperator {
-
+
+ /**
+ * This method is either called directly by the checkpoint coordinator, or called
+ * when all incoming channels have reported a barrier
+ */
void triggerCheckpoint(long checkpointId, long timestamp) throws Exception;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6685b0ba/flink-staging/flink-hbase/src/test/java/org/apache/flink/addons/hbase/example/HBaseWriteStreamExample.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-hbase/src/test/java/org/apache/flink/addons/hbase/example/HBaseWriteStreamExample.java b/flink-staging/flink-hbase/src/test/java/org/apache/flink/addons/hbase/example/HBaseWriteStreamExample.java
index 1e113c0..74c6c57 100644
--- a/flink-staging/flink-hbase/src/test/java/org/apache/flink/addons/hbase/example/HBaseWriteStreamExample.java
+++ b/flink-staging/flink-hbase/src/test/java/org/apache/flink/addons/hbase/example/HBaseWriteStreamExample.java
@@ -48,17 +48,22 @@ public class HBaseWriteStreamExample {
// data stream with random numbers
DataStream<String> dataStream = env.addSource(new SourceFunction<String>() {
+ private static final long serialVersionUID = 1L;
+
+ private volatile boolean isRunning = true;
@Override
- public boolean reachedEnd() throws Exception {
- return false;
+ public void run(Object checkpointLock, Collector<String> out) throws Exception {
+ while (isRunning) {
+ out.collect(String.valueOf(Math.floor(Math.random() * 100)));
+ }
+
}
@Override
- public String next() throws Exception {
- return String.valueOf(Math.floor(Math.random() * 100));
+ public void cancel() {
+ isRunning = false;
}
-
});
dataStream.write(new HBaseOutputFormat(), 0L);
http://git-wip-us.apache.org/repos/asf/flink/blob/6685b0ba/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerExample.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerExample.java b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerExample.java
index 4dd5577..a29e937 100644
--- a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerExample.java
+++ b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerExample.java
@@ -22,6 +22,7 @@ import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.connectors.kafka.api.KafkaSink;
import org.apache.flink.streaming.util.serialization.JavaDefaultStringSchema;
+import org.apache.flink.util.Collector;
public class KafkaProducerExample {
@@ -39,29 +40,25 @@ public class KafkaProducerExample {
@SuppressWarnings({ "unused", "serial" })
DataStream<String> stream1 = env.addSource(new SourceFunction<String>() {
-
- private int index = 0;
-
@Override
- public boolean reachedEnd() throws Exception {
- return index >= 20;
+ public void run(Object checkpointLock, Collector<String> collector) throws Exception {
+ for (int i = 0; i < 20; i++) {
+ collector.collect("message #" + i);
+ Thread.sleep(100L);
+ }
+
+ collector.collect("q");
}
@Override
- public String next() throws Exception {
- if (index < 20) {
- String result = "message #" + index;
- index++;
- return result;
- }
-
- return "q";
+ public void cancel() {
}
+
}).addSink(
new KafkaSink<String>(host + ":" + port, topic, new JavaDefaultStringSchema())
)
- .setParallelism(3);
+ .setParallelism(3);
env.execute();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6685b0ba/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/api/KafkaSource.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/api/KafkaSource.java b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/api/KafkaSource.java
index 4a7ec15..2fa2c26 100644
--- a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/api/KafkaSource.java
+++ b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/api/KafkaSource.java
@@ -32,12 +32,13 @@ import kafka.javaapi.consumer.ConsumerConnector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.ConnectorSource;
import org.apache.flink.streaming.util.serialization.DeserializationSchema;
+import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Source that listens to a Kafka topic using the high level Kafka API.
- *
+ *
* @param <OUT>
* Type of the messages on the topic.
*/
@@ -59,9 +60,7 @@ public class KafkaSource<OUT> extends ConnectorSource<OUT> {
private static final long ZOOKEEPER_DEFAULT_SYNC_TIME = 200;
private static final String DEFAULT_GROUP_ID = "flink-group";
- // We must read this in reachedEnd() to check for the end. We keep it to return it in
- // next()
- private OUT nextElement;
+ private volatile boolean isRunning = false;
/**
* Creates a KafkaSource that consumes a topic.
@@ -78,14 +77,15 @@ public class KafkaSource<OUT> extends ConnectorSource<OUT> {
* Synchronization time with zookeeper.
*/
public KafkaSource(String zookeeperAddress,
- String topicId, String groupId,
- DeserializationSchema<OUT> deserializationSchema,
- long zookeeperSyncTimeMillis) {
+ String topicId,
+ String groupId,
+ DeserializationSchema<OUT> deserializationSchema,
+ long zookeeperSyncTimeMillis) {
this(zookeeperAddress, topicId, groupId, deserializationSchema, zookeeperSyncTimeMillis, null);
}
/**
* Creates a KafkaSource that consumes a topic.
- *
+ *
* @param zookeeperAddress
* Address of the Zookeeper host (with port number).
* @param topicId
@@ -100,9 +100,9 @@ public class KafkaSource<OUT> extends ConnectorSource<OUT> {
* Custom properties for Kafka
*/
public KafkaSource(String zookeeperAddress,
- String topicId, String groupId,
- DeserializationSchema<OUT> deserializationSchema,
- long zookeeperSyncTimeMillis, Properties customProperties) {
+ String topicId, String groupId,
+ DeserializationSchema<OUT> deserializationSchema,
+ long zookeeperSyncTimeMillis, Properties customProperties) {
super(deserializationSchema);
Preconditions.checkNotNull(zookeeperAddress, "ZK address is null");
Preconditions.checkNotNull(topicId, "Topic ID is null");
@@ -178,42 +178,31 @@ public class KafkaSource<OUT> extends ConnectorSource<OUT> {
}
@Override
- public void open(Configuration config) throws Exception {
- super.open(config);
- initializeConnection();
- }
-
- @Override
- public void close() throws Exception {
- super.close();
- if (consumer != null) {
+ public void run(Object checkpointLock, Collector<OUT> collector) throws Exception {
+ isRunning = true;
+ try {
+ while (isRunning && consumerIterator.hasNext()) {
+ OUT out = schema.deserialize(consumerIterator.next().message());
+ if (schema.isEndOfStream(out)) {
+ break;
+ }
+ collector.collect(out);
+ }
+ } finally {
consumer.shutdown();
}
}
@Override
- public boolean reachedEnd() throws Exception {
- if (nextElement != null) {
- return false;
- } else if (consumerIterator.hasNext()) {
- OUT out = schema.deserialize(consumerIterator.next().message());
- if (schema.isEndOfStream(out)) {
- return true;
- }
- nextElement = out;
- }
- return false;
+ public void open(Configuration config) throws Exception {
+ initializeConnection();
}
@Override
- public OUT next() throws Exception {
- if (!reachedEnd()) {
- OUT result = nextElement;
- nextElement = null;
- return result;
- } else {
- throw new RuntimeException("Source exhausted");
+ public void cancel() {
+ isRunning = false;
+ if (consumer != null) {
+ consumer.shutdown();
}
}
-
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6685b0ba/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaITCase.java b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaITCase.java
index 51682ab..d695b09 100644
--- a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaITCase.java
+++ b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaITCase.java
@@ -70,7 +70,6 @@ import org.apache.flink.streaming.connectors.kafka.util.KafkaLocalSystemTime;
import org.apache.flink.streaming.util.serialization.DeserializationSchema;
import org.apache.flink.streaming.util.serialization.JavaDefaultStringSchema;
import org.apache.flink.util.Collector;
-import org.apache.flink.util.StringUtils;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.BeforeClass;
@@ -316,9 +315,9 @@ public class KafkaITCase {
Assert.assertTrue("The offset seems incorrect, got " + o2, o2 > 50L);
Assert.assertTrue("The offset seems incorrect, got " + o3, o3 > 50L);
/** Once we have proper shutdown of streaming jobs, enable these tests
- Assert.assertEquals("The offset seems incorrect", 99L, PersistentKafkaSource.getOffset(zk, standardCC.groupId(), topicName, 0));
- Assert.assertEquals("The offset seems incorrect", 99L, PersistentKafkaSource.getOffset(zk, standardCC.groupId(), topicName, 1));
- Assert.assertEquals("The offset seems incorrect", 99L, PersistentKafkaSource.getOffset(zk, standardCC.groupId(), topicName, 2));*/
+ Assert.assertEquals("The offset seems incorrect", 99L, PersistentKafkaSource.getOffset(zk, standardCC.groupId(), topicName, 0));
+ Assert.assertEquals("The offset seems incorrect", 99L, PersistentKafkaSource.getOffset(zk, standardCC.groupId(), topicName, 1));
+ Assert.assertEquals("The offset seems incorrect", 99L, PersistentKafkaSource.getOffset(zk, standardCC.groupId(), topicName, 2));*/
LOG.info("Manipulating offsets");
@@ -342,18 +341,22 @@ public class KafkaITCase {
DataStream<Tuple2<Integer, Integer>> source = env.addSource(
new PersistentKafkaSource<Tuple2<Integer, Integer>>(topicName, new Utils.TypeInformationSerializationSchema<Tuple2<Integer, Integer>>(new Tuple2<Integer, Integer>(1,1), env.getConfig()), cc)
)
- //add a sleeper mapper. Since there is no good way of "shutting down" a running topology, we have
- // to play this trick. The problem is that we have to wait until all checkpoints are confirmed
- .map(new MapFunction<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>>() {
- @Override
- public Tuple2<Integer, Integer> map(Tuple2<Integer, Integer> value) throws Exception {
- Thread.sleep(150);
- return value;
- }
- }).setParallelism(3);
+ //add a sleeper mapper. Since there is no good way of "shutting down" a running topology, we have
+ // to play this trick. The problem is that we have to wait until all checkpoints are confirmed
+ .map(new MapFunction<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>>() {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public Tuple2<Integer, Integer> map(Tuple2<Integer, Integer> value) throws Exception {
+ Thread.sleep(150);
+ return value;
+ }
+ }).setParallelism(3);
// verify data
DataStream<Integer> validIndexes = source.flatMap(new RichFlatMapFunction<Tuple2<Integer, Integer>, Integer>() {
+ private static final long serialVersionUID = 1L;
+
int[] values = new int[valuesCount];
int count = 0;
@@ -391,33 +394,32 @@ public class KafkaITCase {
private void writeSequence(StreamExecutionEnvironment env, String topicName, final int from, final int to) throws Exception {
LOG.info("Writing sequence from {} to {} to topic {}", from, to, topicName);
-
DataStream<Tuple2<Integer, Integer>> stream = env.addSource(new RichParallelSourceFunction<Tuple2<Integer, Integer>>() {
private static final long serialVersionUID = 1L;
- int cnt = from;
- int partition;
-
- @Override
- public void open(Configuration parameters) throws Exception {
- super.open(parameters);
- partition = getRuntimeContext().getIndexOfThisSubtask();
-
- }
+ boolean running = true;
@Override
- public boolean reachedEnd() throws Exception {
- return cnt > to;
+ public void run(Object checkpointLock, Collector<Tuple2<Integer, Integer>> collector) throws Exception {
+ LOG.info("Starting source.");
+ int cnt = from;
+ int partition = getRuntimeContext().getIndexOfThisSubtask();
+ while (running) {
+ LOG.info("Writing " + cnt + " to partition " + partition);
+ collector.collect(new Tuple2<Integer, Integer>(getRuntimeContext().getIndexOfThisSubtask(), cnt));
+ if (cnt == to) {
+ LOG.info("Writer reached end.");
+ return;
+ }
+ cnt++;
+ }
}
@Override
- public Tuple2<Integer, Integer> next() throws Exception {
- LOG.info("Writing " + cnt + " to partition " + partition);
- Tuple2<Integer, Integer> result = new Tuple2<Integer, Integer>(getRuntimeContext().getIndexOfThisSubtask(), cnt);
- cnt++;
- return result;
+ public void cancel() {
+ LOG.info("Source got cancel()");
+ running = false;
}
}).setParallelism(3);
-
stream.addSink(new KafkaSink<Tuple2<Integer, Integer>>(brokerConnectionStrings,
topicName,
new Utils.TypeInformationSerializationSchema<Tuple2<Integer, Integer>>(new Tuple2<Integer, Integer>(1, 1), env.getConfig()),
@@ -428,6 +430,8 @@ public class KafkaITCase {
}
private static class T2Partitioner implements SerializableKafkaPartitioner {
+ private static final long serialVersionUID = 1L;
+
@Override
public int partition(Object key, int numPartitions) {
if(numPartitions != 3) {
@@ -451,6 +455,8 @@ public class KafkaITCase {
DataStreamSource<Tuple2<Long, String>> consuming = env.addSource(
new KafkaSource<Tuple2<Long, String>>(zookeeperConnectionString, topic, "myFlinkGroup", new Utils.TypeInformationSerializationSchema<Tuple2<Long, String>>(new Tuple2<Long, String>(1L, ""), env.getConfig()), 5000));
consuming.addSink(new SinkFunction<Tuple2<Long, String>>() {
+ private static final long serialVersionUID = 1L;
+
int elCnt = 0;
int start = -1;
BitSet validator = new BitSet(101);
@@ -483,17 +489,25 @@ public class KafkaITCase {
// add producing topology
DataStream<Tuple2<Long, String>> stream = env.addSource(new SourceFunction<Tuple2<Long, String>>() {
private static final long serialVersionUID = 1L;
- int cnt = 0;
+ boolean running = true;
@Override
- public boolean reachedEnd() throws Exception {
- return false;
+ public void run(Object checkpointLock, Collector<Tuple2<Long, String>> collector) throws Exception {
+ LOG.info("Starting source.");
+ int cnt = 0;
+ while (running) {
+ collector.collect(new Tuple2<Long, String>(1000L + cnt, "kafka-" + cnt++));
+ try {
+ Thread.sleep(100);
+ } catch (InterruptedException ignored) {
+ }
+ }
}
@Override
- public Tuple2<Long, String> next() throws Exception {
- Thread.sleep(100);
- return new Tuple2<Long, String>(1000L + cnt, "kafka-" + cnt++);
+ public void cancel() {
+ LOG.info("Source got cancel()");
+ running = false;
}
});
stream.addSink(new KafkaSink<Tuple2<Long, String>>(brokerConnectionStrings, topic, new Utils.TypeInformationSerializationSchema<Tuple2<Long, String>>(new Tuple2<Long, String>(1L, ""), env.getConfig())));
@@ -519,6 +533,8 @@ public class KafkaITCase {
standardCC
));
consuming.addSink(new RichSinkFunction<Tuple2<Long, String>>() {
+ private static final long serialVersionUID = 1L;
+
int elCnt = 0;
int start = -1;
BitSet validator = new BitSet(101);
@@ -557,17 +573,27 @@ public class KafkaITCase {
// add producing topology
DataStream<Tuple2<Long, String>> stream = env.addSource(new SourceFunction<Tuple2<Long, String>>() {
private static final long serialVersionUID = 1L;
- int cnt = 0;
+ boolean running = true;
@Override
- public boolean reachedEnd() throws Exception {
- return false;
+ public void run(Object checkpointLock, Collector<Tuple2<Long, String>> collector) throws Exception {
+ LOG.info("Starting source.");
+ int cnt = 0;
+ while (running) {
+ collector.collect(new Tuple2<Long, String>(1000L + cnt, "kafka-" + cnt++));
+ LOG.info("Produced " + cnt);
+
+ try {
+ Thread.sleep(100);
+ } catch (InterruptedException ignored) {
+ }
+ }
}
@Override
- public Tuple2<Long, String> next() throws Exception {
- Thread.sleep(100);
- return new Tuple2<Long, String>(1000L + cnt, "kafka-" + cnt++);
+ public void cancel() {
+ LOG.info("Source got cancel()");
+ running = false;
}
});
stream.addSink(new KafkaSink<Tuple2<Long, String>>(brokerConnectionStrings, topic, new Utils.TypeInformationSerializationSchema<Tuple2<Long, String>>(new Tuple2<Long, String>(1L, ""), env.getConfig())));
@@ -608,6 +634,8 @@ public class KafkaITCase {
new PersistentKafkaSource<Tuple2<Long, byte[]>>(topic, serSchema, cc));
consuming.addSink(new SinkFunction<Tuple2<Long, byte[]>>() {
+ private static final long serialVersionUID = 1L;
+
int elCnt = 0;
@Override
@@ -631,45 +659,47 @@ public class KafkaITCase {
// add producing topology
DataStream<Tuple2<Long, byte[]>> stream = env.addSource(new RichSourceFunction<Tuple2<Long, byte[]>>() {
private static final long serialVersionUID = 1L;
- boolean running = true;
- long cnt;
- transient Random rnd;
+ boolean running;
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
- cnt = 0;
- rnd = new Random(1337);
-
+ running = true;
}
@Override
- public boolean reachedEnd() throws Exception {
- return cnt > 10;
- }
-
- @Override
- public Tuple2<Long, byte[]> next() throws Exception {
- Thread.sleep(100);
-
- if (cnt < 10) {
- byte[] wl = new byte[Math.abs(rnd.nextInt(1024 * 1024 * 30))];
- Tuple2<Long, byte[]> result = new Tuple2<Long, byte[]>(cnt++, wl);
+ public void run(Object checkpointLock, Collector<Tuple2<Long, byte[]>> collector) throws Exception {
+ LOG.info("Starting source.");
+ long cnt = 0;
+ Random rnd = new Random(1337);
+ while (running) {
+ //
+ byte[] wl = new byte[Math.abs(rnd.nextInt(1024 * 1024 * 30))];
+ collector.collect(new Tuple2<Long, byte[]>(cnt++, wl));
LOG.info("Emitted cnt=" + (cnt - 1) + " with byte.length = " + wl.length);
- return result;
- } else if (cnt == 10) {
- Tuple2<Long, byte[]> result = new Tuple2<Long, byte[]>(-1L, new byte[]{1});
- cnt++;
- return result;
- } else {
- throw new RuntimeException("Source is exhausted.");
+ try {
+ Thread.sleep(100);
+ } catch (InterruptedException ignored) {
+ }
+ if(cnt == 10) {
+ LOG.info("Send end signal");
+ // signal end
+ collector.collect(new Tuple2<Long, byte[]>(-1L, new byte[]{1}));
+ running = false;
+ }
}
}
+
+ @Override
+ public void cancel() {
+ LOG.info("Source got cancel()");
+ running = false;
+ }
});
stream.addSink(new KafkaSink<Tuple2<Long, byte[]>>(brokerConnectionStrings, topic,
- new Utils.TypeInformationSerializationSchema<Tuple2<Long, byte[]>>(new Tuple2<Long, byte[]>(0L, new byte[]{0}), env.getConfig()))
+ new Utils.TypeInformationSerializationSchema<Tuple2<Long, byte[]>>(new Tuple2<Long, byte[]>(0L, new byte[]{0}), env.getConfig()))
);
tryExecute(env, "big topology test");
@@ -694,6 +724,8 @@ public class KafkaITCase {
new Utils.TypeInformationSerializationSchema<Tuple2<Long, String>>(new Tuple2<Long, String>(1L, ""), env.getConfig()),
standardCC));
consuming.addSink(new SinkFunction<Tuple2<Long, String>>() {
+ private static final long serialVersionUID = 1L;
+
int start = -1;
BitSet validator = new BitSet(101);
@@ -741,17 +773,25 @@ public class KafkaITCase {
// add producing topology
DataStream<Tuple2<Long, String>> stream = env.addSource(new SourceFunction<Tuple2<Long, String>>() {
private static final long serialVersionUID = 1L;
- int cnt = 0;
+ boolean running = true;
@Override
- public boolean reachedEnd() throws Exception {
- return false;
+ public void run(Object checkpointLock, Collector<Tuple2<Long, String>> collector) throws Exception {
+ LOG.info("Starting source.");
+ int cnt = 0;
+ while (running) {
+ collector.collect(new Tuple2<Long, String>(1000L + cnt, "kafka-" + cnt++));
+ try {
+ Thread.sleep(100);
+ } catch (InterruptedException ignored) {
+ }
+ }
}
@Override
- public Tuple2<Long, String> next() throws Exception {
- Thread.sleep(100);
- return new Tuple2<Long, String>(1000L + cnt, "kafka-" + cnt++);
+ public void cancel() {
+ LOG.info("Source got cancel()");
+ running = false;
}
});
stream.addSink(new KafkaSink<Tuple2<Long, String>>(brokerConnectionStrings, topic, new Utils.TypeInformationSerializationSchema<Tuple2<Long, String>>(new Tuple2<Long, String>(1L, ""), env.getConfig()), new CustomPartitioner()));
@@ -765,6 +805,7 @@ public class KafkaITCase {
* This is for a topic with 3 partitions and Tuple2<Long, String>
*/
private static class CustomPartitioner implements SerializableKafkaPartitioner {
+ private static final long serialVersionUID = 1L;
@Override
public int partition(Object key, int numPartitions) {
@@ -794,6 +835,8 @@ public class KafkaITCase {
DataStreamSource<String> consuming = env.addSource(
new PersistentKafkaSource<String>(topic, new JavaDefaultStringSchema(), standardCC));
consuming.addSink(new SinkFunction<String>() {
+ private static final long serialVersionUID = 1L;
+
int elCnt = 0;
int start = -1;
BitSet validator = new BitSet(101);
@@ -824,17 +867,24 @@ public class KafkaITCase {
DataStream<String> stream = env.addSource(new SourceFunction<String>() {
private static final long serialVersionUID = 1L;
boolean running = true;
- int cnt = 0;
@Override
- public boolean reachedEnd() throws Exception {
- return false;
+ public void run(Object checkpointLock, Collector<String> collector) throws Exception {
+ LOG.info("Starting source.");
+ int cnt = 0;
+ while (running) {
+ collector.collect("kafka-" + cnt++);
+ try {
+ Thread.sleep(100);
+ } catch (InterruptedException ignored) {
+ }
+ }
}
@Override
- public String next() throws Exception {
- Thread.sleep(100);
- return "kafka-" + cnt++;
+ public void cancel() {
+ LOG.info("Source got cancel()");
+ running = false;
}
});
stream.addSink(new KafkaSink<String>(brokerConnectionStrings, topic, new JavaDefaultStringSchema()));
@@ -856,26 +906,35 @@ public class KafkaITCase {
StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(1);
DataStream<String> stream = env.addSource(new SourceFunction<String>() {
+ private static final long serialVersionUID = 1L;
- private int cnt = 0;
-
- @Override
- public boolean reachedEnd() throws Exception {
- return cnt == 200;
- }
+ boolean running = true;
@Override
- public String next() throws Exception {
- String msg = "kafka-" + cnt++;
- LOG.info("sending message = "+msg);
+ public void run(Object checkpointLock, Collector<String> collector) throws Exception {
+ LOG.info("Starting source.");
+ int cnt = 0;
+ while (running) {
+ String msg = "kafka-" + cnt++;
+ collector.collect(msg);
+ LOG.info("sending message = "+msg);
+
+ if ((cnt - 1) % 20 == 0) {
+ LOG.debug("Sending message #{}", cnt - 1);
+ }
+ if(cnt == 200) {
+ LOG.info("Stopping to produce after 200 msgs");
+ break;
+ }
- if ((cnt - 1) % 20 == 0) {
- LOG.debug("Sending message #{}", cnt - 1);
}
-
- return msg;
}
+ @Override
+ public void cancel() {
+ LOG.info("Source got chancel()");
+ running = false;
+ }
});
stream.addSink(new KafkaSink<String>(brokerConnectionStrings, topic, new JavaDefaultStringSchema()))
.setParallelism(1);
@@ -928,6 +987,8 @@ public class KafkaITCase {
consuming.setParallelism(1);
consuming.addSink(new SinkFunction<String>() {
+ private static final long serialVersionUID = 1L;
+
int elCnt = 0;
int start = 0;
int numOfMessagesToBeCorrect = 100;
@@ -987,13 +1048,13 @@ public class KafkaITCase {
while (!(t instanceof SuccessException)) {
if(t == null) {
LOG.warn("Test failed with exception", good);
- Assert.fail("Test failed with: " + StringUtils.stringifyException(good));
+ Assert.fail("Test failed with: " + good.getMessage());
}
t = t.getCause();
if (limit++ == 20) {
LOG.warn("Test failed with exception", good);
- Assert.fail("Test failed with: " + StringUtils.stringifyException(good));
+ Assert.fail("Test failed with: " + good.getMessage());
}
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6685b0ba/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSource.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSource.java b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSource.java
index d706b8c..93812c2 100644
--- a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSource.java
+++ b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSource.java
@@ -27,6 +27,7 @@ import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;
+import org.apache.flink.util.Collector;
public class RMQSource<OUT> extends ConnectorSource<OUT> {
private static final long serialVersionUID = 1L;
@@ -40,7 +41,7 @@ public class RMQSource<OUT> extends ConnectorSource<OUT> {
private transient QueueingConsumer consumer;
private transient QueueingConsumer.Delivery delivery;
- OUT out;
+ private transient volatile boolean running;
public RMQSource(String HOST_NAME, String QUEUE_NAME,
DeserializationSchema<OUT> deserializationSchema) {
@@ -70,6 +71,7 @@ public class RMQSource<OUT> extends ConnectorSource<OUT> {
@Override
public void open(Configuration config) throws Exception {
initializeConnection();
+ running = true;
}
@Override
@@ -84,47 +86,21 @@ public class RMQSource<OUT> extends ConnectorSource<OUT> {
}
@Override
- public boolean reachedEnd() throws Exception {
- if (out != null) {
- return true;
- }
- try {
+ public void run(Object checkpointLock, Collector<OUT> out) throws Exception {
+ while (running) {
delivery = consumer.nextDelivery();
- } catch (Exception e) {
- throw new RuntimeException("Error while reading message from RMQ source from " + QUEUE_NAME
- + " at " + HOST_NAME, e);
- }
- out = schema.deserialize(delivery.getBody());
- if (schema.isEndOfStream(out)) {
- out = null;
- return false;
+ OUT result = schema.deserialize(delivery.getBody());
+ if (schema.isEndOfStream(result)) {
+ break;
+ }
+
+ out.collect(result);
}
- return true;
}
@Override
- public OUT next() throws Exception {
- if (out != null) {
- OUT result = out;
- out = null;
- return result;
- }
-
- try {
- delivery = consumer.nextDelivery();
- } catch (Exception e) {
- throw new RuntimeException("Error while reading message from RMQ source from " + QUEUE_NAME
- + " at " + HOST_NAME, e);
- }
-
- out = schema.deserialize(delivery.getBody());
- if (schema.isEndOfStream(out)) {
- throw new RuntimeException("RMQ source is at end for " + QUEUE_NAME + " at " + HOST_NAME);
- }
- OUT result = out;
- out = null;
- return result;
+ public void cancel() {
+ running = false;
}
-
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6685b0ba/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-twitter/src/main/java/org/apache/flink/streaming/connectors/twitter/TwitterSource.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-twitter/src/main/java/org/apache/flink/streaming/connectors/twitter/TwitterSource.java b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-twitter/src/main/java/org/apache/flink/streaming/connectors/twitter/TwitterSource.java
index 0b47985..a1a6d9b 100644
--- a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-twitter/src/main/java/org/apache/flink/streaming/connectors/twitter/TwitterSource.java
+++ b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-twitter/src/main/java/org/apache/flink/streaming/connectors/twitter/TwitterSource.java
@@ -26,6 +26,7 @@ import java.util.concurrent.LinkedBlockingQueue;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -40,7 +41,8 @@ import com.twitter.hbc.httpclient.auth.OAuth1;
/**
* Implementation of {@link SourceFunction} specialized to emit tweets from
- * Twitter.
+ * Twitter. This is not a parallel source because the Twitter API only allows
+ * two concurrent connections.
*/
public class TwitterSource extends RichSourceFunction<String> {
@@ -56,6 +58,8 @@ public class TwitterSource extends RichSourceFunction<String> {
private int maxNumberOfTweets;
private int currentNumberOfTweets;
+ private transient volatile boolean isRunning;
+
/**
* Create {@link TwitterSource} for streaming
*
@@ -86,6 +90,7 @@ public class TwitterSource extends RichSourceFunction<String> {
public void open(Configuration parameters) throws Exception {
initializeConnection();
currentNumberOfTweets = 0;
+ isRunning = true;
}
/**
@@ -148,13 +153,16 @@ public class TwitterSource extends RichSourceFunction<String> {
client.connect();
}
- private void closeConnection() {
+ @Override
+ public void close() {
if (LOG.isInfoEnabled()) {
LOG.info("Initiating connection close");
}
- client.stop();
+ if (client != null) {
+ client.stop();
+ }
if (LOG.isInfoEnabled()) {
LOG.info("Connection closed successfully");
@@ -201,30 +209,26 @@ public class TwitterSource extends RichSourceFunction<String> {
}
@Override
- public boolean reachedEnd() throws Exception {
- if (maxNumberOfTweets != -1 && currentNumberOfTweets >= maxNumberOfTweets){
- closeConnection();
- return true;
- }
+ public void run(Object checkpointLock, Collector<String> out) throws Exception {
+ while (isRunning) {
+ if (client.isDone()) {
+ if (LOG.isErrorEnabled()) {
+ LOG.error("Client connection closed unexpectedly: {}", client.getExitEvent()
+ .getMessage());
+ }
+ break;
+ }
- if (client.isDone()) {
- if (LOG.isErrorEnabled()) {
- LOG.error("Client connection closed unexpectedly: {}", client.getExitEvent()
- .getMessage());
+ out.collect(queue.take());
+
+ if (maxNumberOfTweets != -1 && currentNumberOfTweets >= maxNumberOfTweets) {
+ break;
}
- return true;
}
-
- return false;
}
@Override
- public String next() throws Exception {
- if (reachedEnd()) {
- throw new RuntimeException("Twitter stream end reached.");
- }
-
- String msg = queue.take();
- return msg;
+ public void cancel() {
+ isRunning = false;
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6685b0ba/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-twitter/src/main/java/org/apache/flink/streaming/connectors/twitter/TwitterStreaming.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-twitter/src/main/java/org/apache/flink/streaming/connectors/twitter/TwitterStreaming.java b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-twitter/src/main/java/org/apache/flink/streaming/connectors/twitter/TwitterStreaming.java
index e500fef..a80c32a 100644
--- a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-twitter/src/main/java/org/apache/flink/streaming/connectors/twitter/TwitterStreaming.java
+++ b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-twitter/src/main/java/org/apache/flink/streaming/connectors/twitter/TwitterStreaming.java
@@ -1,19 +1,19 @@
/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
+* Licensed to the Apache Software Foundation (ASF) under one or more
+* contributor license agreements. See the NOTICE file distributed with
+* this work for additional information regarding copyright ownership.
+* The ASF licenses this file to You under the Apache License, Version 2.0
+* (the "License"); you may not use this file except in compliance with
+* the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
package org.apache.flink.streaming.connectors.twitter;
@@ -68,7 +68,7 @@ public class TwitterStreaming {
if (LOG.isErrorEnabled()) {
LOG.error("Field not found");
}
- }
+ }
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6685b0ba/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-twitter/src/main/java/org/apache/flink/streaming/connectors/twitter/TwitterTopology.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-twitter/src/main/java/org/apache/flink/streaming/connectors/twitter/TwitterTopology.java b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-twitter/src/main/java/org/apache/flink/streaming/connectors/twitter/TwitterTopology.java
index 4bc6df0..d5e8c41 100644
--- a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-twitter/src/main/java/org/apache/flink/streaming/connectors/twitter/TwitterTopology.java
+++ b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-twitter/src/main/java/org/apache/flink/streaming/connectors/twitter/TwitterTopology.java
@@ -1,19 +1,19 @@
/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
+* Licensed to the Apache Software Foundation (ASF) under one or more
+* contributor license agreements. See the NOTICE file distributed with
+* this work for additional information regarding copyright ownership.
+* The ASF licenses this file to You under the Apache License, Version 2.0
+* (the "License"); you may not use this file except in compliance with
+* the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
package org.apache.flink.streaming.connectors.twitter;
@@ -26,15 +26,15 @@ import org.apache.flink.util.Collector;
import org.apache.sling.commons.json.JSONException;
/**
- * This program demonstrate the use of TwitterSource.
- * Its aim is to count the frequency of the languages of tweets
- */
+* This program demonstrate the use of TwitterSource.
+* Its aim is to count the frequency of the languages of tweets
+*/
public class TwitterTopology {
private static final int NUMBEROFTWEETS = 100;
-
+
/**
- * FlatMapFunction to determine the language of tweets if possible
+ * FlatMapFunction to determine the language of tweets if possible
*/
public static class SelectLanguageFlatMap extends
JSONParseFlatMap<String, String> {
@@ -76,7 +76,7 @@ public class TwitterTopology {
.flatMap(new SelectLanguageFlatMap())
.map(new MapFunction<String, Tuple2<String, Integer>>() {
private static final long serialVersionUID = 1L;
-
+
@Override
public Tuple2<String, Integer> map(String value) throws Exception {
return new Tuple2<String, Integer>(value, 1);
http://git-wip-us.apache.org/repos/asf/flink/blob/6685b0ba/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FileMonitoringFunction.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FileMonitoringFunction.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FileMonitoringFunction.java
index f74d81c..ecbcee5 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FileMonitoringFunction.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FileMonitoringFunction.java
@@ -21,20 +21,18 @@ import java.io.IOException;
import java.net.URI;
import java.util.ArrayList;
import java.util.HashMap;
-import java.util.LinkedList;
import java.util.List;
import java.util.Map;
-import java.util.Queue;
import org.apache.flink.api.java.tuple.Tuple3;
-import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.FileStatus;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.fs.Path;
+import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class FileMonitoringFunction extends RichSourceFunction<Tuple3<String, Long, Long>> {
+public class FileMonitoringFunction implements SourceFunction<Tuple3<String, Long, Long>> {
private static final long serialVersionUID = 1L;
private static final Logger LOG = LoggerFactory.getLogger(FileMonitoringFunction.class);
@@ -51,28 +49,50 @@ public class FileMonitoringFunction extends RichSourceFunction<Tuple3<String, Lo
private long interval;
private WatchType watchType;
- private FileSystem fileSystem;
private Map<String, Long> offsetOfFiles;
private Map<String, Long> modificationTimes;
- private Queue<Tuple3<String, Long, Long>> pendingFiles;
+ private volatile boolean isRunning;
public FileMonitoringFunction(String path, long interval, WatchType watchType) {
this.path = path;
this.interval = interval;
this.watchType = watchType;
+ this.modificationTimes = new HashMap<String, Long>();
+ this.offsetOfFiles = new HashMap<String, Long>();
}
@Override
- public void open(Configuration parameters) throws Exception {
- super.open(parameters);
- this.modificationTimes = new HashMap<String, Long>();
- this.offsetOfFiles = new HashMap<String, Long>();
- this.pendingFiles = new LinkedList<Tuple3<String, Long, Long>>();
- fileSystem = FileSystem.get(new URI(path));
+ public void run(Object checkpointLock, Collector<Tuple3<String, Long, Long>> collector) throws Exception {
+ isRunning = true;
+ FileSystem fileSystem = FileSystem.get(new URI(path));
+
+ while (isRunning) {
+ List<String> files = listNewFiles(fileSystem);
+ for (String filePath : files) {
+ if (watchType == WatchType.ONLY_NEW_FILES
+ || watchType == WatchType.REPROCESS_WITH_APPENDED) {
+ collector.collect(new Tuple3<String, Long, Long>(filePath, 0L, -1L));
+ offsetOfFiles.put(filePath, -1L);
+ } else if (watchType == WatchType.PROCESS_ONLY_APPENDED) {
+ long offset = 0;
+ long fileSize = fileSystem.getFileStatus(new Path(filePath)).getLen();
+ if (offsetOfFiles.containsKey(filePath)) {
+ offset = offsetOfFiles.get(filePath);
+ }
+
+ collector.collect(new Tuple3<String, Long, Long>(filePath, offset, fileSize));
+ offsetOfFiles.put(filePath, fileSize);
+
+ LOG.info("File processed: {}, {}, {}", filePath, offset, fileSize);
+ }
+ }
+
+ Thread.sleep(interval);
+ }
}
- private List<String> listNewFiles() throws IOException {
+ private List<String> listNewFiles(FileSystem fileSystem) throws IOException {
List<String> files = new ArrayList<String>();
FileStatus[] statuses = fileSystem.listStatus(new Path(path));
@@ -105,44 +125,8 @@ public class FileMonitoringFunction extends RichSourceFunction<Tuple3<String, Lo
}
}
-
- @Override
- public boolean reachedEnd() throws Exception {
- return false;
- }
-
@Override
- public Tuple3<String, Long, Long> next() throws Exception {
- if (pendingFiles.size() > 0) {
- return pendingFiles.poll();
- } else {
- while (true) {
- List<String> files = listNewFiles();
- for (String filePath : files) {
- if (watchType == WatchType.ONLY_NEW_FILES
- || watchType == WatchType.REPROCESS_WITH_APPENDED) {
- pendingFiles.add(new Tuple3<String, Long, Long>(filePath, 0L, -1L));
- offsetOfFiles.put(filePath, -1L);
- } else if (watchType == WatchType.PROCESS_ONLY_APPENDED) {
- long offset = 0;
- long fileSize = fileSystem.getFileStatus(new Path(filePath)).getLen();
- if (offsetOfFiles.containsKey(filePath)) {
- offset = offsetOfFiles.get(filePath);
- }
-
- pendingFiles.add(new Tuple3<String, Long, Long>(filePath, offset, fileSize));
- offsetOfFiles.put(filePath, fileSize);
-
- LOG.info("File added to queue: {}, {}, {}", filePath, offset, fileSize);
- }
- }
- if (files.size() > 0) {
- break;
- }
- Thread.sleep(interval);
- }
- }
-
- return pendingFiles.poll();
+ public void cancel() {
+ isRunning = false;
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6685b0ba/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FileSourceFunction.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FileSourceFunction.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FileSourceFunction.java
index c6f4421..19ee7e1 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FileSourceFunction.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FileSourceFunction.java
@@ -24,6 +24,7 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.io.InputSplit;
import org.apache.flink.runtime.jobgraph.tasks.InputSplitProvider;
import org.apache.flink.streaming.runtime.tasks.StreamingRuntimeContext;
+import org.apache.flink.util.Collector;
import java.util.Iterator;
import java.util.NoSuchElementException;
@@ -34,11 +35,12 @@ public class FileSourceFunction<OUT> extends RichParallelSourceFunction<OUT> {
private TypeInformation<OUT> typeInfo;
private transient TypeSerializer<OUT> serializer;
- private InputSplitProvider provider;
private InputFormat<OUT, InputSplit> format;
- private Iterator<InputSplit> splitIterator;
- private transient OUT nextElement;
+ private transient InputSplitProvider provider;
+ private transient Iterator<InputSplit> splitIterator;
+
+ private volatile boolean isRunning;
@SuppressWarnings("unchecked")
public FileSourceFunction(InputFormat<OUT, ?> format, TypeInformation<OUT> typeInfo) {
@@ -58,7 +60,7 @@ public class FileSourceFunction<OUT> extends RichParallelSourceFunction<OUT> {
if (splitIterator.hasNext()) {
format.open(splitIterator.next());
}
-
+ isRunning = true;
}
@Override
@@ -115,30 +117,25 @@ public class FileSourceFunction<OUT> extends RichParallelSourceFunction<OUT> {
}
@Override
- public boolean reachedEnd() throws Exception {
- if (nextElement != null) {
- return false;
- }
- nextElement = serializer.createInstance();
- nextElement = format.nextRecord(nextElement);
- if (nextElement == null && splitIterator.hasNext()) {
- format.open(splitIterator.next());
- return reachedEnd();
- } else if (nextElement == null) {
- return true;
+ public void run(Object checkpointLock, Collector<OUT> out) throws Exception {
+ isRunning = true;
+
+ while (isRunning) {
+ OUT nextElement = serializer.createInstance();
+ nextElement = format.nextRecord(nextElement);
+ if (nextElement == null && splitIterator.hasNext()) {
+ format.open(splitIterator.next());
+ continue;
+ } else if (nextElement == null) {
+ break;
+ }
+ out.collect(nextElement);
}
- return false;
}
@Override
- public OUT next() throws Exception {
- if (reachedEnd()) {
- throw new RuntimeException("End of FileSource reached.");
- }
-
- OUT result = nextElement;
- nextElement = null;
- return result;
+ public void cancel() {
+ isRunning = false;
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6685b0ba/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FromElementsFunction.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FromElementsFunction.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FromElementsFunction.java
index 6654361..0b323bc 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FromElementsFunction.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FromElementsFunction.java
@@ -21,15 +21,15 @@ import java.util.Arrays;
import java.util.Collection;
import java.util.Iterator;
-import org.apache.flink.configuration.Configuration;
+import org.apache.flink.util.Collector;
-public class FromElementsFunction<T> extends RichSourceFunction<T> {
+public class FromElementsFunction<T> implements SourceFunction<T> {
private static final long serialVersionUID = 1L;
- private transient Iterator<T> iterator;
-
private Iterable<T> iterable;
+ private volatile boolean isRunning;
+
public FromElementsFunction(T... elements) {
this.iterable = Arrays.asList(elements);
}
@@ -43,19 +43,17 @@ public class FromElementsFunction<T> extends RichSourceFunction<T> {
}
@Override
- public void open(Configuration parameters) throws Exception {
- super.open(parameters);
- iterator = iterable.iterator();
- }
+ public void run(Object checkpointLock, Collector<T> out) throws Exception {
+ isRunning = true;
+ Iterator<T> it = iterable.iterator();
- @Override
- public boolean reachedEnd() throws Exception {
- return !iterator.hasNext();
+ while (isRunning && it.hasNext()) {
+ out.collect(it.next());
+ }
}
@Override
- public T next() throws Exception {
- return iterator.next();
+ public void cancel() {
+ isRunning = false;
}
-
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6685b0ba/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FromIteratorFunction.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FromIteratorFunction.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FromIteratorFunction.java
index 125b88b..4ee1334 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FromIteratorFunction.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FromIteratorFunction.java
@@ -17,6 +17,8 @@
package org.apache.flink.streaming.api.functions.source;
+import org.apache.flink.util.Collector;
+
import java.util.Iterator;
public class FromIteratorFunction<T> implements SourceFunction<T> {
@@ -25,17 +27,22 @@ public class FromIteratorFunction<T> implements SourceFunction<T> {
Iterator<T> iterator;
+ private volatile boolean isRunning;
+
public FromIteratorFunction(Iterator<T> iterator) {
this.iterator = iterator;
}
@Override
- public boolean reachedEnd() throws Exception {
- return !iterator.hasNext();
+ public void run(Object checkpointLock, Collector<T> out) throws Exception {
+ isRunning = true;
+ while (isRunning && iterator.hasNext()) {
+ out.collect(iterator.next());
+ }
}
@Override
- public T next() throws Exception {
- return iterator.next();
+ public void cancel() {
+ isRunning = false;
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6685b0ba/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FromSplittableIteratorFunction.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FromSplittableIteratorFunction.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FromSplittableIteratorFunction.java
index fd86858..97f5c06 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FromSplittableIteratorFunction.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/FromSplittableIteratorFunction.java
@@ -18,6 +18,7 @@
package org.apache.flink.streaming.api.functions.source;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.util.Collector;
import org.apache.flink.util.SplittableIterator;
import java.util.Iterator;
@@ -26,8 +27,11 @@ public class FromSplittableIteratorFunction<T> extends RichParallelSourceFunctio
private static final long serialVersionUID = 1L;
- SplittableIterator<T> fullIterator;
- Iterator<T> iterator;
+ private SplittableIterator<T> fullIterator;
+
+ private transient Iterator<T> iterator;
+
+ private volatile boolean isRunning;
public FromSplittableIteratorFunction(SplittableIterator<T> iterator) {
this.fullIterator = iterator;
@@ -38,15 +42,20 @@ public class FromSplittableIteratorFunction<T> extends RichParallelSourceFunctio
int numberOfSubTasks = getRuntimeContext().getNumberOfParallelSubtasks();
int indexofThisSubTask = getRuntimeContext().getIndexOfThisSubtask();
iterator = fullIterator.split(numberOfSubTasks)[indexofThisSubTask];
+ isRunning = true;
}
@Override
- public boolean reachedEnd() throws Exception {
- return !iterator.hasNext();
+ public void run(Object checkpointLock, Collector<T> out) throws Exception {
+ isRunning = true;
+
+ while (isRunning && iterator.hasNext()) {
+ out.collect(iterator.next());
+ }
}
@Override
- public T next() throws Exception {
- return iterator.next();
+ public void cancel() {
+ isRunning = false;
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6685b0ba/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/SocketTextStreamFunction.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/SocketTextStreamFunction.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/SocketTextStreamFunction.java
index 6a1f930..8e5c3b3 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/SocketTextStreamFunction.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/SocketTextStreamFunction.java
@@ -26,6 +26,7 @@ import java.net.Socket;
import java.net.SocketException;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -44,11 +45,7 @@ public class SocketTextStreamFunction extends RichSourceFunction<String> {
private static final int CONNECTION_TIMEOUT_TIME = 0;
private static final int CONNECTION_RETRY_SLEEP = 1000;
- private transient StringBuffer buffer;
- private transient BufferedReader reader;
-
- private boolean socketClosed;
- private transient String nextElement;
+ private volatile boolean isRunning = false;
public SocketTextStreamFunction(String hostname, int port, char delimiter, long maxRetry) {
this.hostname = hostname;
@@ -63,102 +60,89 @@ public class SocketTextStreamFunction extends RichSourceFunction<String> {
super.open(parameters);
socket = new Socket();
socket.connect(new InetSocketAddress(hostname, port), CONNECTION_TIMEOUT_TIME);
- buffer = new StringBuffer();
- reader = new BufferedReader(new InputStreamReader(
- socket.getInputStream()));
- socketClosed = false;
}
@Override
- public void close() throws Exception {
- super.close();
- if (reader != null) {
- reader.close();
- }
- if (socket != null && !socket.isClosed()) {
- try {
- socket.close();
- } catch (IOException e) {
- if (LOG.isErrorEnabled()) {
- LOG.error("Could not close open socket");
- }
- }
- }
-
+ public void run(Object checkpointLock, Collector<String> collector) throws Exception {
+ streamFromSocket(collector, socket);
}
- public String blockingRead(Socket socket) throws Exception {
-
- while (true) {
- int data;
- try {
- data = reader.read();
- } catch (SocketException e) {
- socketClosed = true;
- break;
- }
+ public void streamFromSocket(Collector<String> collector, Socket socket) throws Exception {
+ isRunning = true;
+ try {
+ StringBuffer buffer = new StringBuffer();
+ BufferedReader reader = new BufferedReader(new InputStreamReader(
+ socket.getInputStream()));
+
+ while (isRunning) {
+ int data;
+ try {
+ data = reader.read();
+ } catch (SocketException e) {
+ if (!isRunning) {
+ break;
+ } else {
+ throw e;
+ }
+ }
- if (data == -1) {
- socket.close();
- long retry = 0;
- boolean success = false;
- while (retry < maxRetry && !success) {
- if (!retryForever) {
- retry++;
+ if (data == -1) {
+ socket.close();
+ long retry = 0;
+ boolean success = false;
+ while (retry < maxRetry && !success) {
+ if (!retryForever) {
+ retry++;
+ }
+ LOG.warn("Lost connection to server socket. Retrying in "
+ + (CONNECTION_RETRY_SLEEP / 1000) + " seconds...");
+ try {
+ socket = new Socket();
+ socket.connect(new InetSocketAddress(hostname, port),
+ CONNECTION_TIMEOUT_TIME);
+ success = true;
+ } catch (ConnectException ce) {
+ Thread.sleep(CONNECTION_RETRY_SLEEP);
+ }
}
- LOG.warn("Lost connection to server socket. Retrying in "
- + (CONNECTION_RETRY_SLEEP / 1000) + " seconds...");
- try {
- socket = new Socket();
- socket.connect(new InetSocketAddress(hostname, port),
- CONNECTION_TIMEOUT_TIME);
- success = true;
- } catch (ConnectException ce) {
- Thread.sleep(CONNECTION_RETRY_SLEEP);
+
+ if (success) {
+ LOG.info("Server socket is reconnected.");
+ } else {
+ LOG.error("Could not reconnect to server socket.");
+ break;
}
+ reader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
+ continue;
}
- if (success) {
- LOG.info("Server socket is reconnected.");
- } else {
- LOG.error("Could not reconnect to server socket.");
- break;
+ if (data == delimiter) {
+ collector.collect(buffer.toString());
+ buffer = new StringBuffer();
+ } else if (data != '\r') { // ignore carriage return
+ buffer.append((char) data);
}
- reader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
- continue;
}
- if (data == delimiter) {
- String result = buffer.toString();
- buffer = new StringBuffer();
- return result;
- } else if (data != '\r') { // ignore carriage return
- buffer.append((char) data);
+ if (buffer.length() > 0) {
+ collector.collect(buffer.toString());
}
+ } finally {
+ socket.close();
}
-
- return null;
}
-
@Override
- public boolean reachedEnd() throws Exception {
- if (socketClosed) {
- return false;
- }
-
- nextElement = blockingRead(socket);
-
- return nextElement == null;
- }
-
- @Override
- public String next() throws Exception {
- if (nextElement == null) {
- reachedEnd();
+ public void cancel() {
+ isRunning = false;
+ if (socket != null && !socket.isClosed()) {
+ try {
+ socket.close();
+ } catch (IOException e) {
+ if (LOG.isErrorEnabled()) {
+ LOG.error("Could not close open socket");
+ }
+ }
}
-
- return nextElement;
}
-
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6685b0ba/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/SourceFunction.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/SourceFunction.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/SourceFunction.java
index e6ffdc1..8c349e9 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/SourceFunction.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/SourceFunction.java
@@ -19,66 +19,81 @@
package org.apache.flink.streaming.api.functions.source;
import org.apache.flink.api.common.functions.Function;
+import org.apache.flink.util.Collector;
import java.io.Serializable;
/**
* Base interface for all stream data sources in Flink. The contract of a stream source
- * is similar to an iterator - it is consumed as in the following pseudo code:
+ * is the following: When the source should start emitting elements the {@link #run} method
+ * is called with a {@link org.apache.flink.util.Collector} that can be used for emitting elements.
+ * The run method can run for as long as necessary. The source must, however, react to an
+ * invocation of {@link #cancel} by breaking out of its main loop.
*
- * <pre>{@code
- * StreamSource<T> source = ...;
- * Collector<T> out = ...;
- * while (!source.reachedEnd()) {
- * out.collect(source.next());
- * }
+ * <b>Note about checkpointed sources</b>
+ * <p>
+ * Sources that also implement the {@link org.apache.flink.streaming.api.checkpoint.Checkpointed}
+ * interface must ensure that state checkpointing, updating of internal state and emission of
+ * elements are not done concurrently. This is achieved by using the provided checkpointing lock
+ * object to protect update of state and emission of elements in a synchronized block.
+ * </p>
+ *
+ * <p>
+ * This is the basic pattern one should follow when implementing a (checkpointed) source:
+ * </p>
+ *
+ * <pre>
+ * {@code
+ * public class ExampleSource<T> implements SourceFunction<T>, Checkpointed<Long> {
+ * private long count = 0L;
+ * private volatile boolean isRunning;
+ *
+ * @Override
+ * public void run(Object checkpointLock, Collector<T> out) {
+ * isRunning = true;
+ * while (isRunning && count < 1000) {
+ * synchronized (checkpointLock) {
+ * out.collect(count);
+ * count++;
+ * }
+ * }
+ * }
+ *
+ * @Override
+ * public void cancel() {
+ * isRunning = false;
+ * }
+ *
+ * @Override
+ * public Long snapshotState(long checkpointId, long checkpointTimestamp) { return count; }
+ *
+ * @Override
+ * public void restoreState(Long state) { this.count = state; }
* }
* </pre>
- *
- * <b>Note about blocking behavior</b>
- * <p>This implementations of the methods in the stream sources must have certain guarantees about
- * blocking behavior. One of the two characteristics must be fulfilled.</p>
- * <ul>
- * <li>The methods must react to thread interrupt calls and break out of blocking calls with
- * an {@link InterruptedException}.</li>
- * <li>The method may ignore interrupt calls and/or swallow InterruptedExceptions, if it is guaranteed
- * that the method returns quasi immediately irrespectively of the input. This is true for example
- * for file streams, where the call is guaranteed to return after a very short I/O delay in
- * the order of milliseconds.</li>
- * </ul>
- *
- * @param <T> The type of the records produced by this source.
+ *
+ * @param <T> The type of the elements produced by this source.
*/
public interface SourceFunction<T> extends Function, Serializable {
-
+
/**
- * Checks whether the stream has reached its end.
+ * Starts the source. You can use the {@link org.apache.flink.util.Collector} parameter to emit
+ * elements. 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/or emitting
+ * elements. Also, the update of state and emission of elements must happen in the same
+ * synchronized block.
*
- * <p>This method must obey the contract about blocking behavior declared in the
- * description of this class.</p>
- *
- * @return True, if the end of the stream has been reached, false if more data is available.
- *
- * @throws InterruptedException The calling thread may be interrupted to pull the function out of this
- * method during checkpoints.
- * @throws Exception Any other exception that is thrown causes the source to fail and results in failure of
- * the streaming program, or triggers recovery, depending on the program setup.
+ * @param checkpointLock The object to synchronize on when updating state and emitting elements.
+ * @param out The collector to use for emitting elements
*/
- boolean reachedEnd() throws Exception;
-
+ void run(final Object checkpointLock, Collector<T> out) throws Exception;
/**
- * Produces the next record.
- *
- * <p>This method must obey the contract about blocking behavior declared in the
- * description of this class.</p>
- *
- * @return The next record produced by this stream source.
- *
- * @throws InterruptedException The calling thread may be interrupted to pull the function out of this
- * method during checkpoints.
- * @throws Exception Any other exception that is thrown causes the source to fail and results in failure of
- * the streaming program, or triggers recovery, depending on the program setup.
+ * Cancels the source. Most sources will have a while loop inside the
+ * {@link #run} method. You need to ensure that the source will break out of this loop. This
+ * can be achieved by having a volatile field "isRunning" that is checked in the loop and that
+ * is set to false in this method.
*/
- T next() throws Exception;
+ void cancel();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6685b0ba/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java
index b55e5d6..7bb7780 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java
@@ -41,7 +41,7 @@ public abstract class AbstractStreamOperator<OUT> implements StreamOperator<OUT>
protected ChainingStrategy chainingStrategy = ChainingStrategy.HEAD;
@Override
- public final void setup(Output<OUT> output, RuntimeContext runtimeContext) {
+ public void setup(Output<OUT> output, RuntimeContext runtimeContext) {
this.output = output;
this.executionConfig = runtimeContext.getExecutionConfig();
this.runtimeContext = runtimeContext;
http://git-wip-us.apache.org/repos/asf/flink/blob/6685b0ba/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/AbstractUdfStreamOperator.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/AbstractUdfStreamOperator.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/AbstractUdfStreamOperator.java
index 09d1ef6..852bfde 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/AbstractUdfStreamOperator.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/AbstractUdfStreamOperator.java
@@ -19,6 +19,7 @@
package org.apache.flink.streaming.api.operators;
import org.apache.flink.api.common.functions.Function;
+import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.functions.util.FunctionUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.checkpoint.CheckpointCommitter;
@@ -43,9 +44,15 @@ public abstract class AbstractUdfStreamOperator<OUT, F extends Function & Serial
}
@Override
+ public final void setup(Output<OUT> output, RuntimeContext runtimeContext) {
+ super.setup(output, runtimeContext);
+ FunctionUtils.setFunctionRuntimeContext(userFunction, runtimeContext);
+ }
+
+
+ @Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
- FunctionUtils.setFunctionRuntimeContext(userFunction, runtimeContext);
FunctionUtils.openFunction(userFunction, parameters);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6685b0ba/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamSource.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamSource.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamSource.java
index 9cdfb01..e63349a 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamSource.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/operators/StreamSource.java
@@ -18,7 +18,11 @@
package org.apache.flink.streaming.api.operators;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.util.Collector;
+/**
+ * {@link StreamOperator} for streaming sources.
+ */
public class StreamSource<OUT> extends AbstractUdfStreamOperator<OUT, SourceFunction<OUT>> implements StreamOperator<OUT> {
private static final long serialVersionUID = 1L;
@@ -29,19 +33,11 @@ public class StreamSource<OUT> extends AbstractUdfStreamOperator<OUT, SourceFunc
this.chainingStrategy = ChainingStrategy.HEAD;
}
- public void run() throws Exception {
- while (true) {
-
- synchronized (userFunction) {
- if (userFunction.reachedEnd()) {
- break;
- }
-
- OUT result = userFunction.next();
+ public void run(Object lockingObject, Collector<OUT> collector) throws Exception {
+ userFunction.run(lockingObject, collector);
+ }
- output.collect(result);
- }
- Thread.yield();
- }
+ public void cancel() {
+ userFunction.cancel();
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6685b0ba/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTask.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTask.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTask.java
index 95cbc87..186fb3f 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTask.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTask.java
@@ -22,6 +22,18 @@ import org.apache.flink.streaming.api.operators.StreamSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+/**
+ * Task for executing streaming sources.
+ *
+ * One important aspect of this is that the checkpointing and the emission of elements must never
+ * occur at the same time. The execution must be serial. This is achieved by having the contract
+ * with the StreamFunction that it must only modify its state or emit elements in
+ * a synchronized block that locks on the checkpointLock Object. Also, the modification of the state
+ * and the emission of elements must happen in the same block of code that is protected by the
+ * synchronized block.
+ *
+ * @param <OUT> Type of the output elements of this source.
+ */
public class SourceStreamTask<OUT> extends StreamTask<OUT, StreamSource<OUT>> {
private static final Logger LOG = LoggerFactory.getLogger(SourceStreamTask.class);
@@ -40,7 +52,7 @@ public class SourceStreamTask<OUT> extends StreamTask<OUT, StreamSource<OUT>> {
openOperator();
operatorOpen = true;
- streamOperator.run();
+ streamOperator.run(checkpointLock, outputHandler.getOutput());
closeOperator();
operatorOpen = false;
@@ -71,4 +83,10 @@ public class SourceStreamTask<OUT> extends StreamTask<OUT, StreamSource<OUT>> {
}
}
+
+ @Override
+ public void cancel() {
+ streamOperator.cancel();
+ super.cancel();
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6685b0ba/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
index d458dac..bb641d9 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
@@ -50,7 +50,7 @@ public abstract class StreamTask<OUT, O extends StreamOperator<OUT>> extends Abs
private static final Logger LOG = LoggerFactory.getLogger(StreamTask.class);
- private final Object checkpointLock = new Object();
+ protected final Object checkpointLock = new Object();
protected StreamConfig configuration;
@@ -191,9 +191,6 @@ public abstract class StreamTask<OUT, O extends StreamOperator<OUT>> extends Abs
// Checkpoint and Restore
// ------------------------------------------------------------------------
- /**
- * Re-injects the user states into the map. Also set the state on the functions.
- */
@Override
public void setInitialState(StateHandle<Serializable> stateHandle) throws Exception {
// here, we later resolve the state handle into the actual state by
@@ -230,13 +227,9 @@ public abstract class StreamTask<OUT, O extends StreamOperator<OUT>> extends Abs
}
}
- /**
- * This method is either called directly by the checkpoint coordinator, or called
- * when all incoming channels have reported a barrier
- */
@Override
public void triggerCheckpoint(long checkpointId, long timestamp) throws Exception {
-
+
synchronized (checkpointLock) {
if (isRunning) {
try {
@@ -293,6 +286,7 @@ public abstract class StreamTask<OUT, O extends StreamOperator<OUT>> extends Abs
}
}
}
+
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/6685b0ba/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/ChainedRuntimeContextTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/ChainedRuntimeContextTest.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/ChainedRuntimeContextTest.java
index 084973f..fef8a31 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/ChainedRuntimeContextTest.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/ChainedRuntimeContextTest.java
@@ -25,6 +25,7 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
import org.apache.flink.streaming.util.TestStreamEnvironment;
+import org.apache.flink.util.Collector;
import org.junit.Test;
@SuppressWarnings("serial")
@@ -46,13 +47,11 @@ public class ChainedRuntimeContextTest {
private static class TestSource extends RichParallelSourceFunction<Integer> {
@Override
- public boolean reachedEnd() throws Exception {
- return true;
+ public void run(Object checkpointLock, Collector<Integer> out) throws Exception {
}
@Override
- public Integer next() throws Exception {
- return null;
+ public void cancel() {
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/6685b0ba/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/TypeFillTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/TypeFillTest.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/TypeFillTest.java
index ca6057c..1dbbc00 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/TypeFillTest.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/TypeFillTest.java
@@ -115,18 +115,17 @@ public class TypeFillTest {
}
private class TestSource<T> implements SourceFunction<T> {
-
+ private static final long serialVersionUID = 1L;
@Override
- public boolean reachedEnd() throws Exception {
- return false;
+ public void run(Object checkpointLock, Collector<T> out) throws Exception {
+
}
@Override
- public T next() throws Exception {
- return null;
- }
+ public void cancel() {
+ }
}
private class TestMap<T, O> implements MapFunction<T, O> {