* Fixed the type-mapping for the `JsonDeserializer`.
* Disabled the caching / out-buffering in Streams.
* Both changes are necessary, to see in `Top10ApplicationIT`, that
messages are processed: _Compare the outcome of the test before and
after this commit!_
* Nonetheless, the integration-test still fails, because the assertion
is not true.
props.put(
JsonDeserializer.TYPE_MAPPINGS,
"word:" + Key.class.getName() + "," +
props.put(
JsonDeserializer.TYPE_MAPPINGS,
"word:" + Key.class.getName() + "," +
- "counter:" + Entry.class.getName());
- props.put(JsonDeserializer.REMOVE_TYPE_INFO_HEADERS, Boolean.FALSE);
- props.put(
- JsonSerializer.TYPE_MAPPINGS,
+ "counter:" + Entry.class.getName() + "," +
"user:" + User.class.getName() + "," +
"ranking:" + Ranking.class.getName());
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
"user:" + User.class.getName() + "," +
"ranking:" + Ranking.class.getName());
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+ if (properties.getCommitInterval() != null)
+ props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, properties.getCommitInterval());
+ if (properties.getCacheMaxBytes() != null)
+ props.put(StreamsConfig.STATESTORE_CACHE_MAX_BYTES_CONFIG, properties.getCacheMaxBytes());
+ props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
private String applicationId = "top10";
private String inputTopic = "countings";
private String outputTopic = "top10";
private String applicationId = "top10";
private String inputTopic = "countings";
private String outputTopic = "top10";
+ private Integer commitInterval;
+ private Integer cacheMaxBytes;
"spring.kafka.consumer.properties.spring.json.trusted.packages=de.juplo.kafka.wordcount.top10 ",
"logging.level.root=WARN",
"logging.level.de.juplo=DEBUG",
"spring.kafka.consumer.properties.spring.json.trusted.packages=de.juplo.kafka.wordcount.top10 ",
"logging.level.root=WARN",
"logging.level.de.juplo=DEBUG",
+ "logging.level.org.apache.kafka.clients=INFO",
+ "logging.level.org.apache.kafka.streams=INFO",
"juplo.wordcount.top10.bootstrap-server=${spring.embedded.kafka.brokers}",
"juplo.wordcount.top10.bootstrap-server=${spring.embedded.kafka.brokers}",
+ "juplo.wordcount.top10.commit-interval=100",
+ "juplo.wordcount.top10.cacheMaxBytes=0",
"juplo.wordcount.top10.input-topic=" + Top10ApplicationIT.TOPIC_IN,
"juplo.wordcount.top10.output-topic=" + Top10ApplicationIT.TOPIC_OUT })
@EmbeddedKafka(topics = { Top10ApplicationIT.TOPIC_IN, Top10ApplicationIT.TOPIC_OUT })
"juplo.wordcount.top10.input-topic=" + Top10ApplicationIT.TOPIC_IN,
"juplo.wordcount.top10.output-topic=" + Top10ApplicationIT.TOPIC_OUT })
@EmbeddedKafka(topics = { Top10ApplicationIT.TOPIC_IN, Top10ApplicationIT.TOPIC_OUT })