Fix commits on partition revocation

Complete the remainder of `onPartitionsRevoked()` if the commits fail
fatally.

It is possible that commits might fail when using manual acks; we must
complete the remainder of the method if the commits fail.

**I will backport as needed**

Author: garyrussell

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

@@ -2267,8 +2267,14 @@ public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
 					else {
 						this.userListener.onPartitionsRevoked(partitions);
 					}
-					// Wait until now to commit, in case the user listener added acks
-					commitPendingAcks();
+					try {
+						// Wait until now to commit, in case the user listener added acks
+						commitPendingAcks();
+					}
+					catch (Exception e) {
+						ListenerConsumer.this.logger.error(e, () -> "Fatal commit error after revocation "
+								+ partitions);
+					}
 					if (this.consumerAwareListener != null) {
 						this.consumerAwareListener.onPartitionsRevokedAfterCommit(ListenerConsumer.this.consumer,
 								partitions);

spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java

@@ -58,6 +58,7 @@
 import java.util.stream.Collectors;
 
 import org.apache.commons.logging.LogFactory;
+import org.apache.kafka.clients.consumer.CommitFailedException;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
@@ -2747,6 +2748,98 @@ void testFatalErrorOnFencedInstanceException() throws Exception {
 		container.stop();
 	}
 
+	@SuppressWarnings({ "unchecked", "rawtypes" })
+	@Test
+	void testCommitFailsOnRevoke() throws Exception {
+		ConsumerFactory<Integer, String> cf = mock(ConsumerFactory.class);
+		Consumer<Integer, String> consumer = mock(Consumer.class);
+		given(cf.createConsumer(eq("grp"), eq("clientId"), isNull(), any())).willReturn(consumer);
+		Map<String, Object> cfProps = new LinkedHashMap<>();
+		given(cf.getConfigurationProperties()).willReturn(cfProps);
+		final Map<TopicPartition, List<ConsumerRecord<Integer, String>>> records = new HashMap<>();
+		TopicPartition topicPartition0 = new TopicPartition("foo", 0);
+		records.put(topicPartition0, Arrays.asList(
+				new ConsumerRecord<>("foo", 0, 0L, 1, "foo"),
+				new ConsumerRecord<>("foo", 0, 1L, 1, "bar")));
+		records.put(new TopicPartition("foo", 1), Arrays.asList(
+				new ConsumerRecord<>("foo", 1, 0L, 1, "foo"),
+				new ConsumerRecord<>("foo", 1, 1L, 1, "bar")));
+		ConsumerRecords<Integer, String> consumerRecords = new ConsumerRecords<>(records);
+		ConsumerRecords<Integer, String> emptyRecords = new ConsumerRecords<>(Collections.emptyMap());
+		AtomicBoolean first = new AtomicBoolean(true);
+		AtomicInteger rebalance = new AtomicInteger();
+		AtomicReference<ConsumerRebalanceListener> rebal = new AtomicReference<>();
+		CountDownLatch latch = new CountDownLatch(2);
+		given(consumer.poll(any(Duration.class))).willAnswer(i -> {
+			Thread.sleep(50);
+			int call = rebalance.getAndIncrement();
+			if (call == 0) {
+				rebal.get().onPartitionsRevoked(Collections.emptyList());
+				rebal.get().onPartitionsAssigned(records.keySet());
+			}
+			else if (call == 1) {
+				rebal.get().onPartitionsRevoked(Collections.singletonList(topicPartition0));
+				rebal.get().onPartitionsAssigned(Collections.emptyList());
+			}
+			latch.countDown();
+			return first.getAndSet(false) ? consumerRecords : emptyRecords;
+		});
+		willAnswer(invoc -> {
+			rebal.set(invoc.getArgument(1));
+			return null;
+		}).given(consumer).subscribe(any(Collection.class), any(ConsumerRebalanceListener.class));
+		List<Map<TopicPartition, OffsetAndMetadata>> commits = new ArrayList<>();
+		AtomicBoolean firstCommit = new AtomicBoolean(true);
+		AtomicInteger commitCount = new AtomicInteger();
+		willAnswer(invoc -> {
+			commits.add(invoc.getArgument(0, Map.class));
+			if (!firstCommit.getAndSet(false)) {
+				throw new CommitFailedException();
+			}
+			return null;
+		}).given(consumer).commitSync(any(), any());
+		ContainerProperties containerProps = new ContainerProperties("foo");
+		containerProps.setGroupId("grp");
+		containerProps.setAckMode(AckMode.MANUAL);
+		containerProps.setClientId("clientId");
+		containerProps.setIdleEventInterval(100L);
+		AtomicReference<Acknowledgment> acknowledgment = new AtomicReference<>();
+		class AckListener implements AcknowledgingMessageListener {
+			// not a lambda https://bugs.openjdk.java.net/browse/JDK-8074381
+
+			@Override
+			public void onMessage(ConsumerRecord data, Acknowledgment ack) {
+				acknowledgment.set(ack);
+			}
+
+			@Override
+			public void onMessage(Object data) {
+			}
+
+		}
+		containerProps.setMessageListener(new AckListener());
+		containerProps.setConsumerRebalanceListener(new ConsumerAwareRebalanceListener() {
+
+			@Override
+			public void onPartitionsRevokedBeforeCommit(Consumer<?, ?> consumer,
+					Collection<TopicPartition> partitions) {
+
+				if (acknowledgment.get() != null) {
+					acknowledgment.get().acknowledge();
+				}
+			}
+
+		});
+		Properties consumerProps = new Properties();
+		containerProps.setKafkaConsumerProperties(consumerProps);
+		KafkaMessageListenerContainer<Integer, String> container =
+				new KafkaMessageListenerContainer<>(cf, containerProps);
+		container.start();
+		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(container.getAssignedPartitions()).hasSize(1);
+		container.stop();
+	}
+
 	private Consumer<?, ?> spyOnConsumer(KafkaMessageListenerContainer<Integer, String> container) {
 		Consumer<?, ?> consumer = spy(
 				KafkaTestUtils.getPropertyValue(container, "listenerConsumer.consumer", Consumer.class));