X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Ftest%2Fjava%2Fde%2Fjuplo%2Fkafka%2Fwordcount%2Ftop10%2FTop10ApplicationIT.java;h=f5ef236c19a26095b1a118262fb860314229b28d;hb=refs%2Fheads%2Ftop10;hp=5f0e8179c4d623b89b935ffd8e50e780b4ae1def;hpb=5030aed19804a0c48f1968208e176657bdd147de;p=demos%2Fkafka%2Fwordcount diff --git a/src/test/java/de/juplo/kafka/wordcount/top10/Top10ApplicationIT.java b/src/test/java/de/juplo/kafka/wordcount/top10/Top10ApplicationIT.java index 5f0e817..51e424e 100644 --- a/src/test/java/de/juplo/kafka/wordcount/top10/Top10ApplicationIT.java +++ b/src/test/java/de/juplo/kafka/wordcount/top10/Top10ApplicationIT.java @@ -3,7 +3,7 @@ package de.juplo.kafka.wordcount.top10; import de.juplo.kafka.wordcount.counter.TestCounter; import de.juplo.kafka.wordcount.counter.TestWord; import de.juplo.kafka.wordcount.query.TestRanking; -import de.juplo.kafka.wordcount.query.TestUser; +import de.juplo.kafka.wordcount.query.TestStats; import lombok.extern.slf4j.Slf4j; import org.apache.kafka.streams.state.KeyValueBytesStoreSupplier; import org.apache.kafka.streams.state.Stores; @@ -40,7 +40,7 @@ import static org.awaitility.Awaitility.await; "spring.kafka.consumer.auto-offset-reset=earliest", "spring.kafka.consumer.key-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer", "spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer", - "spring.kafka.consumer.properties.spring.json.type.mapping=user:de.juplo.kafka.wordcount.query.TestUser,ranking:de.juplo.kafka.wordcount.query.TestRanking", + "spring.kafka.consumer.properties.spring.json.type.mapping=stats:de.juplo.kafka.wordcount.query.TestStats,ranking:de.juplo.kafka.wordcount.query.TestRanking", "logging.level.root=WARN", "logging.level.de.juplo=DEBUG", "logging.level.org.apache.kafka.clients=INFO", @@ -110,32 +110,32 @@ public class Top10ApplicationIT @DisplayName("Await the expected number of messages") @Test - public void testAwaitExpectedNumberOfMessagesForUsers() + public void testAwaitExpectedNumberOfMessages() { await("Expected number of messages") .atMost(Duration.ofSeconds(5)) .untilAsserted(() -> consumer.enforceAssertion( - receivedMessages -> TestData.assertExpectedNumberOfMessagesForUsers(receivedMessages))); + receivedMessages -> TestData.assertExpectedNumberOfMessages(receivedMessages))); } @DisplayName("Await the expected final output messages") @Test - public void testAwaitExpectedLastMessagesForUsers() + public void testAwaitExpectedLastMessages() { await("Expected final output messages") .atMost(Duration.ofSeconds(5)) .untilAsserted(() -> consumer.enforceAssertion( - receivedMessages -> TestData.assertExpectedLastMessagesForUsers(receivedMessages))); + receivedMessages -> TestData.assertExpectedLastMessages(receivedMessages))); } static class Consumer { - private final MultiValueMap received = new LinkedMultiValueMap<>(); + private final MultiValueMap received = new LinkedMultiValueMap<>(); @KafkaListener(groupId = "TEST", topics = TOPIC_OUT) public synchronized void receive( - @Header(KafkaHeaders.RECEIVED_KEY) TestUser user, + @Header(KafkaHeaders.RECEIVED_KEY) TestStats user, @Payload TestRanking ranking) { log.debug("Received message: {} -> {}", user, ranking); @@ -143,7 +143,7 @@ public class Top10ApplicationIT } synchronized void enforceAssertion( - java.util.function.Consumer> assertion) + java.util.function.Consumer> assertion) { assertion.accept(received); }