X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Ftest%2Fjava%2Fde%2Fjuplo%2Fkafka%2FApplicationTests.java;h=a632a89c41f936b88f1f23742e78183279138ef3;hb=b3777fba0ae679d9e2c9d36626fa208a952f83e8;hp=caa25c54313ae2fd4b01e4820bcf135c62d57c5d;hpb=83a4bf324f5a7ec6010a7921118ec7d6e8f997cf;p=demos%2Fkafka%2Ftraining

diff --git a/src/test/java/de/juplo/kafka/ApplicationTests.java b/src/test/java/de/juplo/kafka/ApplicationTests.java
index caa25c5..a632a89 100644
--- a/src/test/java/de/juplo/kafka/ApplicationTests.java
+++ b/src/test/java/de/juplo/kafka/ApplicationTests.java
@@ -26,7 +26,6 @@ import java.util.*;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.function.BiConsumer;
-import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
@@ -43,6 +42,7 @@ import static org.awaitility.Awaitility.*;
 		properties = {
 				"consumer.bootstrap-server=${spring.embedded.kafka.brokers}",
 				"consumer.topic=" + TOPIC,
+				"consumer.commit-interval=1s",
 				"spring.mongodb.embedded.version=4.4.13" })
 @EmbeddedKafka(topics = TOPIC, partitions = PARTITIONS)
 @EnableAutoConfiguration
@@ -63,15 +63,18 @@ class ApplicationTests
 	@Autowired
 	KafkaConsumer<String, Long> kafkaConsumer;
 	@Autowired
-	KafkaConsumer<Bytes, Bytes> offsetConsumer;
+	PartitionStatisticsRepository partitionStatisticsRepository;
 	@Autowired
 	ApplicationProperties properties;
 	@Autowired
 	ExecutorService executor;
 	@Autowired
 	PartitionStatisticsRepository repository;
+	@Autowired
+	KeyCountingRebalanceListener keyCountingRebalanceListener;
+	@Autowired
+	KeyCountingRecordHandler keyCountingRecordHandler;
 
-	Consumer<ConsumerRecord<String, Long>> testHandler;
 	EndlessConsumer<String, Long> endlessConsumer;
 	Map<TopicPartition, Long> oldOffsets;
 	Map<TopicPartition, Long> newOffsets;
@@ -177,10 +180,13 @@ class ApplicationTests
 
 	void doForCurrentOffsets(BiConsumer<TopicPartition, Long> consumer)
 	{
-		offsetConsumer.assign(partitions());
-		partitions().forEach(tp -> consumer.accept(tp, offsetConsumer.position(tp)));
-		offsetConsumer.unsubscribe();
-	}
+		partitions().forEach(tp ->
+		{
+			String partition = Integer.toString(tp.partition());
+			Optional<Long> offset = partitionStatisticsRepository.findById(partition).map(document -> document.offset);
+			consumer.accept(tp, offset.orElse(0l));
+		});
+		}
 
 	List<TopicPartition> partitions()
 	{
@@ -237,8 +243,6 @@ class ApplicationTests
 	@BeforeEach
 	public void init()
 	{
-		testHandler = record -> {} ;
-
 		oldOffsets = new HashMap<>();
 		newOffsets = new HashMap<>();
 		receivedRecords = new HashSet<>();
@@ -249,23 +253,25 @@ class ApplicationTests
 			newOffsets.put(tp, offset - 1);
 		});
 
-		Consumer<ConsumerRecord<String, Long>> captureOffsetAndExecuteTestHandler =
-				record ->
-				{
-					newOffsets.put(
-							new TopicPartition(record.topic(), record.partition()),
-							record.offset());
-					receivedRecords.add(record);
-					testHandler.accept(record);
+		TestRecordHandler<String, Long> captureOffsetAndExecuteTestHandler =
+				new TestRecordHandler<String, Long>(keyCountingRecordHandler) {
+					@Override
+					public void onNewRecord(ConsumerRecord<String, Long> record)
+					{
+						newOffsets.put(
+								new TopicPartition(record.topic(), record.partition()),
+								record.offset());
+						receivedRecords.add(record);
+					}
 				};
 
 		endlessConsumer =
 				new EndlessConsumer<>(
 						executor,
-						repository,
 						properties.getClientId(),
 						properties.getTopic(),
 						kafkaConsumer,
+						keyCountingRebalanceListener,
 						captureOffsetAndExecuteTestHandler);
 
 		endlessConsumer.start();
@@ -306,18 +312,5 @@ class ApplicationTests
 
 			return new KafkaProducer<>(props);
 		}
-
-		@Bean
-		KafkaConsumer<Bytes, Bytes> offsetConsumer(ApplicationProperties properties)
-		{
-			Properties props = new Properties();
-			props.put("bootstrap.servers", properties.getBootstrapServer());
-			props.put("client.id", "OFFSET-CONSUMER");
-			props.put("group.id", properties.getGroupId());
-			props.put("key.deserializer", BytesDeserializer.class.getName());
-			props.put("value.deserializer", BytesDeserializer.class.getName());
-
-			return new KafkaConsumer<>(props);
-		}
 	}
 }