From 86d8373084fa71f1fede3b029d0424df9e1fa9d0 Mon Sep 17 00:00:00 2001
From: Mark <markinnork7@gmail.com>
Date: Tue, 30 Jan 2018 19:42:32 +0200
Subject: [PATCH] Polish synchronization in KafkaMessageSource

* Remove volatile from running flag

As running is always accessed through `KafkaMessageSource` monitor use existing `synchronized` to guard access to it.
Also move setting `running=true` from consumerMonitor in `createConsumer()` method to `doReceive()` as it's guarded by needed lock

* Extract common stopping consumer logic
---
 .../kafka/inbound/KafkaMessageSource.java     | 25 +++++++++----------
 1 file changed, 12 insertions(+), 13 deletions(-)

diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java
index 666586b6e3..5a0bc32601 100644
--- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java
+++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java
@@ -72,6 +72,7 @@ import org.springframework.util.Assert;
  * @param <V> the value type.
  *
  * @author Gary Russell
+ * @author Mark Norkin
  * @since 3.0.1
  *
  */
@@ -108,7 +109,7 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object>
 
 	private volatile Consumer<K, V> consumer;
 
-	private volatile boolean running;
+	private boolean running;
 
 	public KafkaMessageSource(ConsumerFactory<K, V> consumerFactory, String... topics) {
 		this(consumerFactory, new KafkaAckCallbackFactory<>(), topics);
@@ -259,12 +260,7 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object>
 
 	@Override
 	public synchronized void stop() {
-		synchronized (this.consumerMonitor) {
-			if (this.consumer != null) {
-				this.consumer.close(30, TimeUnit.SECONDS);
-				this.consumer = null;
-			}
-		}
+		stopConsumer();
 		this.running = false;
 	}
 
@@ -272,6 +268,7 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object>
 	protected synchronized Object doReceive() {
 		if (this.consumer == null) {
 			createConsumer();
+			this.running = true;
 		}
 		ConsumerRecord<K, V> record;
 		TopicPartition topicPartition;
@@ -333,17 +330,19 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object>
 				}
 
 			});
-			this.running = true;
 		}
 	}
 
 	@Override
 	public synchronized void destroy() {
-		if (this.consumer != null) {
-			Consumer<K, V> consumer2 = this.consumer;
-			this.consumer = null;
-			synchronized (this.consumerMonitor) {
-				consumer2.close(30, TimeUnit.SECONDS);
+		stopConsumer();
+	}
+
+	private void stopConsumer() {
+		synchronized (this.consumerMonitor) {
+			if (this.consumer != null) {
+				this.consumer.close(30, TimeUnit.SECONDS);
+				this.consumer = null;
 			}
 		}
 	}