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

Updating with changes related to core GH-1484

Browse Source 
Resolves #447
This commit is contained in:
Oleg Zhurakousky
2018-09-19 10:44:56 +02:00
parent cc61d916b8
commit 5446485cbf
9 changed files with 26 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBindingProperties.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016 the original author or authors.
* Copyright 2016-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -16,10 +16,13 @@
package org.springframework.cloud.stream.binder.kafka.properties;
import org.springframework.cloud.stream.binder.BinderSpecificPropertiesProvider;
/**
* @author Marius Bogoevici
* @author Oleg Zhurakousky
*/
public class KafkaBindingProperties {
public class KafkaBindingProperties implements BinderSpecificPropertiesProvider{
private KafkaConsumerProperties consumer = new KafkaConsumerProperties();

3
spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaExtendedBindingProperties.java
View File

@@ -20,6 +20,7 @@ import java.util.HashMap;
import java.util.Map;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.cloud.stream.binder.BinderSpecificPropertiesProvider;
import org.springframework.cloud.stream.binder.ExtendedBindingProperties;
/**
@@ -91,7 +92,7 @@ public class KafkaExtendedBindingProperties
}
@Override
public Class<?> getExtendedPropertiesEntryClass() {
public Class<? extends BinderSpecificPropertiesProvider> getExtendedPropertiesEntryClass() {
return KafkaBindingProperties.class;
}

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

@@ -19,6 +19,7 @@ package org.springframework.cloud.stream.binder.kafka.streams;
import org.apache.kafka.streams.kstream.GlobalKTable;
import org.springframework.cloud.stream.binder.AbstractBinder;
import org.springframework.cloud.stream.binder.BinderSpecificPropertiesProvider;
import org.springframework.cloud.stream.binder.Binding;
import org.springframework.cloud.stream.binder.DefaultBinding;
import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
@@ -96,7 +97,7 @@ public class GlobalKTableBinder extends
}
@Override
public Class<?> getExtendedPropertiesEntryClass() {
public Class<? extends BinderSpecificPropertiesProvider> getExtendedPropertiesEntryClass() {
return this.kafkaStreamsExtendedBindingProperties.getExtendedPropertiesEntryClass();
}

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

@@ -23,6 +23,7 @@ import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Produced;
import org.springframework.cloud.stream.binder.AbstractBinder;
import org.springframework.cloud.stream.binder.BinderSpecificPropertiesProvider;
import org.springframework.cloud.stream.binder.Binding;
import org.springframework.cloud.stream.binder.DefaultBinding;
import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
@@ -139,7 +140,7 @@ class KStreamBinder extends
}
@Override
public Class<?> getExtendedPropertiesEntryClass() {
public Class<? extends BinderSpecificPropertiesProvider> getExtendedPropertiesEntryClass() {
return this.kafkaStreamsExtendedBindingProperties.getExtendedPropertiesEntryClass();
}
}

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

@@ -19,6 +19,7 @@ package org.springframework.cloud.stream.binder.kafka.streams;
import org.apache.kafka.streams.kstream.KTable;
import org.springframework.cloud.stream.binder.AbstractBinder;
import org.springframework.cloud.stream.binder.BinderSpecificPropertiesProvider;
import org.springframework.cloud.stream.binder.Binding;
import org.springframework.cloud.stream.binder.DefaultBinding;
import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
@@ -95,7 +96,7 @@ class KTableBinder extends
}
@Override
public Class<?> getExtendedPropertiesEntryClass() {
public Class<? extends BinderSpecificPropertiesProvider> getExtendedPropertiesEntryClass() {
return this.kafkaStreamsExtendedBindingProperties.getExtendedPropertiesEntryClass();
}
}

4
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsBindingProperties.java
View File

@@ -16,10 +16,12 @@
package org.springframework.cloud.stream.binder.kafka.streams.properties;
import org.springframework.cloud.stream.binder.BinderSpecificPropertiesProvider;
/**
* @author Marius Bogoevici
*/
public class KafkaStreamsBindingProperties {
public class KafkaStreamsBindingProperties implements BinderSpecificPropertiesProvider {
private KafkaStreamsConsumerProperties consumer = new KafkaStreamsConsumerProperties();

3
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsExtendedBindingProperties.java
View File

@@ -20,6 +20,7 @@ import java.util.HashMap;
import java.util.Map;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.cloud.stream.binder.BinderSpecificPropertiesProvider;
import org.springframework.cloud.stream.binder.ExtendedBindingProperties;
/**
@@ -65,7 +66,7 @@ public class KafkaStreamsExtendedBindingProperties
}
@Override
public Class<?> getExtendedPropertiesEntryClass() {
public Class<? extends BinderSpecificPropertiesProvider> getExtendedPropertiesEntryClass() {
return KafkaStreamsBindingProperties.class;
}
}

6
spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
View File

@@ -55,6 +55,7 @@ import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.NoSuchBeanDefinitionException;
import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder;
import org.springframework.cloud.stream.binder.BinderHeaders;
import org.springframework.cloud.stream.binder.BinderSpecificPropertiesProvider;
import org.springframework.cloud.stream.binder.DefaultPollableMessageSource;
import org.springframework.cloud.stream.binder.EmbeddedHeaderUtils;
import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
@@ -220,7 +221,7 @@ public class KafkaMessageChannelBinder extends
}
@Override
public Class<?> getExtendedPropertiesEntryClass() {
public Class<? extends BinderSpecificPropertiesProvider> getExtendedPropertiesEntryClass() {
return this.extendedBindingProperties.getExtendedPropertiesEntryClass();
}
@@ -675,6 +676,7 @@ public class KafkaMessageChannelBinder extends
return new RawRecordHeaderErrorMessageStrategy();
}
@SuppressWarnings("unchecked")
@Override
protected MessageHandler getErrorMessageHandler(final ConsumerDestination destination, final String group,
final ExtendedConsumerProperties<KafkaConsumerProperties> properties) {
@@ -687,7 +689,7 @@ public class KafkaMessageChannelBinder extends
new ExtendedProducerProperties<>(dlqProducerProperties));
final KafkaTemplate<?,?> kafkaTemplate = new KafkaTemplate<>(producerFactory);
@SuppressWarnings({"unchecked", "rawtypes"})
@SuppressWarnings("rawtypes")
DlqSender<?,?> dlqSender = new DlqSender(kafkaTemplate);
return message -> {

4
spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java
View File

@@ -69,6 +69,7 @@ public class KafkaBinderConfiguration {
@Autowired
private KafkaExtendedBindingProperties kafkaExtendedBindingProperties;
@SuppressWarnings("rawtypes")
@Autowired
private ProducerListener producerListener;
@@ -85,6 +86,7 @@ public class KafkaBinderConfiguration {
return new KafkaTopicProvisioner(configurationProperties, this.kafkaProperties);
}
@SuppressWarnings("unchecked")
@Bean
KafkaMessageChannelBinder kafkaMessageChannelBinder(KafkaBinderConfigurationProperties configurationProperties,
KafkaTopicProvisioner provisioningProvider, @Nullable ListenerContainerCustomizer<AbstractMessageListenerContainer<?, ?>> listenerContainerCustomizer) {
@@ -96,6 +98,7 @@ public class KafkaBinderConfiguration {
return kafkaMessageChannelBinder;
}
@SuppressWarnings("rawtypes")
@Bean
@ConditionalOnMissingBean(ProducerListener.class)
ProducerListener producerListener() {
@@ -154,6 +157,7 @@ public class KafkaBinderConfiguration {
}
@SuppressWarnings("unused")
public static class JaasConfigurationProperties {
private JaasLoginModuleConfiguration kafka;