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/k-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBindingProperties.java b/k-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/k-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/k-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java b/k-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/k-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