Update SNAPSHOT to 3.1.0-M3

* Customizing producer/consumer factories

Adding hooks by providing Producer and Consumer config
customizers to perform advanced configuration on the producer
and consumer factories.

Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/960

* Addressing PR review comments

* Further PR updates

README.adoc

@@ -127,7 +127,11 @@ spring.cloud.stream.kafka.binder.replicationFactor::
 The replication factor of auto-created topics if `autoCreateTopics` is active.
 Can be overridden on each binding.
 +
-Default: `1`.
+NOTE: If you are using Kafka broker versions prior to 2.4, then this value should be set to at least `1`.
+Starting with version 3.0.8, the binder uses `-1` as the default value, which indicates that the broker 'default.replication.factor' property will be used to determine the number of replicas.
+Check with your Kafka broker admins to see if there is a policy in place that requires a minimum replication factor, if that's the case then, typically, the `default.replication.factor` will match that value and `-1` should be used, unless you need a replication factor greater than the minimum.
++
+Default: `-1`.
 spring.cloud.stream.kafka.binder.autoCreateTopics::
 If set to `true`, the binder creates new topics automatically.
 If set to `false`, the binder relies on the topics being already configured.
@@ -161,6 +165,11 @@ If this custom `BinderHeaderMapper` bean is not made available to the binder usi
 +
 Default: none.
 
+spring.cloud.stream.kafka.binder.considerDownWhenAnyPartitionHasNoLeader::
+Flag to set the binder health as `down`, when any partitions on the topic, regardless of the consumer that is receiving data from it, is found without a leader.
++
+Default: `false`.
+
 [[kafka-consumer-properties]]
 ==== Kafka Consumer Properties
 
@@ -192,6 +201,8 @@ By default, offsets are committed after all records in the batch of records retu
 The number of records returned by a poll can be controlled with the `max.poll.records` Kafka property, which is set through the consumer `configuration` property.
 Setting this to `true` may cause a degradation in performance, but doing so reduces the likelihood of redelivered records when a failure occurs.
 Also, see the binder `requiredAcks` property, which also affects the performance of committing offsets.
+This property is deprecated as of 3.1 in favor of using `ackMode`.
+If the `ackMode` is not set and batch mode is not enabled, `RECORD` ackMode will be used.
 +
 Default: `false`.
 autoCommitOffset::
@@ -200,9 +211,14 @@ If set to `false`, a header with the key `kafka_acknowledgment` of the type `org
 Applications may use this header for acknowledging messages.
 See the examples section for details.
 When this property is set to `false`, Kafka binder sets the ack mode to `org.springframework.kafka.listener.AbstractMessageListenerContainer.AckMode.MANUAL` and the application is responsible for acknowledging records.
-Also see `ackEachRecord`.
+Also see `ackEachRecord`. This property is deprecated as of 3.1. See `ackMode` for more details.
 +
 Default: `true`.
+ackMode::
+Specify the container ack mode.
+This is based on the AckMode enumeration defined in Spring Kafka.
+If `ackEachRecord` property is set to `true` and consumer is not in batch mode, then this will use the ack mode of `RECORD`, otherwise, use the provided ack mode using this property.
+
 autoCommitOnError::
 Effective only if `autoCommitOffset` is set to `true`.
 If set to `false`, it suppresses auto-commits for messages that result in errors and commits only for successful messages. It allows a stream to automatically replay from the last successfully processed message, in case of persistent failures.
@@ -298,7 +314,7 @@ topic.replication-factor::
 The replication factor to use when provisioning topics. Overrides the binder-wide setting.
 Ignored if `replicas-assignments` is present.
 +
-Default: none (the binder-wide default of 1 is used).
+Default: none (the binder-wide default of -1 is used).
 pollTimeout::
 Timeout used for polling in pollable consumers.
 +
@@ -394,7 +410,7 @@ topic.replication-factor::
 The replication factor to use when provisioning topics. Overrides the binder-wide setting.
 Ignored if `replicas-assignments` is present.
 +
-Default: none (the binder-wide default of 1 is used).
+Default: none (the binder-wide default of -1 is used).
 useTopicHeader::
 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.
@@ -419,7 +435,7 @@ If a topic already exists with a larger number of partitions than the maximum of
 
 compression::
 Set the `compression.type` producer property.
-Supported values are `none`, `gzip`, `snappy` and `lz4`.
+Supported values are `none`, `gzip`, `snappy`, `lz4` and `zstd`.
 If you override the `kafka-clients` jar to 2.1.0 (or later), as discussed in the https://docs.spring.io/spring-kafka/docs/2.2.x/reference/html/deps-for-21x.html[Spring for Apache Kafka documentation], and wish to use `zstd` compression, use `spring.cloud.stream.kafka.bindings.<binding-name>.producer.configuration.compression.type=zstd`.
 +
 Default: `none`.
@@ -741,6 +757,20 @@ public interface KafkaBindingRebalanceListener {
 
 You cannot set the `resetOffsets` consumer property to `true` when you provide a rebalance listener.
 
+[[consumer-producer-config-customizer]]
+=== Customizing Consumer and Producer configuration
+
+If you want advanced customization of consumer and producer configuration that is used for creating `ConsumerFactory` and `ProducerFactory` in Kafka,
+you can implement the following customizers.
+
+* ConsusumerConfigCustomizer
+* ProducerConfigCustomizer
+
+Both of these interfaces provide a way to configure the config map used for consumer and producer properties.
+For example, if you want to gain access to a bean that is defined at the application level, you can inject that in the implementation of the `configure` method.
+When the binder discovers that these customizers are available as beans, it will invoke the `configure` method right before creating the consumer and producer factories.
+
+
 = Appendices
 [appendix]
 [[building]]

docs/pom.xml

@@ -7,7 +7,7 @@
 	<parent>
 		<groupId>org.springframework.cloud</groupId>
 		<artifactId>spring-cloud-stream-binder-kafka-parent</artifactId>
-		<version>3.1.0-M2</version>
+		<version>3.1.0-M3</version>
 	</parent>
 	<packaging>jar</packaging>
 	<name>spring-cloud-stream-binder-kafka-docs</name>

docs/src/main/asciidoc/_configprops.adoc

@@ -19,6 +19,7 @@
 |spring.cloud.stream.kafka.binder.auto-create-topics | true | 
 |spring.cloud.stream.kafka.binder.brokers | [localhost] | 
 |spring.cloud.stream.kafka.binder.configuration |  | Arbitrary kafka properties that apply to both producers and consumers.
+|spring.cloud.stream.kafka.binder.consider-down-when-any-partition-has-no-leader | false | 
 |spring.cloud.stream.kafka.binder.consumer-properties |  | Arbitrary kafka consumer properties.
 |spring.cloud.stream.kafka.binder.header-mapper-bean-name |  | The bean name of a custom header mapper to use instead of a {@link org.springframework.kafka.support.DefaultKafkaHeaderMapper}.
 |spring.cloud.stream.kafka.binder.headers | [] | 
@@ -26,7 +27,7 @@
 |spring.cloud.stream.kafka.binder.jaas |  | 
 |spring.cloud.stream.kafka.binder.min-partition-count | 1 | 
 |spring.cloud.stream.kafka.binder.producer-properties |  | Arbitrary kafka producer properties.
-|spring.cloud.stream.kafka.binder.replication-factor | 1 | 
+|spring.cloud.stream.kafka.binder.replication-factor | -1 | 
 |spring.cloud.stream.kafka.binder.required-acks | 1 | 
 |spring.cloud.stream.kafka.binder.transaction.producer.batch-timeout |  | 
 |spring.cloud.stream.kafka.binder.transaction.producer.buffer-size |  | 
@@ -53,6 +54,7 @@
 |spring.cloud.stream.metrics.properties |  | Application properties that should be added to the metrics payload For example: `spring.application**`.
 |spring.cloud.stream.metrics.schedule-interval | 60s | Interval expressed as Duration for scheduling metrics snapshots publishing. Defaults to 60 seconds
 |spring.cloud.stream.override-cloud-connectors | false | This property is only applicable when the cloud profile is active and Spring Cloud Connectors are provided with the application. If the property is false (the default), the binder detects a suitable bound service (for example, a RabbitMQ service bound in Cloud Foundry for the RabbitMQ binder) and uses it for creating connections (usually through Spring Cloud Connectors). When set to true, this property instructs binders to completely ignore the bound services and rely on Spring Boot properties (for example, relying on the spring.rabbitmq.* properties provided in the environment for the RabbitMQ binder). The typical usage of this property is to be nested in a customized environment when connecting to multiple systems.
+|spring.cloud.stream.pollable-source | none | A semi-colon delimited list of binding names of pollable sources. Binding names follow the same naming convention as functions. For example, name '...pollable-source=foobar' will be accessible as 'foobar-iin-0'' binding
 |spring.cloud.stream.poller.cron |  | Cron expression value for the Cron Trigger.
 |spring.cloud.stream.poller.fixed-delay | 1000 | Fixed delay for default poller.
 |spring.cloud.stream.poller.initial-delay | 0 | Initial delay for periodic triggers.

docs/src/main/asciidoc/overview.adoc

@@ -106,7 +106,11 @@ spring.cloud.stream.kafka.binder.replicationFactor::
 The replication factor of auto-created topics if `autoCreateTopics` is active.
 Can be overridden on each binding.
 +
-Default: `1`.
+NOTE: If you are using Kafka broker versions prior to 2.4, then this value should be set to at least `1`.
+Starting with version 3.0.8, the binder uses `-1` as the default value, which indicates that the broker 'default.replication.factor' property will be used to determine the number of replicas.
+Check with your Kafka broker admins to see if there is a policy in place that requires a minimum replication factor, if that's the case then, typically, the `default.replication.factor` will match that value and `-1` should be used, unless you need a replication factor greater than the minimum.
++
+Default: `-1`.
 spring.cloud.stream.kafka.binder.autoCreateTopics::
 If set to `true`, the binder creates new topics automatically.
 If set to `false`, the binder relies on the topics being already configured.
@@ -140,6 +144,11 @@ If this custom `BinderHeaderMapper` bean is not made available to the binder usi
 +
 Default: none.
 
+spring.cloud.stream.kafka.binder.considerDownWhenAnyPartitionHasNoLeader::
+Flag to set the binder health as `down`, when any partitions on the topic, regardless of the consumer that is receiving data from it, is found without a leader.
++
+Default: `false`.
+
 [[kafka-consumer-properties]]
 ==== Kafka Consumer Properties
 
@@ -171,6 +180,8 @@ By default, offsets are committed after all records in the batch of records retu
 The number of records returned by a poll can be controlled with the `max.poll.records` Kafka property, which is set through the consumer `configuration` property.
 Setting this to `true` may cause a degradation in performance, but doing so reduces the likelihood of redelivered records when a failure occurs.
 Also, see the binder `requiredAcks` property, which also affects the performance of committing offsets.
+This property is deprecated as of 3.1 in favor of using `ackMode`.
+If the `ackMode` is not set and batch mode is not enabled, `RECORD` ackMode will be used.
 +
 Default: `false`.
 autoCommitOffset::
@@ -179,9 +190,14 @@ If set to `false`, a header with the key `kafka_acknowledgment` of the type `org
 Applications may use this header for acknowledging messages.
 See the examples section for details.
 When this property is set to `false`, Kafka binder sets the ack mode to `org.springframework.kafka.listener.AbstractMessageListenerContainer.AckMode.MANUAL` and the application is responsible for acknowledging records.
-Also see `ackEachRecord`.
+Also see `ackEachRecord`. This property is deprecated as of 3.1. See `ackMode` for more details.
 +
 Default: `true`.
+ackMode::
+Specify the container ack mode.
+This is based on the AckMode enumeration defined in Spring Kafka.
+If `ackEachRecord` property is set to `true` and consumer is not in batch mode, then this will use the ack mode of `RECORD`, otherwise, use the provided ack mode using this property.
+
 autoCommitOnError::
 Effective only if `autoCommitOffset` is set to `true`.
 If set to `false`, it suppresses auto-commits for messages that result in errors and commits only for successful messages. It allows a stream to automatically replay from the last successfully processed message, in case of persistent failures.
@@ -277,7 +293,7 @@ topic.replication-factor::
 The replication factor to use when provisioning topics. Overrides the binder-wide setting.
 Ignored if `replicas-assignments` is present.
 +
-Default: none (the binder-wide default of 1 is used).
+Default: none (the binder-wide default of -1 is used).
 pollTimeout::
 Timeout used for polling in pollable consumers.
 +
@@ -373,7 +389,7 @@ topic.replication-factor::
 The replication factor to use when provisioning topics. Overrides the binder-wide setting.
 Ignored if `replicas-assignments` is present.
 +
-Default: none (the binder-wide default of 1 is used).
+Default: none (the binder-wide default of -1 is used).
 useTopicHeader::
 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.
@@ -398,7 +414,7 @@ If a topic already exists with a larger number of partitions than the maximum of
 
 compression::
 Set the `compression.type` producer property.
-Supported values are `none`, `gzip`, `snappy` and `lz4`.
+Supported values are `none`, `gzip`, `snappy`, `lz4` and `zstd`.
 If you override the `kafka-clients` jar to 2.1.0 (or later), as discussed in the https://docs.spring.io/spring-kafka/docs/2.2.x/reference/html/deps-for-21x.html[Spring for Apache Kafka documentation], and wish to use `zstd` compression, use `spring.cloud.stream.kafka.bindings.<binding-name>.producer.configuration.compression.type=zstd`.
 +
 Default: `none`.
@@ -719,3 +735,17 @@ public interface KafkaBindingRebalanceListener {
 ====
 
 You cannot set the `resetOffsets` consumer property to `true` when you provide a rebalance listener.
+
+[[consumer-producer-config-customizer]]
+=== Customizing Consumer and Producer configuration
+
+If you want advanced customization of consumer and producer configuration that is used for creating `ConsumerFactory` and `ProducerFactory` in Kafka,
+you can implement the following customizers.
+
+* ConsusumerConfigCustomizer
+* ProducerConfigCustomizer
+
+Both of these interfaces provide a way to configure the config map used for consumer and producer properties.
+For example, if you want to gain access to a bean that is defined at the application level, you can inject that in the implementation of the `configure` method.
+When the binder discovers that these customizers are available as beans, it will invoke the `configure` method right before creating the consumer and producer factories.
+

docs/src/main/asciidoc/spring-cloud-stream-binder-kafka.adoc

@@ -32,7 +32,7 @@ Sabby Anandan, Marius Bogoevici, Eric Bottard, Mark Fisher, Ilayaperumal Gopinat
 // ======================================================================================
 
 
-*{spring-cloud-stream-version}*
+*{project-version}*
 
 
 = Reference Guide

pom.xml

@@ -2,12 +2,12 @@
 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
 	<modelVersion>4.0.0</modelVersion>
 	<artifactId>spring-cloud-stream-binder-kafka-parent</artifactId>
-	<version>3.1.0-M2</version>
+	<version>3.1.0-M3</version>
 	<packaging>pom</packaging>
 	<parent>
 		<groupId>org.springframework.cloud</groupId>
 		<artifactId>spring-cloud-build</artifactId>
-		<version>3.0.0-M3</version>
+		<version>3.0.0-M4</version>
 		<relativePath />
 	</parent>
 	<properties>
@@ -15,8 +15,8 @@
 		<spring-kafka.version>2.5.3.RELEASE</spring-kafka.version>
 		<spring-integration-kafka.version>3.3.0.RELEASE</spring-integration-kafka.version>
 		<kafka.version>2.5.0</kafka.version>
-		<spring-cloud-schema-registry.version>1.1.0-M2</spring-cloud-schema-registry.version>
-		<spring-cloud-stream.version>3.1.0-M2</spring-cloud-stream.version>
+		<spring-cloud-schema-registry.version>1.1.0-M3</spring-cloud-schema-registry.version>
+		<spring-cloud-stream.version>3.1.0-M3</spring-cloud-stream.version>
 		<maven-checkstyle-plugin.failsOnError>true</maven-checkstyle-plugin.failsOnError>
 		<maven-checkstyle-plugin.failsOnViolation>true</maven-checkstyle-plugin.failsOnViolation>
 		<maven-checkstyle-plugin.includeTestSourceDirectory>true</maven-checkstyle-plugin.includeTestSourceDirectory>
@@ -196,6 +196,14 @@
 						<enabled>false</enabled>
 					</snapshots>
 				</repository>
+				<repository>
+				    <id>rsocket-snapshots</id>
+				    <name>RSocket Snapshots</name>
+				    <url>https://oss.jfrog.org/oss-snapshot-local</url>
+				    <snapshots>
+				        <enabled>true</enabled>
+				    </snapshots>
+				</repository>
 			</repositories>
 			<pluginRepositories>
 				<pluginRepository>

spring-cloud-starter-stream-kafka/pom.xml

@@ -4,7 +4,7 @@
 	<parent>
 		<groupId>org.springframework.cloud</groupId>
 		<artifactId>spring-cloud-stream-binder-kafka-parent</artifactId>
-		<version>3.1.0-M2</version>
+		<version>3.1.0-M3</version>
 	</parent>
 	<artifactId>spring-cloud-starter-stream-kafka</artifactId>
 	<description>Spring Cloud Starter Stream Kafka</description>

spring-cloud-stream-binder-kafka-core/pom.xml

@@ -5,7 +5,7 @@
 	<parent>
 		<groupId>org.springframework.cloud</groupId>
 		<artifactId>spring-cloud-stream-binder-kafka-parent</artifactId>
-		<version>3.1.0-M2</version>
+		<version>3.1.0-M3</version>
 	</parent>
 	<artifactId>spring-cloud-stream-binder-kafka-core</artifactId>
 	<description>Spring Cloud Stream Kafka Binder Core</description>

spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java

@@ -53,6 +53,7 @@ import org.springframework.util.StringUtils;
  * @author Rafal Zukowski
  * @author Aldo Sinanaj
  * @author Lukasz Kaminski
+ * @author Chukwubuikem Ume-Ugwa
  */
 @ConfigurationProperties(prefix = "spring.cloud.stream.kafka.binder")
 public class KafkaBinderConfigurationProperties {
@@ -90,9 +91,11 @@ public class KafkaBinderConfigurationProperties {
 
 	private boolean autoAddPartitions;
 
+	private boolean considerDownWhenAnyPartitionHasNoLeader;
+
 	private String requiredAcks = "1";
 
-	private short replicationFactor = 1;
+	private short replicationFactor = -1;
 
 	private int minPartitionCount = 1;
 
@@ -363,6 +366,14 @@ public class KafkaBinderConfigurationProperties {
 		this.authorizationExceptionRetryInterval = authorizationExceptionRetryInterval;
 	}
 
+	public boolean isConsiderDownWhenAnyPartitionHasNoLeader() {
+		return this.considerDownWhenAnyPartitionHasNoLeader;
+	}
+
+	public void setConsiderDownWhenAnyPartitionHasNoLeader(boolean considerDownWhenAnyPartitionHasNoLeader) {
+		this.considerDownWhenAnyPartitionHasNoLeader = considerDownWhenAnyPartitionHasNoLeader;
+	}
+
 	/**
 	 * Domain class that models transaction capabilities in Kafka.
 	 */

spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java

@@ -19,6 +19,8 @@ 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.
  *
@@ -88,6 +90,7 @@ public class KafkaConsumerProperties {
 	 * 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;
 
 	/**
@@ -101,8 +104,15 @@ public class KafkaConsumerProperties {
 	 * 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;
+
 	/**
 	 * Effective only if autoCommitOffset is set to true.
 	 * If set to false, it suppresses auto-commits for messages that result in errors and commits only for successful messages.
@@ -111,6 +121,7 @@ public class KafkaConsumerProperties {
 	 * If not set (the default), it effectively has the same value as enableDlq,
 	 * auto-committing erroneous messages if they are sent to a DLQ and not committing them otherwise.
 	 */
+	@Deprecated
 	private Boolean autoCommitOnError;
 
 	/**
@@ -205,11 +216,20 @@ public class KafkaConsumerProperties {
 	 *
 	 * 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;
 	}
@@ -220,15 +240,35 @@ public class KafkaConsumerProperties {
 	 * 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
 	 *
@@ -280,11 +320,21 @@ public class KafkaConsumerProperties {
 	 * If set to true, it always auto-commits (if auto-commit is enabled).
 	 * If not set (the default), it effectively has the same value as enableDlq,
 	 * auto-committing erroneous messages if they are sent to a DLQ and not committing them otherwise.
+	 *
+	 * @deprecated in favor of using an error handler and customize the container with that error handler.
 	 */
+	@Deprecated
 	public Boolean getAutoCommitOnError() {
 		return this.autoCommitOnError;
 	}
 
+	/**
+	 *
+	 * @param autoCommitOnError commit on error
+	 *
+	 * @deprecated in favor of using an error handler and customize the container with that error handler.
+	 */
+	@Deprecated
 	public void setAutoCommitOnError(Boolean autoCommitOnError) {
 		this.autoCommitOnError = autoCommitOnError;
 	}

spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaProducerProperties.java

@@ -121,7 +121,7 @@ public class KafkaProducerProperties {
 	/**
 	 * @return compression type {@link CompressionType}
 	 *
-	 * Set the compression.type producer property. Supported values are none, gzip, snappy and lz4.
+	 * Set the compression.type producer property. Supported values are none, gzip, snappy, lz4 and zstd.
 	 * See {@link CompressionType} for more details.
 	 */
 	@NotNull
@@ -304,11 +304,10 @@ public class KafkaProducerProperties {
 		 */
 		lz4,
 
-		// /** // TODO: uncomment and fix docs when kafka-clients 2.1.0 or newer is the
-		// default
-		// * zstd compression
-		// */
-		// zstd
+		/**
+		 * zstd compression.
+		 */
+		zstd,
 
 	}
 

spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java

@@ -25,14 +25,20 @@ 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;
@@ -40,6 +46,7 @@ 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;
 
@@ -262,7 +269,7 @@ public class KafkaTopicProvisioner implements
 		if (ObjectUtils
 				.isEmpty(adminProps.get(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG))
 				|| !kafkaConnectionString
-						.equals(binderProps.getDefaultKafkaConnectionString())) {
+				.equals(binderProps.getDefaultKafkaConnectionString())) {
 			adminProps.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,
 					kafkaConnectionString);
 		}
@@ -367,13 +374,42 @@ public class KafkaTopicProvisioner implements
 
 		Set<String> names = namesFutures.get(this.operationTimeout, TimeUnit.SECONDS);
 		if (names.contains(topicName)) {
+			//check if topic.properties are different from Topic Configuration in Kafka
+			if (this.configurationProperties.isAutoCreateTopics()) {
+				ConfigResource topicConfigResource = new ConfigResource(ConfigResource.Type.TOPIC, topicName);
+				DescribeConfigsResult describeConfigsResult = adminClient
+						.describeConfigs(Collections.singletonList(topicConfigResource));
+				KafkaFuture<Map<ConfigResource, Config>> topicConfigurationFuture = describeConfigsResult.all();
+				Map<ConfigResource, Config> topicConfigMap = topicConfigurationFuture
+						.get(this.operationTimeout, TimeUnit.SECONDS);
+				Config config = topicConfigMap.get(topicConfigResource);
+				final List<AlterConfigOp> updatedConfigEntries = topicProperties.getProperties().entrySet().stream()
+						.filter(propertiesEntry -> {
+							// Property is new and should be added
+							if (config.get(propertiesEntry.getKey()) == null) {
+								return true;
+							}
+							else {
+								// Property changed and should be updated
+								return !config.get(propertiesEntry.getKey()).value().equals(propertiesEntry.getValue());
+							}
+
+						}).map(propertyEntry -> new ConfigEntry(propertyEntry.getKey(), propertyEntry.getValue()))
+						.map(configEntry -> new AlterConfigOp(configEntry, AlterConfigOp.OpType.SET))
+						.collect(Collectors
+								.toList());
+				Map<ConfigResource, Collection<AlterConfigOp>> alterConfigForTopics = new HashMap<>();
+				alterConfigForTopics.put(topicConfigResource, updatedConfigEntries);
+				AlterConfigsResult alterConfigsResult = adminClient.incrementalAlterConfigs(alterConfigForTopics);
+				alterConfigsResult.all().get(this.operationTimeout, TimeUnit.SECONDS);
+			}
 			// only consider minPartitionCount for resizing if autoAddPartitions is true
 			int effectivePartitionCount = this.configurationProperties
 					.isAutoAddPartitions()
-							? Math.max(
-									this.configurationProperties.getMinPartitionCount(),
-									partitionCount)
-							: partitionCount;
+					? Math.max(
+					this.configurationProperties.getMinPartitionCount(),
+					partitionCount)
+					: partitionCount;
 			DescribeTopicsResult describeTopicsResult = adminClient
 					.describeTopics(Collections.singletonList(topicName));
 			KafkaFuture<Map<String, TopicDescription>> topicDescriptionsFuture = describeTopicsResult
@@ -426,7 +462,7 @@ public class KafkaTopicProvisioner implements
 							topicProperties.getReplicationFactor() != null
 									? topicProperties.getReplicationFactor()
 									: this.configurationProperties
-											.getReplicationFactor());
+									.getReplicationFactor());
 				}
 				if (topicProperties.getProperties().size() > 0) {
 					newTopic.configs(topicProperties.getProperties());
@@ -494,7 +530,7 @@ public class KafkaTopicProvisioner implements
 					try (AdminClient adminClient = AdminClient
 							.create(this.adminClientProperties)) {
 						final DescribeTopicsResult describeTopicsResult = adminClient
-							.describeTopics(Collections.singletonList(topicName));
+								.describeTopics(Collections.singletonList(topicName));
 
 						describeTopicsResult.all().get();
 					}

spring-cloud-stream-binder-kafka-streams/pom.xml

@@ -10,7 +10,7 @@
 	<parent>
 		<groupId>org.springframework.cloud</groupId>
 		<artifactId>spring-cloud-stream-binder-kafka-parent</artifactId>
-		<version>3.1.0-M2</version>
+		<version>3.1.0-M3</version>
 	</parent>
 
 	<properties>

spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionStateStoreTests.java

@@ -30,6 +30,7 @@ import org.apache.kafka.streams.state.StoreBuilder;
 import org.apache.kafka.streams.state.Stores;
 import org.apache.kafka.streams.state.WindowStore;
 import org.junit.ClassRule;
+import org.junit.Ignore;
 import org.junit.Test;
 
 import org.springframework.boot.SpringApplication;
@@ -54,13 +55,14 @@ public class KafkaStreamsFunctionStateStoreTests {
 	private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule.getEmbeddedKafka();
 
 	@Test
+	@Ignore
 	public void testKafkaStreamsFuncionWithMultipleStateStores() throws Exception {
 		SpringApplication app = new SpringApplication(StateStoreTestApplication.class);
 		app.setWebApplicationType(WebApplicationType.NONE);
 
 		try (ConfigurableApplicationContext context = app.run("--server.port=0",
 				"--spring.jmx.enabled=false",
-				"--spring.cloud.stream.function.definition=process;hello",
+				"--spring.cloud.stream.function.definition=process",
 				"--spring.cloud.stream.bindings.process-in-0.destination=words",
 				"--spring.cloud.stream.bindings.hello-in-0.destination=words",
 				"--spring.cloud.stream.kafka.streams.binder.functions.process.applicationId=testKafkaStreamsFuncionWithMultipleStateStores-123",

spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/MultipleFunctionsInSameAppTests.java

@@ -31,6 +31,7 @@ import org.apache.kafka.streams.kstream.KStream;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 import org.junit.ClassRule;
+import org.junit.Ignore;
 import org.junit.Test;
 
 import org.springframework.boot.SpringApplication;
@@ -49,6 +50,7 @@ import org.springframework.util.Assert;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
+@Ignore
 public class MultipleFunctionsInSameAppTests {
 
 	@ClassRule

spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/StreamToGlobalKTableFunctionTests.java

@@ -36,6 +36,7 @@ import org.apache.kafka.streams.kstream.KTable;
 import org.apache.kafka.streams.processor.TimestampExtractor;
 import org.apache.kafka.streams.processor.WallclockTimestampExtractor;
 import org.junit.ClassRule;
+import org.junit.Ignore;
 import org.junit.Test;
 
 import org.springframework.boot.SpringApplication;
@@ -56,6 +57,7 @@ import org.springframework.kafka.test.utils.KafkaTestUtils;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
+@Ignore
 public class StreamToGlobalKTableFunctionTests {
 
 	@ClassRule

spring-cloud-stream-binder-kafka/pom.xml

@@ -10,7 +10,7 @@
 	<parent>
 		<groupId>org.springframework.cloud</groupId>
 		<artifactId>spring-cloud-stream-binder-kafka-parent</artifactId>
-		<version>3.1.0-M2</version>
+		<version>3.1.0-M3</version>
 	</parent>
 
 	<dependencies>

spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicator.java

@@ -47,6 +47,7 @@ import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
  * @author Laur Aliste
  * @author Soby Chacko
  * @author Vladislav Fefelov
+ * @author Chukwubuikem Ume-Ugwa
  */
 public class KafkaBinderHealthIndicator implements HealthIndicator, DisposableBean {
 
@@ -63,6 +64,8 @@ public class KafkaBinderHealthIndicator implements HealthIndicator, DisposableBe
 
 	private Consumer<?, ?> metadataConsumer;
 
+	private boolean considerDownWhenAnyPartitionHasNoLeader;
+
 	public KafkaBinderHealthIndicator(KafkaMessageChannelBinder binder,
 			ConsumerFactory<?, ?> consumerFactory) {
 		this.binder = binder;
@@ -77,6 +80,10 @@ public class KafkaBinderHealthIndicator implements HealthIndicator, DisposableBe
 		this.timeout = timeout;
 	}
 
+	public void setConsiderDownWhenAnyPartitionHasNoLeader(boolean considerDownWhenAnyPartitionHasNoLeader) {
+		this.considerDownWhenAnyPartitionHasNoLeader = considerDownWhenAnyPartitionHasNoLeader;
+	}
+
 	@Override
 	public Health health() {
 		Future<Health> future = executor.submit(this::buildHealthStatus);
@@ -99,57 +106,59 @@ public class KafkaBinderHealthIndicator implements HealthIndicator, DisposableBe
 		}
 	}
 
-	private synchronized  Consumer<?, ?> initMetadataConsumer() {
+	private void initMetadataConsumer() {
 		if (this.metadataConsumer == null) {
 			this.metadataConsumer = this.consumerFactory.createConsumer();
 		}
-		return this.metadataConsumer;
 	}
 
 	private Health buildHealthStatus() {
 		try {
 			initMetadataConsumer();
-			synchronized (this.metadataConsumer) {
-				Set<String> downMessages = new HashSet<>();
-				final Map<String, KafkaMessageChannelBinder.TopicInformation> topicsInUse = KafkaBinderHealthIndicator.this.binder
-						.getTopicsInUse();
-				if (topicsInUse.isEmpty()) {
-					try {
-						this.metadataConsumer.listTopics(Duration.ofSeconds(this.timeout));
-					}
-					catch (Exception e) {
-						return Health.down().withDetail("No topic information available",
-								"Kafka broker is not reachable").build();
-					}
-					return Health.unknown().withDetail("No bindings found",
-							"Kafka binder may not be bound to destinations on the broker").build();
+			Set<String> downMessages = new HashSet<>();
+			final Map<String, KafkaMessageChannelBinder.TopicInformation> topicsInUse = KafkaBinderHealthIndicator.this.binder
+					.getTopicsInUse();
+			if (topicsInUse.isEmpty()) {
+				try {
+					this.metadataConsumer.listTopics(Duration.ofSeconds(this.timeout));
 				}
-				else {
-					for (String topic : topicsInUse.keySet()) {
-						KafkaMessageChannelBinder.TopicInformation topicInformation = topicsInUse
-								.get(topic);
-						if (!topicInformation.isTopicPattern()) {
-							List<PartitionInfo> partitionInfos = this.metadataConsumer
-									.partitionsFor(topic);
-							for (PartitionInfo partitionInfo : partitionInfos) {
-								if (topicInformation.getPartitionInfos()
-										.contains(partitionInfo)
-										&& partitionInfo.leader().id() == -1) {
-									downMessages.add(partitionInfo.toString());
-								}
+				catch (Exception e) {
+					return Health.down().withDetail("No topic information available",
+							"Kafka broker is not reachable").build();
+				}
+				return Health.unknown().withDetail("No bindings found",
+						"Kafka binder may not be bound to destinations on the broker").build();
+			}
+			else {
+				for (String topic : topicsInUse.keySet()) {
+					KafkaMessageChannelBinder.TopicInformation topicInformation = topicsInUse
+							.get(topic);
+					if (!topicInformation.isTopicPattern()) {
+						List<PartitionInfo> partitionInfos = this.metadataConsumer
+								.partitionsFor(topic);
+						for (PartitionInfo partitionInfo : partitionInfos) {
+							if (topicInformation.getPartitionInfos()
+									.contains(partitionInfo)
+									&& partitionInfo.leader() == null ||
+									(partitionInfo.leader() != null && partitionInfo.leader().id() == -1)) {
+								downMessages.add(partitionInfo.toString());
+							}
+							else if (this.considerDownWhenAnyPartitionHasNoLeader &&
+									partitionInfo.leader() == null || (partitionInfo.leader() != null && partitionInfo.leader().id() == -1)) {
+								downMessages.add(partitionInfo.toString());
 							}
 						}
 					}
 				}
-				if (downMessages.isEmpty()) {
-					return Health.up().build();
-				}
-				else {
-					return Health.down()
-							.withDetail("Following partitions in use have no leaders: ",
-									downMessages.toString())
-							.build();
-				}
+			}
+			if (downMessages.isEmpty()) {
+				return Health.up().build();
+			}
+			else {
+				return Health.down()
+						.withDetail("Following partitions in use have no leaders: ",
+								downMessages.toString())
+						.build();
 			}
 		}
 		catch (Exception ex) {

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

@@ -64,6 +64,8 @@ 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;
@@ -221,6 +223,10 @@ public class KafkaMessageChannelBinder extends
 
 	private ClientFactoryCustomizer clientFactoryCustomizer;
 
+	private ProducerConfigCustomizer producerConfigCustomizer;
+
+	private ConsumerConfigCustomizer consumerConfigCustomizer;
+
 	public KafkaMessageChannelBinder(
 			KafkaBinderConfigurationProperties configurationProperties,
 			KafkaTopicProvisioner provisioningProvider) {
@@ -492,8 +498,6 @@ public class KafkaMessageChannelBinder extends
 			String transactionIdPrefix,
 			ExtendedProducerProperties<KafkaProducerProperties> producerProperties, String beanName) {
 		Map<String, Object> props = new HashMap<>();
-		props.put(ProducerConfig.RETRIES_CONFIG, 0);
-		props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
 		props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
 		props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
 				ByteArraySerializer.class);
@@ -531,6 +535,9 @@ public class KafkaMessageChannelBinder extends
 		if (!ObjectUtils.isEmpty(kafkaProducerProperties.getConfiguration())) {
 			props.putAll(kafkaProducerProperties.getConfiguration());
 		}
+		if (this.producerConfigCustomizer != null) {
+			this.producerConfigCustomizer.configure(props);
+		}
 		DefaultKafkaProducerFactory<byte[], byte[]> producerFactory = new DefaultKafkaProducerFactory<>(
 				props);
 		if (transactionIdPrefix != null) {
@@ -657,19 +664,18 @@ public class KafkaMessageChannelBinder extends
 		}
 		messageListenerContainer.setBeanName(destination + ".container");
 		// end of these won't be needed...
-		if (!extendedConsumerProperties.getExtension().isAutoCommitOffset()) {
-			messageListenerContainer.getContainerProperties()
-					.setAckMode(ContainerProperties.AckMode.MANUAL);
-			messageListenerContainer.getContainerProperties().setAckOnError(false);
+		ContainerProperties.AckMode ackMode = extendedConsumerProperties.getExtension().getAckMode();
+		if (ackMode == null && extendedConsumerProperties.getExtension().isAckEachRecord()) {
+			ackMode = ContainerProperties.AckMode.RECORD;
 		}
-		else {
-			messageListenerContainer.getContainerProperties()
-					.setAckOnError(isAutoCommitOnError(extendedConsumerProperties));
-			if (extendedConsumerProperties.getExtension().isAckEachRecord()) {
+		if (ackMode != null) {
+			if ((extendedConsumerProperties.isBatchMode() && ackMode != ContainerProperties.AckMode.RECORD) ||
+					!extendedConsumerProperties.isBatchMode()) {
 				messageListenerContainer.getContainerProperties()
-						.setAckMode(ContainerProperties.AckMode.RECORD);
+						.setAckMode(ackMode);
 			}
 		}
+
 		if (this.logger.isDebugEnabled()) {
 			this.logger.debug("Listened partitions: "
 					+ StringUtils.collectionToCommaDelimitedString(listenedPartitions));
@@ -1343,6 +1349,9 @@ public class KafkaMessageChannelBinder extends
 					consumerProperties.getExtension().getStartOffset().name());
 		}
 
+		if (this.consumerConfigCustomizer != null) {
+			this.consumerConfigCustomizer.configure(props);
+		}
 		DefaultKafkaConsumerFactory<Object, Object> factory = new DefaultKafkaConsumerFactory<>(props);
 		factory.setBeanName(beanName);
 		if (this.clientFactoryCustomizer != null) {
@@ -1395,6 +1404,14 @@ public class KafkaMessageChannelBinder extends
 		return stringWriter.getBuffer().toString();
 	}
 
+	public void setConsumerConfigCustomizer(ConsumerConfigCustomizer consumerConfigCustomizer) {
+		this.consumerConfigCustomizer = consumerConfigCustomizer;
+	}
+
+	public void setProducerConfigCustomizer(ProducerConfigCustomizer producerConfigCustomizer) {
+		this.producerConfigCustomizer = producerConfigCustomizer;
+	}
+
 	private final class ProducerConfigurationMessageHandler
 			extends KafkaProducerMessageHandler<byte[], byte[]> {
 

spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaNullConverter.java

@@ -38,6 +38,11 @@ public class KafkaNullConverter extends AbstractMessageConverter {
 		super(Collections.singletonList(MimeTypeUtils.ALL));
 	}
 
+	@Override
+	protected boolean supportsMimeType(MessageHeaders headers) {
+		return true;
+	}
+
 	@Override
 	protected boolean supports(Class<?> aClass) {
 		return KafkaNull.class.equals(aClass);

spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/ConsumerConfigCustomizer.java

@@ -0,0 +1,32 @@
+/*
+ * Copyright 2020-2020 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.config;
+
+import java.util.Map;
+
+/**
+ * This customizer is called by the binder to customize consumer configuration in
+ * Kafka Consumer factory.
+ *
+ * @author Soby Chacko
+ * @since 3.0.9
+ */
+@FunctionalInterface
+public interface ConsumerConfigCustomizer {
+
+	void configure(Map<String, Object> consumerProperties);
+}

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

@@ -118,7 +118,10 @@ public class KafkaBinderConfiguration {
 			@Nullable ConsumerEndpointCustomizer<KafkaMessageDrivenChannelAdapter<?, ?>> consumerCustomizer,
 			ObjectProvider<KafkaBindingRebalanceListener> rebalanceListener,
 			ObjectProvider<DlqPartitionFunction> dlqPartitionFunction,
-			ObjectProvider<ClientFactoryCustomizer> clientFactoryCustomizer) {
+			ObjectProvider<ClientFactoryCustomizer> clientFactoryCustomizer,
+			ObjectProvider<ConsumerConfigCustomizer> consumerConfigCustomizer,
+			ObjectProvider<ProducerConfigCustomizer> producerConfigCustomizer
+			) {
 
 		KafkaMessageChannelBinder kafkaMessageChannelBinder = new KafkaMessageChannelBinder(
 				configurationProperties, provisioningProvider,
@@ -130,6 +133,8 @@ public class KafkaBinderConfiguration {
 		kafkaMessageChannelBinder.setProducerMessageHandlerCustomizer(messageHandlerCustomizer);
 		kafkaMessageChannelBinder.setConsumerEndpointCustomizer(consumerCustomizer);
 		kafkaMessageChannelBinder.setClientFactoryCustomizer(clientFactoryCustomizer.getIfUnique());
+		kafkaMessageChannelBinder.setConsumerConfigCustomizer(consumerConfigCustomizer.getIfUnique());
+		kafkaMessageChannelBinder.setProducerConfigCustomizer(producerConfigCustomizer.getIfUnique());
 		return kafkaMessageChannelBinder;
 	}
 
@@ -238,7 +243,6 @@ public class KafkaBinderConfiguration {
 		}
 
 		@ConditionalOnClass(name = "org.springframework.kafka.core.MicrometerConsumerListener")
-		@ConditionalOnBean(MeterRegistry.class)
 		protected class KafkaMicrometer {
 
 			@Bean

spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderHealthIndicatorConfiguration.java

@@ -37,6 +37,7 @@ import org.springframework.util.ObjectUtils;
  * Configuration class for Kafka binder health indicator beans.
  *
  * @author Oleg Zhurakousky
+ * @author Chukwubuikem Ume-Ugwa
  */
 
 @Configuration
@@ -66,6 +67,7 @@ class KafkaBinderHealthIndicatorConfiguration {
 		KafkaBinderHealthIndicator indicator = new KafkaBinderHealthIndicator(
 				kafkaMessageChannelBinder, consumerFactory);
 		indicator.setTimeout(configurationProperties.getHealthTimeout());
+		indicator.setConsiderDownWhenAnyPartitionHasNoLeader(configurationProperties.isConsiderDownWhenAnyPartitionHasNoLeader());
 		return indicator;
 	}
 

spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/ProducerConfigCustomizer.java

@@ -0,0 +1,32 @@
+/*
+ * Copyright 2020-2020 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.config;
+
+import java.util.Map;
+
+/**
+ * This customizer is called by the binder to customize producer configuration in
+ * Kafka Producer factory.
+ *
+ * @author Soby Chacko
+ * @since 3.0.9
+ */
+@FunctionalInterface
+public interface ProducerConfigCustomizer {
+
+	void configure(Map<String, Object> consumerProperties);
+}

spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicatorTest.java

@@ -42,6 +42,7 @@ import static org.assertj.core.api.Assertions.assertThat;
  * @author Gary Russell
  * @author Laur Aliste
  * @author Soby Chacko
+ * @author Chukwubuikem Ume-Ugwa
  */
 public class KafkaBinderHealthIndicatorTest {
 
@@ -97,7 +98,7 @@ public class KafkaBinderHealthIndicatorTest {
 
 	@Test
 	public void kafkaBinderIsDown() {
-		final List<PartitionInfo> partitions = partitions(new Node(-1, null, 0));
+		final List<PartitionInfo> partitions = partitions(null);
 		topicsInUse.put(TEST_TOPIC, new KafkaMessageChannelBinder.TopicInformation(
 				"group2-healthIndicator", partitions, false));
 		org.mockito.BDDMockito.given(consumer.partitionsFor(TEST_TOPIC))
@@ -106,6 +107,33 @@ public class KafkaBinderHealthIndicatorTest {
 		assertThat(health.getStatus()).isEqualTo(Status.DOWN);
 	}
 
+	@Test
+	public void kafkaBinderIsDownWhenConsiderDownWhenAnyPartitionHasNoLeaderIsTrue() {
+		final List<PartitionInfo> partitions = partitions(new Node(0, null, 0));
+		partitions.add(new PartitionInfo(TEST_TOPIC, 0, null, null, null));
+		indicator.setConsiderDownWhenAnyPartitionHasNoLeader(true);
+		topicsInUse.put(TEST_TOPIC, new KafkaMessageChannelBinder.TopicInformation(
+				"group2-healthIndicator", partitions, false));
+		org.mockito.BDDMockito.given(consumer.partitionsFor(TEST_TOPIC))
+				.willReturn(partitions);
+		Health health = indicator.health();
+		assertThat(health.getStatus()).isEqualTo(Status.DOWN);
+	}
+
+	@Test
+	public void kafkaBinderIsUpWhenConsiderDownWhenAnyPartitionHasNoLeaderIsFalse() {
+		Node node = new Node(0, null, 0);
+		final List<PartitionInfo> partitions = partitions(node);
+		partitions.add(new PartitionInfo(TEST_TOPIC, 0, null, null, null));
+		indicator.setConsiderDownWhenAnyPartitionHasNoLeader(false);
+		topicsInUse.put(TEST_TOPIC, new KafkaMessageChannelBinder.TopicInformation(
+				"group2-healthIndicator", partitions(node), false));
+		org.mockito.BDDMockito.given(consumer.partitionsFor(TEST_TOPIC))
+				.willReturn(partitions);
+		Health health = indicator.health();
+		assertThat(health.getStatus()).isEqualTo(Status.UP);
+	}
+
 	@Test(timeout = 5000)
 	public void kafkaBinderDoesNotAnswer() {
 		final List<PartitionInfo> partitions = partitions(new Node(-1, null, 0));

spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java

@@ -1430,7 +1430,8 @@ public class KafkaBinderTests extends
 		final KafkaProducerProperties.CompressionType[] codecs = new KafkaProducerProperties.CompressionType[] {
 				KafkaProducerProperties.CompressionType.none,
 				KafkaProducerProperties.CompressionType.gzip,
-				KafkaProducerProperties.CompressionType.snappy };
+				KafkaProducerProperties.CompressionType.snappy,
+				KafkaProducerProperties.CompressionType.zstd};
 		byte[] testPayload = new byte[2048];
 		Arrays.fill(testPayload, (byte) 65);
 		Binder binder = getBinder();
@@ -1638,7 +1639,7 @@ public class KafkaBinderTests extends
 				moduleOutputChannel, createProducerProperties());
 
 		ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
-		consumerProperties.getExtension().setAutoCommitOffset(false);
+		consumerProperties.getExtension().setAckMode(ContainerProperties.AckMode.MANUAL);
 
 		Binding<MessageChannel> consumerBinding = binder.bindConsumer(
 				"testManualAckSucceedsWhenAutoCommitOffsetIsTurnedOff", "test",
@@ -1736,7 +1737,7 @@ public class KafkaBinderTests extends
 		QueueChannel inbound1 = new QueueChannel();
 		ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
 		consumerProperties.getExtension().setAutoRebalanceEnabled(false);
-		consumerProperties.getExtension().setAckEachRecord(true);
+		consumerProperties.getExtension().setAckMode(ContainerProperties.AckMode.RECORD);
 		Binding<MessageChannel> consumerBinding1 = binder.bindConsumer(testDestination,
 				"test1", inbound1, consumerProperties);
 		QueueChannel inbound2 = new QueueChannel();

spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/bootstrap/KafkaBinderMeterRegistryTest.java

@@ -0,0 +1,104 @@
+/*
+ * 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.bootstrap;
+
+import java.util.function.Function;
+
+import io.micrometer.core.instrument.MeterRegistry;
+import org.junit.ClassRule;
+import org.junit.Test;
+
+import org.springframework.boot.WebApplicationType;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.builder.SpringApplicationBuilder;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.kafka.test.rule.EmbeddedKafkaRule;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+
+/**
+ * @author Soby Chacko
+ */
+public class KafkaBinderMeterRegistryTest {
+
+	@ClassRule
+	public static EmbeddedKafkaRule embeddedKafka = new EmbeddedKafkaRule(1, true, 10);
+
+	@Test
+	public void testMetricsWithSingleBinder() {
+		ConfigurableApplicationContext applicationContext = new SpringApplicationBuilder(SimpleApplication.class)
+				.web(WebApplicationType.NONE)
+				.run("--spring.cloud.stream.bindings.uppercase-in-0.destination=inputTopic",
+						"--spring.cloud.stream.bindings.uppercase-in-0.group=inputGroup",
+						"--spring.cloud.stream.bindings.uppercase-out-0.destination=outputTopic",
+						"--spring.cloud.stream.kafka.binder.brokers" + "="
+								+ embeddedKafka.getEmbeddedKafka().getBrokersAsString());
+
+		final MeterRegistry meterRegistry = applicationContext.getBean(MeterRegistry.class);
+		assertMeterRegistry(meterRegistry);
+		applicationContext.close();
+	}
+
+	@Test
+	public void testMetricsWithMultiBinders() {
+		ConfigurableApplicationContext applicationContext = new SpringApplicationBuilder(SimpleApplication.class)
+				.web(WebApplicationType.NONE)
+				.run("--spring.cloud.stream.bindings.uppercase-in-0.destination=inputTopic",
+						"--spring.cloud.stream.bindings.uppercase-in-0.group=inputGroup",
+						"--spring.cloud.stream.bindings.uppercase-in-0.binder=kafka1",
+						"--spring.cloud.stream.bindings.uppercase-out-0.destination=outputTopic",
+						"--spring.cloud.stream.bindings.uppercase-out-0.binder=kafka2",
+						"--spring.cloud.stream.binders.kafka1.type=kafka",
+						"--spring.cloud.stream.binders.kafka2.type=kafka",
+						"--spring.cloud.stream.binders.kafka1.environment"
+								+ ".spring.cloud.stream.kafka.binder.brokers" + "="
+								+ embeddedKafka.getEmbeddedKafka().getBrokersAsString(),
+						"--spring.cloud.stream.binders.kafka2.environment"
+								+ ".spring.cloud.stream.kafka.binder.brokers" + "="
+								+ embeddedKafka.getEmbeddedKafka().getBrokersAsString());
+
+		final MeterRegistry meterRegistry = applicationContext.getBean(MeterRegistry.class);
+		assertMeterRegistry(meterRegistry);
+		applicationContext.close();
+	}
+
+	private void assertMeterRegistry(MeterRegistry meterRegistry) {
+		assertThat(meterRegistry).isNotNull();
+
+		// assert kafka binder metrics
+		assertThat(meterRegistry.get("spring.cloud.stream.binder.kafka.offset")
+				.tag("group", "inputGroup")
+				.tag("topic", "inputTopic").gauge().value()).isNotNull();
+
+		// assert consumer metrics
+		assertThatCode(() -> meterRegistry.get("kafka.consumer.connection.count").meter()).doesNotThrowAnyException();
+
+		// assert producer metrics
+		assertThatCode(() -> meterRegistry.get("kafka.producer.connection.count").meter()).doesNotThrowAnyException();
+	}
+
+	@SpringBootApplication
+	static class SimpleApplication {
+
+		@Bean
+		public Function<String, String> uppercase() {
+			return String::toUpperCase;
+		}
+	}
+}

spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/bootstrap/MultiBinderMeterRegistryTest.java

@@ -1,70 +0,0 @@
-/*
- * 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.bootstrap;
-
-import io.micrometer.core.instrument.MeterRegistry;
-import org.junit.ClassRule;
-import org.junit.Test;
-
-import org.springframework.boot.WebApplicationType;
-import org.springframework.boot.autoconfigure.SpringBootApplication;
-import org.springframework.boot.builder.SpringApplicationBuilder;
-import org.springframework.cloud.stream.annotation.EnableBinding;
-import org.springframework.cloud.stream.messaging.Sink;
-import org.springframework.context.ConfigurableApplicationContext;
-import org.springframework.kafka.test.rule.EmbeddedKafkaRule;
-
-import static org.assertj.core.api.Assertions.assertThat;
-
-/**
- * @author Soby Chacko
- */
-public class MultiBinderMeterRegistryTest {
-
-	@ClassRule
-	public static EmbeddedKafkaRule embeddedKafka = new EmbeddedKafkaRule(1, true, 10);
-
-	@Test
-	public void testMetricsWorkWithMultiBinders() {
-		ConfigurableApplicationContext applicationContext = new SpringApplicationBuilder(
-				SimpleApplication.class).web(WebApplicationType.NONE).run(
-						"--spring.cloud.stream.bindings.input.destination=foo",
-						"--spring.cloud.stream.bindings.input.binder=inbound",
-						"--spring.cloud.stream.bindings.input.group=testGroupabc",
-						"--spring.cloud.stream.binders.inbound.type=kafka",
-						"--spring.cloud.stream.binders.inbound.environment"
-								+ ".spring.cloud.stream.kafka.binder.brokers" + "="
-								+ embeddedKafka.getEmbeddedKafka().getBrokersAsString());
-
-		final MeterRegistry meterRegistry = applicationContext.getBean(MeterRegistry.class);
-
-		assertThat(meterRegistry).isNotNull();
-
-		assertThat(meterRegistry.get("spring.cloud.stream.binder.kafka.offset")
-				.tag("group", "testGroupabc")
-				.tag("topic", "foo").gauge().value()).isNotNull();
-
-		applicationContext.close();
-	}
-
-	@SpringBootApplication
-	@EnableBinding(Sink.class)
-	static class SimpleApplication {
-
-	}
-
-}

spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaBinderTopicPropertiesUpdateTest.java

@@ -0,0 +1,123 @@
+/*
+ * Copyright 2018-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.integration;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.admin.Config;
+import org.apache.kafka.clients.admin.DescribeConfigsResult;
+import org.apache.kafka.common.KafkaFuture;
+import org.apache.kafka.common.config.ConfigResource;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.cloud.stream.annotation.EnableBinding;
+import org.springframework.cloud.stream.annotation.Input;
+import org.springframework.cloud.stream.annotation.Output;
+import org.springframework.cloud.stream.annotation.StreamListener;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.kafka.test.rule.EmbeddedKafkaRule;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.SubscribableChannel;
+import org.springframework.messaging.handler.annotation.SendTo;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit4.SpringRunner;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Heiko Does
+ */
+@RunWith(SpringRunner.class)
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE, properties = {
+		"spring.cloud.stream.kafka.bindings.standard-out.producer.topic.properties.retention.ms=9001",
+		"spring.cloud.stream.kafka.default.producer.topic.properties.retention.ms=-1",
+		"spring.cloud.stream.kafka.bindings.standard-in.consumer.topic.properties.retention.ms=9001",
+		"spring.cloud.stream.kafka.default.consumer.topic.properties.retention.ms=-1"
+})
+@DirtiesContext
+public class KafkaBinderTopicPropertiesUpdateTest {
+
+	private static final String KAFKA_BROKERS_PROPERTY = "spring.cloud.stream.kafka.binder.brokers";
+
+	@ClassRule
+	public static EmbeddedKafkaRule kafkaEmbedded = new EmbeddedKafkaRule(1, true, "standard-in", "standard-out");
+
+	@BeforeClass
+	public static void setup() {
+		System.setProperty(KAFKA_BROKERS_PROPERTY,
+				kafkaEmbedded.getEmbeddedKafka().getBrokersAsString());
+	}
+
+	@AfterClass
+	public static void clean() {
+		System.clearProperty(KAFKA_BROKERS_PROPERTY);
+	}
+
+	@Autowired
+	private ConfigurableApplicationContext context;
+
+	@Test
+	public void testKafkaBinderUpdateTopicConfiguration() throws Exception {
+		Map<String, Object> adminClientConfig = new HashMap<>();
+		adminClientConfig.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaEmbedded.getEmbeddedKafka().getBrokersAsString());
+		AdminClient adminClient = AdminClient.create(adminClientConfig);
+		ConfigResource standardInConfigResource = new ConfigResource(ConfigResource.Type.TOPIC, "standard-in");
+		ConfigResource standardOutConfigResource = new ConfigResource(ConfigResource.Type.TOPIC, "standard-out");
+		DescribeConfigsResult describeConfigsResult = adminClient.describeConfigs(Arrays
+				.asList(standardInConfigResource, standardOutConfigResource));
+		KafkaFuture<Map<ConfigResource, Config>> kafkaFuture = describeConfigsResult.all();
+		Map<ConfigResource, Config> configResourceConfigMap = kafkaFuture.get(3, TimeUnit.SECONDS);
+		Config standardInTopicConfig = configResourceConfigMap.get(standardInConfigResource);
+		assertThat(standardInTopicConfig.get("retention.ms").value()).isEqualTo("9001");
+
+		Config standardOutTopicConfig = configResourceConfigMap.get(standardOutConfigResource);
+		assertThat(standardOutTopicConfig.get("retention.ms").value()).isEqualTo("9001");
+	}
+
+	@EnableBinding(CustomBindingForTopicPropertiesUpdateTesting.class)
+	@EnableAutoConfiguration
+	public static class KafkaMetricsTestConfig {
+
+		@StreamListener("standard-in")
+		@SendTo("standard-out")
+		public String process(String payload) {
+			return payload;
+		}
+	}
+
+	interface CustomBindingForTopicPropertiesUpdateTesting {
+
+		@Input("standard-in")
+		SubscribableChannel standardIn();
+
+		@Output("standard-out")
+		MessageChannel standardOut();
+	}
+
+}

spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaConfigCustomizationTests.java

@@ -0,0 +1,180 @@
+/*
+ * Copyright 2020-2020 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.integration;
+
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerInterceptor;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerInterceptor;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.TopicPartition;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+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.context.annotation.Bean;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.test.rule.EmbeddedKafkaRule;
+import org.springframework.kafka.test.utils.KafkaTestUtils;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit4.SpringRunner;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Soby Chacko
+ *
+ * Based on: https://github.com/spring-projects/spring-kafka/issues/897#issuecomment-466060097
+ */
+@RunWith(SpringRunner.class)
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE, properties = {"spring.cloud.function.definition=process",
+		"spring.cloud.stream.bindings.process-in-0.group=KafkaConfigCustomizationTests.group"})
+@DirtiesContext
+public class KafkaConfigCustomizationTests {
+
+	private static final String KAFKA_BROKERS_PROPERTY = "spring.cloud.stream.kafka.binder.brokers";
+
+	@ClassRule
+	public static EmbeddedKafkaRule kafkaEmbedded = new EmbeddedKafkaRule(1, true);
+
+	static final CountDownLatch countDownLatch = new CountDownLatch(2);
+
+	@BeforeClass
+	public static void setup() {
+		System.setProperty(KAFKA_BROKERS_PROPERTY,
+				kafkaEmbedded.getEmbeddedKafka().getBrokersAsString());
+	}
+
+	@AfterClass
+	public static void clean() {
+		System.clearProperty(KAFKA_BROKERS_PROPERTY);
+	}
+
+	@Test
+	public void testBothConsumerAndProducerConfigsCanBeCustomized() throws InterruptedException {
+		Map<String, Object> producerProps = KafkaTestUtils
+				.producerProps(kafkaEmbedded.getEmbeddedKafka());
+		KafkaTemplate<Integer, String> template = new KafkaTemplate<>(
+				new DefaultKafkaProducerFactory<>(producerProps));
+		template.send("process-in-0", "test-foo");
+		template.flush();
+		assertThat(countDownLatch.await(10, TimeUnit.SECONDS)).isTrue();
+	}
+
+	@SpringBootApplication
+	public static class ConfigCustomizerTestConfig {
+
+		@Bean
+		public Function<String, String> process() {
+			return payload -> payload;
+		}
+
+		@Bean
+		public ConsumerConfigCustomizer consumerConfigCustomizer() {
+			return consumerProperties -> {
+				consumerProperties.put(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, MyConsumerInterceptor.class.getName());
+				consumerProperties.put("foo.bean", foo());
+			};
+		}
+
+		@Bean
+		public ProducerConfigCustomizer producerConfigCustomizer() {
+			return producerProperties -> {
+				producerProperties.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, MyProducerInterceptor.class.getName());
+				producerProperties.put("foo.bean", foo());
+			};
+		}
+
+		@Bean
+		public Foo foo() {
+			return new Foo();
+		}
+	}
+
+	public static class Foo {
+
+		public void foo(String what) {
+			KafkaConfigCustomizationTests.countDownLatch.countDown();
+		}
+
+	}
+
+	public static class MyConsumerInterceptor implements ConsumerInterceptor<String, String> {
+
+		private Foo foo;
+
+		@Override
+		public void configure(Map<String, ?> configs) {
+			this.foo = (Foo) configs.get("foo.bean");
+		}
+
+		@Override
+		public ConsumerRecords<String, String> onConsume(ConsumerRecords<String, String> records) {
+			this.foo.foo("consumer interceptor");
+			return records;
+		}
+
+		@Override
+		public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {
+
+		}
+
+		@Override
+		public void close() {
+		}
+
+	}
+
+	public static class MyProducerInterceptor implements ProducerInterceptor<String, String> {
+
+		private Foo foo;
+
+		@Override
+		public void configure(Map<String, ?> configs) {
+			this.foo = (Foo) configs.get("foo.bean");
+		}
+
+		@Override
+		public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
+			this.foo.foo("producer interceptor");
+			return record;
+		}
+
+		@Override
+		public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
+		}
+
+		@Override
+		public void close() {
+		}
+	}
+}