[FLINK-4097] Fix NullPointerException on CassandraSinkBase and...

[FLINK-4097] Fix NullPointerException on CassandraSinkBase and CassandraTupleWriteAheadSink's close()

This closes #2144

flink-streaming-connectors/flink-connector-cassandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/CassandraSinkBase.java

@@ -81,12 +81,16 @@ public abstract class CassandraSinkBase<IN, V> extends RichSinkFunction<IN> {
 	@Override
 	public void close() {
 		try {
-			session.close();
+			if (session != null) {
+				session.close();
+			}
 		} catch (Exception e) {
 			LOG.error("Error while closing session.", e);
 		}
 		try {
-			cluster.close();
+			if (cluster != null) {
+				cluster.close();
+			}
 		} catch (Exception e) {
 			LOG.error("Error while closing cluster.", e);
 		}

flink-streaming-connectors/flink-connector-cassandra/src/main/java/org/apache/flink/streaming/connectors/cassandra/CassandraTupleWriteAheadSink.java

@@ -94,12 +94,16 @@ public class CassandraTupleWriteAheadSink<IN extends Tuple> extends GenericWrite
 	public void close() throws Exception {
 		super.close();
 		try {
-			session.close();
+			if (session != null) {
+				session.close();
+			}
 		} catch (Exception e) {
 			LOG.error("Error while closing session.", e);
 		}
 		try {
-			cluster.close();
+			if (cluster != null) {
+				cluster.close();
+			}
 		} catch (Exception e) {
 			LOG.error("Error while closing cluster.", e);
 		}