Fix test for previous back ported fix

Author: garyrussell

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

@@ -1755,8 +1755,7 @@ public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
 						commitPendingAcks();
 					}
 					catch (Exception e) {
-						ListenerConsumer.this.logger.error(e, () -> "Fatal commit error after revocation "
-								+ partitions);
+						ListenerConsumer.this.logger.error("Fatal commit error after revocation " + partitions, e);
 					}
 					if (this.consumerAwareListener != null) {
 						this.consumerAwareListener.onPartitionsRevokedAfterCommit(ListenerConsumer.this.consumer,

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

@@ -2260,7 +2260,7 @@ public void testCommitErrorHandlerCalled() throws Exception {
 
 	@SuppressWarnings({ "unchecked", "rawtypes" })
 	@Test
-	void testCommitFailsOnRevoke() throws Exception {
+	public 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);
@@ -2271,7 +2271,8 @@ void testCommitFailsOnRevoke() throws Exception {
 		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(
+		TopicPartition topicPartition1 = new TopicPartition("foo", 1);
+		records.put(topicPartition1, Arrays.asList(
 				new ConsumerRecord<>("foo", 1, 0L, 1, "foo"),
 				new ConsumerRecord<>("foo", 1, 1L, 1, "bar")));
 		ConsumerRecords<Integer, String> consumerRecords = new ConsumerRecords<>(records);
@@ -2289,7 +2290,7 @@ void testCommitFailsOnRevoke() throws Exception {
 			}
 			else if (call == 1) {
 				rebal.get().onPartitionsRevoked(Collections.singletonList(topicPartition0));
-				rebal.get().onPartitionsAssigned(Collections.emptyList());
+				rebal.get().onPartitionsAssigned(Collections.singletonList(topicPartition1));
 			}
 			latch.countDown();
 			return first.getAndSet(false) ? consumerRecords : emptyRecords;
@@ -2299,11 +2300,10 @@ else if (call == 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)) {
+			if (commitCount.incrementAndGet() == 2) {
 				throw new CommitFailedException();
 			}
 			return null;
@@ -2314,6 +2314,7 @@ else if (call == 1) {
 		containerProps.setClientId("clientId");
 		containerProps.setIdleEventInterval(100L);
 		AtomicReference<Acknowledgment> acknowledgment = new AtomicReference<>();
+		AtomicBoolean listenerCalled = new AtomicBoolean();
 		class AckListener implements AcknowledgingMessageListener {
 			// not a lambda https://bugs.openjdk.java.net/browse/JDK-8074381
 
@@ -2339,14 +2340,18 @@ public void onPartitionsRevokedBeforeCommit(Consumer<?, ?> consumer,
 				}
 			}
 
+			@Override
+			public void onPartitionsRevokedAfterCommit(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
+				listenerCalled.set(true);
+			}
+
 		});
-		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);
+		assertThat(listenerCalled.get()).isTrue();
 		container.stop();
 	}