Skip to content

Commit d6f2f4c

Browse files
garyrussellartembilan
authored andcommitted
GH-881: Null check in closeProducers()
Fixes #881 `getAssignedPartitions()` can return `null`. # Conflicts: # spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java * Add import for `@Nullable`
1 parent 66a7162 commit d6f2f4c

File tree

1 file changed

+15
-11
lines changed

1 file changed

+15
-11
lines changed

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

Lines changed: 15 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -62,6 +62,7 @@
6262
import org.springframework.kafka.support.TopicPartitionInitialOffset.SeekPosition;
6363
import org.springframework.kafka.support.TransactionSupport;
6464
import org.springframework.kafka.transaction.KafkaTransactionManager;
65+
import org.springframework.lang.Nullable;
6566
import org.springframework.scheduling.SchedulingAwareRunnable;
6667
import org.springframework.scheduling.TaskScheduler;
6768
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
@@ -153,6 +154,7 @@ public void setClientIdSuffix(String clientIdSuffix) {
153154
* @return the {@link TopicPartition}s currently assigned to this container,
154155
* either explicitly or by Kafka; may be null if not assigned yet.
155156
*/
157+
@Nullable
156158
public Collection<TopicPartition> getAssignedPartitions() {
157159
ListenerConsumer listenerConsumer = this.listenerConsumer;
158160
if (listenerConsumer != null) {
@@ -1258,17 +1260,19 @@ public void seekToEnd(String topic, int partition) {
12581260
this.seeks.add(new TopicPartitionInitialOffset(topic, partition, SeekPosition.END));
12591261
}
12601262

1261-
private void closeProducers(Collection<TopicPartition> partitions) {
1262-
ProducerFactory<?, ?> producerFactory = this.kafkaTxManager.getProducerFactory();
1263-
partitions.forEach(tp -> {
1264-
try {
1265-
producerFactory.closeProducerFor(zombieFenceTxIdSuffix(tp.topic(), tp.partition()));
1266-
}
1267-
catch (Exception e) {
1268-
this.logger.error("Failed to close producer with transaction id suffix: "
1269-
+ zombieFenceTxIdSuffix(tp.topic(), tp.partition()), e);
1270-
}
1271-
});
1263+
private void closeProducers(@Nullable Collection<TopicPartition> partitions) {
1264+
if (partitions != null) {
1265+
ProducerFactory<?, ?> producerFactory = this.kafkaTxManager.getProducerFactory();
1266+
partitions.forEach(tp -> {
1267+
try {
1268+
producerFactory.closeProducerFor(zombieFenceTxIdSuffix(tp.topic(), tp.partition()));
1269+
}
1270+
catch (Exception e) {
1271+
this.logger.error("Failed to close producer with transaction id suffix: "
1272+
+ zombieFenceTxIdSuffix(tp.topic(), tp.partition()), e);
1273+
}
1274+
});
1275+
}
12721276
}
12731277

12741278
private String zombieFenceTxIdSuffix(String topic, int partition) {

0 commit comments

Comments
 (0)