"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.use.type.headers=false",
- "spring.kafka.consumer.properties.spring.json.key.default.type=de.juplo.kafka.wordcount.query.TestUser",
- "spring.kafka.consumer.properties.spring.json.value.default.type=de.juplo.kafka.wordcount.query.TestRanking",
+ "spring.kafka.consumer.properties.spring.json.type.mapping=user:de.juplo.kafka.wordcount.query.TestUser,ranking:de.juplo.kafka.wordcount.query.TestRanking",
"logging.level.root=WARN",
"logging.level.de.juplo=DEBUG",
"logging.level.org.apache.kafka.clients=INFO",
{
await("Expected messages")
.atMost(Duration.ofSeconds(5))
- .untilAsserted(() -> TestData.assertExpectedMessages(consumer.getReceivedMessages()));
+ .untilAsserted(() -> consumer.enforceAssertion(
+ receivedMessages -> TestData.assertExpectedMessages(receivedMessages)));
}
@DisplayName("Await the expected number of messages")
{
await("Expected number of messages")
.atMost(Duration.ofSeconds(5))
- .untilAsserted(() -> TestData.assertExpectedNumberOfMessagesForUsers(consumer.getReceivedMessages()));
+ .untilAsserted(() -> consumer.enforceAssertion(
+ receivedMessages -> TestData.assertExpectedNumberOfMessagesForUsers(receivedMessages)));
}
@DisplayName("Await the expected final output messages")
{
await("Expected final output messages")
.atMost(Duration.ofSeconds(5))
- .untilAsserted(() -> TestData.assertExpectedLastMessagesForUsers(consumer.getReceivedMessages()));
+ .untilAsserted(() -> consumer.enforceAssertion(
+ receivedMessages -> TestData.assertExpectedLastMessagesForUsers(receivedMessages)));
}
received.add(user, ranking);
}
- synchronized MultiValueMap<TestUser, TestRanking> getReceivedMessages()
+ synchronized void enforceAssertion(
+ java.util.function.Consumer<MultiValueMap<TestUser, TestRanking>> assertion)
{
- return received;
+ assertion.accept(received);
}
}