GH-2489: Retry Commits If Necessary on Revoke

Closes #2489

* Retry commits that have failed temporarily due to rebalance in progress when onPartitionsRevoked is called.
* Adjust expectations for unit tests where commits are retried (previous expectation accepted the defect that failed commits for subsequently revoked partitions were not retried)

Author: mikael-carlstedt

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

@@ -3446,6 +3446,7 @@ public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
 					}
 					try {
 						// Wait until now to commit, in case the user listener added acks
+						checkRebalanceCommits();
 						commitPendingAcks();
 						fixTxOffsetsIfNeeded();
 					}

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

@@ -3351,7 +3351,7 @@ void testCommitRebalanceInProgressBatch() throws Exception {
 			assertThat(commits).hasSize(3);
 			assertThat(commits.get(0)).hasSize(2); // assignment
 			assertThat(commits.get(1)).hasSize(2); // batch commit
-			assertThat(commits.get(2)).hasSize(1); // re-commit
+			assertThat(commits.get(2)).hasSize(2); // GH-2489: offsets for both partition should be re-committed before partition 1 is revoked
 		});
 	}
 
@@ -3364,7 +3364,7 @@ void testCommitRebalanceInProgressRecord() throws Exception {
 			assertThat(commits.get(2)).hasSize(1);
 			assertThat(commits.get(3)).hasSize(1);
 			assertThat(commits.get(4)).hasSize(1);
-			assertThat(commits.get(5)).hasSize(1); // re-commit
+			assertThat(commits.get(5)).hasSize(2); // GH-2489: offsets for both partition should be re-committed before partition 1 is revoked
 			assertThat(commits.get(5).get(new TopicPartition("foo", 1)))
 				.isNotNull()
 				.extracting(om -> om.offset())