From 64ea989b08ad0a3103aaec3bd27017c65d201c1f Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Tue, 27 Nov 2018 21:41:00 -0500 Subject: [PATCH] Kafka Streams environment properties changes When Kafka Streams binder is used in multi binder environments, the properties defined under environment is not propagated to the auto configuration class. The environment processing only takes place when the actual binder configuration is instantiated (for example, KStreamConfiguration), and therefore the environment properties are unavailable during the earlier autoconfiguration. This change makes the environment properties availble during auto configuration. Resolves #504 --- ...StreamsBinderSupportAutoConfiguration.java | 51 ++++++++++++++++++- 1 file changed, 50 insertions(+), 1 deletion(-) diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java index 5387302a0..5f4fb74d7 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java @@ -34,16 +34,20 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; import org.springframework.boot.autoconfigure.kafka.KafkaProperties; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.cloud.stream.binder.BinderConfiguration; import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsBinderConfigurationProperties; import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsExtendedBindingProperties; import org.springframework.cloud.stream.binder.kafka.streams.serde.CompositeNonNativeSerde; import org.springframework.cloud.stream.binding.BindingService; import org.springframework.cloud.stream.binding.StreamListenerResultAdapter; +import org.springframework.cloud.stream.config.BinderProperties; import org.springframework.cloud.stream.config.BindingServiceConfiguration; import org.springframework.cloud.stream.config.BindingServiceProperties; import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; import org.springframework.context.annotation.Bean; +import org.springframework.core.env.ConfigurableEnvironment; import org.springframework.core.env.Environment; +import org.springframework.core.env.MapPropertySource; import org.springframework.kafka.config.KafkaStreamsConfiguration; import org.springframework.kafka.core.CleanupConfig; import org.springframework.util.ObjectUtils; @@ -61,12 +65,57 @@ import org.springframework.util.StringUtils; @AutoConfigureAfter(BindingServiceConfiguration.class) public class KafkaStreamsBinderSupportAutoConfiguration { + private static final String KSTREAM_BINDER_TYPE = "kstream"; + private static final String KTABLE_BINDER_TYPE = "ktable"; + private static final String GLOBALKTABLE_BINDER_TYPE = "globalktable"; + @Bean @ConfigurationProperties(prefix = "spring.cloud.stream.kafka.streams.binder") - public KafkaStreamsBinderConfigurationProperties binderConfigurationProperties(KafkaProperties kafkaProperties) { + public KafkaStreamsBinderConfigurationProperties binderConfigurationProperties(KafkaProperties kafkaProperties, + ConfigurableEnvironment environment, + BindingServiceProperties bindingServiceProperties) { + final Map binderConfigurations = getBinderConfigurations(bindingServiceProperties); + for (Map.Entry entry : binderConfigurations.entrySet()) { + final BinderConfiguration binderConfiguration = entry.getValue(); + final String binderType = binderConfiguration.getBinderType(); + if (binderType.equals(KSTREAM_BINDER_TYPE) || + binderType.equals(KTABLE_BINDER_TYPE) || + binderType.equals(GLOBALKTABLE_BINDER_TYPE)) { + Map binderProperties = new HashMap<>(); + this.flatten(null, binderConfiguration.getProperties(), binderProperties); + environment.getPropertySources().addFirst(new MapPropertySource("kafkaStreamsBinderEnv", binderProperties)); + } + } return new KafkaStreamsBinderConfigurationProperties(kafkaProperties); } + //TODO: Lifted from core - good candidate for exposing as a utility method in core. + private static Map getBinderConfigurations(BindingServiceProperties bindingServiceProperties) { + + Map binderConfigurations = new HashMap<>(); + Map declaredBinders = bindingServiceProperties.getBinders(); + + for (Map.Entry binderEntry : declaredBinders.entrySet()) { + BinderProperties binderProperties = binderEntry.getValue(); + binderConfigurations.put(binderEntry.getKey(), + new BinderConfiguration(binderProperties.getType(), binderProperties.getEnvironment(), + binderProperties.isInheritEnvironment(), binderProperties.isDefaultCandidate())); + } + return binderConfigurations; + } + + //TODO: Lifted from core - good candidate for exposing as a utility method in core. + @SuppressWarnings("unchecked") + private void flatten(String propertyName, Object value, Map flattenedProperties) { + if (value instanceof Map) { + ((Map) value) + .forEach((k, v) -> flatten((propertyName != null ? propertyName + "." : "") + k, v, flattenedProperties)); + } + else { + flattenedProperties.put(propertyName, value.toString()); + } + } + @Bean public KafkaStreamsConfiguration kafkaStreamsConfiguration(KafkaStreamsBinderConfigurationProperties binderConfigurationProperties, Environment environment) {