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

@@ -3490,6 +3490,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

@@ -3366,7 +3366,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
 		});
 	}
 
@@ -3379,7 +3379,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())