Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

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

Browse Source 
Fixes https://github.com/spring-projects/spring-kafka/issues/522

* Polishing - PR Comment

**Cherry-pick to 2.0.x & 1.3.x**
This commit is contained in:
Gary Russell
2017-12-18 16:02:01 -05:00
committed by Artem Bilan
parent 0262d87e7c
commit 7d1aa4eadc
1 changed files with 15 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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 class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
* 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 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
public String toString() {
return "KafkaMessageListenerContainer [id=" + getBeanName()
+ (this.clientIdSuffix != null ? ", clientIndex=" + this.clientIdSuffix : "")
+ ", topicPartitions=" + getAssignedPartitions()
+ ", topicPartitions="
+ (getAssignedPartitions() == null ? "none assigned" : getAssignedPartitions())
+ "]";
}