GH-522: Fix NPE in listener container

Fixes #522

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

@@ -97,9 +97,9 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 
 	private final TopicPartitionInitialOffset[] topicPartitions;
 
-	private ListenerConsumer listenerConsumer;
+	private volatile ListenerConsumer listenerConsumer;
 
-	private ListenableFuture<?> listenerConsumerFuture;
+	private volatile ListenableFuture<?> listenerConsumerFuture;
 
 	private GenericMessageListener<?> listener;
 
@@ -179,11 +179,17 @@ public void setClientIdSuffix(String clientIdSuffix) {
 	 * either explicitly or by Kafka; may be null if not assigned yet.
 	 */
 	public Collection<TopicPartition> getAssignedPartitions() {
-		if (this.listenerConsumer.definedPartitions != null) {
-			return Collections.unmodifiableCollection(this.listenerConsumer.definedPartitions.keySet());
-		}
-		else if (this.listenerConsumer.assignedPartitions != null) {
-			return Collections.unmodifiableCollection(this.listenerConsumer.assignedPartitions);
+		ListenerConsumer listenerConsumer = this.listenerConsumer;
+		if (listenerConsumer != null) {
+			if (listenerConsumer.definedPartitions != null) {
+				return Collections.unmodifiableCollection(listenerConsumer.definedPartitions.keySet());
+			}
+			else if (listenerConsumer.assignedPartitions != null) {
+				return Collections.unmodifiableCollection(listenerConsumer.assignedPartitions);
+			}
+			else {
+				return null;
+			}
 		}
 		else {
 			return null;