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

Fix close producers after a rebalance

Browse Source 
With a transactional container if committing the offsets fails
during a rebalance, we would leave the producers open.

Move the close to a finally block.

**cherry-pick to 2.0.x**
This commit is contained in:
Gary Russell
2019-04-11 10:56:59 -04:00
committed by Artem Bilan
parent 994fb29e16
commit 0bbb23a57f
1 changed files with 16 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -472,19 +472,23 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
@Override
public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
if (this.consumerAwareListener != null) {
this.consumerAwareListener.onPartitionsRevokedBeforeCommit(consumer, partitions);
try {
if (this.consumerAwareListener != null) {
this.consumerAwareListener.onPartitionsRevokedBeforeCommit(consumer, partitions);
}
else {
this.userListener.onPartitionsRevoked(partitions);
}
// Wait until now to commit, in case the user listener added acks
commitPendingAcks();
if (this.consumerAwareListener != null) {
this.consumerAwareListener.onPartitionsRevokedAfterCommit(consumer, partitions);
}
}
else {
this.userListener.onPartitionsRevoked(partitions);
}
// Wait until now to commit, in case the user listener added acks
commitPendingAcks();
if (this.consumerAwareListener != null) {
this.consumerAwareListener.onPartitionsRevokedAfterCommit(consumer, partitions);
}
if (ListenerConsumer.this.kafkaTxManager != null) {
closeProducers(partitions);
finally {
if (ListenerConsumer.this.kafkaTxManager != null) {
closeProducers(partitions);
}
}
}