[FLINK-1625] [streaming] [api-breaking] Added proper cancellation to …

…StreamInvokables + Sink- and SourceFunction interfaces extended with cancel method

flink-staging/flink-streaming/flink-streaming-connectors/src/main/java/org/apache/flink/streaming/connectors/flume/FlumeSink.java

@@ -133,6 +133,13 @@ public void close() {
 		client.client.close();
 	}
 
+	@Override
+	public void cancel() {
+		if (client != null) {
+			client.client.close();
+		}
+	}
+
 	@Override
 	public void open(Configuration config) {
 		client = new FlinkRpcClientFacade();

flink-staging/flink-streaming/flink-streaming-connectors/src/main/java/org/apache/flink/streaming/connectors/flume/FlumeSource.java

@@ -130,12 +130,16 @@ public void configureAvroSource(Collector<OUT> collector) {
 	 *            The Collector for sending data to the datastream
 	 */
 	@Override
-	public void invoke(Collector<OUT> collector) throws Exception {
+	public void run(Collector<OUT> collector) throws Exception {
 		configureAvroSource(collector);
 		avroSource.start();
 		while (!finished) {
 			this.wait();
 		}
 	}
 
+	@Override
+	public void cancel() {
+	}
+
 }

flink-staging/flink-streaming/flink-streaming-connectors/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerExample.java

@@ -41,14 +41,20 @@ public static void main(String[] args) throws Exception {
 		@SuppressWarnings({ "unused", "serial" })
 		DataStream<String> stream1 = env.addSource(new SourceFunction<String>() {
 			@Override
-			public void invoke(Collector<String> collector) throws Exception {
+			public void run(Collector<String> collector) throws Exception {
 				for (int i = 0; i < 100; i++) {
 					collector.collect("message #" + i);
 					Thread.sleep(100L);
 				}
 
 				collector.collect(new String("q"));
 			}
+
+			@Override
+			public void cancel() {				
+			}
+
+
 		}).addSink(
 				new KafkaSink<String>(topic, host + ":" + port, new JavaDefaultStringSchema())
 		)

flink-staging/flink-streaming/flink-streaming-connectors/src/main/java/org/apache/flink/streaming/connectors/kafka/api/KafkaSink.java

@@ -33,9 +33,9 @@
 
 /**
  * Sink that emits its inputs to a Kafka topic.
- *
+ * 
  * @param <IN>
- * 		Type of the sink input
+ *            Type of the sink input
  */
 public class KafkaSink<IN> extends RichSinkFunction<IN> {
 	private static final long serialVersionUID = 1L;
@@ -49,31 +49,33 @@ public class KafkaSink<IN> extends RichSinkFunction<IN> {
 	private KafkaPartitioner<IN> partitioner;
 
 	/**
-	 * Creates a KafkaSink for a given topic. The partitioner distributes the messages between the partitions of the topics.
-	 *
+	 * Creates a KafkaSink for a given topic. The partitioner distributes the
+	 * messages between the partitions of the topics.
+	 * 
 	 * @param topicId
-	 * 		ID of the Kafka topic.
+	 *            ID of the Kafka topic.
 	 * @param brokerAddr
-	 * 		Address of the Kafka broker (with port number).
+	 *            Address of the Kafka broker (with port number).
 	 * @param serializationSchema
-	 * 		User defined serialization schema.
+	 *            User defined serialization schema.
 	 */
 	public KafkaSink(String topicId, String brokerAddr,
 			SerializationSchema<IN, byte[]> serializationSchema) {
 		this(topicId, brokerAddr, serializationSchema, new KafkaDistributePartitioner<IN>());
 	}
 
 	/**
-	 * Creates a KafkaSink for a given topic. The sink produces its input into the topic.
-	 *
+	 * Creates a KafkaSink for a given topic. The sink produces its input into
+	 * the topic.
+	 * 
 	 * @param topicId
-	 * 		ID of the Kafka topic.
+	 *            ID of the Kafka topic.
 	 * @param brokerAddr
-	 * 		Address of the Kafka broker (with port number).
+	 *            Address of the Kafka broker (with port number).
 	 * @param serializationSchema
-	 * 		User defined serialization schema.
+	 *            User defined serialization schema.
 	 * @param partitioner
-	 * 		User defined partitioner.
+	 *            User defined partitioner.
 	 */
 	public KafkaSink(String topicId, String brokerAddr,
 			SerializationSchema<IN, byte[]> serializationSchema, KafkaPartitioner<IN> partitioner) {
@@ -111,9 +113,9 @@ public void initialize() {
 
 	/**
 	 * Called when new data arrives to the sink, and forwards it to Kafka.
-	 *
+	 * 
 	 * @param next
-	 * 		The incoming data
+	 *            The incoming data
 	 */
 	@Override
 	public void invoke(IN next) {
@@ -132,4 +134,9 @@ public void close() {
 		}
 	}
 
+	@Override
+	public void cancel() {
+		close();
+	}
+
 }

flink-staging/flink-streaming/flink-streaming-connectors/src/main/java/org/apache/flink/streaming/connectors/kafka/api/KafkaSource.java

@@ -53,6 +53,8 @@ public class KafkaSource<OUT> extends ConnectorSource<OUT> {
 	private long zookeeperSyncTimeMillis;
 	private static final long ZOOKEEPER_DEFAULT_SYNC_TIME = 200;
 
+	private volatile boolean isRunning = false;
+
 	/**
 	 * Creates a KafkaSource that consumes a topic.
 	 * 
@@ -107,21 +109,31 @@ private void initializeConnection() {
 	 *            The Collector for sending data to the dataStream
 	 */
 	@Override
-	public void invoke(Collector<OUT> collector) throws Exception {
-
-		while (consumerIterator.hasNext()) {
-			OUT out = schema.deserialize(consumerIterator.next().message());
-			if (schema.isEndOfStream(out)) {
-				break;
+	public void run(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);
 			}
-			collector.collect(out);
+		} finally {
+			consumer.shutdown();
 		}
-		consumer.shutdown();
-
 	}
 
 	@Override
-	public void open(Configuration config) {
+	public void open(Configuration config) throws Exception {
 		initializeConnection();
 	}
+
+	@Override
+	public void cancel() {
+		isRunning = false;
+		if (consumer != null) {
+			consumer.shutdown();
+		}
+	}
 }

flink-staging/flink-streaming/flink-streaming-connectors/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSink.java

@@ -108,4 +108,9 @@ public void close() {
 		closeChannel();
 	}
 
+	@Override
+	public void cancel() {
+		close();
+	}
+
 }

flink-staging/flink-streaming/flink-streaming-connectors/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSource.java

@@ -46,6 +46,8 @@ public class RMQSource<OUT> extends ConnectorSource<OUT> {
 	private transient QueueingConsumer consumer;
 	private transient QueueingConsumer.Delivery delivery;
 
+	private volatile boolean isRunning = false;
+
 	OUT out;
 
 	public RMQSource(String HOST_NAME, String QUEUE_NAME,
@@ -80,42 +82,46 @@ private void initializeConnection() {
 	 *            The Collector for sending data to the dataStream
 	 */
 	@Override
-	public void invoke(Collector<OUT> collector) throws Exception {
-
-		while (true) {
-
-			try {
-				delivery = consumer.nextDelivery();
-			} catch (Exception e) {
-				if (LOG.isErrorEnabled()) {
-					LOG.error("Cannot recieve RMQ message {} at {}", QUEUE_NAME, HOST_NAME);
+	public void run(Collector<OUT> collector) throws Exception {
+		isRunning = true;
+		try {
+			while (isRunning) {
+
+				try {
+					delivery = consumer.nextDelivery();
+				} catch (Exception e) {
+					if (LOG.isErrorEnabled()) {
+						LOG.error("Cannot recieve RMQ message {} at {}", QUEUE_NAME, HOST_NAME);
+					}
 				}
-			}
 
-			out = schema.deserialize(delivery.getBody());
-			if (schema.isEndOfStream(out)) {
-				break;
-			} else {
-				collector.collect(out);
+				out = schema.deserialize(delivery.getBody());
+				if (schema.isEndOfStream(out)) {
+					break;
+				} else {
+					collector.collect(out);
+				}
 			}
+		} finally {
+			connection.close();
 		}
 
 	}
 
 	@Override
-	public void open(Configuration config) {
+	public void open(Configuration config) throws Exception {
 		initializeConnection();
 	}
 
 	@Override
-	public void close() {
+	public void cancel() {
+		isRunning = false;
 		try {
 			connection.close();
 		} catch (IOException e) {
 			throw new RuntimeException("Error while closing RMQ connection with " + QUEUE_NAME
 					+ " at " + HOST_NAME, e);
 		}
-
 	}
 
 }

flink-staging/flink-streaming/flink-streaming-connectors/src/main/java/org/apache/flink/streaming/connectors/twitter/TwitterSource.java

@@ -57,6 +57,8 @@ public class TwitterSource extends RichParallelSourceFunction<String> {
 	private boolean streaming;
 	private int numberOfTweets;
 
+	private volatile boolean isRunning = false;
+
 	/**
 	 * Create {@link TwitterSource} for streaming
 	 * 
@@ -90,20 +92,20 @@ public void open(Configuration parameters) throws Exception {
 	}
 
 	@Override
-	public void invoke(Collector<String> collector) throws Exception {
-
-		if (streaming) {
-			collectMessages(collector);
-		} else {
-			collectFiniteMessages(collector);
+	public void run(Collector<String> collector) throws Exception {
+		isRunning = true;
+		try {
+			if (streaming) {
+				collectMessages(collector);
+			} else {
+				collectFiniteMessages(collector);
+			}
+		} finally {
+			closeConnection();
+			isRunning = false;
 		}
 	}
 
-	@Override
-	public void close() throws Exception {
-		closeConnection();
-	}
-
 	/**
 	 * Initialize Hosebird Client to be able to consume Twitter's Streaming API
 	 */
@@ -196,7 +198,7 @@ protected void collectMessages(Collector<String> collector) {
 			LOG.info("Tweet-stream begins");
 		}
 
-		while (true) {
+		while (isRunning) {
 			collectOneMessage(collector);
 		}
 	}
@@ -246,7 +248,8 @@ private void closeConnection() {
 	/**
 	 * Get the size of the queue in which the tweets are contained temporarily.
 	 * 
-	 * @return the size of the queue in which the tweets are contained temporarily
+	 * @return the size of the queue in which the tweets are contained
+	 *         temporarily
 	 */
 	public int getQueueSize() {
 		return queueSize;
@@ -280,4 +283,10 @@ public int getWaitSec() {
 	public void setWaitSec(int waitSec) {
 		this.waitSec = waitSec;
 	}
+
+	@Override
+	public void cancel() {
+		isRunning = false;
+		closeConnection();
+	}
 }

flink-staging/flink-streaming/flink-streaming-connectors/src/main/java/org/apache/flink/streaming/connectors/twitter/TwitterStreaming.java

@@ -47,6 +47,10 @@ public void invoke(Tuple5<Long, Integer, String, String, String> tuple) {
 			System.out.println("");
 		}
 
+		@Override
+		public void cancel() {
+		}
+
 	}
 
 	public static class SelectDataFlatMap extends

flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/function/sink/FileSinkFunction.java

@@ -115,4 +115,13 @@ protected void flush() {
 	 */
 	protected abstract void resetParameters();
 
+	@Override
+	public void cancel() {
+		try {
+			close();
+		} catch (IOException e) {
+			throw new RuntimeException(e);
+		}
+	}
+
 }

flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/function/sink/PrintSinkFunction.java

@@ -84,15 +84,19 @@ public void invoke(IN record) {
 	}
 
 	@Override
-	public void close() throws Exception {
+	public void close() {
 		this.stream = null;
 		this.prefix = null;
-		super.close();
 	}
 
 	@Override
 	public String toString() {
 		return "Print to " + (target == STD_OUT ? "System.out" : "System.err");
 	}
 
+	@Override
+	public void cancel() {
+		close();
+	}
+
 }

flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/function/sink/SinkFunction.java

@@ -23,6 +23,8 @@
 
 public interface SinkFunction<IN> extends Function, Serializable {
 
-	public abstract void invoke(IN value) throws Exception;
+	public void invoke(IN value) throws Exception;
+
+	public void cancel();
 
 }