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

@@ -487,11 +487,15 @@ public ConsumerRebalanceListener createRebalanceListener(final Consumer<K, V> co
 
 				@Override
 				public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
-					getContainerProperties().getConsumerRebalanceListener().onPartitionsRevoked(partitions);
-					// Wait until now to commit, in case the user listener added acks
-					commitPendingAcks();
-					if (ListenerConsumer.this.kafkaTxManager != null) {
-						closeProducers(partitions);
+					try {
+						getContainerProperties().getConsumerRebalanceListener().onPartitionsRevoked(partitions);
+						// Wait until now to commit, in case the user listener added acks
+						commitPendingAcks();
+					}
+					finally {
+						if (ListenerConsumer.this.kafkaTxManager != null) {
+							closeProducers(partitions);
+						}
 					}
 				}