`requests`-Service für das SumUp-Beispiel implementiert

[demos/kafka/training] / src / test / java / de / juplo / kafka / ApplicationTests.java

diff --git a/src/test/java/de/juplo/kafka/ApplicationTests.java b/src/test/java/de/juplo/kafka/ApplicationTests.java
index 50627da..491335a 100644 (file)
--- a/src/test/java/de/juplo/kafka/ApplicationTests.java
+++ b/src/test/java/de/juplo/kafka/ApplicationTests.java
@@ -28,8 +28,7 @@ import java.util.function.BiConsumer;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
-import static de.juplo.kafka.ApplicationTests.PARTITIONS;
-import static de.juplo.kafka.ApplicationTests.TOPIC;
+import static de.juplo.kafka.ApplicationTests.*;
 import static org.assertj.core.api.Assertions.*;
 import static org.awaitility.Awaitility.*;
 
@@ -38,15 +37,16 @@ import static org.awaitility.Awaitility.*;
 @TestMethodOrder(MethodOrderer.OrderAnnotation.class)
 @TestPropertySource(
                properties = {
-                               "consumer.bootstrap-server=${spring.embedded.kafka.brokers}",
-                               "consumer.topic=" + TOPIC,
-                               "consumer.commit-interval=1s" })
-@EmbeddedKafka(topics = TOPIC, partitions = PARTITIONS)
+                               "sumup.requests.bootstrap-server=${spring.embedded.kafka.brokers}",
+                               "sumup.requests.topic-in=" + INPUT_TOPIC,
+                               "sumup.requests.commit-interval=1s" })
+@EmbeddedKafka(topics = { INPUT_TOPIC, OUTPUT_TOPIC }, partitions = PARTITIONS)
 @EnableAutoConfiguration
 @Slf4j
 class ApplicationTests
 {
-       public static final String TOPIC = "FOO";
+       public static final String INPUT_TOPIC = "FOO";
+       public static final String OUTPUT_TOPIC = "BAR";
        public static final int PARTITIONS = 10;
 
 
@@ -55,9 +55,9 @@ class ApplicationTests
        @Autowired
        Serializer valueSerializer;
        @Autowired
-       KafkaProducer<String, Bytes> kafkaProducer;
+       KafkaProducer<String, Bytes> testProducer;
        @Autowired
-       KafkaConsumer<String, Long> kafkaConsumer;
+       KafkaConsumer<String, Integer> kafkaConsumer;
        @Autowired
        KafkaConsumer<Bytes, Bytes> offsetConsumer;
        @Autowired
@@ -67,10 +67,10 @@ class ApplicationTests
        @Autowired
        RecordHandler noopRecordHandler;
 
-       EndlessConsumer<String, Long> endlessConsumer;
+       EndlessConsumer<String, Integer> endlessConsumer;
        Map<TopicPartition, Long> oldOffsets;
        Map<TopicPartition, Long> newOffsets;
-       Set<ConsumerRecord<String, Long>> receivedRecords;
+       Set<ConsumerRecord<String, Integer>> receivedRecords;
 
 
        /** Tests methods */
@@ -80,8 +80,8 @@ class ApplicationTests
        {
                send100Messages((partition, key, counter) ->
                {
-                       Bytes value = new Bytes(valueSerializer.serialize(TOPIC, counter));
-                       return new ProducerRecord<>(TOPIC, partition, key, value);
+                       Bytes value = new Bytes(valueSerializer.serialize(INPUT_TOPIC, counter));
+                       return new ProducerRecord<>(INPUT_TOPIC, partition, key, value);
                });
 
                await("100 records received")
@@ -109,9 +109,9 @@ class ApplicationTests
                send100Messages((partition, key, counter) ->
                {
                        Bytes value = counter == 77
-                                       ? new Bytes(stringSerializer.serialize(TOPIC, "BOOM!"))
-                                       : new Bytes(valueSerializer.serialize(TOPIC, counter));
-                       return new ProducerRecord<>(TOPIC, partition, key, value);
+                                       ? new Bytes(stringSerializer.serialize(INPUT_TOPIC, "BOOM!"))
+                                       : new Bytes(valueSerializer.serialize(INPUT_TOPIC, counter));
+                       return new ProducerRecord<>(INPUT_TOPIC, partition, key, value);
                });
 
                await("Consumer failed")
@@ -206,19 +206,19 @@ class ApplicationTests
                return
                                IntStream
                                                .range(0, PARTITIONS)
-                                               .mapToObj(partition -> new TopicPartition(TOPIC, partition))
+                                               .mapToObj(partition -> new TopicPartition(INPUT_TOPIC, partition))
                                                .collect(Collectors.toList());
        }
 
 
        public interface RecordGenerator<K, V>
        {
-               public ProducerRecord<String, Bytes> generate(int partition, String key, long counter);
+               public ProducerRecord<String, Bytes> generate(int partition, String key, int counter);
        }
 
        void send100Messages(RecordGenerator recordGenerator)
        {
-               long i = 0;
+               int i = 0;
 
                for (int partition = 0; partition < 10; partition++)
                {
@@ -227,7 +227,7 @@ class ApplicationTests
                                ProducerRecord<String, Bytes> record =
                                                recordGenerator.generate(partition, Integer.toString(partition*10+key%2), ++i);
 
-                               kafkaProducer.send(record, (metadata, e) ->
+                               testProducer.send(record, (metadata, e) ->
                                {
                                        if (metadata != null)
                                        {
@@ -267,10 +267,10 @@ class ApplicationTests
                        newOffsets.put(tp, offset - 1);
                });
 
-               TestRecordHandler<String, Long> captureOffsetAndExecuteTestHandler =
-                               new TestRecordHandler<String, Long>(noopRecordHandler) {
+               TestRecordHandler<String, Integer> captureOffsetAndExecuteTestHandler =
+                               new TestRecordHandler<String, Integer>(noopRecordHandler) {
                                        @Override
-                                       public void onNewRecord(ConsumerRecord<String, Long> record)
+                                       public void onNewRecord(ConsumerRecord<String, Integer> record)
                                        {
                                                newOffsets.put(
                                                                new TopicPartition(record.topic(), record.partition()),
@@ -283,7 +283,7 @@ class ApplicationTests
                                new EndlessConsumer<>(
                                                executor,
                                                properties.getClientId(),
-                                               properties.getTopic(),
+                                               properties.getTopicIn(),
                                                kafkaConsumer,
                                                captureOffsetAndExecuteTestHandler);
 
@@ -309,13 +309,13 @@ class ApplicationTests
        public static class Configuration
        {
                @Bean
-               Serializer<Long> serializer()
+               Serializer<Integer> valueSerializer()
                {
-                       return new LongSerializer();
+                       return new IntegerSerializer();
                }
 
                @Bean
-               KafkaProducer<String, Bytes> kafkaProducer(ApplicationProperties properties)
+               KafkaProducer<String, Bytes> testProducer(ApplicationProperties properties)
                {
                        Properties props = new Properties();
                        props.put("bootstrap.servers", properties.getBootstrapServer());
@@ -338,5 +338,16 @@ class ApplicationTests
 
                        return new KafkaConsumer<>(props);
                }
+
+               @Bean
+               KafkaProducer<String, String> kafkaProducer(ApplicationProperties properties)
+               {
+                       Properties props = new Properties();
+                       props.put("bootstrap.servers", properties.getBootstrapServer());
+                       props.put("key.serializer", StringSerializer.class.getName());
+                       props.put("value.serializer", StringSerializer.class.getName());
+
+                       return new KafkaProducer<>(props);
+               }
        }
 }