Refaktorisierungen des Testfalls gemerged (Branch 'deserialization')

diff --combined src/test/java/de/juplo/kafka/ApplicationTests.java
index 6c25bcd,05eebd0..62906b3
--- 1/src/test/java/de/juplo/kafka/ApplicationTests.java
--- 2/src/test/java/de/juplo/kafka/ApplicationTests.java
+++ b/src/test/java/de/juplo/kafka/ApplicationTests.java
@@@ -15,7 -15,6 +15,7 @@@ import org.springframework.boot.test.co
  import org.springframework.boot.test.context.TestConfiguration;
  import org.springframework.context.annotation.Bean;
  import org.springframework.context.annotation.Import;
 +import org.springframework.kafka.support.serializer.JsonSerializer;
  import org.springframework.kafka.test.context.EmbeddedKafka;
  import org.springframework.test.context.TestPropertySource;
  import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
@@@ -25,8 -24,8 +25,8 @@@ import java.util.*
  import java.util.concurrent.ExecutionException;
  import java.util.concurrent.ExecutorService;
  import java.util.function.BiConsumer;
 +import java.util.function.BiFunction;
  import java.util.function.Consumer;
 -import java.util.function.Function;
  import java.util.stream.Collectors;
  import java.util.stream.IntStream;
  
@@@ -57,7 -56,7 +57,7 @@@ class ApplicationTest
        @Autowired
        KafkaProducer<String, Bytes> kafkaProducer;
        @Autowired
 -      KafkaConsumer<String, Long> kafkaConsumer;
 +      KafkaConsumer<String, ClientMessage> kafkaConsumer;
        @Autowired
        KafkaConsumer<Bytes, Bytes> offsetConsumer;
        @Autowired
@@@ -65,11 -64,11 +65,11 @@@
        @Autowired
        ExecutorService executor;
  
 -      Consumer<ConsumerRecord<String, Long>> testHandler;
 -      EndlessConsumer<String, Long> endlessConsumer;
 +      Consumer<ConsumerRecord<String, ClientMessage>> testHandler;
 +      EndlessConsumer<String, ClientMessage> endlessConsumer;
        Map<TopicPartition, Long> oldOffsets;
        Map<TopicPartition, Long> newOffsets;
 -      Set<ConsumerRecord<String, Long>> receivedRecords;
 +      Set<ConsumerRecord<String, ClientMessage>> receivedRecords;
  
  
        /** Tests methods */
@@@ -78,7 -77,11 +78,11 @@@
        @Order(1) // << The poistion pill is not skipped. Hence, this test must run first
        void commitsCurrentOffsetsOnSuccess() throws ExecutionException, InterruptedException
        {
-               send100Messages((key, counter) -> serialize(key, counter));
+               send100Messages((partition, key, counter) ->
+               {
 -                      Bytes value = new Bytes(valueSerializer.serialize(TOPIC, counter));
++                      Bytes value = serialize(key, counter);
+                       return new ProducerRecord<>(TOPIC, partition, key, value);
+               });
  
                await("100 records received")
                                .atMost(Duration.ofSeconds(30))
@@@ -101,10 -104,13 +105,13 @@@
        @Order(2)
        void commitsOffsetOfErrorForReprocessingOnError()
        {
-               send100Messages((key, counter) ->
-                               counter == 77
-                                               ? new Bytes(stringSerializer.serialize(TOPIC, "BOOM!"))
-                                               : serialize(key, counter));
+               send100Messages((partition, key, counter) ->
+               {
+                       Bytes value = counter == 77
+                                       ? new Bytes(stringSerializer.serialize(TOPIC, "BOOM!"))
 -                                      : new Bytes(valueSerializer.serialize(TOPIC, counter));
++                                      : serialize(key, counter);
+                       return new ProducerRecord<>(TOPIC, partition, key, value);
+               });
  
                await("Consumer failed")
                                .atMost(Duration.ofSeconds(30))
@@@ -186,7 -192,12 +193,12 @@@
        }
  
  
-       void send100Messages(BiFunction<Integer, Long, Bytes> messageGenerator)
+       public interface RecordGenerator<K, V>
+       {
+               public ProducerRecord<String, Bytes> generate(int partition, String key, long counter);
+       }
+ 
+       void send100Messages(RecordGenerator recordGenerator)
        {
                long i = 0;
  
@@@ -194,16 -205,9 +206,10 @@@
                {
                        for (int key = 0; key < 10; key++)
                        {
-                               Bytes value = messageGenerator.apply(key, ++i);
- 
                                ProducerRecord<String, Bytes> record =
-                                               new ProducerRecord<>(
-                                                               TOPIC,
-                                                               partition,
-                                                               Integer.toString(key%2),
-                                                               value);
+                                               recordGenerator.generate(partition, Integer.toString(partition*10+key%2), ++i);
  
 +                              record.headers().add("__TypeId__", "message".getBytes());
                                kafkaProducer.send(record, (metadata, e) ->
                                {
                                        if (metadata != null)
@@@ -228,14 -232,6 +234,14 @@@
                }
        }
  
-       Bytes serialize(Integer key, Long value)
++      Bytes serialize(String key, Long value)
 +      {
 +              ClientMessage message = new ClientMessage();
-               message.setClient(key.toString());
++              message.setClient(key);
 +              message.setMessage(value.toString());
 +              return new Bytes(valueSerializer.serialize(TOPIC, message));
 +      }
 +
  
        @BeforeEach
        public void init()
@@@ -252,7 -248,7 +258,7 @@@
                        newOffsets.put(tp, offset - 1);
                });
  
 -              Consumer<ConsumerRecord<String, Long>> captureOffsetAndExecuteTestHandler =
 +              Consumer<ConsumerRecord<String, ClientMessage>> captureOffsetAndExecuteTestHandler =
                                record ->
                                {
                                        newOffsets.put(
@@@ -292,9 -288,9 +298,9 @@@
        public static class Configuration
        {
                @Bean
 -              Serializer<Long> serializer()
 +              Serializer<ClientMessage> serializer()
                {
 -                      return new LongSerializer();
 +                      return new JsonSerializer<>();
                }
  
                @Bean