test : add read uncommited test

Signed-off-by: Su Ko <rhtn1128@gmail.com>

Author: bandalgomsu

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

@@ -1232,4 +1232,69 @@ public void afterRecord(
 		container.stop();
 		pf.destroy();
 	}
+
+	@SuppressWarnings("unchecked")
+	@Test
+	void testFixTxOffsetsWithReadUncommitted() throws Exception {
+		Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "txReadUncommittedTest", false);
+		props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_uncommitted");
+		DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
+		ContainerProperties containerProps = new ContainerProperties(topic7);
+		containerProps.setGroupId("txReadUncommittedTest");
+		containerProps.setPollTimeout(500L);
+		containerProps.setIdleEventInterval(500L);
+		containerProps.setFixTxOffsets(true);
+
+		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
+		DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
+		pf.setTransactionIdPrefix("readUncommitted.");
+
+		final KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
+		final AtomicInteger messageCount = new AtomicInteger();
+		final CountDownLatch latch = new CountDownLatch(3);
+		containerProps.setMessageListener((MessageListener<Integer, String>) message -> {
+			messageCount.incrementAndGet();
+			latch.countDown();
+		});
+
+		@SuppressWarnings({ "rawtypes" })
+		KafkaTransactionManager tm = new KafkaTransactionManager(pf);
+		containerProps.setKafkaAwareTransactionManager(tm);
+		KafkaMessageListenerContainer<Integer, String> container =
+				new KafkaMessageListenerContainer<>(cf, containerProps);
+		container.setBeanName("testFixTxOffsetsWithReadUncommitted");
+
+		AtomicReference<Map<TopicPartition, OffsetAndMetadata>> committed = new AtomicReference<>();
+		CountDownLatch idleLatch = new CountDownLatch(1);
+		container.setApplicationEventPublisher(event -> {
+			if (event instanceof ListenerContainerIdleEvent) {
+				Consumer<?, ?> consumer = ((ListenerContainerIdleEvent) event).getConsumer();
+				committed.set(consumer.committed(
+						Collections.singleton(new TopicPartition(topic7, 0))));
+				idleLatch.countDown();
+			}
+		});
+
+		container.start();
+
+		template.setDefaultTopic(topic7);
+		template.executeInTransaction(t -> {
+			template.sendDefault(0, 0, "msg1");
+			template.sendDefault(0, 0, "msg2");
+			template.sendDefault(0, 0, "msg3");
+			return null;
+		});
+
+		assertThat(latch.await(60, TimeUnit.SECONDS)).isTrue();
+		assertThat(idleLatch.await(10, TimeUnit.SECONDS)).isTrue();
+
+		assertThat(messageCount.get()).isGreaterThanOrEqualTo(3);
+		TopicPartition partition0 = new TopicPartition(topic7, 0);
+		assertThat(committed.get().get(partition0)).isNotNull();
+		assertThat(committed.get().get(partition0).offset()).isGreaterThanOrEqualTo(3L);
+
+		container.stop();
+		pf.destroy();
+	}
+
 }