[FLINK-3081] Properly stop periodic Kafka committer

This closes #1410

flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer.java

@@ -414,14 +414,18 @@ public class FlinkKafkaConsumer<T> extends RichParallelSourceFunction<T>
 				// same here.
 				long commitInterval = Long.valueOf(props.getProperty("auto.commit.interval.ms", "60000"));
 				offsetCommitter = new PeriodicOffsetCommitter(commitInterval, this);
+				offsetCommitter.setDaemon(true);
 				offsetCommitter.start();
 				LOG.info("Starting periodic offset committer, with commit interval of {}ms", commitInterval);
 			}
 
-			fetcher.run(sourceContext, deserializer, lastOffsets);
-
-			if (offsetCommitter != null) {
-				offsetCommitter.close();
+			try {
+				fetcher.run(sourceContext, deserializer, lastOffsets);
+			} finally {
+				if (offsetCommitter != null) {
+					offsetCommitter.close();
+					offsetCommitter.join();
+				}
 			}
 		}
 		else {