JsonDeserializer.TYPE_MAPPINGS,
"word:" + Word.class.getName() + "," +
"counter:" + WordCounter.class.getName());
- propertyMap.put(JsonDeserializer.REMOVE_TYPE_INFO_HEADERS, Boolean.FALSE);
return propertyMap;
}
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
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);
}
import de.juplo.kafka.wordcount.splitter.TestInputWord;
import de.juplo.kafka.wordcount.top10.TestOutputWord;
import de.juplo.kafka.wordcount.top10.TestOutputWordCounter;
-import org.apache.kafka.common.header.Header;
-import org.apache.kafka.common.header.Headers;
import org.apache.kafka.streams.KeyValue;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;
.forEach(keyValue -> expectedMessages.add(keyValue.key, keyValue.value));
return expectedMessages;
}
-
- static String parseHeader(Headers headers, String key)
- {
- Header header = headers.lastHeader(key);
- if (header == null)
- {
- return key + "=null";
- }
- else
- {
- return key + "=" + new String(header.value());
- }
- }
}