configuration) {
+ this.kafkaProducerProperties.setConfiguration(configuration);
+ }
+
+ public KafkaTopicProperties getTopic() {
+ return this.kafkaProducerProperties.getTopic();
+ }
+
+ public void setTopic(KafkaTopicProperties topic) {
+ this.kafkaProducerProperties.setTopic(topic);
+ }
+
+ public KafkaProducerProperties getExtension() {
+ return this.kafkaProducerProperties;
+ }
+
+ }
+
+}
diff --git a/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBindingProperties.java b/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBindingProperties.java
new file mode 100644
index 000000000..e84b0b2c7
--- /dev/null
+++ b/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBindingProperties.java
@@ -0,0 +1,63 @@
+/*
+ * 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.
+ * 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.properties;
+
+import org.springframework.cloud.stream.binder.BinderSpecificPropertiesProvider;
+
+/**
+ * Container object for Kafka specific extended producer and consumer binding properties.
+ *
+ * @author Marius Bogoevici
+ * @author Oleg Zhurakousky
+ */
+public class KafkaBindingProperties implements BinderSpecificPropertiesProvider {
+
+ /**
+ * Consumer specific binding properties. @see {@link KafkaConsumerProperties}.
+ */
+ private KafkaConsumerProperties consumer = new KafkaConsumerProperties();
+
+ /**
+ * Producer specific binding properties. @see {@link KafkaProducerProperties}.
+ */
+ private KafkaProducerProperties producer = new KafkaProducerProperties();
+
+ /**
+ * @return {@link KafkaConsumerProperties}
+ * Consumer specific binding properties. @see {@link KafkaConsumerProperties}.
+ */
+ public KafkaConsumerProperties getConsumer() {
+ return this.consumer;
+ }
+
+ public void setConsumer(KafkaConsumerProperties consumer) {
+ this.consumer = consumer;
+ }
+
+ /**
+ * @return {@link KafkaProducerProperties}
+ * Producer specific binding properties. @see {@link KafkaProducerProperties}.
+ */
+ public KafkaProducerProperties getProducer() {
+ return this.producer;
+ }
+
+ public void setProducer(KafkaProducerProperties producer) {
+ this.producer = producer;
+ }
+
+}
diff --git a/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java b/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java
new file mode 100644
index 000000000..0a4d2561d
--- /dev/null
+++ b/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java
@@ -0,0 +1,545 @@
+/*
+ * Copyright 2016-2021 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.properties;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.springframework.kafka.listener.ContainerProperties;
+
+/**
+ * Extended consumer properties for Kafka binder.
+ *
+ * @author Marius Bogoevici
+ * @author Ilayaperumal Gopinathan
+ * @author Soby Chacko
+ * @author Gary Russell
+ * @author Aldo Sinanaj
+ *
+ *
+ * Thanks to Laszlo Szabo for providing the initial patch for generic property support.
+ *
+ */
+public class KafkaConsumerProperties {
+
+ /**
+ * Enumeration for starting consumer offset.
+ */
+ public enum StartOffset {
+
+ /**
+ * Starting from earliest offset.
+ */
+ earliest(-2L),
+ /**
+ * Starting from latest offset.
+ */
+ latest(-1L);
+
+ private final long referencePoint;
+
+ StartOffset(long referencePoint) {
+ this.referencePoint = referencePoint;
+ }
+
+ public long getReferencePoint() {
+ return this.referencePoint;
+ }
+
+ }
+
+ /**
+ * Standard headers for the message.
+ */
+ public enum StandardHeaders {
+
+ /**
+ * No headers.
+ */
+ none,
+ /**
+ * Message header representing ID.
+ */
+ id,
+ /**
+ * Message header representing timestamp.
+ */
+ timestamp,
+ /**
+ * Indicating both ID and timestamp headers.
+ */
+ both
+
+ }
+
+ /**
+ * When true the offset is committed after each record, otherwise the offsets for the complete set of records
+ * received from the poll() are committed after all records have been processed.
+ */
+ @Deprecated
+ private boolean ackEachRecord;
+
+ /**
+ * When true, topic partitions is automatically rebalanced between the members of a consumer group.
+ * When false, each consumer is assigned a fixed set of partitions based on spring.cloud.stream.instanceCount and spring.cloud.stream.instanceIndex.
+ */
+ private boolean autoRebalanceEnabled = true;
+
+ /**
+ * Whether to autocommit offsets when a message has been processed.
+ * If set to false, a header with the key kafka_acknowledgment of the type org.springframework.kafka.support.Acknowledgment header
+ * is present in the inbound message. Applications may use this header for acknowledging messages.
+ */
+ @Deprecated
+ private boolean autoCommitOffset = true;
+
+ /**
+ * Controlling the container acknowledgement mode. This is the preferred way to control the ack mode on the
+ * container instead of the deprecated autoCommitOffset property.
+ */
+ private ContainerProperties.AckMode ackMode;
+
+ /**
+ * Flag to enable auto commit on error in polled consumers.
+ */
+ private Boolean autoCommitOnError;
+
+ /**
+ * The starting offset for new groups. Allowed values: earliest and latest.
+ */
+ private StartOffset startOffset;
+
+ /**
+ * Whether to reset offsets on the consumer to the value provided by startOffset.
+ * Must be false if a KafkaRebalanceListener is provided.
+ */
+ private boolean resetOffsets;
+
+ /**
+ * When set to true, it enables DLQ behavior for the consumer.
+ * By default, messages that result in errors are forwarded to a topic named error.name-of-destination.name-of-group.
+ * The DLQ topic name can be configurable by setting the dlqName property.
+ */
+ private boolean enableDlq;
+
+ /**
+ * The name of the DLQ topic to receive the error messages.
+ */
+ private String dlqName;
+
+ /**
+ * Number of partitions to use on the DLQ.
+ */
+ private Integer dlqPartitions;
+
+ /**
+ * Using this, DLQ-specific producer properties can be set.
+ * All the properties available through kafka producer properties can be set through this property.
+ */
+ private KafkaProducerProperties dlqProducerProperties = new KafkaProducerProperties();
+
+ /**
+ * @deprecated No longer used by the binder.
+ */
+ @Deprecated
+ private int recoveryInterval = 5000;
+
+ /**
+ * List of trusted packages to provide the header mapper.
+ */
+ private String[] trustedPackages;
+
+ /**
+ * Indicates which standard headers are populated by the inbound channel adapter.
+ * Allowed values: none, id, timestamp, or both.
+ */
+ private StandardHeaders standardHeaders = StandardHeaders.none;
+
+ /**
+ * The name of a bean that implements RecordMessageConverter.
+ */
+ private String converterBeanName;
+
+ /**
+ * The interval, in milliseconds, between events indicating that no messages have recently been received.
+ */
+ private long idleEventInterval = 30_000;
+
+ /**
+ * When true, the destination is treated as a regular expression Pattern used to match topic names by the broker.
+ */
+ private boolean destinationIsPattern;
+
+ /**
+ * Map with a key/value pair containing generic Kafka consumer properties.
+ * In addition to having Kafka consumer properties, other configuration properties can be passed here.
+ */
+ private Map configuration = new HashMap<>();
+
+ /**
+ * Various topic level properties. @see {@link KafkaTopicProperties} for more details.
+ */
+ private KafkaTopicProperties topic = new KafkaTopicProperties();
+
+ /**
+ * Timeout used for polling in pollable consumers.
+ */
+ private long pollTimeout = org.springframework.kafka.listener.ConsumerProperties.DEFAULT_POLL_TIMEOUT;
+
+ /**
+ * Transaction manager bean name - overrides the binder's transaction configuration.
+ */
+ private String transactionManager;
+
+ /**
+ * Set to false to NOT commit the offset of a successfully recovered recovered in the after rollback processor.
+ */
+ private boolean txCommitRecovered = true;
+
+ /**
+ * CommonErrorHandler bean name per consumer binding.
+ * @since 3.2
+ */
+ private String commonErrorHandlerBeanName;
+
+ /**
+ * @return if each record needs to be acknowledged.
+ *
+ * When true the offset is committed after each record, otherwise the offsets for the complete set of records
+ * received from the poll() are committed after all records have been processed.
+ *
+ * @deprecated since 3.1 in favor of using {@link #ackMode}
+ */
+ @Deprecated
+ public boolean isAckEachRecord() {
+ return this.ackEachRecord;
+ }
+
+ /**
+ * @param ackEachRecord
+ *
+ * @deprecated in favor of using {@link #ackMode}
+ */
+ @Deprecated
+ public void setAckEachRecord(boolean ackEachRecord) {
+ this.ackEachRecord = ackEachRecord;
+ }
+
+ /**
+ * @return is autocommit offset enabled
+ *
+ * Whether to autocommit offsets when a message has been processed.
+ * If set to false, a header with the key kafka_acknowledgment of the type org.springframework.kafka.support.Acknowledgment header
+ * is present in the inbound message. Applications may use this header for acknowledging messages.
+ *
+ * @deprecated since 3.1 in favor of using {@link #ackMode}
+ */
+ @Deprecated
+ public boolean isAutoCommitOffset() {
+ return this.autoCommitOffset;
+ }
+
+ /**
+ * @param autoCommitOffset
+ *
+ * @deprecated in favor of using {@link #ackMode}
+ */
+ @Deprecated
+ public void setAutoCommitOffset(boolean autoCommitOffset) {
+ this.autoCommitOffset = autoCommitOffset;
+ }
+
+ /**
+ * @return Container's ack mode.
+ */
+ public ContainerProperties.AckMode getAckMode() {
+ return this.ackMode;
+ }
+
+ public void setAckMode(ContainerProperties.AckMode ackMode) {
+ this.ackMode = ackMode;
+ }
+
+ /**
+ * @return start offset
+ *
+ * The starting offset for new groups. Allowed values: earliest and latest.
+ */
+ public StartOffset getStartOffset() {
+ return this.startOffset;
+ }
+
+ public void setStartOffset(StartOffset startOffset) {
+ this.startOffset = startOffset;
+ }
+
+ /**
+ * @return if resetting offset is enabled
+ *
+ * Whether to reset offsets on the consumer to the value provided by startOffset.
+ * Must be false if a KafkaRebalanceListener is provided.
+ */
+ public boolean isResetOffsets() {
+ return this.resetOffsets;
+ }
+
+ public void setResetOffsets(boolean resetOffsets) {
+ this.resetOffsets = resetOffsets;
+ }
+
+ /**
+ * @return is DLQ enabled.
+ *
+ * When set to true, it enables DLQ behavior for the consumer.
+ * By default, messages that result in errors are forwarded to a topic named error.name-of-destination.name-of-group.
+ * The DLQ topic name can be configurable by setting the dlqName property.
+ */
+ public boolean isEnableDlq() {
+ return this.enableDlq;
+ }
+
+ public void setEnableDlq(boolean enableDlq) {
+ this.enableDlq = enableDlq;
+ }
+
+ /**
+ * @return is autocommit on error in polled consumers.
+ *
+ * This property accessor is only used in polled consumers.
+ */
+ public Boolean getAutoCommitOnError() {
+ return this.autoCommitOnError;
+ }
+
+ /**
+ *
+ * @param autoCommitOnError commit on error in polled consumers.
+ *
+ */
+ public void setAutoCommitOnError(Boolean autoCommitOnError) {
+ this.autoCommitOnError = autoCommitOnError;
+ }
+
+ /**
+ * No longer used.
+ * @return the interval.
+ * @deprecated No longer used by the binder
+ */
+ @Deprecated
+ public int getRecoveryInterval() {
+ return this.recoveryInterval;
+ }
+
+ /**
+ * No longer used.
+ * @param recoveryInterval the interval.
+ * @deprecated No longer needed by the binder
+ */
+ @Deprecated
+ public void setRecoveryInterval(int recoveryInterval) {
+ this.recoveryInterval = recoveryInterval;
+ }
+
+ /**
+ * @return is auto rebalance enabled
+ *
+ * When true, topic partitions is automatically rebalanced between the members of a consumer group.
+ * When false, each consumer is assigned a fixed set of partitions based on spring.cloud.stream.instanceCount and spring.cloud.stream.instanceIndex.
+ */
+ public boolean isAutoRebalanceEnabled() {
+ return this.autoRebalanceEnabled;
+ }
+
+ public void setAutoRebalanceEnabled(boolean autoRebalanceEnabled) {
+ this.autoRebalanceEnabled = autoRebalanceEnabled;
+ }
+
+ /**
+ * @return a map of configuration
+ *
+ * Map with a key/value pair containing generic Kafka consumer properties.
+ * In addition to having Kafka consumer properties, other configuration properties can be passed here.
+ */
+ public Map getConfiguration() {
+ return this.configuration;
+ }
+
+ public void setConfiguration(Map configuration) {
+ this.configuration = configuration;
+ }
+
+ /**
+ * @return dlq name
+ *
+ * The name of the DLQ topic to receive the error messages.
+ */
+ public String getDlqName() {
+ return this.dlqName;
+ }
+
+ public void setDlqName(String dlqName) {
+ this.dlqName = dlqName;
+ }
+
+ /**
+ * @return number of partitions on the DLQ topic
+ *
+ * Number of partitions to use on the DLQ.
+ */
+ public Integer getDlqPartitions() {
+ return this.dlqPartitions;
+ }
+
+ public void setDlqPartitions(Integer dlqPartitions) {
+ this.dlqPartitions = dlqPartitions;
+ }
+
+ /**
+ * @return trusted packages
+ *
+ * List of trusted packages to provide the header mapper.
+ */
+ public String[] getTrustedPackages() {
+ return this.trustedPackages;
+ }
+
+ public void setTrustedPackages(String[] trustedPackages) {
+ this.trustedPackages = trustedPackages;
+ }
+
+ /**
+ * @return dlq producer properties
+ *
+ * Using this, DLQ-specific producer properties can be set.
+ * All the properties available through kafka producer properties can be set through this property.
+ */
+ public KafkaProducerProperties getDlqProducerProperties() {
+ return this.dlqProducerProperties;
+ }
+
+ public void setDlqProducerProperties(KafkaProducerProperties dlqProducerProperties) {
+ this.dlqProducerProperties = dlqProducerProperties;
+ }
+
+ /**
+ * @return standard headers
+ *
+ * Indicates which standard headers are populated by the inbound channel adapter.
+ * Allowed values: none, id, timestamp, or both.
+ */
+ public StandardHeaders getStandardHeaders() {
+ return this.standardHeaders;
+ }
+
+ public void setStandardHeaders(StandardHeaders standardHeaders) {
+ this.standardHeaders = standardHeaders;
+ }
+
+ /**
+ * @return converter bean name
+ *
+ * The name of a bean that implements RecordMessageConverter.
+ */
+ public String getConverterBeanName() {
+ return this.converterBeanName;
+ }
+
+ public void setConverterBeanName(String converterBeanName) {
+ this.converterBeanName = converterBeanName;
+ }
+
+ /**
+ * @return idle event interval
+ *
+ * The interval, in milliseconds, between events indicating that no messages have recently been received.
+ */
+ public long getIdleEventInterval() {
+ return this.idleEventInterval;
+ }
+
+ public void setIdleEventInterval(long idleEventInterval) {
+ this.idleEventInterval = idleEventInterval;
+ }
+
+ /**
+ * @return is destination given through a pattern
+ *
+ * When true, the destination is treated as a regular expression Pattern used to match topic names by the broker.
+ */
+ public boolean isDestinationIsPattern() {
+ return this.destinationIsPattern;
+ }
+
+ public void setDestinationIsPattern(boolean destinationIsPattern) {
+ this.destinationIsPattern = destinationIsPattern;
+ }
+
+ /**
+ * @return topic properties
+ *
+ * Various topic level properties. @see {@link KafkaTopicProperties} for more details.
+ */
+ public KafkaTopicProperties getTopic() {
+ return this.topic;
+ }
+
+ public void setTopic(KafkaTopicProperties topic) {
+ this.topic = topic;
+ }
+
+ /**
+ * @return timeout in pollable consumers
+ *
+ * Timeout used for polling in pollable consumers.
+ */
+ public long getPollTimeout() {
+ return this.pollTimeout;
+ }
+
+ public void setPollTimeout(long pollTimeout) {
+ this.pollTimeout = pollTimeout;
+ }
+
+ /**
+ * @return the transaction manager bean name.
+ *
+ * Transaction manager bean name (must be {@code KafkaAwareTransactionManager}.
+ */
+ public String getTransactionManager() {
+ return this.transactionManager;
+ }
+
+ public void setTransactionManager(String transactionManager) {
+ this.transactionManager = transactionManager;
+ }
+
+ public boolean isTxCommitRecovered() {
+ return this.txCommitRecovered;
+ }
+
+ public void setTxCommitRecovered(boolean txCommitRecovered) {
+ this.txCommitRecovered = txCommitRecovered;
+ }
+
+ public String getCommonErrorHandlerBeanName() {
+ return commonErrorHandlerBeanName;
+ }
+
+ public void setCommonErrorHandlerBeanName(String commonErrorHandlerBeanName) {
+ this.commonErrorHandlerBeanName = commonErrorHandlerBeanName;
+ }
+}
diff --git a/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaExtendedBindingProperties.java b/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaExtendedBindingProperties.java
new file mode 100644
index 000000000..99357cc2a
--- /dev/null
+++ b/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaExtendedBindingProperties.java
@@ -0,0 +1,55 @@
+/*
+ * 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.
+ * 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.properties;
+
+import java.util.Map;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.cloud.stream.binder.AbstractExtendedBindingProperties;
+import org.springframework.cloud.stream.binder.BinderSpecificPropertiesProvider;
+
+/**
+ * Kafka specific extended binding properties class that extends from
+ * {@link AbstractExtendedBindingProperties}.
+ *
+ * @author Marius Bogoevici
+ * @author Gary Russell
+ * @author Soby Chacko
+ * @author Oleg Zhurakousky
+ */
+@ConfigurationProperties("spring.cloud.stream.kafka")
+public class KafkaExtendedBindingProperties extends
+ AbstractExtendedBindingProperties {
+
+ private static final String DEFAULTS_PREFIX = "spring.cloud.stream.kafka.default";
+
+ @Override
+ public String getDefaultsPrefix() {
+ return DEFAULTS_PREFIX;
+ }
+
+ @Override
+ public Map getBindings() {
+ return this.doGetBindings();
+ }
+
+ @Override
+ public Class getExtendedPropertiesEntryClass() {
+ return KafkaBindingProperties.class;
+ }
+
+}
diff --git a/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaProducerProperties.java b/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaProducerProperties.java
new file mode 100644
index 000000000..54ca36713
--- /dev/null
+++ b/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaProducerProperties.java
@@ -0,0 +1,327 @@
+/*
+ * 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.
+ * 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.properties;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import jakarta.validation.constraints.NotNull;
+
+import org.springframework.expression.Expression;
+
+/**
+ * Extended producer properties for Kafka binder.
+ *
+ * @author Marius Bogoevici
+ * @author Henryk Konsek
+ * @author Gary Russell
+ * @author Aldo Sinanaj
+ */
+public class KafkaProducerProperties {
+
+ /**
+ * Upper limit, in bytes, of how much data the Kafka producer attempts to batch before sending.
+ */
+ private int bufferSize = 16384;
+
+ /**
+ * Set the compression.type producer property. Supported values are none, gzip, snappy and lz4.
+ * See {@link CompressionType} for more details.
+ */
+ private CompressionType compressionType = CompressionType.none;
+
+ /**
+ * Whether the producer is synchronous.
+ */
+ private boolean sync;
+
+ /**
+ * A SpEL expression evaluated against the outgoing message used to evaluate the time to wait
+ * for ack when synchronous publish is enabled.
+ */
+ private Expression sendTimeoutExpression;
+
+ /**
+ * How long the producer waits to allow more messages to accumulate in the same batch before sending the messages.
+ */
+ private int batchTimeout;
+
+ /**
+ * A SpEL expression evaluated against the outgoing message used to populate the key of the produced Kafka message.
+ */
+ private Expression messageKeyExpression;
+
+ /**
+ * A comma-delimited list of simple patterns to match Spring messaging headers
+ * to be mapped to the Kafka Headers in the ProducerRecord.
+ */
+ private String[] headerPatterns;
+
+ /**
+ * Map with a key/value pair containing generic Kafka producer properties.
+ */
+ private Map configuration = new HashMap<>();
+
+ /**
+ * Various topic level properties. @see {@link KafkaTopicProperties} for more details.
+ */
+ private KafkaTopicProperties topic = new KafkaTopicProperties();
+
+ /**
+ * Set to true to override the default binding destination (topic name) with the value of the
+ * KafkaHeaders.TOPIC message header in the outbound message. If the header is not present,
+ * the default binding destination is used.
+ */
+ private boolean useTopicHeader;
+
+ /**
+ * The bean name of a MessageChannel to which successful send results should be sent;
+ * the bean must exist in the application context.
+ */
+ private String recordMetadataChannel;
+
+ /**
+ * Transaction manager bean name - overrides the binder's transaction configuration.
+ */
+ private String transactionManager;
+
+ /*
+ * Timeout value in seconds for the duration to wait when closing the producer.
+ * If not set this defaults to 30 seconds.
+ */
+ private int closeTimeout;
+
+ /**
+ * Set to true to disable transactions.
+ */
+ private boolean allowNonTransactional;
+
+ /**
+ * @return buffer size
+ *
+ * Upper limit, in bytes, of how much data the Kafka producer attempts to batch before sending.
+ */
+ public int getBufferSize() {
+ return this.bufferSize;
+ }
+
+ public void setBufferSize(int bufferSize) {
+ this.bufferSize = bufferSize;
+ }
+
+ /**
+ * @return compression type {@link CompressionType}
+ *
+ * Set the compression.type producer property. Supported values are none, gzip, snappy, lz4 and zstd.
+ * See {@link CompressionType} for more details.
+ */
+ @NotNull
+ public CompressionType getCompressionType() {
+ return this.compressionType;
+ }
+
+ public void setCompressionType(CompressionType compressionType) {
+ this.compressionType = compressionType;
+ }
+
+ /**
+ * @return if synchronous sending is enabled
+ *
+ * Whether the producer is synchronous.
+ */
+ public boolean isSync() {
+ return this.sync;
+ }
+
+ public void setSync(boolean sync) {
+ this.sync = sync;
+ }
+
+ /**
+ * @return timeout expression for send
+ *
+ * A SpEL expression evaluated against the outgoing message used to evaluate the time to wait
+ * for ack when synchronous publish is enabled.
+ */
+ public Expression getSendTimeoutExpression() {
+ return this.sendTimeoutExpression;
+ }
+
+ public void setSendTimeoutExpression(Expression sendTimeoutExpression) {
+ this.sendTimeoutExpression = sendTimeoutExpression;
+ }
+
+ /**
+ * @return batch timeout
+ *
+ * How long the producer waits to allow more messages to accumulate in the same batch before sending the messages.
+ */
+ public int getBatchTimeout() {
+ return this.batchTimeout;
+ }
+
+ public void setBatchTimeout(int batchTimeout) {
+ this.batchTimeout = batchTimeout;
+ }
+
+ /**
+ * @return message key expression
+ *
+ * A SpEL expression evaluated against the outgoing message used to populate the key of the produced Kafka message.
+ */
+ public Expression getMessageKeyExpression() {
+ return this.messageKeyExpression;
+ }
+
+ public void setMessageKeyExpression(Expression messageKeyExpression) {
+ this.messageKeyExpression = messageKeyExpression;
+ }
+
+ /**
+ * @return header patterns
+ *
+ * A comma-delimited list of simple patterns to match Spring messaging headers
+ * to be mapped to the Kafka Headers in the ProducerRecord.
+ */
+ public String[] getHeaderPatterns() {
+ return this.headerPatterns;
+ }
+
+ public void setHeaderPatterns(String[] headerPatterns) {
+ this.headerPatterns = headerPatterns;
+ }
+
+ /**
+ * @return map of configuration
+ *
+ * Map with a key/value pair containing generic Kafka producer properties.
+ */
+ public Map getConfiguration() {
+ return this.configuration;
+ }
+
+ public void setConfiguration(Map configuration) {
+ this.configuration = configuration;
+ }
+
+ /**
+ * @return topic properties
+ *
+ * Various topic level properties. @see {@link KafkaTopicProperties} for more details.
+ */
+ public KafkaTopicProperties getTopic() {
+ return this.topic;
+ }
+
+ public void setTopic(KafkaTopicProperties topic) {
+ this.topic = topic;
+ }
+
+ /**
+ * @return if using topic header
+ *
+ * Set to true to override the default binding destination (topic name) with the value of the
+ * KafkaHeaders.TOPIC message header in the outbound message. If the header is not present,
+ * the default binding destination is used.
+ */
+ public boolean isUseTopicHeader() {
+ return this.useTopicHeader;
+ }
+
+ public void setUseTopicHeader(boolean useTopicHeader) {
+ this.useTopicHeader = useTopicHeader;
+ }
+
+ /**
+ * @return record metadata channel
+ *
+ * The bean name of a MessageChannel to which successful send results should be sent;
+ * the bean must exist in the application context.
+ */
+ public String getRecordMetadataChannel() {
+ return this.recordMetadataChannel;
+ }
+
+ public void setRecordMetadataChannel(String recordMetadataChannel) {
+ this.recordMetadataChannel = recordMetadataChannel;
+ }
+
+ /**
+ * @return the transaction manager bean name.
+ *
+ * Transaction manager bean name (must be {@code KafkaAwareTransactionManager}.
+ */
+ public String getTransactionManager() {
+ return this.transactionManager;
+ }
+
+ public void setTransactionManager(String transactionManager) {
+ this.transactionManager = transactionManager;
+ }
+
+ /*
+ * @return timeout in seconds for closing the producer
+ */
+ public int getCloseTimeout() {
+ return this.closeTimeout;
+ }
+
+ public void setCloseTimeout(int closeTimeout) {
+ this.closeTimeout = closeTimeout;
+ }
+
+ public boolean isAllowNonTransactional() {
+ return this.allowNonTransactional;
+ }
+
+ public void setAllowNonTransactional(boolean allowNonTransactional) {
+ this.allowNonTransactional = allowNonTransactional;
+ }
+
+ /**
+ * Enumeration for compression types.
+ */
+ public enum CompressionType {
+
+ /**
+ * No compression.
+ */
+ none,
+
+ /**
+ * gzip based compression.
+ */
+ gzip,
+
+ /**
+ * snappy based compression.
+ */
+ snappy,
+
+ /**
+ * lz4 compression.
+ */
+ lz4,
+
+ /**
+ * zstd compression.
+ */
+ zstd,
+
+ }
+
+}
diff --git a/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaTopicProperties.java b/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaTopicProperties.java
new file mode 100644
index 000000000..e9a1425b7
--- /dev/null
+++ b/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaTopicProperties.java
@@ -0,0 +1,62 @@
+/*
+ * Copyright 2019-2019 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.properties;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Properties for configuring topics.
+ *
+ * @author Aldo Sinanaj
+ * @since 2.2
+ *
+ */
+public class KafkaTopicProperties {
+
+ private Short replicationFactor;
+
+ private Map> replicasAssignments = new HashMap<>();
+
+ private Map properties = new HashMap<>();
+
+ public Short getReplicationFactor() {
+ return replicationFactor;
+ }
+
+ public void setReplicationFactor(Short replicationFactor) {
+ this.replicationFactor = replicationFactor;
+ }
+
+ public Map> getReplicasAssignments() {
+ return replicasAssignments;
+ }
+
+ public void setReplicasAssignments(Map> replicasAssignments) {
+ this.replicasAssignments = replicasAssignments;
+ }
+
+ public Map getProperties() {
+ return properties;
+ }
+
+ public void setProperties(Map properties) {
+ this.properties = properties;
+ }
+
+}
diff --git a/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/AdminClientConfigCustomizer.java b/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/AdminClientConfigCustomizer.java
new file mode 100644
index 000000000..25b2fe79e
--- /dev/null
+++ b/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/AdminClientConfigCustomizer.java
@@ -0,0 +1,31 @@
+/*
+ * Copyright 2021-2021 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.provisioning;
+
+import java.util.Map;
+
+/**
+ * Customizer for configuring AdminClient.
+ *
+ * @author Soby Chacko
+ * @since 3.1.2
+ */
+@FunctionalInterface
+public interface AdminClientConfigCustomizer {
+
+ void configure(Map adminClientProperties);
+}
diff --git a/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java b/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java
new file mode 100644
index 000000000..ade077417
--- /dev/null
+++ b/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java
@@ -0,0 +1,661 @@
+/*
+ * Copyright 2014-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.
+ * 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.provisioning;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.admin.AlterConfigOp;
+import org.apache.kafka.clients.admin.AlterConfigsResult;
+import org.apache.kafka.clients.admin.Config;
+import org.apache.kafka.clients.admin.ConfigEntry;
+import org.apache.kafka.clients.admin.CreatePartitionsResult;
+import org.apache.kafka.clients.admin.CreateTopicsResult;
+import org.apache.kafka.clients.admin.DescribeConfigsResult;
+import org.apache.kafka.clients.admin.DescribeTopicsResult;
+import org.apache.kafka.clients.admin.ListTopicsResult;
+import org.apache.kafka.clients.admin.NewPartitions;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.admin.TopicDescription;
+import org.apache.kafka.common.KafkaFuture;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.config.ConfigResource;
+import org.apache.kafka.common.errors.TopicExistsException;
+import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
+
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
+import org.springframework.cloud.stream.binder.BinderException;
+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.KafkaProducerProperties;
+import org.springframework.cloud.stream.binder.kafka.properties.KafkaTopicProperties;
+import org.springframework.cloud.stream.binder.kafka.utils.KafkaTopicUtils;
+import org.springframework.cloud.stream.provisioning.ConsumerDestination;
+import org.springframework.cloud.stream.provisioning.ProducerDestination;
+import org.springframework.cloud.stream.provisioning.ProvisioningException;
+import org.springframework.cloud.stream.provisioning.ProvisioningProvider;
+import org.springframework.retry.RetryOperations;
+import org.springframework.retry.backoff.ExponentialBackOffPolicy;
+import org.springframework.retry.policy.SimpleRetryPolicy;
+import org.springframework.retry.support.RetryTemplate;
+import org.springframework.util.Assert;
+import org.springframework.util.CollectionUtils;
+import org.springframework.util.ObjectUtils;
+import org.springframework.util.StringUtils;
+
+/**
+ * Kafka implementation for {@link ProvisioningProvider}.
+ *
+ * @author Soby Chacko
+ * @author Gary Russell
+ * @author Ilayaperumal Gopinathan
+ * @author Simon Flandergan
+ * @author Oleg Zhurakousky
+ * @author Aldo Sinanaj
+ */
+public class KafkaTopicProvisioner implements
+ // @checkstyle:off
+ ProvisioningProvider, ExtendedProducerProperties>,
+ // @checkstyle:on
+ InitializingBean {
+
+ private static final Log logger = LogFactory.getLog(KafkaTopicProvisioner.class);
+
+ private static final int DEFAULT_OPERATION_TIMEOUT = 30;
+
+ private final KafkaBinderConfigurationProperties configurationProperties;
+
+ private final int operationTimeout = DEFAULT_OPERATION_TIMEOUT;
+
+ private final Map adminClientProperties;
+
+ private RetryOperations metadataRetryOperations;
+
+ /**
+ * Create an instance.
+ * @param kafkaBinderConfigurationProperties the binder configuration properties.
+ * @param kafkaProperties the boot Kafka properties used to build the
+ * @param adminClientConfigCustomizer to customize {@link AdminClient}.
+ * {@link AdminClient}.
+ */
+ public KafkaTopicProvisioner(
+ KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties,
+ KafkaProperties kafkaProperties,
+ AdminClientConfigCustomizer adminClientConfigCustomizer) {
+ Assert.isTrue(kafkaProperties != null, "KafkaProperties cannot be null");
+ this.configurationProperties = kafkaBinderConfigurationProperties;
+ this.adminClientProperties = kafkaProperties.buildAdminProperties();
+ normalalizeBootPropsWithBinder(this.adminClientProperties, kafkaProperties,
+ kafkaBinderConfigurationProperties);
+ // If the application provides an AdminConfig customizer
+ // and overrides properties, that takes precedence.
+ if (adminClientConfigCustomizer != null) {
+ adminClientConfigCustomizer.configure(this.adminClientProperties);
+ }
+ }
+
+ /**
+ * Mutator for metadata retry operations.
+ * @param metadataRetryOperations the retry configuration
+ */
+ public void setMetadataRetryOperations(RetryOperations metadataRetryOperations) {
+ this.metadataRetryOperations = metadataRetryOperations;
+ }
+
+ @Override
+ public void afterPropertiesSet() {
+ if (this.metadataRetryOperations == null) {
+ RetryTemplate retryTemplate = new RetryTemplate();
+
+ SimpleRetryPolicy simpleRetryPolicy = new SimpleRetryPolicy();
+ simpleRetryPolicy.setMaxAttempts(10);
+ retryTemplate.setRetryPolicy(simpleRetryPolicy);
+
+ ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
+ backOffPolicy.setInitialInterval(100);
+ backOffPolicy.setMultiplier(2);
+ backOffPolicy.setMaxInterval(1000);
+ retryTemplate.setBackOffPolicy(backOffPolicy);
+ this.metadataRetryOperations = retryTemplate;
+ }
+ }
+
+ @Override
+ public ProducerDestination provisionProducerDestination(final String name,
+ ExtendedProducerProperties properties) {
+
+ if (logger.isInfoEnabled()) {
+ logger.info("Using kafka topic for outbound: " + name);
+ }
+ KafkaTopicUtils.validateTopicName(name);
+ try (AdminClient adminClient = createAdminClient()) {
+ createTopic(adminClient, name, properties.getPartitionCount(), false,
+ properties.getExtension().getTopic());
+ int partitions = 0;
+ Map