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
MultiValueMap<User, Ranking> 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);