GH-522: Fix NPE in listener container (#524)

Fixes #522

* Polishing - PR Comment

**Cherry-pick to 2.0.x & 1.3.x**

Author: garyrussell

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

@@ -98,9 +98,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;
 
@@ -180,11 +180,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;
@@ -294,7 +300,8 @@ private void publishNonResponsiveConsumerEvent(long timeSinceLastPoll, Consumer<
 	public String toString() {
 		return "KafkaMessageListenerContainer [id=" + getBeanName()
 				+ (this.clientIdSuffix != null ? ", clientIndex=" + this.clientIdSuffix : "")
-				+ ", topicPartitions=" + getAssignedPartitions()
+				+ ", topicPartitions="
+				+ (getAssignedPartitions() == null ? "none assigned" : getAssignedPartitions())
 				+ "]";
 	}