Deprecation Code Cleanup

* Because `transactionManager` property is deprecated in ContainerProperties via a
  previous commit, there are some test variants that we may not need any longer.
  Cleaning them up and updating the tests where we were still using the deprecated transaction manager.

Author: sobychacko

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

@@ -72,12 +72,12 @@
 import org.springframework.kafka.transaction.KafkaAwareTransactionManager;
 import org.springframework.lang.Nullable;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
-import org.springframework.transaction.PlatformTransactionManager;
 import org.springframework.transaction.support.TransactionSynchronizationManager;
 
 /**
  * @author Gary Russell
  * @author Wang Zhiyang
+ * @author Soby Chacko
  *
  * @since 2.2.4
  *
@@ -662,103 +662,6 @@ public void failure(ConsumerRecords records, Exception exception, Consumer consu
 		}
 	}
 
-	@Test
-	@SuppressWarnings({ "rawtypes", "unchecked" })
-	void testInterceptInTxNonKafkaTM() throws InterruptedException {
-		ConsumerFactory consumerFactory = mock(ConsumerFactory.class);
-		final Consumer consumer = mock(Consumer.class);
-		TopicPartition tp0 = new TopicPartition("foo", 0);
-		ConsumerRecord record1 = new ConsumerRecord("foo", 0, 0L, "bar", "baz");
-		ConsumerRecords records = new ConsumerRecords(
-				Collections.singletonMap(tp0, Collections.singletonList(record1)));
-		ConsumerRecords empty = new ConsumerRecords(Collections.emptyMap());
-		AtomicInteger firstOrSecondPoll = new AtomicInteger();
-		willAnswer(invocation -> {
-			Thread.sleep(10);
-			return firstOrSecondPoll.incrementAndGet() < 2 ? records : empty;
-		}).given(consumer).poll(any());
-		List<TopicPartition> assignments = List.of(tp0);
-		willAnswer(invocation -> {
-			((ConsumerRebalanceListener) invocation.getArgument(1))
-				.onPartitionsAssigned(assignments);
-			return null;
-		}).given(consumer).subscribe(any(Collection.class), any());
-		given(consumer.position(any())).willReturn(0L);
-		given(consumerFactory.createConsumer("grp", "", "-0", KafkaTestUtils.defaultPropertyOverrides()))
-			.willReturn(consumer);
-		ContainerProperties containerProperties = new ContainerProperties("foo");
-		containerProperties.setGroupId("grp");
-		containerProperties.setMessageListener((MessageListener) rec -> {
-		});
-		containerProperties.setMissingTopicsFatal(false);
-		List<String> order = new ArrayList<>();
-		AtomicReference<CountDownLatch> latch = new AtomicReference<>(new CountDownLatch(2));
-		PlatformTransactionManager tm = mock(PlatformTransactionManager.class);
-		willAnswer(inv -> {
-			order.add("tx");
-			latch.get().countDown();
-			return null;
-		}).given(tm).getTransaction(any());
-		containerProperties.setTransactionManager(tm);
-		ConcurrentMessageListenerContainer container = new ConcurrentMessageListenerContainer(consumerFactory,
-				containerProperties);
-		AtomicReference<CountDownLatch> successCalled = new AtomicReference<>(new CountDownLatch(1));
-		container.setRecordInterceptor(new RecordInterceptor() {
-
-			@Override
-			@Nullable
-			public ConsumerRecord intercept(ConsumerRecord rec, Consumer consumer) {
-				order.add("interceptor");
-				latch.get().countDown();
-				return rec;
-			}
-
-			@Override
-			public void success(ConsumerRecord record, Consumer consumer) {
-				order.add("success");
-				successCalled.get().countDown();
-			}
-
-		});
-		container.setBatchInterceptor(new BatchInterceptor() {
-
-			@Override
-			@Nullable
-			public ConsumerRecords intercept(ConsumerRecords recs, Consumer consumer) {
-				order.add("b.interceptor");
-				latch.get().countDown();
-				return new ConsumerRecords(Collections.singletonMap(tp0, Collections.singletonList(record1)));
-			}
-
-			@Override
-			public void success(ConsumerRecords records, Consumer consumer) {
-				order.add("b.success");
-				successCalled.get().countDown();
-			}
-
-		});
-		container.setInterceptBeforeTx(false);
-		container.start();
-		try {
-			assertThat(latch.get().await(10, TimeUnit.SECONDS)).isTrue();
-			assertThat(successCalled.get().await(10, TimeUnit.SECONDS)).isTrue();
-			assertThat(order).containsExactly("tx", "interceptor", "success");
-			container.stop();
-			latch.set(new CountDownLatch(2));
-			successCalled.set(new CountDownLatch(1));
-			container.getContainerProperties().setMessageListener((BatchMessageListener) recs -> {
-			});
-			firstOrSecondPoll.set(0);
-			container.start();
-			assertThat(latch.get().await(10, TimeUnit.SECONDS)).isTrue();
-			assertThat(successCalled.get().await(10, TimeUnit.SECONDS)).isTrue();
-			assertThat(order).containsExactly("tx", "interceptor", "success", "tx", "b.interceptor", "b.success");
-		}
-		finally {
-			container.stop();
-		}
-	}
-
 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	@Test
 	void testNoCommitOnAssignmentWithEarliest() throws InterruptedException {

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

@@ -102,6 +102,7 @@ records, mock(Consumer.class), mock(MessageListenerContainer.class), mock(Runnab
 		assertThat(output).contains("Record not found in batch: topic-42@123;");
 	}
 
+	@SuppressWarnings({ "rawtypes", "unchecked" })
 	@Test
 	void testExceptionDuringCommit() {
 		CommonErrorHandler mockEH = mock(CommonErrorHandler.class);

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

@@ -125,10 +125,10 @@
 import org.springframework.kafka.test.context.EmbeddedKafka;
 import org.springframework.kafka.test.utils.ContainerTestUtils;
 import org.springframework.kafka.test.utils.KafkaTestUtils;
+import org.springframework.kafka.transaction.KafkaAwareTransactionManager;
 import org.springframework.lang.NonNull;
 import org.springframework.lang.Nullable;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
-import org.springframework.transaction.PlatformTransactionManager;
 import org.springframework.util.backoff.FixedBackOff;
 
 /**
@@ -3899,7 +3899,7 @@ public void testInvokeRecordInterceptorAllSkipped(AckMode ackMode, boolean early
 		});
 		containerProps.setClientId("clientId");
 		if (early) {
-			containerProps.setTransactionManager(mock(PlatformTransactionManager.class));
+			containerProps.setKafkaAwareTransactionManager(mock(KafkaAwareTransactionManager.class));
 		}
 
 		RecordInterceptor<Integer, String> recordInterceptor = spy(new RecordInterceptor<Integer, String>() {
@@ -3982,7 +3982,7 @@ public void testInvokeBatchInterceptorAllSkipped(boolean early) throws Exception
 		});
 		containerProps.setClientId("clientId");
 		if (early) {
-			containerProps.setTransactionManager(mock(PlatformTransactionManager.class));
+			containerProps.setKafkaAwareTransactionManager(mock(KafkaAwareTransactionManager.class));
 		}
 
 		BatchInterceptor<Integer, String> interceptor = spy(new BatchInterceptor<Integer, String>() {

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

@@ -95,16 +95,14 @@
 import org.springframework.kafka.transaction.KafkaTransactionManager;
 import org.springframework.messaging.MessageHeaders;
 import org.springframework.transaction.TransactionDefinition;
-import org.springframework.transaction.TransactionException;
-import org.springframework.transaction.support.AbstractPlatformTransactionManager;
 import org.springframework.transaction.support.DefaultTransactionDefinition;
-import org.springframework.transaction.support.DefaultTransactionStatus;
 import org.springframework.util.backoff.FixedBackOff;
 
 /**
  * @author Gary Russell
  * @author Artem Bilan
  * @author Wang Zhiyang
+ * @author Soby Chacko
  *
  * @since 1.3
  *
@@ -462,7 +460,7 @@ public void testConsumeAndProduceTransactionExternalTM() throws Exception {
 		given(pf.createProducer(isNull())).willReturn(producer);
 		ContainerProperties props = new ContainerProperties(new TopicPartitionOffset("foo", 0));
 		props.setGroupId("group");
-		props.setTransactionManager(new SomeOtherTransactionManager());
+		props.setKafkaAwareTransactionManager(new KafkaTransactionManager<>(pf));
 		final KafkaTemplate template = new KafkaTemplate(pf);
 		ConsumerGroupMetadata meta = mock(ConsumerGroupMetadata.class);
 		props.setMessageListener((MessageListener<String, String>) m -> {
@@ -595,11 +593,6 @@ public void testFixLagKTM() throws InterruptedException {
 		testFixLagGuts(topic6, 1);
 	}
 
-	@Test
-	public void testFixLagOtherTM() throws InterruptedException {
-		testFixLagGuts(topic7, 2);
-	}
-
 	@SuppressWarnings({"unchecked"})
 	private void testFixLagGuts(String topic, int whichTm) throws InterruptedException {
 		Map<String, Object> props = KafkaTestUtils.consumerProps("txTest2", "false", embeddedKafka);
@@ -619,8 +612,6 @@ private void testFixLagGuts(String topic, int whichTm) throws InterruptedExcepti
 		case 1:
 			containerProps.setKafkaAwareTransactionManager(new KafkaTransactionManager<>(pf));
 			break;
-		case 2:
-			containerProps.setTransactionManager(new SomeOtherTransactionManager());
 		}
 
 		final KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
@@ -1082,28 +1073,4 @@ void testNoAfterRollbackWhenFenced() throws Exception {
 		assertThatIllegalStateException().isThrownBy(container::start);
 	}
 
-	public static class SomeOtherTransactionManager extends AbstractPlatformTransactionManager {
-
-		@Override
-		protected Object doGetTransaction() throws TransactionException {
-			return new Object();
-		}
-
-		@Override
-		protected void doBegin(Object transaction, TransactionDefinition definition) throws TransactionException {
-			//noop
-		}
-
-		@Override
-		protected void doCommit(DefaultTransactionStatus status) throws TransactionException {
-			//noop
-		}
-
-		@Override
-		protected void doRollback(DefaultTransactionStatus status) throws TransactionException {
-			//noop
-		}
-
-	}
-
 }