SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix NPE in Kafka Streams binder

Browse Source 
* Fix NPE in Kafka Streams binder

Fix NPE when user provided consumer properties are missing in kafka streams binder

Resolves #343

* Addressing PR review comments
This commit is contained in:
Soby Chacko
2018-03-28 13:32:37 -04:00
committed by Artem Bilan
parent 11a275a299
commit 710ff2c292
1 changed files with 4 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java
View File

@@ -221,7 +221,7 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene
enableNativeDecodingForKTableAlways(parameterType, bindingProperties);
StreamsConfig streamsConfig = null;
//Retrieve the StreamsConfig created for this method if available.
//Otherwise, carete the StreamsBuilderFactory and get the underlying config.
//Otherwise, create the StreamsBuilderFactory and get the underlying config.
if (!methodStreamsBuilderFactoryBeanMap.containsKey(method)) {
streamsConfig = buildStreamsBuilderAndRetrieveConfig(method, applicationContext, bindingProperties);
}
@@ -291,7 +291,8 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene
Serde<?> keySerde, Serde<?> valueSerde) {
KStream<?, ?> stream = streamsBuilder.stream(bindingServiceProperties.getBindingDestination(inboundName),
Consumed.with(keySerde, valueSerde));
if (bindingProperties.getConsumer().isUseNativeDecoding()){
final boolean nativeDecoding = bindingServiceProperties.getConsumerProperties(inboundName).isUseNativeDecoding();
if (nativeDecoding){
LOG.info("Native decoding is enabled for " + inboundName + ". Inbound deserialization done at the broker.");
}
else {
@@ -300,7 +301,7 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene
stream = stream.map((key, value) -> {
KeyValue<Object, Object> keyValue;
String contentType = bindingProperties.getContentType();
if (!StringUtils.isEmpty(contentType) && !bindingProperties.getConsumer().isUseNativeDecoding()) {
if (!StringUtils.isEmpty(contentType) && !nativeDecoding) {
Message<?> message = MessageBuilder.withPayload(value)
.setHeader(MessageHeaders.CONTENT_TYPE, contentType).build();
keyValue = new KeyValue<>(key, message);