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 (file)
--- 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 (file)
--- 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);