[hotfix][Kafka 0.9] Avoid committing offsets to closed consumer

fkaufer · Mar 21, 2016 · 35892ed · 35892ed
1 parent 6078923
commit 35892ed
Showing 1 changed file with 4 additions and 0 deletions.
diff --git a/...a-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer09.java b/...a-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer09.java
@@ -387,6 +387,10 @@ public void close() throws Exception {
 
 	@Override
 	protected void commitOffsets(HashMap<KafkaTopicPartition, Long> checkpointOffsets) {
+		if(!running) {
+			LOG.warn("Unable to commit offsets on closed consumer");
+			return;
+		}
 		Map<TopicPartition, OffsetAndMetadata> kafkaCheckpointOffsets = convertToCommitMap(checkpointOffsets);
 		synchronized (this.consumer) {
 			this.consumer.commitSync(kafkaCheckpointOffsets);