[FLINK-6311] [kinesis] NPE in FlinkKinesisConsumer if source was closed before run

This closes #3738.

flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/internals/KinesisDataFetcher.java

@@ -149,7 +149,7 @@ public class KinesisDataFetcher<T> {
 	private final KinesisProxyInterface kinesis;
 
 	/** Thread that executed runFetcher() */
-	private Thread mainThread;
+	private volatile Thread mainThread;
 
 	/**
 	 * The current number of shards that are actively read by this fetcher.
@@ -408,7 +408,10 @@ public class KinesisDataFetcher<T> {
 	 */
 	public void shutdownFetcher() {
 		running = false;
-		mainThread.interrupt(); // the main thread may be sleeping for the discovery interval
+
+		if (mainThread != null) {
+			mainThread.interrupt(); // the main thread may be sleeping for the discovery interval
+		}
 
 		if (LOG.isInfoEnabled()) {
 			LOG.info("Shutting down the shard consumer threads of subtask {} ...", indexOfThisConsumerSubtask);