X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Ftest%2Fjava%2Fde%2Fjuplo%2Fkafka%2Fwordcount%2Ftop10%2FTop10StreamProcessorTopologyTest.java;fp=src%2Ftest%2Fjava%2Fde%2Fjuplo%2Fkafka%2Fwordcount%2Ftop10%2FTop10StreamProcessorTopologyTest.java;h=84cfb1e283eb240c09c6ff4fe1af76358445a10c;hb=6500abf8a211334e869de7cc2f354368f0263bc1;hp=f2a9eca4f2624937d567d26b5543dcd658820f0b;hpb=d1ac99074b22f357152644d4593b234540d1c952;p=demos%2Fkafka%2Fwordcount diff --git a/src/test/java/de/juplo/kafka/wordcount/top10/Top10StreamProcessorTopologyTest.java b/src/test/java/de/juplo/kafka/wordcount/top10/Top10StreamProcessorTopologyTest.java index f2a9eca..84cfb1e 100644 --- a/src/test/java/de/juplo/kafka/wordcount/top10/Top10StreamProcessorTopologyTest.java +++ b/src/test/java/de/juplo/kafka/wordcount/top10/Top10StreamProcessorTopologyTest.java @@ -20,10 +20,7 @@ import java.util.Map; import java.util.Properties; import java.util.stream.Stream; -import static de.juplo.kafka.wordcount.top10.TestData.parseHeader; import static de.juplo.kafka.wordcount.top10.Top10ApplicationConfiguration.serializationConfig; -import static org.springframework.kafka.support.mapping.AbstractJavaTypeMapper.DEFAULT_CLASSID_FIELD_NAME; -import static org.springframework.kafka.support.mapping.AbstractJavaTypeMapper.KEY_DEFAULT_CLASSID_FIELD_NAME; @Slf4j @@ -84,16 +81,7 @@ public class Top10StreamProcessorTopologyTest MultiValueMap receivedMessages = new LinkedMultiValueMap<>(); out .readRecordsToList() - .forEach(record -> - { - log.debug( - "OUT: {} -> {}, {}, {}", - record.key(), - record.value(), - parseHeader(record.headers(), KEY_DEFAULT_CLASSID_FIELD_NAME), - parseHeader(record.headers(), DEFAULT_CLASSID_FIELD_NAME)); - receivedMessages.add(record.key(), record.value()); - }); + .forEach(record -> receivedMessages.add(record.key(), record.value())); TestData.assertExpectedMessages(receivedMessages);