From e522450f8ec91f64105f5d1f13b7c83818ee6bbe Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 11 May 2022 16:59:39 -0400 Subject: [PATCH] GH-2298: Refactor Common Binder Code to Core Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2298 Polishing per review comments; javadocs. --- .../support}/ConsumerConfigCustomizer.java | 5 +- .../support}/ProducerConfigCustomizer.java | 5 +- .../binder/kafka/utils/BindingUtils.java | 230 ++++++++++++++++++ .../reactorkafka/ReactorKafkaBinder.java | 175 +++---------- .../ReactorKafkaBinderConfiguration.java | 8 +- .../kafka/KafkaMessageChannelBinder.java | 164 ++----------- .../config/KafkaBinderConfiguration.java | 2 + .../KafkaConfigCustomizationTests.java | 4 +- 8 files changed, 305 insertions(+), 288 deletions(-) rename binders/kafka-binder/{spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config => spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/support}/ConsumerConfigCustomizer.java (88%) rename binders/kafka-binder/{spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config => spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/support}/ProducerConfigCustomizer.java (88%) create mode 100644 binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/utils/BindingUtils.java diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/ConsumerConfigCustomizer.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/support/ConsumerConfigCustomizer.java similarity index 88% rename from binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/ConsumerConfigCustomizer.java rename to binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/support/ConsumerConfigCustomizer.java index adc184803..135d85d51 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/ConsumerConfigCustomizer.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/support/ConsumerConfigCustomizer.java @@ -1,5 +1,5 @@ /* - * Copyright 2020-2020 the original author or authors. + * Copyright 2020-2022 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. @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.cloud.stream.binder.kafka.config; +package org.springframework.cloud.stream.binder.kafka.support; import java.util.Map; @@ -29,4 +29,5 @@ import java.util.Map; public interface ConsumerConfigCustomizer { void configure(Map consumerProperties, String bindingName, String destination); + } diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/ProducerConfigCustomizer.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/support/ProducerConfigCustomizer.java similarity index 88% rename from binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/ProducerConfigCustomizer.java rename to binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/support/ProducerConfigCustomizer.java index 9dd5cd9d6..9a260cdf4 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/ProducerConfigCustomizer.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/support/ProducerConfigCustomizer.java @@ -1,5 +1,5 @@ /* - * Copyright 2020-2020 the original author or authors. + * Copyright 2020-2022 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. @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.cloud.stream.binder.kafka.config; +package org.springframework.cloud.stream.binder.kafka.support; import java.util.Map; @@ -29,4 +29,5 @@ import java.util.Map; public interface ProducerConfigCustomizer { void configure(Map producerProperties, String bindingName, String destination); + } diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/utils/BindingUtils.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/utils/BindingUtils.java new file mode 100644 index 000000000..e9a45fb90 --- /dev/null +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/utils/BindingUtils.java @@ -0,0 +1,230 @@ +/* + * Copyright 2022-2022 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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder.kafka.utils; + +import java.util.HashMap; +import java.util.Map; + +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.common.serialization.ByteArrayDeserializer; +import org.apache.kafka.common.serialization.ByteArraySerializer; + +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.NoSuchBeanDefinitionException; +import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; +import org.springframework.cloud.stream.binder.ExtendedProducerProperties; +import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties; +import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties; +import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties.StandardHeaders; +import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties; +import org.springframework.context.ApplicationContext; +import org.springframework.kafka.support.DefaultKafkaHeaderMapper; +import org.springframework.kafka.support.KafkaHeaderMapper; +import org.springframework.kafka.support.converter.MessageConverter; +import org.springframework.kafka.support.converter.MessagingMessageConverter; +import org.springframework.lang.Nullable; +import org.springframework.util.Assert; +import org.springframework.util.ObjectUtils; + +/** + * Binding Utilities. + * + * @author Gary Russell + * @since 4.0 + * + */ +public final class BindingUtils { + + private BindingUtils() { + } + + /** + * Get the message converter for consumer bindings from the application context. If + * the binding properties do not contain a bean name, a default + * {@link MessagingMessageConverter} is returned; if the binder properties contain a + * header mapper bean name, it is used in the default converter, otherwise a + * {@link DefaultKafkaHeaderMapper} is used. + * @param applicationContext the application context. + * @param extendedConsumerProperties the consumer binding properties. + * @param configurationProperties the binder properties. + * @return the converter + * @throws IllegalStateException if a bean name is specified but not found. + */ + public static MessageConverter getConsumerMessageConverter(ApplicationContext applicationContext, + ExtendedConsumerProperties extendedConsumerProperties, + KafkaBinderConfigurationProperties configurationProperties) { + + MessageConverter messageConverter; + if (extendedConsumerProperties.getExtension().getConverterBeanName() == null) { + MessagingMessageConverter mmc = new MessagingMessageConverter(); + StandardHeaders standardHeaders = extendedConsumerProperties.getExtension() + .getStandardHeaders(); + mmc.setGenerateMessageId(StandardHeaders.id.equals(standardHeaders) + || StandardHeaders.both.equals(standardHeaders)); + mmc.setGenerateTimestamp( + StandardHeaders.timestamp.equals(standardHeaders) + || StandardHeaders.both.equals(standardHeaders)); + KafkaHeaderMapper headerMapper = getHeaderMapper(applicationContext, configurationProperties); + if (headerMapper == null) { + headerMapper = new DefaultKafkaHeaderMapper(); + } + mmc.setHeaderMapper(headerMapper); + messageConverter = mmc; + } + else { + try { + messageConverter = applicationContext.getBean( + extendedConsumerProperties.getExtension().getConverterBeanName(), MessageConverter.class); + } + catch (NoSuchBeanDefinitionException ex) { + throw new IllegalStateException( + "Converter bean not present in application context", ex); + } + } + return messageConverter; + } + + /** + * Get the header mapper bean, if the binder properties contains a bean name; if not + * look for a bean with name {@code kafkaBinderHeaderMapper} is looked up; if that + * doesn't exist, null is returned. + * @param applicationContext the application context. + * @param configurationProperties the binder properties. + * @return the mapper. + */ + @Nullable + public static KafkaHeaderMapper getHeaderMapper(ApplicationContext applicationContext, + KafkaBinderConfigurationProperties configurationProperties) { + + KafkaHeaderMapper mapper = null; + if (configurationProperties.getHeaderMapperBeanName() != null) { + mapper = applicationContext.getBean( + configurationProperties.getHeaderMapperBeanName(), + KafkaHeaderMapper.class); + } + if (mapper == null) { + //First, try to see if there is a bean named headerMapper registered by other frameworks using the binder (for e.g. spring cloud sleuth) + try { + mapper = applicationContext.getBean("kafkaBinderHeaderMapper", KafkaHeaderMapper.class); + } + catch (BeansException be) { + } + } + return mapper; + } + + /** + * Create the Kafka configuration map for a consumer binding. With anonymous bindings + * (those without a {@code group} property, which are given a {@code UUID.toString()} + * in the group id) consumption begins from the current end of the topic, otherwise + * consumption starts from the beginning, the first time the binding consumes. + * @param anonymous true if this is for an anonymous binding. + * @param consumerGroup the group. + * @param consumerProperties the binding properties. + * @param configurationProperties the binder properties. + * @return the config map. + */ + public static Map createConsumerConfigs(boolean anonymous, String consumerGroup, + ExtendedConsumerProperties consumerProperties, + KafkaBinderConfigurationProperties configurationProperties) { + + Map props = new HashMap<>(); + props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, + ByteArrayDeserializer.class); + props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, + ByteArrayDeserializer.class); + props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); + props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 100); + props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, + anonymous ? "latest" : "earliest"); + props.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroup); + + Map mergedConfig = configurationProperties.mergedConsumerConfiguration(); + if (!ObjectUtils.isEmpty(mergedConfig)) { + props.putAll(mergedConfig); + } + if (ObjectUtils.isEmpty(props.get(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG))) { + props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, + configurationProperties.getKafkaConnectionString()); + } + Map config = consumerProperties.getExtension().getConfiguration(); + if (!ObjectUtils.isEmpty(config)) { + Assert.state(!config.containsKey(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG), + ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG + " cannot be overridden at the binding level; " + + "use multiple binders instead"); + props.putAll(config); + } + if (!ObjectUtils.isEmpty(consumerProperties.getExtension().getStartOffset())) { + props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, + consumerProperties.getExtension().getStartOffset().name()); + } + return props; + } + + /** + * Create the Kafka configuration map for a producer binding. + * @param producerProperties the binding properties. + * @param configurationProperties the binder properties. + * @return the config map. + */ + public static Map createProducerConfigs( + ExtendedProducerProperties producerProperties, + KafkaBinderConfigurationProperties configurationProperties) { + + Map props = new HashMap<>(); + props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class); + props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, + ByteArraySerializer.class); + props.put(ProducerConfig.ACKS_CONFIG, + String.valueOf(configurationProperties.getRequiredAcks())); + Map mergedConfig = configurationProperties + .mergedProducerConfiguration(); + if (!ObjectUtils.isEmpty(mergedConfig)) { + props.putAll(mergedConfig); + } + if (ObjectUtils.isEmpty(props.get(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG))) { + props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, + configurationProperties.getKafkaConnectionString()); + } + final KafkaProducerProperties kafkaProducerProperties = producerProperties.getExtension(); + if (ObjectUtils.isEmpty(props.get(ProducerConfig.BATCH_SIZE_CONFIG))) { + props.put(ProducerConfig.BATCH_SIZE_CONFIG, + String.valueOf(kafkaProducerProperties.getBufferSize())); + } + if (ObjectUtils.isEmpty(props.get(ProducerConfig.LINGER_MS_CONFIG))) { + props.put(ProducerConfig.LINGER_MS_CONFIG, + String.valueOf(kafkaProducerProperties.getBatchTimeout())); + } + if (ObjectUtils.isEmpty(props.get(ProducerConfig.COMPRESSION_TYPE_CONFIG))) { + props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, + kafkaProducerProperties.getCompressionType().toString()); + } + Map configs = producerProperties.getExtension().getConfiguration(); + Assert.state(!configs.containsKey(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG), + ProducerConfig.BOOTSTRAP_SERVERS_CONFIG + " cannot be overridden at the binding level; " + + "use multiple binders instead"); + if (!ObjectUtils.isEmpty(configs)) { + props.putAll(configs); + } + if (!ObjectUtils.isEmpty(kafkaProducerProperties.getConfiguration())) { + props.putAll(kafkaProducerProperties.getConfiguration()); + } + return props; + } + +} diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinder.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinder.java index 798777dec..c1bc8a424 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinder.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinder.java @@ -17,19 +17,14 @@ package org.springframework.cloud.stream.binder.reactorkafka; import java.util.Collections; -import java.util.HashMap; import java.util.Map; import java.util.UUID; import java.util.concurrent.atomic.AtomicReference; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.apache.kafka.clients.consumer.ConsumerConfig; -import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.clients.producer.RecordMetadata; -import org.apache.kafka.common.serialization.ByteArrayDeserializer; -import org.apache.kafka.common.serialization.ByteArraySerializer; import org.reactivestreams.Subscription; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -41,7 +36,6 @@ import reactor.kafka.sender.SenderOptions; import reactor.kafka.sender.SenderRecord; import reactor.kafka.sender.SenderResult; -import org.springframework.beans.factory.NoSuchBeanDefinitionException; import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder; import org.springframework.cloud.stream.binder.BinderSpecificPropertiesProvider; import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; @@ -49,24 +43,25 @@ import org.springframework.cloud.stream.binder.ExtendedProducerProperties; import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder; import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties; import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties; -import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties.StandardHeaders; import org.springframework.cloud.stream.binder.kafka.properties.KafkaExtendedBindingProperties; import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties; import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner; +import org.springframework.cloud.stream.binder.kafka.support.ConsumerConfigCustomizer; +import org.springframework.cloud.stream.binder.kafka.support.ProducerConfigCustomizer; +import org.springframework.cloud.stream.binder.kafka.utils.BindingUtils; import org.springframework.cloud.stream.provisioning.ConsumerDestination; import org.springframework.cloud.stream.provisioning.ProducerDestination; import org.springframework.context.Lifecycle; import org.springframework.integration.core.MessageProducer; import org.springframework.integration.endpoint.MessageProducerSupport; import org.springframework.integration.handler.AbstractMessageHandler; -import org.springframework.kafka.support.DefaultKafkaHeaderMapper; +import org.springframework.kafka.support.converter.MessageConverter; import org.springframework.kafka.support.converter.MessagingMessageConverter; import org.springframework.kafka.support.converter.RecordMessageConverter; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import org.springframework.util.Assert; -import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; /** @@ -86,6 +81,10 @@ public class ReactorKafkaBinder private KafkaExtendedBindingProperties extendedBindingProperties = new KafkaExtendedBindingProperties(); + private ConsumerConfigCustomizer consumerConfigCustomizer; + + private ProducerConfigCustomizer producerConfigCustomizer; + public ReactorKafkaBinder(KafkaBinderConfigurationProperties configurationProperties, KafkaTopicProvisioner provisioner) { @@ -93,21 +92,28 @@ public class ReactorKafkaBinder this.configurationProperties = configurationProperties; } + public void setConsumerConfigCustomizer(ConsumerConfigCustomizer consumerConfigCustomizer) { + this.consumerConfigCustomizer = consumerConfigCustomizer; + } + + public void setProducerConfigCustomizer(ProducerConfigCustomizer producerConfigCustomizer) { + this.producerConfigCustomizer = producerConfigCustomizer; + } @Override protected MessageHandler createProducerMessageHandler(ProducerDestination destination, ExtendedProducerProperties producerProperties, MessageChannel errorChannel) throws Exception { - Map configs = createProducerConfigs(producerProperties); - // TODO: Move config customizers to core -// if (this.producerConfigCustomizer != null) { -// this.producerConfigCustomizer.configure(props, bindingNameHolder.get(), destination); -// bindingNameHolder.remove(); -// } + Map configs = BindingUtils.createProducerConfigs(producerProperties, + this.configurationProperties); + if (this.producerConfigCustomizer != null) { + this.producerConfigCustomizer.configure(configs, producerProperties.getBindingName(), + destination.getName()); + } SenderOptions opts = SenderOptions.create(configs); - // TODO bean for converter. + // TODO bean for converter; MCB doesn't use one on the producer side. RecordMessageConverter converter = new MessagingMessageConverter(); return new ReactorMessageHandler(opts, converter, destination.getName()); } @@ -119,14 +125,16 @@ public class ReactorKafkaBinder boolean anonymous = !StringUtils.hasText(group); String consumerGroup = anonymous ? "anonymous." + UUID.randomUUID().toString() : group; - Map configs = createConsumerConfigs(anonymous, consumerGroup, properties); + Map configs = BindingUtils.createConsumerConfigs(anonymous, consumerGroup, properties, + this.configurationProperties); - // TODO: Move config customizers to core -// if (this.consumerConfigCustomizer != null) { -// this.consumerConfigCustomizer.configure(configs, bindingNameHolder.get(), destination); -// } + if (this.consumerConfigCustomizer != null) { + this.consumerConfigCustomizer.configure(configs, properties.getBindingName(), destination.getName()); + } - RecordMessageConverter converter = getMessageConverter(properties); + MessageConverter converter = BindingUtils.getConsumerMessageConverter(getApplicationContext(), properties, + this.configurationProperties); + Assert.isInstanceOf(RecordMessageConverter.class, converter); ReceiverOptions opts = ReceiverOptions.create(configs) .addAssignListener(parts -> System.out.println("Assigned: " + parts)) .subscription(Collections.singletonList(destination.getName())); @@ -143,7 +151,8 @@ public class ReactorKafkaBinder Flux> flux = receiver .receive() .doOnSubscribe(subs -> this.subscription = subs) - .map(record -> (Message) converter.toMessage(record, null, null, null)); + .map(record -> (Message) ((RecordMessageConverter) converter) + .toMessage(record, null, null, null)); subscribeToPublisher(flux); } @@ -158,126 +167,6 @@ public class ReactorKafkaBinder }; } - /* - * TODO: Copied (and modified) from Kafka binder - refactor to core - */ - private RecordMessageConverter getMessageConverter( - final ExtendedConsumerProperties extendedConsumerProperties) { - - RecordMessageConverter messageConverter; - if (extendedConsumerProperties.getExtension().getConverterBeanName() == null) { - MessagingMessageConverter mmc = new MessagingMessageConverter(); - StandardHeaders standardHeaders = extendedConsumerProperties.getExtension() - .getStandardHeaders(); - mmc.setGenerateMessageId(StandardHeaders.id.equals(standardHeaders) - || StandardHeaders.both.equals(standardHeaders)); - mmc.setGenerateTimestamp( - StandardHeaders.timestamp.equals(standardHeaders) - || StandardHeaders.both.equals(standardHeaders)); - mmc.setHeaderMapper(new DefaultKafkaHeaderMapper()); //TODO - messageConverter = mmc; - } - else { - try { - messageConverter = getApplicationContext().getBean( - extendedConsumerProperties.getExtension().getConverterBeanName(), - RecordMessageConverter.class); - } - catch (NoSuchBeanDefinitionException ex) { - throw new IllegalStateException( - "Converter bean not present in application context", ex); - } - } - return messageConverter; - } - - /* - * TODO: Copied from Kafka binder - refactor to core - */ - private Map createConsumerConfigs(boolean anonymous, String consumerGroup, - ExtendedConsumerProperties consumerProperties) { - - Map props = new HashMap<>(); - props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, - ByteArrayDeserializer.class); - props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, - ByteArrayDeserializer.class); - props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); - props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 100); - props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, - anonymous ? "latest" : "earliest"); - props.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroup); - - Map mergedConfig = this.configurationProperties - .mergedConsumerConfiguration(); - if (!ObjectUtils.isEmpty(mergedConfig)) { - props.putAll(mergedConfig); - } - if (ObjectUtils.isEmpty(props.get(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG))) { - props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, - this.configurationProperties.getKafkaConnectionString()); - } - Map config = consumerProperties.getExtension().getConfiguration(); - if (!ObjectUtils.isEmpty(config)) { - Assert.state(!config.containsKey(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG), - ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG + " cannot be overridden at the binding level; " - + "use multiple binders instead"); - props.putAll(config); - } - if (!ObjectUtils.isEmpty(consumerProperties.getExtension().getStartOffset())) { - props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, - consumerProperties.getExtension().getStartOffset().name()); - } - return props; - } - - /* - * TODO: Copied from Kafka binder - refactor to core - */ - private Map createProducerConfigs( - ExtendedProducerProperties producerProperties) { - - Map props = new HashMap<>(); - props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class); - props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, - ByteArraySerializer.class); - props.put(ProducerConfig.ACKS_CONFIG, - String.valueOf(this.configurationProperties.getRequiredAcks())); - Map mergedConfig = this.configurationProperties - .mergedProducerConfiguration(); - if (!ObjectUtils.isEmpty(mergedConfig)) { - props.putAll(mergedConfig); - } - if (ObjectUtils.isEmpty(props.get(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG))) { - props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, - this.configurationProperties.getKafkaConnectionString()); - } - final KafkaProducerProperties kafkaProducerProperties = producerProperties.getExtension(); - if (ObjectUtils.isEmpty(props.get(ProducerConfig.BATCH_SIZE_CONFIG))) { - props.put(ProducerConfig.BATCH_SIZE_CONFIG, - String.valueOf(kafkaProducerProperties.getBufferSize())); - } - if (ObjectUtils.isEmpty(props.get(ProducerConfig.LINGER_MS_CONFIG))) { - props.put(ProducerConfig.LINGER_MS_CONFIG, - String.valueOf(kafkaProducerProperties.getBatchTimeout())); - } - if (ObjectUtils.isEmpty(props.get(ProducerConfig.COMPRESSION_TYPE_CONFIG))) { - props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, - kafkaProducerProperties.getCompressionType().toString()); - } - Map configs = producerProperties.getExtension().getConfiguration(); - Assert.state(!configs.containsKey(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG), - ProducerConfig.BOOTSTRAP_SERVERS_CONFIG + " cannot be overridden at the binding level; " - + "use multiple binders instead"); - if (!ObjectUtils.isEmpty(configs)) { - props.putAll(configs); - } - if (!ObjectUtils.isEmpty(kafkaProducerProperties.getConfiguration())) { - props.putAll(kafkaProducerProperties.getConfiguration()); - } - return props; - } - @Override public KafkaConsumerProperties getExtendedConsumerProperties(String channelName) { return this.extendedBindingProperties.getExtendedConsumerProperties(channelName); diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderConfiguration.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderConfiguration.java index 1c88515f5..ceb2547ee 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderConfiguration.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderConfiguration.java @@ -25,6 +25,8 @@ import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfi import org.springframework.cloud.stream.binder.kafka.properties.KafkaExtendedBindingProperties; import org.springframework.cloud.stream.binder.kafka.provisioning.AdminClientConfigCustomizer; import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner; +import org.springframework.cloud.stream.binder.kafka.support.ConsumerConfigCustomizer; +import org.springframework.cloud.stream.binder.kafka.support.ProducerConfigCustomizer; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -56,10 +58,14 @@ public class ReactorKafkaBinderConfiguration { @Bean ReactorKafkaBinder reactorKafkaBinder(KafkaBinderConfigurationProperties configurationProperties, KafkaTopicProvisioner provisioningProvider, - KafkaExtendedBindingProperties extendedBindingProperties) { + KafkaExtendedBindingProperties extendedBindingProperties, + ObjectProvider consumerConfigCustomizer, + ObjectProvider producerConfigCustomizer) { ReactorKafkaBinder reactorKafkaBinder = new ReactorKafkaBinder(configurationProperties, provisioningProvider); reactorKafkaBinder.setExtendedBindingProperties(extendedBindingProperties); + reactorKafkaBinder.setConsumerConfigCustomizer(consumerConfigCustomizer.getIfUnique()); + reactorKafkaBinder.setProducerConfigCustomizer(producerConfigCustomizer.getIfUnique()); return reactorKafkaBinder; } diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index e4dbc06cb..3716c0a28 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java @@ -24,7 +24,6 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; import java.util.Collections; -import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Set; @@ -49,12 +48,8 @@ import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.header.Headers; import org.apache.kafka.common.header.internals.RecordHeader; import org.apache.kafka.common.header.internals.RecordHeaders; -import org.apache.kafka.common.serialization.ByteArrayDeserializer; -import org.apache.kafka.common.serialization.ByteArraySerializer; -import org.springframework.beans.BeansException; 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; @@ -66,14 +61,14 @@ import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder; import org.springframework.cloud.stream.binder.HeaderMode; import org.springframework.cloud.stream.binder.MessageValues; import org.springframework.cloud.stream.binder.kafka.config.ClientFactoryCustomizer; -import org.springframework.cloud.stream.binder.kafka.config.ConsumerConfigCustomizer; -import org.springframework.cloud.stream.binder.kafka.config.ProducerConfigCustomizer; import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties; import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties; -import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties.StandardHeaders; import org.springframework.cloud.stream.binder.kafka.properties.KafkaExtendedBindingProperties; import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties; import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner; +import org.springframework.cloud.stream.binder.kafka.support.ConsumerConfigCustomizer; +import org.springframework.cloud.stream.binder.kafka.support.ProducerConfigCustomizer; +import org.springframework.cloud.stream.binder.kafka.utils.BindingUtils; import org.springframework.cloud.stream.binder.kafka.utils.DlqDestinationResolver; import org.springframework.cloud.stream.binder.kafka.utils.DlqPartitionFunction; import org.springframework.cloud.stream.binding.DefaultPartitioningInterceptor; @@ -448,21 +443,7 @@ public class KafkaMessageChannelBinder extends AbstractApplicationContext applicationContext = getApplicationContext(); handler.setApplicationContext(applicationContext); - KafkaHeaderMapper mapper = null; - if (this.configurationProperties.getHeaderMapperBeanName() != null) { - mapper = applicationContext.getBean( - this.configurationProperties.getHeaderMapperBeanName(), - KafkaHeaderMapper.class); - } - if (mapper == null) { - //First, try to see if there is a bean named headerMapper registered by other frameworks using the binder (for e.g. spring cloud sleuth) - try { - mapper = applicationContext.getBean("kafkaBinderHeaderMapper", KafkaHeaderMapper.class); - } - catch (BeansException be) { - // Pass through - } - } + KafkaHeaderMapper mapper = BindingUtils.getHeaderMapper(applicationContext, this.configurationProperties); /* * Even if the user configures a bean, we must not use it if the header mode is @@ -530,44 +511,10 @@ public class KafkaMessageChannelBinder extends protected DefaultKafkaProducerFactory getProducerFactory( String transactionIdPrefix, ExtendedProducerProperties producerProperties, String beanName, String destination) { - Map props = new HashMap<>(); - props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class); - props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, - ByteArraySerializer.class); - props.put(ProducerConfig.ACKS_CONFIG, - String.valueOf(this.configurationProperties.getRequiredAcks())); - Map mergedConfig = this.configurationProperties - .mergedProducerConfiguration(); - if (!ObjectUtils.isEmpty(mergedConfig)) { - props.putAll(mergedConfig); - } - if (ObjectUtils.isEmpty(props.get(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG))) { - props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, - this.configurationProperties.getKafkaConnectionString()); - } + + Map props = BindingUtils.createProducerConfigs(producerProperties, + this.configurationProperties); final KafkaProducerProperties kafkaProducerProperties = producerProperties.getExtension(); - if (ObjectUtils.isEmpty(props.get(ProducerConfig.BATCH_SIZE_CONFIG))) { - props.put(ProducerConfig.BATCH_SIZE_CONFIG, - String.valueOf(kafkaProducerProperties.getBufferSize())); - } - if (ObjectUtils.isEmpty(props.get(ProducerConfig.LINGER_MS_CONFIG))) { - props.put(ProducerConfig.LINGER_MS_CONFIG, - String.valueOf(kafkaProducerProperties.getBatchTimeout())); - } - if (ObjectUtils.isEmpty(props.get(ProducerConfig.COMPRESSION_TYPE_CONFIG))) { - props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, - kafkaProducerProperties.getCompressionType().toString()); - } - Map configs = producerProperties.getExtension().getConfiguration(); - Assert.state(!configs.containsKey(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG), - ProducerConfig.BOOTSTRAP_SERVERS_CONFIG + " cannot be overridden at the binding level; " - + "use multiple binders instead"); - if (!ObjectUtils.isEmpty(configs)) { - props.putAll(configs); - } - if (!ObjectUtils.isEmpty(kafkaProducerProperties.getConfiguration())) { - props.putAll(kafkaProducerProperties.getConfiguration()); - } if (this.producerConfigCustomizer != null) { this.producerConfigCustomizer.configure(props, producerProperties.getBindingName(), destination); } @@ -1082,29 +1029,8 @@ public class KafkaMessageChannelBinder extends private MessageConverter getMessageConverter( final ExtendedConsumerProperties extendedConsumerProperties) { - MessageConverter messageConverter; - if (extendedConsumerProperties.getExtension().getConverterBeanName() == null) { - MessagingMessageConverter mmc = new MessagingMessageConverter(); - StandardHeaders standardHeaders = extendedConsumerProperties.getExtension() - .getStandardHeaders(); - mmc.setGenerateMessageId(StandardHeaders.id.equals(standardHeaders) - || StandardHeaders.both.equals(standardHeaders)); - mmc.setGenerateTimestamp( - StandardHeaders.timestamp.equals(standardHeaders) - || StandardHeaders.both.equals(standardHeaders)); - messageConverter = mmc; - } - else { - try { - messageConverter = getApplicationContext().getBean( - extendedConsumerProperties.getExtension().getConverterBeanName(), - MessageConverter.class); - } - catch (NoSuchBeanDefinitionException ex) { - throw new IllegalStateException( - "Converter bean not present in application context", ex); - } - } + MessageConverter messageConverter = BindingUtils.getConsumerMessageConverter(getApplicationContext(), + extendedConsumerProperties, this.configurationProperties); if (messageConverter instanceof MessagingMessageConverter) { ((MessagingMessageConverter) messageConverter).setHeaderMapper(getHeaderMapper(extendedConsumerProperties)); } @@ -1113,36 +1039,26 @@ public class KafkaMessageChannelBinder extends private KafkaHeaderMapper getHeaderMapper( final ExtendedConsumerProperties extendedConsumerProperties) { - KafkaHeaderMapper mapper = null; - if (this.configurationProperties.getHeaderMapperBeanName() != null) { - mapper = getApplicationContext().getBean( - this.configurationProperties.getHeaderMapperBeanName(), - KafkaHeaderMapper.class); - } + + KafkaHeaderMapper mapper = BindingUtils.getHeaderMapper(getApplicationContext(), this.configurationProperties); if (mapper == null) { - //First, try to see if there is a bean named headerMapper registered by other frameworks using the binder (for e.g. spring cloud sleuth) - try { - mapper = getApplicationContext().getBean("kafkaBinderHeaderMapper", KafkaHeaderMapper.class); - } - catch (BeansException be) { - BinderHeaderMapper headerMapper = new BinderHeaderMapper() { + BinderHeaderMapper headerMapper = new BinderHeaderMapper() { - @Override - public void toHeaders(Headers source, Map headers) { - super.toHeaders(source, headers); - if (headers.size() > 0) { - headers.put(BinderHeaders.NATIVE_HEADERS_PRESENT, Boolean.TRUE); - } + @Override + public void toHeaders(Headers source, Map headers) { + super.toHeaders(source, headers); + if (headers.size() > 0) { + headers.put(BinderHeaders.NATIVE_HEADERS_PRESENT, Boolean.TRUE); } - - }; - String[] trustedPackages = extendedConsumerProperties.getExtension() - .getTrustedPackages(); - if (!StringUtils.isEmpty(trustedPackages)) { - headerMapper.addTrustedPackages(trustedPackages); } - mapper = headerMapper; + + }; + String[] trustedPackages = extendedConsumerProperties.getExtension() + .getTrustedPackages(); + if (!ObjectUtils.isEmpty(trustedPackages)) { + headerMapper.addTrustedPackages(trustedPackages); } + mapper = headerMapper; } return mapper; } @@ -1418,37 +1334,9 @@ public class KafkaMessageChannelBinder extends protected ConsumerFactory createKafkaConsumerFactory(boolean anonymous, String consumerGroup, ExtendedConsumerProperties consumerProperties, String beanName, String destination) { - Map props = new HashMap<>(); - props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, - ByteArrayDeserializer.class); - props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, - ByteArrayDeserializer.class); - props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); - props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 100); - props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, - anonymous ? "latest" : "earliest"); - props.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroup); - Map mergedConfig = this.configurationProperties - .mergedConsumerConfiguration(); - if (!ObjectUtils.isEmpty(mergedConfig)) { - props.putAll(mergedConfig); - } - if (ObjectUtils.isEmpty(props.get(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG))) { - props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, - this.configurationProperties.getKafkaConnectionString()); - } - Map config = consumerProperties.getExtension().getConfiguration(); - if (!ObjectUtils.isEmpty(config)) { - Assert.state(!config.containsKey(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG), - ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG + " cannot be overridden at the binding level; " - + "use multiple binders instead"); - props.putAll(config); - } - if (!ObjectUtils.isEmpty(consumerProperties.getExtension().getStartOffset())) { - props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, - consumerProperties.getExtension().getStartOffset().name()); - } + Map props = BindingUtils.createConsumerConfigs(anonymous, consumerGroup, consumerProperties, + this.configurationProperties); if (this.consumerConfigCustomizer != null) { this.consumerConfigCustomizer.configure(props, consumerProperties.getBindingName(), destination); diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java index 7b31b806f..d89fed11b 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java @@ -37,6 +37,8 @@ import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfi import org.springframework.cloud.stream.binder.kafka.properties.KafkaExtendedBindingProperties; import org.springframework.cloud.stream.binder.kafka.provisioning.AdminClientConfigCustomizer; import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner; +import org.springframework.cloud.stream.binder.kafka.support.ConsumerConfigCustomizer; +import org.springframework.cloud.stream.binder.kafka.support.ProducerConfigCustomizer; import org.springframework.cloud.stream.binder.kafka.utils.DlqDestinationResolver; import org.springframework.cloud.stream.binder.kafka.utils.DlqPartitionFunction; import org.springframework.cloud.stream.config.ConsumerEndpointCustomizer; diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaConfigCustomizationTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaConfigCustomizationTests.java index ba09b90e7..f82cd8890 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaConfigCustomizationTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaConfigCustomizationTests.java @@ -36,8 +36,8 @@ import org.junit.jupiter.api.extension.ExtendWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.boot.test.context.SpringBootTest; -import org.springframework.cloud.stream.binder.kafka.config.ConsumerConfigCustomizer; -import org.springframework.cloud.stream.binder.kafka.config.ProducerConfigCustomizer; +import org.springframework.cloud.stream.binder.kafka.support.ConsumerConfigCustomizer; +import org.springframework.cloud.stream.binder.kafka.support.ProducerConfigCustomizer; import org.springframework.context.annotation.Bean; import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate;