From 760bf31b310c3dcdeddcc5fe89de23ea965c0f7c Mon Sep 17 00:00:00 2001
From: zentol <s.motsu@web.de>
Date: Wed, 10 Feb 2016 14:14:18 +0100
Subject: [PATCH 01/10] [FLINK-3332] Add Exactly-Once Cassandra connector

---
 .../api/writer/ResultPartitionWriter.java     |   2 +-
 .../flink-connector-cassandra/pom.xml         | 138 ++++++++++++
 .../cassandra/CassandraExactlyOnceSink.java   | 100 +++++++++
 .../CassandraExactlyOnceSinkExample.java      |  80 +++++++
 .../CassandraExactlyOnceSinkTest.java         | 192 +++++++++++++++++
 .../src/test/resources/cassandra.yaml         |  40 ++++
 .../src/test/resources/log4j-test.properties  |  29 +++
 flink-streaming-connectors/pom.xml            |   1 +
 .../operators/GenericExactlyOnceSink.java     | 173 +++++++++++++++
 .../operators/ExactlyOnceSinkTestBase.java    | 203 ++++++++++++++++++
 .../operators/GenericExactlyOnceSinkTest.java | 107 +++++++++
 11 files changed, 1064 insertions(+), 1 deletion(-)
 create mode 100644 flink-streaming-connectors/flink-connector-cassandra/pom.xml
 create mode 100644 flink-streaming-connectors/flink-connector-cassandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/CassandraExactlyOnceSink.java
 create mode 100644 flink-streaming-connectors/flink-connector-cassandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/example/CassandraExactlyOnceSinkExample.java
 create mode 100644 flink-streaming-connectors/flink-connector-cassandra/src/test/java/org/apache/flink/streaming/connectors/cassandra/CassandraExactlyOnceSinkTest.java
 create mode 100644 flink-streaming-connectors/flink-connector-cassandra/src/test/resources/cassandra.yaml
 create mode 100644 flink-streaming-connectors/flink-connector-cassandra/src/test/resources/log4j-test.properties
 create mode 100644 flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericExactlyOnceSink.java
 create mode 100644 flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/ExactlyOnceSinkTestBase.java
 create mode 100644 flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/GenericExactlyOnceSinkTest.java

diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/ResultPartitionWriter.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/ResultPartitionWriter.java
index 79c21c6a5a601..cfab34d829ef4 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/ResultPartitionWriter.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/ResultPartitionWriter.java
@@ -37,7 +37,7 @@
  * The {@link ResultPartitionWriter} is the runtime API for producing results. It
  * supports two kinds of data to be sent: buffers and events.
  */
-public final class ResultPartitionWriter implements EventListener<TaskEvent> {
+public class ResultPartitionWriter implements EventListener<TaskEvent> {
 
 	private final ResultPartition partition;
 
diff --git a/flink-streaming-connectors/flink-connector-cassandra/pom.xml b/flink-streaming-connectors/flink-connector-cassandra/pom.xml
new file mode 100644
index 0000000000000..6d9eab026db3c
--- /dev/null
+++ b/flink-streaming-connectors/flink-connector-cassandra/pom.xml
@@ -0,0 +1,138 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+		 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+		 xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+
+	<modelVersion>4.0.0</modelVersion>
+
+	<parent>
+		<groupId>org.apache.flink</groupId>
+		<artifactId>flink-streaming-connectors</artifactId>
+		<version>1.0-SNAPSHOT</version>
+		<relativePath>..</relativePath>
+	</parent>
+
+	<artifactId>flink-connector-cassandra_2.10</artifactId>
+	<name>flink-connector-cassandra</name>
+
+	<packaging>jar</packaging>
+
+	<build>
+		<plugins>
+			<plugin>
+				<groupId>org.apache.maven.plugins</groupId>
+				<artifactId>maven-surefire-plugin</artifactId>
+				<configuration>
+					<reuseForks>true</reuseForks>
+					<forkCount>1</forkCount>
+				</configuration>
+			</plugin>
+		</plugins>
+	</build>
+
+	<!-- Allow users to pass custom connector versions -->
+	<properties>
+		<cassandra.version>2.2.0</cassandra.version>
+		<driver.version>3.0.0</driver.version>
+	</properties>
+
+	<dependencies>
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-streaming-java_2.10</artifactId>
+			<version>${project.version}</version>
+		</dependency>
+		<dependency>
+			<groupId>com.datastax.cassandra</groupId>
+			<artifactId>cassandra-driver-core</artifactId>
+			<version>${driver.version}</version>
+			<exclusions>
+				<exclusion>
+					<groupId>org.slf4j</groupId>
+					<artifactId>log4j-over-slf4j</artifactId>
+				</exclusion>
+				<exclusion>
+					<groupId>com.google.guava</groupId>
+					<artifactId>guava</artifactId>
+				</exclusion>
+				<exclusion>
+					<groupId>ch.qos.logback</groupId>
+					<artifactId>logback-classic</artifactId>
+				</exclusion>
+			</exclusions>
+		</dependency>
+		<dependency>
+			<groupId>com.datastax.cassandra</groupId>
+			<artifactId>cassandra-driver-mapping</artifactId>
+			<version>${driver.version}</version>
+			<exclusions>
+				<exclusion>
+					<groupId>org.slf4j</groupId>
+					<artifactId>log4j-over-slf4j</artifactId>
+				</exclusion>
+				<exclusion>
+					<groupId>com.google.guava</groupId>
+					<artifactId>guava</artifactId>
+				</exclusion>
+				<exclusion>
+					<groupId>ch.qos.logback</groupId>
+					<artifactId>logback-classic</artifactId>
+				</exclusion>
+			</exclusions>
+		</dependency>
+		<dependency>
+			<groupId>com.google.guava</groupId>
+			<artifactId>guava</artifactId>
+			<version>${guava.version}</version>
+		</dependency>
+
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-runtime_2.10</artifactId>
+			<version>${project.version}</version>
+			<scope>test</scope>
+			<type>test-jar</type>
+		</dependency>
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-streaming-java_2.10</artifactId>
+			<version>${project.version}</version>
+			<scope>test</scope>
+			<type>test-jar</type>
+		</dependency>
+		<dependency>
+			<groupId>org.apache.cassandra</groupId>
+			<artifactId>cassandra-all</artifactId>
+			<version>${cassandra.version}</version>
+			<scope>test</scope>
+			<exclusions>
+				<exclusion>
+					<groupId>org.slf4j</groupId>
+					<artifactId>log4j-over-slf4j</artifactId>
+				</exclusion>
+				<exclusion>
+					<groupId>ch.qos.logback</groupId>
+					<artifactId>logback-classic</artifactId>
+				</exclusion>
+			</exclusions>
+		</dependency>
+	</dependencies>
+</project>
diff --git a/flink-streaming-connectors/flink-connector-cassandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/CassandraExactlyOnceSink.java b/flink-streaming-connectors/flink-connector-cassandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/CassandraExactlyOnceSink.java
new file mode 100644
index 0000000000000..53b9d699f7a37
--- /dev/null
+++ b/flink-streaming-connectors/flink-connector-cassandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/CassandraExactlyOnceSink.java
@@ -0,0 +1,100 @@
+/**
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.cassandra;
+
+import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.PreparedStatement;
+import com.datastax.driver.core.ResultSet;
+import com.datastax.driver.core.ResultSetFuture;
+import com.datastax.driver.core.Session;
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.streaming.runtime.operators.GenericExactlyOnceSink;
+
+/**
+ * Sink that emits its input elements into a Cassandra database. This sink is integrated with the checkpointing
+ * mechanism to provide exactly once semantics.
+ *
+ * Incoming records are stored within a {@link org.apache.flink.runtime.state.AbstractStateBackend}, and only committed if a
+ * checkpoint is completed. Should a job fail while the data is being committed, no exactly once guarantee can be made.
+ * @param <IN>
+ */
+public class CassandraExactlyOnceSink<IN extends Tuple> extends GenericExactlyOnceSink<IN> {
+	private String host;
+	private String createQuery;
+	private String insertQuery;
+
+	private transient Cluster cluster;
+	private transient Session session;
+	private transient PreparedStatement preparedStatement;
+
+	private transient Throwable exception = null;
+
+	public CassandraExactlyOnceSink(String host, String insertQuery) {
+		this(host, null, insertQuery);
+	}
+
+	public CassandraExactlyOnceSink(String host, String createQuery, String insertQuery) {
+		this.host = host;
+		this.createQuery = createQuery;
+		this.insertQuery = insertQuery;
+	}
+
+	@Override
+	public void close() {
+		session.close();
+		cluster.close();
+	}
+
+	@Override
+	public void open() {
+		cluster = Cluster.builder().addContactPoint(host).build();
+		session = cluster.connect();
+		if (createQuery != null) {
+			session.execute(createQuery);
+		}
+		preparedStatement = session.prepare(insertQuery);
+	}
+
+	@Override
+	protected void sendValue(IN value) throws Exception {
+		//verify that no query failed until now
+		if (exception != null) {
+			throw new Exception(exception);
+		}
+		//set values for prepared statement
+		Object[] fields = new Object[value.getArity()];
+		for (int x = 0; x < value.getArity(); x++) {
+			fields[x] = value.getField(x);
+		}
+		//insert values and send to cassandra
+		ResultSetFuture result = session.executeAsync(preparedStatement.bind(fields));
+		//add callback to detect errors
+		Futures.addCallback(result, new FutureCallback<ResultSet>() {
+			@Override
+			public void onSuccess(ResultSet resultSet) {
+			}
+
+			@Override
+			public void onFailure(Throwable throwable) {
+				exception = throwable;
+			}
+		});
+	}
+}
diff --git a/flink-streaming-connectors/flink-connector-cassandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/example/CassandraExactlyOnceSinkExample.java b/flink-streaming-connectors/flink-connector-cassandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/example/CassandraExactlyOnceSinkExample.java
new file mode 100644
index 0000000000000..0248f88e18aa6
--- /dev/null
+++ b/flink-streaming-connectors/flink-connector-cassandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/example/CassandraExactlyOnceSinkExample.java
@@ -0,0 +1,80 @@
+/**
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.cassandra.example;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.runtime.state.filesystem.FsStateBackend;
+import org.apache.flink.streaming.api.checkpoint.Checkpointed;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.streaming.connectors.cassandra.CassandraExactlyOnceSink;
+
+import java.util.UUID;
+
+public class CassandraExactlyOnceSinkExample {
+	public static void main(String[] args) throws Exception {
+
+		class MySource implements SourceFunction<Tuple2<String, Integer>>, Checkpointed<Integer> {
+			private int counter = 0;
+			private boolean stop = false;
+
+			@Override
+			public void run(SourceContext<Tuple2<String, Integer>> ctx) throws Exception {
+				while (!stop) {
+					Thread.sleep(50);
+					ctx.collect(new Tuple2<>("" + UUID.randomUUID(), counter));
+					counter++;
+				}
+			}
+
+			@Override
+			public void cancel() {
+				stop = true;
+			}
+
+			@Override
+			public Integer snapshotState(long checkpointId, long checkpointTimestamp) throws Exception {
+				return counter;
+			}
+
+			@Override
+			public void restoreState(Integer state) throws Exception {
+				this.counter = state;
+			}
+		}
+
+		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+		env.setParallelism(1);
+		env.enableCheckpointing(1000);
+		env.setNumberOfExecutionRetries(1);
+		env.setStateBackend(new FsStateBackend(System.getProperty("java.io.tmpdir") + "/flink/backend"));
+
+		env
+			.addSource(new MySource())
+			.transform(
+				"Cassandra Sink",
+				TypeExtractor.getForObject(new Tuple2<>("", 0)),
+				new CassandraExactlyOnceSink<Tuple2<String, Integer>>(
+					"127.0.0.1",
+					"CREATE TABLE example.entries (id text PRIMARY KEY, counter int);",
+					"INSERT INTO example.entries (id, counter) VALUES (?, ?)"));
+
+		env.execute();
+	}
+}
diff --git a/flink-streaming-connectors/flink-connector-cassandra/src/test/java/org/apache/flink/streaming/connectors/cassandra/CassandraExactlyOnceSinkTest.java b/flink-streaming-connectors/flink-connector-cassandra/src/test/java/org/apache/flink/streaming/connectors/cassandra/CassandraExactlyOnceSinkTest.java
new file mode 100644
index 0000000000000..3a2d0ea823da0
--- /dev/null
+++ b/flink-streaming-connectors/flink-connector-cassandra/src/test/java/org/apache/flink/streaming/connectors/cassandra/CassandraExactlyOnceSinkTest.java
@@ -0,0 +1,192 @@
+/**
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.cassandra;
+
+import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.ResultSet;
+import com.datastax.driver.core.Row;
+import com.datastax.driver.core.Session;
+import org.apache.cassandra.service.CassandraDaemon;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.streaming.runtime.operators.ExactlyOnceSinkTestBase;
+import org.apache.flink.streaming.runtime.tasks.OneInputStreamTask;
+import org.apache.flink.streaming.runtime.tasks.OneInputStreamTaskTestHarness;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Scanner;
+import java.util.UUID;
+
+public class CassandraExactlyOnceSinkTest extends ExactlyOnceSinkTestBase<Tuple3<String, Integer, Integer>, CassandraExactlyOnceSink<Tuple3<String, Integer, Integer>>> {
+	private static EmbeddedCassandraService cassandra;
+	private static Cluster cluster;
+	private static Session session;
+
+	private static final String CREATE_KEYSPACE_QUERY = "CREATE KEYSPACE flink WITH replication" + "= {'class':'SimpleStrategy', 'replication_factor':3};";
+	private static final String CREATE_TABLE_QUERY = "CREATE TABLE flink.test (id text PRIMARY KEY, counter int, batch_id int);";
+	private static final String CLEAR_TABLE_QUERY = "TRUNCATE flink.test;";
+	private static final String INSERT_DATA_QUERY = "INSERT INTO flink.test (id, counter, batch_id) VALUES (?, ?, ?)";
+	private static final String SELECT_DATA_QUERY = "SELECT * FROM flink.test;";
+
+	private static class EmbeddedCassandraService {
+		CassandraDaemon cassandraDaemon;
+
+		public void start() throws IOException {
+			this.cassandraDaemon = new CassandraDaemon();
+			this.cassandraDaemon.init(null);
+			this.cassandraDaemon.start();
+		}
+
+		public void stop() {
+			this.cassandraDaemon.stop();
+		}
+	}
+
+	@ClassRule
+	public static TemporaryFolder testFolder = new TemporaryFolder();
+
+	@BeforeClass
+	public static void startCassandra() throws IOException {
+		//generate temporary files
+		ClassLoader classLoader = CassandraExactlyOnceSinkTest.class.getClassLoader();
+		File file = new File(classLoader.getResource("cassandra.yaml").getFile());
+		File tmp = testFolder.newFile("cassandra.yaml");
+		testFolder.newFolder("data");
+		testFolder.newFolder("commit");
+		testFolder.newFolder("cache");
+		BufferedWriter b = new BufferedWriter(new FileWriter(tmp));
+
+		//copy cassandra.yaml; inject absolute paths into cassandra.yaml
+		Scanner scanner = new Scanner(file);
+		while (scanner.hasNextLine()) {
+			String line = scanner.nextLine();
+			line = line.replace("$PATH", "'" + tmp.getParentFile());
+			b.write(line + "\n");
+			b.flush();
+		}
+		scanner.close();
+
+
+		// Tell cassandra where the configuration files are.
+		// Use the test configuration file.
+		System.setProperty("cassandra.config", "file:" + File.separator + File.separator + File.separator + tmp.getAbsolutePath());
+
+		cassandra = new EmbeddedCassandraService();
+		cassandra.start();
+
+		cluster = Cluster.builder().addContactPoint("127.0.0.1").build();
+		session = cluster.connect();
+
+		session.execute(CREATE_KEYSPACE_QUERY);
+		session.execute(CREATE_TABLE_QUERY);
+	}
+
+	@After
+	public void deleteSchema() throws Exception {
+		session.executeAsync(CLEAR_TABLE_QUERY);
+	}
+
+	@AfterClass
+	public static void closeCassandra() {
+		session.close();
+		cluster.close();
+		cassandra.stop();
+	}
+
+	protected CassandraExactlyOnceSink<Tuple3<String, Integer, Integer>> createSink() {
+		return new CassandraExactlyOnceSink<>(
+			"127.0.0.1",
+			INSERT_DATA_QUERY);
+	}
+
+	protected TupleTypeInfo<Tuple3<String, Integer, Integer>> createTypeInfo() {
+		return TupleTypeInfo.getBasicTupleTypeInfo(String.class, Integer.class, Integer.class);
+	}
+
+	@Override
+	protected Tuple3<String, Integer, Integer> generateValue(int counter, int checkpointID) {
+		return new Tuple3<>("" + UUID.randomUUID(), counter, checkpointID);
+	}
+
+	@Override
+	protected void verifyResultsIdealCircumstances(
+		OneInputStreamTaskTestHarness<Tuple3<String, Integer, Integer>, Tuple3<String, Integer, Integer>> harness,
+		OneInputStreamTask<Tuple3<String, Integer, Integer>, Tuple3<String, Integer, Integer>> task,
+		CassandraExactlyOnceSink<Tuple3<String, Integer, Integer>> sink) {
+
+		ResultSet result = session.execute(SELECT_DATA_QUERY);
+		ArrayList<Integer> list = new ArrayList<>();
+		for (int x = 1; x <= 600; x++) {
+			list.add(x);
+		}
+
+		for (Row s : result) {
+			list.remove(new Integer(s.getInt("counter")));
+		}
+		Assert.assertTrue("The following ID's were not found in the ResultSet: " + list.toString(), list.isEmpty());
+	}
+
+	@Override
+	protected void verifyResultsDataPersistenceUponMissedNotify(
+		OneInputStreamTaskTestHarness<Tuple3<String, Integer, Integer>, Tuple3<String, Integer, Integer>> harness,
+		OneInputStreamTask<Tuple3<String, Integer, Integer>, Tuple3<String, Integer, Integer>> task,
+		CassandraExactlyOnceSink<Tuple3<String, Integer, Integer>> sink) {
+
+		ResultSet result = session.execute(SELECT_DATA_QUERY);
+		ArrayList<Integer> list = new ArrayList<>();
+		for (int x = 1; x <= 600; x++) {
+			list.add(x);
+		}
+
+		for (Row s : result) {
+			list.remove(new Integer(s.getInt("counter")));
+		}
+		Assert.assertTrue("The following ID's were not found in the ResultSet: " + list.toString(), list.isEmpty());
+	}
+
+	@Override
+	protected void verifyResultsDataDiscardingUponRestore(
+		OneInputStreamTaskTestHarness<Tuple3<String, Integer, Integer>, Tuple3<String, Integer, Integer>> harness,
+		OneInputStreamTask<Tuple3<String, Integer, Integer>, Tuple3<String, Integer, Integer>> task,
+		CassandraExactlyOnceSink<Tuple3<String, Integer, Integer>> sink) {
+
+		ResultSet result = session.execute(SELECT_DATA_QUERY);
+		ArrayList<Integer> list = new ArrayList<>();
+		for (int x = 1; x <= 200; x++) {
+			list.add(x);
+		}
+		for (int x = 401; x <= 600; x++) {
+			list.add(x);
+		}
+
+		for (Row s : result) {
+			list.remove(new Integer(s.getInt("counter")));
+		}
+		Assert.assertTrue("The following ID's were not found in the ResultSet: " + list.toString(), list.isEmpty());
+	}
+}
diff --git a/flink-streaming-connectors/flink-connector-cassandra/src/test/resources/cassandra.yaml b/flink-streaming-connectors/flink-connector-cassandra/src/test/resources/cassandra.yaml
new file mode 100644
index 0000000000000..6cd7d4e2c734d
--- /dev/null
+++ b/flink-streaming-connectors/flink-connector-cassandra/src/test/resources/cassandra.yaml
@@ -0,0 +1,40 @@
+################################################################################
+#  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.
+################################################################################
+cluster_name: 'Test Cluster'
+commitlog_sync: 'periodic'
+commitlog_sync_period_in_ms: 10000
+partitioner: 'org.apache.cassandra.dht.RandomPartitioner'
+endpoint_snitch: 'org.apache.cassandra.locator.SimpleSnitch'
+commitlog_directory: $PATH\commit'
+data_file_directories:
+    - $PATH\data'
+saved_caches_directory: $PATH\cache'
+listen_address: '127.0.0.1'
+seed_provider:
+    - class_name: 'org.apache.cassandra.locator.SimpleSeedProvider'
+      parameters:
+          - seeds: '127.0.0.1'
+native_transport_port: 9042
+start_native_transport: true
+
+read_request_timeout_in_ms: 60000
+range_request_timeout_in_ms: 60000
+write_request_timeout_in_ms: 40000
+cas_contention_timeout_in_ms: 3000
+truncate_request_timeout_in_ms: 60000
+request_timeout_in_ms: 60000
diff --git a/flink-streaming-connectors/flink-connector-cassandra/src/test/resources/log4j-test.properties b/flink-streaming-connectors/flink-connector-cassandra/src/test/resources/log4j-test.properties
new file mode 100644
index 0000000000000..6d70ebddbfdfa
--- /dev/null
+++ b/flink-streaming-connectors/flink-connector-cassandra/src/test/resources/log4j-test.properties
@@ -0,0 +1,29 @@
+################################################################################
+#  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.
+################################################################################
+
+log4j.rootLogger=ERROR, testlogger
+
+log4j.appender.testlogger=org.apache.log4j.ConsoleAppender
+log4j.appender.testlogger.target = System.err
+log4j.appender.testlogger.layout=org.apache.log4j.PatternLayout
+log4j.appender.testlogger.layout.ConversionPattern=%-4r [%t] %-5p %c %x - %m%n
+
+# suppress the irrelevant (wrong) warnings from the netty channel handler
+log4j.logger.org.jboss.netty.channel.DefaultChannelPipeline=ERROR, testlogger
+
+
diff --git a/flink-streaming-connectors/pom.xml b/flink-streaming-connectors/pom.xml
index fceae25f9de80..1249bf24a0b1a 100644
--- a/flink-streaming-connectors/pom.xml
+++ b/flink-streaming-connectors/pom.xml
@@ -44,6 +44,7 @@ under the License.
 		<module>flink-connector-rabbitmq</module>
 		<module>flink-connector-twitter</module>
 		<module>flink-connector-nifi</module>
+		<module>flink-connector-cassandra</module>
 	</modules>
 
 	<!-- See main pom.xml for explanation of profiles -->
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericExactlyOnceSink.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericExactlyOnceSink.java
new file mode 100644
index 0000000000000..066ada83f52db
--- /dev/null
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericExactlyOnceSink.java
@@ -0,0 +1,173 @@
+/**
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.runtime.operators;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.runtime.state.AbstractStateBackend;
+import org.apache.flink.runtime.state.StateHandle;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.StreamTaskState;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * Generic Sink that emits its input elements into an arbitrary backend. This sink is integrated with the checkpointing
+ * mechanism to provide exactly once semantics.
+ *
+ * Incoming records are stored within a {@link org.apache.flink.runtime.state.AbstractStateBackend}, and only committed if a
+ * checkpoint is completed. Should a job fail while the data is being committed, no exactly once guarantee can be made.
+ * @param <IN>
+ */
+public abstract class GenericExactlyOnceSink<IN extends Tuple> extends AbstractStreamOperator<IN> implements OneInputStreamOperator<IN, IN> {
+	private AbstractStateBackend.CheckpointStateOutputView out;
+	private TypeSerializer<IN> serializer;
+	protected TypeInformation<IN> typeInfo;
+
+	private ExactlyOnceState state = new ExactlyOnceState();
+
+	/**
+	 * Saves a handle in the state.
+	 * @param checkpointId
+	 * @throws IOException
+	 */
+	private void saveHandleInState(final long checkpointId) throws IOException {
+		//only add handle if a new OperatorState was created since the last snapshot/notify
+		if (out != null) {
+			out.writeByte(0); //EOF-byte
+			StateHandle<DataInputView> handle = out.closeAndGetHandle();
+			if (state.pendingHandles.containsKey(checkpointId)) {
+				state.pendingHandles.get(checkpointId).add(handle);
+			} else {
+				ArrayList<StateHandle<DataInputView>> list = new ArrayList<>();
+				list.add(handle);
+				state.pendingHandles.put(checkpointId, list);
+			}
+			out = null;
+		}
+	}
+
+	@Override
+	public StreamTaskState snapshotOperatorState(final long checkpointId, final long timestamp) throws Exception {
+		StreamTaskState taskState = super.snapshotOperatorState(checkpointId, timestamp);
+		saveHandleInState(checkpointId);
+		taskState.setFunctionState(state);
+		return taskState;
+	}
+
+	@Override
+	public void restoreState(StreamTaskState state, long recoveryTimestamp) throws Exception {
+		super.restoreState(state, recoveryTimestamp);
+		this.state = (ExactlyOnceState) state.getFunctionState();
+		out = null;
+	}
+
+	@Override
+	public void notifyOfCompletedCheckpoint(long checkpointId) throws Exception {
+		super.notifyOfCompletedCheckpoint(checkpointId);
+		saveHandleInState(checkpointId);
+
+		synchronized (state.pendingHandles) {
+			Set<Long> pastCheckpointIds = state.pendingHandles.keySet();
+			Set<Long> checkpointsToRemove = new HashSet<>();
+			for (Long pastCheckpointId : pastCheckpointIds) {
+				if (pastCheckpointId <= checkpointId) {
+					List<StateHandle<DataInputView>> handles = state.pendingHandles.get(pastCheckpointId);
+					for (StateHandle<DataInputView> handle : handles) {
+						DataInputView in = handle.getState(getUserCodeClassloader());
+						while (in.readByte() == 1) {
+							IN value = serializer.deserialize(in);
+							sendValue(value);
+						}
+						checkpointsToRemove.add(pastCheckpointId);
+					}
+				}
+			}
+			for (Long toRemove : checkpointsToRemove) {
+				state.pendingHandles.remove(toRemove);
+			}
+		}
+	}
+
+	/**
+	 * Write the given element into the backend.
+	 * @param value value to be written
+	 * @throws Exception
+     */
+	protected abstract void sendValue(IN value) throws Exception;
+
+	@Override
+	public void processElement(StreamRecord<IN> element) throws Exception {
+		IN value = element.getValue();
+		if (serializer == null) {
+			typeInfo = TypeExtractor.getForObject(value);
+			serializer = typeInfo.createSerializer(new ExecutionConfig());
+		}
+		//generate initial operator state
+		if (out == null) {
+			out = getStateBackend().createCheckpointStateOutputView(0, 0);
+		}
+		out.writeByte(1);
+		serializer.serialize(value, out);
+	}
+
+	@Override
+	public void processWatermark(Watermark mark) throws Exception {
+		//don't do anything, since no are a sink
+	}
+
+	/**
+	 * This state is used to keep a list of all StateHandles (essentially references to past OperatorStates) that were
+	 * used since the last completed checkpoint.
+	 **/
+	public class ExactlyOnceState implements StateHandle<Serializable> {
+		protected HashMap<Long, ArrayList<StateHandle<DataInputView>>> pendingHandles;
+
+		public ExactlyOnceState() {
+			pendingHandles = new HashMap<>();
+		}
+
+		@Override
+		public HashMap<Long, ArrayList<StateHandle<DataInputView>>> getState(ClassLoader userCodeClassLoader) throws Exception {
+			return pendingHandles;
+		}
+
+		@Override
+		public void discardState() throws Exception {
+			pendingHandles = new HashMap<>();
+		}
+
+		@Override
+		public long getStateSize() throws Exception {
+			return 0;
+		}
+	}
+}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/ExactlyOnceSinkTestBase.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/ExactlyOnceSinkTestBase.java
new file mode 100644
index 0000000000000..a5c3c11534aa2
--- /dev/null
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/ExactlyOnceSinkTestBase.java
@@ -0,0 +1,203 @@
+/**
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.runtime.operators;
+
+import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.streaming.api.graph.StreamConfig;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.OneInputStreamTask;
+import org.apache.flink.streaming.runtime.tasks.OneInputStreamTaskTestHarness;
+import org.apache.flink.streaming.runtime.tasks.StreamTaskState;
+import org.junit.Test;
+
+import java.util.ArrayList;
+
+public abstract class ExactlyOnceSinkTestBase<IN extends Tuple, S extends GenericExactlyOnceSink<IN>> {
+
+	protected class OperatorExposingTask<IN> extends OneInputStreamTask<IN, IN> {
+		public OneInputStreamOperator<IN, IN> getOperator() {
+			return this.headOperator;
+		}
+	}
+
+	protected OperatorExposingTask<IN> createTask() {
+		return new OperatorExposingTask<>();
+	}
+
+	protected abstract S createSink();
+
+	protected abstract TupleTypeInfo<IN> createTypeInfo();
+
+	protected abstract IN generateValue(int counter, int checkpointID);
+
+	protected abstract void verifyResultsIdealCircumstances(
+		OneInputStreamTaskTestHarness<IN, IN> harness, OneInputStreamTask<IN, IN> task, S sink) throws Exception;
+
+	protected abstract void verifyResultsDataPersistenceUponMissedNotify(
+		OneInputStreamTaskTestHarness<IN, IN> harness, OneInputStreamTask<IN, IN> task, S sink) throws Exception;
+
+	protected abstract void verifyResultsDataDiscardingUponRestore(
+		OneInputStreamTaskTestHarness<IN, IN> harness, OneInputStreamTask<IN, IN> task, S sink) throws Exception;
+
+	@Test
+	public void testIdealCircumstances() throws Exception {
+		S sink = createSink();
+		OperatorExposingTask<IN> task = createTask();
+		TupleTypeInfo<IN> info = createTypeInfo();
+		OneInputStreamTaskTestHarness<IN, IN> testHarness = new OneInputStreamTaskTestHarness<>(task, 1, 1, info, info);
+		StreamConfig streamConfig = testHarness.getStreamConfig();
+		streamConfig.setStreamOperator(sink);
+
+		int elementCounter = 1;
+
+		testHarness.invoke();
+		testHarness.waitForTaskRunning();
+
+		ArrayList<StreamTaskState> states = new ArrayList<>();
+
+		for (int x = 0; x < 200; x++) {
+			testHarness.processElement(new StreamRecord<>(generateValue(elementCounter, 0)));
+			elementCounter++;
+		}
+		testHarness.waitForInputProcessing();
+
+		states.add(task.getOperator().snapshotOperatorState(states.size(), 0));
+		task.notifyCheckpointComplete(states.size() - 1);
+
+		for (int x = 0; x < 200; x++) {
+			testHarness.processElement(new StreamRecord<>(generateValue(elementCounter, 1)));
+			elementCounter++;
+		}
+		testHarness.waitForInputProcessing();
+
+		states.add(task.getOperator().snapshotOperatorState(states.size(), 0));
+		task.notifyCheckpointComplete(states.size() - 1);
+
+		for (int x = 0; x < 200; x++) {
+			testHarness.processElement(new StreamRecord<>(generateValue(elementCounter, 2)));
+			elementCounter++;
+		}
+		testHarness.waitForInputProcessing();
+
+		states.add(task.getOperator().snapshotOperatorState(states.size(), 0));
+		task.notifyCheckpointComplete(states.size() - 1);
+
+		testHarness.endInput();
+		testHarness.waitForTaskCompletion();
+
+		verifyResultsIdealCircumstances(testHarness, task, (S) task.getOperator());
+	}
+
+	@Test
+	public void testDataPersistenceUponMissedNotify() throws Exception {
+		S sink = createSink();
+		OperatorExposingTask<IN> task = createTask();
+		TupleTypeInfo<IN> info = createTypeInfo();
+		OneInputStreamTaskTestHarness<IN, IN> testHarness = new OneInputStreamTaskTestHarness<>(task, 1, 1, info, info);
+		StreamConfig streamConfig = testHarness.getStreamConfig();
+		streamConfig.setStreamOperator(sink);
+
+		int elementCounter = 1;
+
+		testHarness.invoke();
+		testHarness.waitForTaskRunning();
+
+		ArrayList<StreamTaskState> states = new ArrayList<>();
+
+		for (int x = 0; x < 200; x++) {
+			testHarness.processElement(new StreamRecord<>(generateValue(elementCounter, 0)));
+			elementCounter++;
+		}
+		testHarness.waitForInputProcessing();
+
+		states.add(task.getOperator().snapshotOperatorState(states.size(), 0));
+		task.notifyCheckpointComplete(states.size() - 1);
+
+		for (int x = 0; x < 200; x++) {
+			testHarness.processElement(new StreamRecord<>(generateValue(elementCounter, 1)));
+			elementCounter++;
+		}
+		testHarness.waitForInputProcessing();
+
+		states.add(task.getOperator().snapshotOperatorState(states.size(), 0));
+
+		for (int x = 0; x < 200; x++) {
+			testHarness.processElement(new StreamRecord<>(generateValue(elementCounter, 2)));
+			elementCounter++;
+		}
+		testHarness.waitForInputProcessing();
+
+		states.add(task.getOperator().snapshotOperatorState(states.size(), 0));
+		task.notifyCheckpointComplete(states.size() - 1);
+
+		testHarness.endInput();
+		testHarness.waitForTaskCompletion();
+
+		verifyResultsDataPersistenceUponMissedNotify(testHarness, task, (S) task.getOperator());
+	}
+
+	@Test
+	public void testDataDiscardingUponRestore() throws Exception {
+		S sink = createSink();
+		OperatorExposingTask<IN> task = createTask();
+		TupleTypeInfo<IN> info = createTypeInfo();
+		OneInputStreamTaskTestHarness<IN, IN> testHarness = new OneInputStreamTaskTestHarness<>(task, 1, 1, info, info);
+		StreamConfig streamConfig = testHarness.getStreamConfig();
+		streamConfig.setStreamOperator(sink);
+
+		int elementCounter = 1;
+
+		testHarness.invoke();
+		testHarness.waitForTaskRunning();
+
+		ArrayList<StreamTaskState> states = new ArrayList<>();
+
+		for (int x = 0; x < 200; x++) {
+			testHarness.processElement(new StreamRecord<>(generateValue(elementCounter, 0)));
+			elementCounter++;
+		}
+		testHarness.waitForInputProcessing();
+
+		states.add(task.getOperator().snapshotOperatorState(states.size(), 0));
+		task.notifyCheckpointComplete(states.size() - 1);
+
+		for (int x = 0; x < 200; x++) {
+			testHarness.processElement(new StreamRecord<>(generateValue(elementCounter, 1)));
+			elementCounter++;
+		}
+		testHarness.waitForInputProcessing();
+
+		task.getOperator().restoreState(states.get(states.size() - 1), 0);
+
+		for (int x = 0; x < 200; x++) {
+			testHarness.processElement(new StreamRecord<>(generateValue(elementCounter, 2)));
+			elementCounter++;
+		}
+		testHarness.waitForInputProcessing();
+
+		states.add(task.getOperator().snapshotOperatorState(states.size(), 0));
+		task.notifyCheckpointComplete(states.size() - 1);
+
+		testHarness.endInput();
+		testHarness.waitForTaskCompletion();
+
+		verifyResultsDataDiscardingUponRestore(testHarness, task, (S) task.getOperator());
+	}
+}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/GenericExactlyOnceSinkTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/GenericExactlyOnceSinkTest.java
new file mode 100644
index 0000000000000..2c167589f5f6e
--- /dev/null
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/GenericExactlyOnceSinkTest.java
@@ -0,0 +1,107 @@
+/**
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.runtime.operators;
+
+import org.apache.flink.api.java.tuple.Tuple1;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.streaming.runtime.tasks.OneInputStreamTask;
+import org.apache.flink.streaming.runtime.tasks.OneInputStreamTaskTestHarness;
+import org.junit.Assert;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class GenericExactlyOnceSinkTest extends ExactlyOnceSinkTestBase<Tuple1<Integer>, GenericExactlyOnceSinkTest.ListSink> {
+	@Override
+	protected ListSink createSink() {
+		return new ListSink();
+	}
+
+	@Override
+	protected TupleTypeInfo<Tuple1<Integer>> createTypeInfo() {
+		return TupleTypeInfo.getBasicTupleTypeInfo(Integer.class);
+	}
+
+	@Override
+	protected Tuple1<Integer> generateValue(int counter, int checkpointID) {
+		return new Tuple1<>(counter);
+	}
+
+	@Override
+	protected void verifyResultsIdealCircumstances(
+		OneInputStreamTaskTestHarness<Tuple1<Integer>, Tuple1<Integer>> harness,
+		OneInputStreamTask<Tuple1<Integer>, Tuple1<Integer>> task, ListSink sink) {
+
+		ArrayList<Integer> list = new ArrayList<>();
+		for (int x = 1; x <= 600; x++) {
+			list.add(x);
+		}
+
+		for (Integer i : sink.values) {
+			list.remove(i);
+		}
+		Assert.assertTrue("The following ID's where not found in the result list: " + list.toString(), list.isEmpty());
+	}
+
+	@Override
+	protected void verifyResultsDataPersistenceUponMissedNotify(
+		OneInputStreamTaskTestHarness<Tuple1<Integer>, Tuple1<Integer>> harness,
+		OneInputStreamTask<Tuple1<Integer>, Tuple1<Integer>> task, ListSink sink) {
+
+		ArrayList<Integer> list = new ArrayList<>();
+		for (int x = 1; x <= 600; x++) {
+			list.add(x);
+		}
+
+		for (Integer i : sink.values) {
+			list.remove(i);
+		}
+		Assert.assertTrue("The following ID's where not found in the result list: " + list.toString(), list.isEmpty());
+	}
+
+	@Override
+	protected void verifyResultsDataDiscardingUponRestore(
+		OneInputStreamTaskTestHarness<Tuple1<Integer>, Tuple1<Integer>> harness,
+		OneInputStreamTask<Tuple1<Integer>, Tuple1<Integer>> task, ListSink sink) {
+
+		ArrayList<Integer> list = new ArrayList<>();
+		for (int x = 1; x <= 200; x++) {
+			list.add(x);
+		}
+		for (int x = 401; x <= 600; x++) {
+			list.add(x);
+		}
+
+		for (Integer i : sink.values) {
+			list.remove(i);
+		}
+		Assert.assertTrue("The following ID's where not found in the result list: " + list.toString(), list.isEmpty());
+	}
+
+	/**
+	 * Simple sink that stores all records in a public list.
+	 */
+	public static class ListSink extends GenericExactlyOnceSink<Tuple1<Integer>> {
+		public List<Integer> values = new ArrayList<>();
+
+		@Override
+		protected void sendValue(Tuple1<Integer> value) throws Exception {
+			values.add(value.f0);
+		}
+	}
+}

From d78a7fe09591a7405edac04e45da9af2e57a9f3e Mon Sep 17 00:00:00 2001
From: zentol <s.motsu@web.de>
Date: Thu, 11 Feb 2016 17:23:17 +0100
Subject: [PATCH 02/10] GenericSink no longer bound to Tuple type

---
 .../runtime/operators/GenericExactlyOnceSink.java   |  2 +-
 .../runtime/operators/ExactlyOnceSinkTestBase.java  | 13 ++++++-------
 2 files changed, 7 insertions(+), 8 deletions(-)

diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericExactlyOnceSink.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericExactlyOnceSink.java
index 066ada83f52db..ad77140c31562 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericExactlyOnceSink.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericExactlyOnceSink.java
@@ -47,7 +47,7 @@
  * checkpoint is completed. Should a job fail while the data is being committed, no exactly once guarantee can be made.
  * @param <IN>
  */
-public abstract class GenericExactlyOnceSink<IN extends Tuple> extends AbstractStreamOperator<IN> implements OneInputStreamOperator<IN, IN> {
+public abstract class GenericExactlyOnceSink<IN> extends AbstractStreamOperator<IN> implements OneInputStreamOperator<IN, IN> {
 	private AbstractStateBackend.CheckpointStateOutputView out;
 	private TypeSerializer<IN> serializer;
 	protected TypeInformation<IN> typeInfo;
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/ExactlyOnceSinkTestBase.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/ExactlyOnceSinkTestBase.java
index a5c3c11534aa2..05452e6aab0b9 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/ExactlyOnceSinkTestBase.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/ExactlyOnceSinkTestBase.java
@@ -17,8 +17,7 @@
  */
 package org.apache.flink.streaming.runtime.operators;
 
-import org.apache.flink.api.java.tuple.Tuple;
-import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.streaming.api.graph.StreamConfig;
 import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
 import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
@@ -29,7 +28,7 @@
 
 import java.util.ArrayList;
 
-public abstract class ExactlyOnceSinkTestBase<IN extends Tuple, S extends GenericExactlyOnceSink<IN>> {
+public abstract class ExactlyOnceSinkTestBase<IN, S extends GenericExactlyOnceSink<IN>> {
 
 	protected class OperatorExposingTask<IN> extends OneInputStreamTask<IN, IN> {
 		public OneInputStreamOperator<IN, IN> getOperator() {
@@ -43,7 +42,7 @@ protected OperatorExposingTask<IN> createTask() {
 
 	protected abstract S createSink();
 
-	protected abstract TupleTypeInfo<IN> createTypeInfo();
+	protected abstract TypeInformation<IN> createTypeInfo();
 
 	protected abstract IN generateValue(int counter, int checkpointID);
 
@@ -60,7 +59,7 @@ protected abstract void verifyResultsDataDiscardingUponRestore(
 	public void testIdealCircumstances() throws Exception {
 		S sink = createSink();
 		OperatorExposingTask<IN> task = createTask();
-		TupleTypeInfo<IN> info = createTypeInfo();
+		TypeInformation<IN> info = createTypeInfo();
 		OneInputStreamTaskTestHarness<IN, IN> testHarness = new OneInputStreamTaskTestHarness<>(task, 1, 1, info, info);
 		StreamConfig streamConfig = testHarness.getStreamConfig();
 		streamConfig.setStreamOperator(sink);
@@ -109,7 +108,7 @@ public void testIdealCircumstances() throws Exception {
 	public void testDataPersistenceUponMissedNotify() throws Exception {
 		S sink = createSink();
 		OperatorExposingTask<IN> task = createTask();
-		TupleTypeInfo<IN> info = createTypeInfo();
+		TypeInformation<IN> info = createTypeInfo();
 		OneInputStreamTaskTestHarness<IN, IN> testHarness = new OneInputStreamTaskTestHarness<>(task, 1, 1, info, info);
 		StreamConfig streamConfig = testHarness.getStreamConfig();
 		streamConfig.setStreamOperator(sink);
@@ -157,7 +156,7 @@ public void testDataPersistenceUponMissedNotify() throws Exception {
 	public void testDataDiscardingUponRestore() throws Exception {
 		S sink = createSink();
 		OperatorExposingTask<IN> task = createTask();
-		TupleTypeInfo<IN> info = createTypeInfo();
+		TypeInformation<IN> info = createTypeInfo();
 		OneInputStreamTaskTestHarness<IN, IN> testHarness = new OneInputStreamTaskTestHarness<>(task, 1, 1, info, info);
 		StreamConfig streamConfig = testHarness.getStreamConfig();
 		streamConfig.setStreamOperator(sink);

From 3b60d8192bcaec8d76d2544761b2d372f766f31c Mon Sep 17 00:00:00 2001
From: zentol <s.motsu@web.de>
Date: Thu, 11 Feb 2016 17:25:17 +0100
Subject: [PATCH 03/10] added missing parameter description

---
 .../connectors/cassandra/CassandraExactlyOnceSink.java          | 2 +-
 .../streaming/runtime/operators/GenericExactlyOnceSink.java     | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/flink-streaming-connectors/flink-connector-cassandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/CassandraExactlyOnceSink.java b/flink-streaming-connectors/flink-connector-cassandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/CassandraExactlyOnceSink.java
index 53b9d699f7a37..29e84bf517b60 100644
--- a/flink-streaming-connectors/flink-connector-cassandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/CassandraExactlyOnceSink.java
+++ b/flink-streaming-connectors/flink-connector-cassandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/CassandraExactlyOnceSink.java
@@ -33,7 +33,7 @@
  *
  * Incoming records are stored within a {@link org.apache.flink.runtime.state.AbstractStateBackend}, and only committed if a
  * checkpoint is completed. Should a job fail while the data is being committed, no exactly once guarantee can be made.
- * @param <IN>
+ * @param <IN> Type of the elements emitted by this sink
  */
 public class CassandraExactlyOnceSink<IN extends Tuple> extends GenericExactlyOnceSink<IN> {
 	private String host;
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericExactlyOnceSink.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericExactlyOnceSink.java
index ad77140c31562..ee5060f66a39b 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericExactlyOnceSink.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericExactlyOnceSink.java
@@ -45,7 +45,7 @@
  *
  * Incoming records are stored within a {@link org.apache.flink.runtime.state.AbstractStateBackend}, and only committed if a
  * checkpoint is completed. Should a job fail while the data is being committed, no exactly once guarantee can be made.
- * @param <IN>
+ * @param <IN> Type of the elements emitted by this sink
  */
 public abstract class GenericExactlyOnceSink<IN> extends AbstractStreamOperator<IN> implements OneInputStreamOperator<IN, IN> {
 	private AbstractStateBackend.CheckpointStateOutputView out;

From 98b3c071d269d39651e239e54ee121ac6e57b1c4 Mon Sep 17 00:00:00 2001
From: zentol <s.motsu@web.de>
Date: Thu, 11 Feb 2016 17:29:03 +0100
Subject: [PATCH 04/10] move checkpointsToRemove out of loop

---
 .../streaming/runtime/operators/GenericExactlyOnceSink.java     | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericExactlyOnceSink.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericExactlyOnceSink.java
index ee5060f66a39b..d1b1378542715 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericExactlyOnceSink.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericExactlyOnceSink.java
@@ -107,8 +107,8 @@ public void notifyOfCompletedCheckpoint(long checkpointId) throws Exception {
 							IN value = serializer.deserialize(in);
 							sendValue(value);
 						}
-						checkpointsToRemove.add(pastCheckpointId);
 					}
+					checkpointsToRemove.add(pastCheckpointId);
 				}
 			}
 			for (Long toRemove : checkpointsToRemove) {

From a1f19a690e752ed8a398a33d4a9b23066e1239d0 Mon Sep 17 00:00:00 2001
From: zentol <s.motsu@web.de>
Date: Thu, 11 Feb 2016 17:31:12 +0100
Subject: [PATCH 05/10] fix typo

---
 .../streaming/runtime/operators/GenericExactlyOnceSink.java     | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericExactlyOnceSink.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericExactlyOnceSink.java
index d1b1378542715..76fd5866ff6c4 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericExactlyOnceSink.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericExactlyOnceSink.java
@@ -141,7 +141,7 @@ public void processElement(StreamRecord<IN> element) throws Exception {
 
 	@Override
 	public void processWatermark(Watermark mark) throws Exception {
-		//don't do anything, since no are a sink
+		//don't do anything, since we are a sink
 	}
 
 	/**

From dd219bf6ebe7a6ffe56b850937f4ee03b9e76812 Mon Sep 17 00:00:00 2001
From: zentol <s.motsu@web.de>
Date: Thu, 11 Feb 2016 17:39:10 +0100
Subject: [PATCH 06/10] remove savehandleInState() call in notify...()

---
 .../streaming/runtime/operators/GenericExactlyOnceSink.java      | 1 -
 1 file changed, 1 deletion(-)

diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericExactlyOnceSink.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericExactlyOnceSink.java
index 76fd5866ff6c4..9f9bedeb6aced 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericExactlyOnceSink.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericExactlyOnceSink.java
@@ -93,7 +93,6 @@ public void restoreState(StreamTaskState state, long recoveryTimestamp) throws E
 	@Override
 	public void notifyOfCompletedCheckpoint(long checkpointId) throws Exception {
 		super.notifyOfCompletedCheckpoint(checkpointId);
-		saveHandleInState(checkpointId);
 
 		synchronized (state.pendingHandles) {
 			Set<Long> pastCheckpointIds = state.pendingHandles.keySet();

From d8becc32789b4235dc3bf427266c92d29fd178ca Mon Sep 17 00:00:00 2001
From: zentol <s.motsu@web.de>
Date: Thu, 11 Feb 2016 17:48:57 +0100
Subject: [PATCH 07/10] properly discard states

---
 .../runtime/operators/GenericExactlyOnceSink.java        | 9 +++++++++
 1 file changed, 9 insertions(+)

diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericExactlyOnceSink.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericExactlyOnceSink.java
index 9f9bedeb6aced..bbeffbf31bdc7 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericExactlyOnceSink.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericExactlyOnceSink.java
@@ -111,7 +111,11 @@ public void notifyOfCompletedCheckpoint(long checkpointId) throws Exception {
 				}
 			}
 			for (Long toRemove : checkpointsToRemove) {
+				List<StateHandle<DataInputView>> handles = state.pendingHandles.get(toRemove);
 				state.pendingHandles.remove(toRemove);
+				for (StateHandle<DataInputView> handle : handles) {
+					handle.discardState();
+				}
 			}
 		}
 	}
@@ -161,6 +165,11 @@ public HashMap<Long, ArrayList<StateHandle<DataInputView>>> getState(ClassLoader
 
 		@Override
 		public void discardState() throws Exception {
+			for (List<StateHandle<DataInputView>> handles : pendingHandles.values()) {
+				for (StateHandle<DataInputView> handle : handles) {
+					handle.discardState();
+				}
+			}
 			pendingHandles = new HashMap<>();
 		}
 

From 7b1e2c526634e833bd5fd0a8e48b2d6867aeecb2 Mon Sep 17 00:00:00 2001
From: zentol <s.motsu@web.de>
Date: Thu, 11 Feb 2016 17:53:20 +0100
Subject: [PATCH 08/10] calculate state size

---
 .../runtime/operators/GenericExactlyOnceSink.java         | 8 +++++++-
 1 file changed, 7 insertions(+), 1 deletion(-)

diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericExactlyOnceSink.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericExactlyOnceSink.java
index bbeffbf31bdc7..77e375ab448b8 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericExactlyOnceSink.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/GenericExactlyOnceSink.java
@@ -175,7 +175,13 @@ public void discardState() throws Exception {
 
 		@Override
 		public long getStateSize() throws Exception {
-			return 0;
+			int stateSize = 0;
+			for (List<StateHandle<DataInputView>> handles : pendingHandles.values()) {
+				for (StateHandle<DataInputView> handle : handles) {
+					stateSize += handle.getStateSize();
+				}
+			}
+			return stateSize;
 		}
 	}
 }

From d60f51d6bb2c72a745b89b4c73f11777e3bc76eb Mon Sep 17 00:00:00 2001
From: zentol <s.motsu@web.de>
Date: Thu, 11 Feb 2016 17:55:27 +0100
Subject: [PATCH 09/10] add sanity checks to cassandra sink constructor

---
 .../connectors/cassandra/CassandraExactlyOnceSink.java      | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/flink-streaming-connectors/flink-connector-cassandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/CassandraExactlyOnceSink.java b/flink-streaming-connectors/flink-connector-cassandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/CassandraExactlyOnceSink.java
index 29e84bf517b60..067d7338ed3f4 100644
--- a/flink-streaming-connectors/flink-connector-cassandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/CassandraExactlyOnceSink.java
+++ b/flink-streaming-connectors/flink-connector-cassandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/CassandraExactlyOnceSink.java
@@ -51,6 +51,12 @@ public CassandraExactlyOnceSink(String host, String insertQuery) {
 	}
 
 	public CassandraExactlyOnceSink(String host, String createQuery, String insertQuery) {
+		if (host == null) {
+			throw new IllegalArgumentException("Host argument must not be null.");
+		}
+		if (insertQuery == null) {
+			throw new IllegalArgumentException("Insert query argument must not be null.");
+		}
 		this.host = host;
 		this.createQuery = createQuery;
 		this.insertQuery = insertQuery;

From e1b145b53ee12e8a25a9d4e52aa47bc938428514 Mon Sep 17 00:00:00 2001
From: zentol <s.motsu@web.de>
Date: Thu, 11 Feb 2016 17:56:12 +0100
Subject: [PATCH 10/10] constructor argument fields set to final

---
 .../connectors/cassandra/CassandraExactlyOnceSink.java      | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/flink-streaming-connectors/flink-connector-cassandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/CassandraExactlyOnceSink.java b/flink-streaming-connectors/flink-connector-cassandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/CassandraExactlyOnceSink.java
index 067d7338ed3f4..dc47afdd5d2e7 100644
--- a/flink-streaming-connectors/flink-connector-cassandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/CassandraExactlyOnceSink.java
+++ b/flink-streaming-connectors/flink-connector-cassandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/CassandraExactlyOnceSink.java
@@ -36,9 +36,9 @@
  * @param <IN> Type of the elements emitted by this sink
  */
 public class CassandraExactlyOnceSink<IN extends Tuple> extends GenericExactlyOnceSink<IN> {
-	private String host;
-	private String createQuery;
-	private String insertQuery;
+	private final String host;
+	private final String createQuery;
+	private final String insertQuery;
 
 	private transient Cluster cluster;
 	private transient Session session;