Fix close producers after a rebalance

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.

Author: garyrussell

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

@@ -1720,21 +1720,25 @@ private class ListenerConsumerRebalanceListener implements ConsumerRebalanceList
 
 			@Override
 			public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
-				if (this.consumerAwareListener != null) {
-					this.consumerAwareListener.onPartitionsRevokedBeforeCommit(ListenerConsumer.this.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(ListenerConsumer.this.consumer,
-							partitions);
+				try {
+					if (this.consumerAwareListener != null) {
+						this.consumerAwareListener.onPartitionsRevokedBeforeCommit(ListenerConsumer.this.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(ListenerConsumer.this.consumer,
+								partitions);
+					}
 				}
-				if (ListenerConsumer.this.kafkaTxManager != null) {
-					closeProducers(partitions);
+				finally {
+					if (ListenerConsumer.this.kafkaTxManager != null) {
+						closeProducers(partitions);
+					}
 				}
 			}