From 38390e893d4129c77c1d0bb43a9d864842b7b4e1 Mon Sep 17 00:00:00 2001
From: Artem Bilan <abilan@vmware.com>
Date: Tue, 26 Oct 2021 14:06:13 -0400
Subject: [PATCH] Restore generic args for anonymous classes

https://build.spring.io/browse/SK-K25X-123/
---
 .../springframework/kafka/core/KafkaTemplateTests.java   | 9 +++++----
 .../ReactiveKafkaProducerTemplateIntegrationTests.java   | 1 -
 2 files changed, 5 insertions(+), 5 deletions(-)

diff --git a/spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java b/spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java
index c88fadba..d74f3cc3 100644
--- a/spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java
+++ b/spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java
@@ -102,7 +102,8 @@ public class KafkaTemplateTests {
 
 	private static Consumer<Integer, String> consumer;
 
-	private static final ProducerFactory.Listener<String, String> noopListener = new ProducerFactory.Listener<>() {
+	private static final ProducerFactory.Listener<String, String> noopListener =
+			new ProducerFactory.Listener<String, String>() {
 
 		@Override
 		public void producerAdded(String id, Producer<String, String> producer) {
@@ -357,7 +358,7 @@ public class KafkaTemplateTests {
 		template.flush();
 		final CountDownLatch latch = new CountDownLatch(1);
 		final AtomicReference<SendResult<Integer, String>> theResult = new AtomicReference<>();
-		future.addCallback(new ListenableFutureCallback<>() {
+		future.addCallback(new ListenableFutureCallback<SendResult<Integer, String>>() {
 
 			@Override
 			public void onSuccess(SendResult<Integer, String> result) {
@@ -391,7 +392,7 @@ public class KafkaTemplateTests {
 		final CountDownLatch latch = new CountDownLatch(1);
 		final AtomicReference<SendResult<Integer, String>> theResult = new AtomicReference<>();
 		AtomicReference<String> value = new AtomicReference<>();
-		future.addCallback(new KafkaSendCallback<>() {
+		future.addCallback(new KafkaSendCallback<Integer, String>() {
 
 			@Override
 			public void onSuccess(SendResult<Integer, String> result) {
@@ -499,7 +500,7 @@ public class KafkaTemplateTests {
 	@Test
 	void testConfigOverridesWithCustomProducerFactory() {
 		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
-		ProducerFactory<String, String> pf = new ProducerFactory<>() {
+		ProducerFactory<String, String> pf = new ProducerFactory<String, String>() {
 
 			@Override
 			public Producer<String, String> createProducer() {
diff --git a/spring-kafka/src/test/java/org/springframework/kafka/core/reactive/ReactiveKafkaProducerTemplateIntegrationTests.java b/spring-kafka/src/test/java/org/springframework/kafka/core/reactive/ReactiveKafkaProducerTemplateIntegrationTests.java
index 24050e21..d3fc8cf9 100644
--- a/spring-kafka/src/test/java/org/springframework/kafka/core/reactive/ReactiveKafkaProducerTemplateIntegrationTests.java
+++ b/spring-kafka/src/test/java/org/springframework/kafka/core/reactive/ReactiveKafkaProducerTemplateIntegrationTests.java
@@ -408,7 +408,6 @@ public class ReactiveKafkaProducerTemplateIntegrationTests {
 		Mono<Void> sendWithFlushMono = reactiveKafkaProducerTemplate
 				.send(Mono.just(SenderRecord
 						.create(new ProducerRecord<>(REACTIVE_INT_KEY_TOPIC, DEFAULT_KEY, DEFAULT_VALUE), null)))
-				.then(reactiveKafkaProducerTemplate.flush())
 				.then();
 
 		StepVerifier.create(sendWithFlushMono)