From: Kai Moritz <kai@juplo.de>
Date: Sun, 14 Aug 2022 09:32:10 +0000 (+0200)
Subject: Typisierung in `GenericApplicationTest` nur noch, wo wirklich nötig
X-Git-Tag: sumup-adder---lvm-2-tage~9^2~8
X-Git-Url: http://juplo.de/gitweb/?a=commitdiff_plain;h=80f616369c011db99eddf42c6ee91e66fd1dfd07;p=demos%2Fkafka%2Ftraining

Typisierung in `GenericApplicationTest` nur noch, wo wirklich nötig

* Es wird nur noch dort mit Typisierung gearbeitet, wo dies unumgänglich
  ist, weil die typisierte Implementierung angesprochen wird.
* Das Versenden der Test-Nachrichten erfolgt als `Bytes` für Schlüssel
  und Nachricht.
* Dadurch muss der `RecordGenerator` nicht mehr typisiert werden.
* Dafür muss die typisierte Implementierung des Testfalls dann Schlüssel
  und Nachricht mit einem passenden Serializer in eine `Bytes`-Payload
  umwandeln.
---

diff --git a/src/test/java/de/juplo/kafka/ApplicationTest.java b/src/test/java/de/juplo/kafka/ApplicationTest.java
index d3ff3b1..81165ab 100644
--- a/src/test/java/de/juplo/kafka/ApplicationTest.java
+++ b/src/test/java/de/juplo/kafka/ApplicationTest.java
@@ -14,8 +14,7 @@ public class ApplicationTest extends GenericApplicationTest<String, Long>
   public ApplicationTest()
   {
     super(
-        new StringSerializer(),
-        new RecordGenerator<> ()
+        new RecordGenerator()
         {
           final StringSerializer stringSerializer = new StringSerializer();
           final LongSerializer longSerializer = new LongSerializer();
@@ -25,7 +24,7 @@ public class ApplicationTest extends GenericApplicationTest<String, Long>
           public void generate(
               int numberOfMessagesToGenerate,
               Set<Integer> poisonPills,
-              Consumer<ProducerRecord<String, Bytes>> messageSender)
+              Consumer<ProducerRecord<Bytes, Bytes>> messageSender)
           {
             int i = 0;
 
@@ -41,11 +40,11 @@ public class ApplicationTest extends GenericApplicationTest<String, Long>
                         ? new Bytes(stringSerializer.serialize(TOPIC, "BOOM!"))
                         : new Bytes(longSerializer.serialize(TOPIC, (long)i));
 
-                ProducerRecord<String, Bytes> record =
+                ProducerRecord<Bytes, Bytes> record =
                     new ProducerRecord<>(
                         TOPIC,
                         partition,
-                        Integer.toString(partition*10+key%2),
+                        new Bytes(stringSerializer.serialize(TOPIC,Integer.toString(partition*10+key%2))),
                         value);
 
                 messageSender.accept(record);
diff --git a/src/test/java/de/juplo/kafka/GenericApplicationTest.java b/src/test/java/de/juplo/kafka/GenericApplicationTest.java
index 6b2b635..68f150f 100644
--- a/src/test/java/de/juplo/kafka/GenericApplicationTest.java
+++ b/src/test/java/de/juplo/kafka/GenericApplicationTest.java
@@ -53,7 +53,7 @@ abstract class GenericApplicationTest<K, V>
 	@Autowired
 	ExecutorService executor;
 
-	KafkaProducer<K, Bytes> testRecordProducer;
+	KafkaProducer<Bytes, Bytes> testRecordProducer;
 	KafkaConsumer<Bytes, Bytes> offsetConsumer;
 	Consumer<ConsumerRecord<K, V>> testHandler;
 	EndlessConsumer<K, V> endlessConsumer;
@@ -62,15 +62,11 @@ abstract class GenericApplicationTest<K, V>
 	Set<ConsumerRecord<K, V>> receivedRecords;
 
 
-	final Serializer<K> keySerializer;
-	final RecordGenerator<K> recordGenerator;
-	final Consumer<ProducerRecord<K, Bytes>> messageSender;
+	final RecordGenerator recordGenerator;
+	final Consumer<ProducerRecord<Bytes, Bytes>> messageSender;
 
-	public GenericApplicationTest(
-			Serializer<K> keySerializer,
-			RecordGenerator<K> recordGenerator)
+	public GenericApplicationTest(RecordGenerator recordGenerator)
 	{
-		this.keySerializer = keySerializer;
 		this.recordGenerator = recordGenerator;
 		this.messageSender = (record) -> sendMessage(record);
 	}
@@ -204,15 +200,15 @@ abstract class GenericApplicationTest<K, V>
 	}
 
 
-	public interface RecordGenerator<K>
+	public interface RecordGenerator
 	{
 		void generate(
 				int numberOfMessagesToGenerate,
 				Set<Integer> poistionPills,
-				Consumer<ProducerRecord<K, Bytes>> messageSender);
+				Consumer<ProducerRecord<Bytes, Bytes>> messageSender);
 	}
 
-	void sendMessage(ProducerRecord<K, Bytes> record)
+	void sendMessage(ProducerRecord<Bytes, Bytes> record)
 	{
 		testRecordProducer.send(record, (metadata, e) ->
 		{
@@ -244,7 +240,7 @@ abstract class GenericApplicationTest<K, V>
 		props = new Properties();
 		props.put("bootstrap.servers", properties.getBootstrapServer());
 		props.put("linger.ms", 100);
-		props.put("key.serializer", keySerializer.getClass().getName());
+		props.put("key.serializer", BytesSerializer.class.getName());
 		props.put("value.serializer", BytesSerializer.class.getName());
 		testRecordProducer = new KafkaProducer<>(props);