import org.springframework.util.MultiValueMap;
import static de.juplo.kafka.wordcount.counter.CounterApplicationConfiguriation.serializationConfig;
-import static de.juplo.kafka.wordcount.counter.TestData.parseHeader;
-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
{
TestData
.getInputMessages()
- .forEach(word -> in.pipeInput(word.getUser(), word));
+ .forEach(kv -> in.pipeInput(kv.key, kv.value));
MultiValueMap<TestOutputWord, TestOutputWordCounter> 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);
}