anchor.diff:  Uncaught exceptions will be treated as errors by RxJava and will cause the `Observable` to complete, disrupting the flow.
   ====
   
  -=== Binder SPI
  -
  -Spring Cloud Stream provides a Binder abstraction for use in connecting to physical destinations.
  +== Binders
  +
  +Spring Cloud Stream provides a Binder abstraction for use in connecting to physical destinations at the external middleware.
   This section provides information about the main concepts behind the Binder SPI, its main components, and implementation-specific details.
   
  -==== Producers and Consumers
  +=== Producers and Consumers
   
   .Producers and Consumers
   image::producers-consumers.png[width=300,scaledwidth="75%"]
  
positive.code: Each group that is represented by consumer bindings for a given destination receives a copy of each message that a producer sends to that destination (i.e., publish-subscribe semantics).
  If there are multiple consumer instances bound using the same group name, then messages will be load-balanced across those consumer instances so that each message sent by a producer is consumed by only a single consumer instance within each group (i.e., queueing semantics).
  
  ==== Kafka Binder
  
  .Kafka Binder
  image::kafka-binder.png[width=300,scaledwidth="50%"]
  
  The Kafka Binder implementation maps the destination to a Kafka topic.
  The consumer group maps directly to the same Kafka concept.
  Spring Cloud Stream does not use the high-level consumer, but implements a similar concept for the simple consumer.
  
  
  ==== RabbitMQ Binder
  
  .RabbitMQ Binder
  image::rabbit-binder.png[width=300,scaledwidth="50%"]
  
  The RabbitMQ Binder implementation maps the destination to a `TopicExchange`.
  For each consumer group, a `Queue` will be bound to that `TopicExchange`.
  Each consumer instance that binds will trigger creation of a corresponding RabbitMQ `Consumer` instance for its group's `Queue`.
  
  == Configuration Options
  
  Spring Cloud Stream supports general configuration options as well as configuration for bindings and binders.
  Some binders allow additional binding properties to support middleware-specific features.
  
  Configuration options can be provided to Spring Cloud Stream applications via any mechanism supported by Spring Boot.
  This includes application arguments, environment variables, and YAML or .properties files.
  
  === Spring Cloud Stream Properties
  
  spring.cloud.stream.instanceCount::
    The number of deployed instances of an application.
  Must be set for partitioning and if using Kafka.
  +
  Default: `1`.
  
  spring.cloud.stream.instanceIndex::
    The instance index of the application: a number from `0` to `instanceCount`-1.
  Used for partitioning and with Kafka.
  Automatically set in Cloud Foundry to match the application's instance index.
  spring.cloud.stream.dynamicDestinations::
    A list of destinations that can be bound dynamically (for example, in a dynamic routing scenario).
  If set, only listed destinations can be bound.
  +
  Default: empty (allowing any destination to be bound).
  
  spring.cloud.stream.defaultBinder::
    The default binder to use, if multiple binders are configured.
  See <<multiple-binders,Multiple Binders on the Classpath>>.
  
  [[binding-properties]]
  === Binding Properties
  
  Binding properties are supplied using the format `spring.cloud.stream.bindings.<channelName>.<property>=<value>`.
  The `<channelName>` represents the name of the channel being configured (e.g., `output` for a `Source`).
  
  In what follows, we indicate where we have omitted the `spring.cloud.stream.bindings.<channelName>.` prefix and focus just on the property name, with the understanding that the prefix will be included at runtime.
  
  ==== Properties for Use of Spring Cloud Stream
  
  The following binding properties are available for both input and output bindings and
  must be prefixed with `spring.cloud.stream.bindings.<channelName>.`.
  
  destination::
      The target destination of a channel on the bound middleware (e.g., the RabbitMQ exchange or Kafka topic).
      If not set, the channel name is used instead.
  group::
      The consumer group of the channel.
  Applies only to inbound bindings.
  See <<consumer-groups,Consumer Groups>>.
  +
  Default: null (indicating an anonymous consumer).
  contentType::
      The content type of the channel.
  //See <<content type management>>.
  +
  Default: null (so that no type coercion is performed).
  binder::
      The binder used by this binding.
  See <<multiple-binders>> for details.
  +
  Default: null (the default binder will be used, if one exists).
  
  ==== Consumer properties
  
  The following binding properties are available for input bindings only and must be prefixed with `spring.cloud.stream.bindings.<channelName>.consumer.`.
  
  concurrency::
    The concurrency of the inbound consumer.
  +
  Default: `1`.
  partitioned::
    Whether the consumer receives data from a partitioned producer.
  +
  Default: `false`.
  headerMode::
      When set to `raw`, disables header parsing on input.
  Effective only for messaging middleware that does not support message headers natively and requires header embedding.
  Useful when inbound data is coming from outside Spring Cloud Stream applications.
  +
  Default: `embeddedHeaders`.
  maxAttempts::
    The number of attempts of re-processing an inbound message.
  Currently ignored by Kafka.
  +
  Default: `3`.
  backOffInitialInterval::
    The backoff initial interval on retry.
  Currently ignored by Kafka.
  +
  Default: `1000`.
  backOffMaxInterval::
    The maximum backoff interval.
  Currently ignored by Kafka.
  +
  Default: `10000`.
  backOffMultiplier::
    The backoff multiplier.
  +
  Default: `2.0`.
  
  ==== Producer Properties
  
  The following binding properties are available for output bindings only and must be prefixed with `spring.cloud.stream.bindings.<channelName>.producer.`.
  
  partitionKeyExpression::
    A SpEL expression that determines how to partition outbound data.
  If set, or if `partitionKeyExtractorClass` is set, outbound data on this channel will be partitioned, and `partitionCount` must be set to a value greater than 1 to be effective.
  The two options are mutually exclusive.
  See <<partitioning>>.
  +
  Default: null.
  partitionKeyExtractorClass::
    A `PartitionKeyExtractorStrategy` implementation.
  If set, or if `partitionKeyExpression` is set, outbound data on this channel will be partitioned, and `partitionCount` must be set to a value greater than 1 to be effective.
  The two options are mutually exclusive.
  See <<partitioning>>.
  +
  Default: null.
  partitionSelectorClass::
    A `PartitionSelectorStrategy` implementation.
  Mutually exclusive with `partitionSelectorExpression`.
  If neither is set, the partition will be selected as the `hashCode(key) % partitionCount`, where `key` is computed via either `partitionKeyExpression` or `partitionKeyExtractorClass`.
  +
  Default: null.
  partitionSelectorExpression::
    A SpEL expression for customizing partition selection.
  Mutually exclusive with `partitionSelectorClass`.
  If neither is set, the partition will be selected as the `hashCode(key) % partitionCount`, where `key` is computed via either `partitionKeyExpression` or `partitionKeyExtractorClass`.
  +
  Default: null.
  partitionCount::
    The number of target partitions for the data, if partitioning is enabled.
  Must be
    set to a value greater than 1 if the producer is partitioned.
  On Kafka, interpreted as a
    hint; the larger of this and the partition count of the target topic is used instead.
  +
  Default: `1`.
  requiredGroups::
    A comma-separated list of groups to which the producer must ensure message delivery even if they start after it has been created (e.g., by pre-creating durable queues in RabbitMQ).
  headerMode::
    When set to `raw`, disables header embedding on output.
  Effective only for messaging middleware that does not support message headers natively and requires header embedding.
  Useful when producing data for non-Spring Cloud Stream applications.
  +
  Default: `embeddedHeaders`.
  
  [[binder-specific-configuration]]
  == Binder-Specific Configuration
  
  The following binder, consumer, and producer properties are specific to binder implementations.
  
  === Rabbit-Specific Settings
  
  ==== RabbitMQ Binder Properties
  
  By default, the RabbitMQ binder uses Spring Boot's `ConnectionFactory`, and it therefore supports all Spring Boot configuration options for RabbitMQ.
  (For reference, consult the http://docs.spring.io/spring-boot/docs/current/reference/htmlsingle/#common-application-properties[Spring Boot documentation].) RabbitMQ configuration options use the `spring.rabbitmq` prefix.
  
  In addition to the Spring Boot options, the RabbitMQ binder supports the following properties:
  
  spring.cloud.stream.rabbit.binder.adminAddresses::
    A comma-separated list of RabbitMQ management plugin URLs.
  Only used when `nodes` contains more than one entry.
  Each entry in this list must have a corresponding entry in `spring.rabbitmq.addresses`.
  +
  Default: empty.
  spring.cloud.stream.rabbit.binder.nodes::
    A comma-separated list of RabbitMQ node names.
  When more than one entry, used to locate the server address where a queue is located.
  Each entry in this list must have a corresponding entry in `spring.rabbitmq.addresses`.
  +
  Default: empty.
  spring.cloud.stream.rabbit.binder.compressionLevel::
    Compression level for compressed bindings.
  See `java.util.zip.Deflater`.
  +
  Default: `1` (BEST_LEVEL).
  
  ==== RabbitMQ Consumer Properties
  
  The following properties are available for Rabbit consumers only and
  must be prefixed with `spring.cloud.stream.rabbit.bindings.<channelName>.consumer.`.
  
  acknowledgeMode::
    The acknowledge mode.
  +
  Default: `AUTO`.
  autoBindDlq::
    Whether to automatically declare the DLQ and bind it to the binder DLX.
  +
  Default: `false`.
  durableSubscription::
    Whether subscription should be durable.
  Only effective if `group` is also set.
  +
  Default: `true`.
  maxConcurrency::
    Default: `1`.
  prefetch::
    Prefetch count.
  +
  Default: `1`.
  prefix::
    A prefix to be added to the name of the `destination` and queues.
  +
  Default: "".
  requeueRejected::
    Whether delivery failures should be requeued.
  +
  Default: `true`.
  requestHeaderPatterns::
    The request headers to be transported.
  +
  Default: `[STANDARD_REQUEST_HEADERS,'*']`.
  replyHeaderPatterns::
    The reply headers to be transported.
  +
  Default: `[STANDARD_REQUEST_HEADERS,'*']`.
  republishToDlq::
    By default, messages which fail after retries are exhausted are rejected.
  If a dead-letter queue (DLQ) is configured, RabbitMQ will route the failed message (unchanged) to the DLQ.
  If set to `true`, the bus will republish failed messages to the DLQ with additional headers, including the exception message and stack trace from the cause of the final failure.
  transacted::
    Whether to use transacted channels.
  +
  Default: `false`.
  txSize::
    The number of deliveries between acks.
  +
  Default: `1`.
  
  ==== Rabbit Producer Properties
  
  The following properties are available for Rabbit producers only and
  must be prefixed with `spring.cloud.stream.rabbit.bindings.<channelName>.producer.`.
  
  autoBindDlq::
    Whether to automatically declare the DLQ and bind it to the binder DLX.
  +
  Default: `false`.
  batchingEnabled::
    Whether to enable message batching by producers.
  +
  Default: `false`.
  batchSize::
    The number of messages to buffer when batching is enabled.
  +
  Default: `100`.
  batchBufferLimit::
    Default: `10000`.
  batchTimeout::
    Default: `5000`.
  compress::
    Whether data should be compressed when sent.
  +
  Default: `false`.
  deliveryMode::
    Delivery mode.
  +
  Default: `PERSISTENT`.
  prefix::
    A prefix to be added to the name of the `destination` exchange.
  +
  Default: "".
  requestHeaderPatterns::
    The request headers to be transported.
  +
  Default: `[STANDARD_REQUEST_HEADERS,'*']`.
  replyHeaderPatterns::
    The reply headers to be transported.
  +
  Default: `[STANDARD_REQUEST_HEADERS,'*']`.
  
  === Kafka-Specific Settings
  
  ==== Kafka Binder Properties
  
  spring.cloud.stream.kafka.binder.brokers::
    A list of brokers to which the Kafka binder will connect.
  +
  Default: `localhost`.
  spring.cloud.stream.kafka.binder.defaultBrokerPort::
    `brokers` allows hosts specified with or without port information (e.g., `host1,host2:port2`).
  This sets the default port when no port is configured in the broker list.
  +
  Default: `9092`.
  spring.cloud.stream.kafka.binder.zkNodes::
    A list of ZooKeeper nodes to which the Kafka binder can connect.
  +
  Default: `localhost`.
  spring.cloud.stream.kafka.binder.defaultZkPort::
    `zkNodes` allows hosts specified with or without port information (e.g., `host1,host2:port2`).
  This sets the default port when no port is configured in the node list.
  +
  Default: `2181`.
  spring.cloud.stream.kafka.binder.headers::
    The list of custom headers that will be transported by the binder.
  +
  Default: empty.
  spring.cloud.stream.kafka.binder.offsetUpdateTimeWindow::
    The frequency, in milliseconds, with which offsets are saved.
  Ignored if `0`.
  +
  Default: `10000`.
  spring.cloud.stream.kafka.binder.offsetUpdateCount::
    The frequency, in number of updates, which which consumed offsets are persisted.
  Ignored if `0`.
  Mutually exclusive with `offsetUpdateTimeWindow`.
  +
  Default: `0`.
  spring.cloud.stream.kafka.binder.requiredAcks::
    The number of required acks on the broker.
  +
  Default: `1`.
  spring.cloud.stream.kafka.binder.minPartitionCount::
    Effective only if `autoCreateTopics` or `autoAddPartitions` is set.
  The global minimum number of partitions that the binder will configure on topics on which it produces/consumes data.
  It can be superseded by the `partitionCount` setting of the producer or by the value of `instanceCount` * `concurrency` settings of the producer (if either is larger).
  +
  Default: `1`.
  spring.cloud.stream.kafka.binder.replicationFactor::
    The replication factor of auto-created topics if `autoCreateTopics` is active.
  +
  Default: `1`.
  spring.cloud.stream.kafka.binder.autoCreateTopics::
    If set to `true`, the binder will create new topics automatically.
  If set to `false`, the binder will rely on the topics being already configured.
  In the latter case, if the topics do not exist, the binder will fail to start.
  Of note, this setting is independent of the `auto.topic.create.enable` setting of the broker and it does not influence it: if the server is set to auto-create topics, they may be created as part of the metadata retrieval request, with default broker settings.
  +
  Default: `true`.
  spring.cloud.stream.kafka.binder.autoAddPartitions::
    If set to `true`, the binder will create add new partitions if required.
  If set to `false`, the binder will rely on the partition size of the topic being already configured.
  If the partition count of the target topic is smaller than the expected value, the binder will fail to start.
  +
  Default: `false`.
  
  
  ==== Kafka Consumer Properties
  
  The following properties are available for Kafka consumers only and
  must be prefixed with `spring.cloud.stream.kafka.bindings.<channelName>.consumer.`.
  
  autoCommitOffset::
    Whether to autocommit offsets when a message has been processed.
  If set to `false`, an `Acknowledgment` header will be available in the message headers for late acknowledgment.
  +
  Default: `true`.
  resetOffsets::
    Whether to reset offsets on the consumer to the value provided by `startOffset`.
  +
  Default: `false`.
  startOffset::
    The starting offset for new groups, or when `resetOffsets` is `true`.
  Allowed values: `earliest`, `latest`.
  +
  Default: null (equivalent to `earliest`).
  enableDlq::
    When set to true, it will send enable DLQ behavior for the consumer.
    Messages that result in errors will be forwarded to a topic named `error.<destination>.<group>`.
    This provides an alternative option to the more common Kafka replay scenario for the case when the number of errors is relatively small and replaying the entire original topic may be too cumbersome.
  +
  Default: `false`.
  
  ==== Kafka Producer Properties
  
  The following properties are available for Kafka producers only and
  must be prefixed with `spring.cloud.stream.kafka.bindings.<channelName>.producer.`.
  
  bufferSize::
    Upper limit, in bytes, of how much data the Kafka producer will attempt to batch before sending.
  +
  Default: `16384`.
  sync::
    Whether the producer is synchronous.
  +
  Default: `false`.
  batchTimeout::
    How long the producer will wait before sending in order to allow more messages to accumulate in the same batch.
  (Normally the producer does not wait at all, and simply sends all the messages that accumulated while the previous send was in progress.) A non-zero value may increase throughput at the expense of latency.
  +
  Default: `0`.
  
  == Binder Detection
  
  Spring Cloud Stream relies on implementations of the Binder SPI to perform the task of connecting channels to message brokers.
  Each Binder implementation typically connects to one type of messaging system.
  Out of the box, Spring Cloud Stream provides binders for Kafka, RabbitMQ, and Redis.
  
  === Classpath Detection
  
  By default, Spring Cloud Stream relies on Spring Boot's auto-configuration to configure the binding process.
  If a single Binder implementation is found on the classpath, Spring Cloud Stream will use it automatically.
  
negative.code:                 host: <host2>
  ----
  
  
  === Implementation strategies
  
  This section details the binder implementation strategies for Kafka and Rabbit MQ, in what concerns mapping the Spring Cloud Stream concepts onto the middleware concepts.
  
  ==== Kafka Binder
  
  .Kafka Binder
  image::kafka-binder.png[width=300,scaledwidth="50%"]
  
  The Kafka Binder implementation maps the destination to a Kafka topic.
  The consumer group maps directly to the same Kafka concept.
  Spring Cloud Stream does not use the high-level consumer, but implements a similar concept for the simple consumer.
  
  
  ==== RabbitMQ Binder
  
  .RabbitMQ Binder
  image::rabbit-binder.png[width=300,scaledwidth="50%"]
  
  The RabbitMQ Binder implementation maps the destination to a `TopicExchange`.
  For each consumer group, a `Queue` will be bound to that `TopicExchange`.
  Each consumer instance that binds will trigger creation of a corresponding RabbitMQ `Consumer` instance for its group's `Queue`.
  
  == Configuration Options
  
  Spring Cloud Stream supports general configuration options as well as configuration for bindings and binders.
  Some binders allow additional binding properties to support middleware-specific features.
  
  Configuration options can be provided to Spring Cloud Stream applications via any mechanism supported by Spring Boot.
  This includes application arguments, environment variables, and YAML or .properties files.
  
  === Spring Cloud Stream Properties
  
  spring.cloud.stream.instanceCount::
    The number of deployed instances of an application.
  Must be set for partitioning and if using Kafka.
  +
  Default: `1`.
  
  spring.cloud.stream.instanceIndex::
    The instance index of the application: a number from `0` to `instanceCount`-1.
  Used for partitioning and with Kafka.
  Automatically set in Cloud Foundry to match the application's instance index.
  spring.cloud.stream.dynamicDestinations::
    A list of destinations that can be bound dynamically (for example, in a dynamic routing scenario).
  If set, only listed destinations can be bound.
  +
  Default: empty (allowing any destination to be bound).
  
  spring.cloud.stream.defaultBinder::
    The default binder to use, if multiple binders are configured.
  See <<multiple-binders,Multiple Binders on the Classpath>>.
  
  [[binding-properties]]
  === Binding Properties
  
  Binding properties are supplied using the format `spring.cloud.stream.bindings.<channelName>.<property>=<value>`.
  The `<channelName>` represents the name of the channel being configured (e.g., `output` for a `Source`).
  
  In what follows, we indicate where we have omitted the `spring.cloud.stream.bindings.<channelName>.` prefix and focus just on the property name, with the understanding that the prefix will be included at runtime.
  
  ==== Properties for Use of Spring Cloud Stream
  
  The following binding properties are available for both input and output bindings and
  must be prefixed with `spring.cloud.stream.bindings.<channelName>.`.
  
  destination::
      The target destination of a channel on the bound middleware (e.g., the RabbitMQ exchange or Kafka topic).
      If not set, the channel name is used instead.
  group::
      The consumer group of the channel.
  Applies only to inbound bindings.
  See <<consumer-groups,Consumer Groups>>.
  +
  Default: null (indicating an anonymous consumer).
  contentType::
      The content type of the channel.
  //See <<content type management>>.
  +
  Default: null (so that no type coercion is performed).
  binder::
      The binder used by this binding.
  See <<multiple-binders>> for details.
  +
  Default: null (the default binder will be used, if one exists).
  
  ==== Consumer properties
  
  The following binding properties are available for input bindings only and must be prefixed with `spring.cloud.stream.bindings.<channelName>.consumer.`.
  
  concurrency::
    The concurrency of the inbound consumer.
  +
  Default: `1`.
  partitioned::
    Whether the consumer receives data from a partitioned producer.
  +
  Default: `false`.
  headerMode::
      When set to `raw`, disables header parsing on input.
  Effective only for messaging middleware that does not support message headers natively and requires header embedding.
  Useful when inbound data is coming from outside Spring Cloud Stream applications.
  +
  Default: `embeddedHeaders`.
  maxAttempts::
    The number of attempts of re-processing an inbound message.
  Currently ignored by Kafka.
  +
  Default: `3`.
  backOffInitialInterval::
    The backoff initial interval on retry.
  Currently ignored by Kafka.
  +
  Default: `1000`.
  backOffMaxInterval::
    The maximum backoff interval.
  Currently ignored by Kafka.
  +
  Default: `10000`.
  backOffMultiplier::
    The backoff multiplier.
  +
  Default: `2.0`.
  
  ==== Producer Properties
  
  The following binding properties are available for output bindings only and must be prefixed with `spring.cloud.stream.bindings.<channelName>.producer.`.
  
  partitionKeyExpression::
    A SpEL expression that determines how to partition outbound data.
  If set, or if `partitionKeyExtractorClass` is set, outbound data on this channel will be partitioned, and `partitionCount` must be set to a value greater than 1 to be effective.
  The two options are mutually exclusive.
  See <<partitioning>>.
  +
  Default: null.
  partitionKeyExtractorClass::
    A `PartitionKeyExtractorStrategy` implementation.
  If set, or if `partitionKeyExpression` is set, outbound data on this channel will be partitioned, and `partitionCount` must be set to a value greater than 1 to be effective.
  The two options are mutually exclusive.
  See <<partitioning>>.
  +
  Default: null.
  partitionSelectorClass::
    A `PartitionSelectorStrategy` implementation.
  Mutually exclusive with `partitionSelectorExpression`.
  If neither is set, the partition will be selected as the `hashCode(key) % partitionCount`, where `key` is computed via either `partitionKeyExpression` or `partitionKeyExtractorClass`.
  +
  Default: null.
  partitionSelectorExpression::
    A SpEL expression for customizing partition selection.
  Mutually exclusive with `partitionSelectorClass`.
  If neither is set, the partition will be selected as the `hashCode(key) % partitionCount`, where `key` is computed via either `partitionKeyExpression` or `partitionKeyExtractorClass`.
  +
  Default: null.
  partitionCount::
    The number of target partitions for the data, if partitioning is enabled.
  Must be
    set to a value greater than 1 if the producer is partitioned.
  On Kafka, interpreted as a
    hint; the larger of this and the partition count of the target topic is used instead.
  +
  Default: `1`.
  requiredGroups::
    A comma-separated list of groups to which the producer must ensure message delivery even if they start after it has been created (e.g., by pre-creating durable queues in RabbitMQ).
  headerMode::
    When set to `raw`, disables header embedding on output.
  Effective only for messaging middleware that does not support message headers natively and requires header embedding.
  Useful when producing data for non-Spring Cloud Stream applications.
  +
  Default: `embeddedHeaders`.
  
  [[binder-specific-configuration]]
  == Binder-Specific Configuration
  
  The following binder, consumer, and producer properties are specific to binder implementations.
  
  === Rabbit-Specific Settings
  
  ==== RabbitMQ Binder Properties
  
  By default, the RabbitMQ binder uses Spring Boot's `ConnectionFactory`, and it therefore supports all Spring Boot configuration options for RabbitMQ.
  (For reference, consult the http://docs.spring.io/spring-boot/docs/current/reference/htmlsingle/#common-application-properties[Spring Boot documentation].) RabbitMQ configuration options use the `spring.rabbitmq` prefix.
  
  In addition to the Spring Boot options, the RabbitMQ binder supports the following properties:
  
  spring.cloud.stream.rabbit.binder.adminAddresses::
    A comma-separated list of RabbitMQ management plugin URLs.
  Only used when `nodes` contains more than one entry.
  Each entry in this list must have a corresponding entry in `spring.rabbitmq.addresses`.
  +
  Default: empty.
  spring.cloud.stream.rabbit.binder.nodes::
    A comma-separated list of RabbitMQ node names.
  When more than one entry, used to locate the server address where a queue is located.
  Each entry in this list must have a corresponding entry in `spring.rabbitmq.addresses`.
  +
  Default: empty.
  spring.cloud.stream.rabbit.binder.compressionLevel::
    Compression level for compressed bindings.
  See `java.util.zip.Deflater`.
  +
  Default: `1` (BEST_LEVEL).
  
  ==== RabbitMQ Consumer Properties
  
  The following properties are available for Rabbit consumers only and
  must be prefixed with `spring.cloud.stream.rabbit.bindings.<channelName>.consumer.`.
  
  acknowledgeMode::
    The acknowledge mode.
  +
  Default: `AUTO`.
  autoBindDlq::
    Whether to automatically declare the DLQ and bind it to the binder DLX.
  +
  Default: `false`.
  durableSubscription::
    Whether subscription should be durable.
  Only effective if `group` is also set.
  +
  Default: `true`.
  maxConcurrency::
    Default: `1`.
  prefetch::
    Prefetch count.
  +
  Default: `1`.
  prefix::
    A prefix to be added to the name of the `destination` and queues.
  +
  Default: "".
  requeueRejected::
    Whether delivery failures should be requeued.
  +
  Default: `true`.
  requestHeaderPatterns::
    The request headers to be transported.
  +
  Default: `[STANDARD_REQUEST_HEADERS,'*']`.
  replyHeaderPatterns::
    The reply headers to be transported.
  +
  Default: `[STANDARD_REQUEST_HEADERS,'*']`.
  republishToDlq::
    By default, messages which fail after retries are exhausted are rejected.
  If a dead-letter queue (DLQ) is configured, RabbitMQ will route the failed message (unchanged) to the DLQ.
  If set to `true`, the bus will republish failed messages to the DLQ with additional headers, including the exception message and stack trace from the cause of the final failure.
  transacted::
    Whether to use transacted channels.
  +
  Default: `false`.
  txSize::
    The number of deliveries between acks.
  +
  Default: `1`.
  
  ==== Rabbit Producer Properties
  
  The following properties are available for Rabbit producers only and
  must be prefixed with `spring.cloud.stream.rabbit.bindings.<channelName>.producer.`.
  
  autoBindDlq::
    Whether to automatically declare the DLQ and bind it to the binder DLX.
  +
  Default: `false`.
  batchingEnabled::
    Whether to enable message batching by producers.
  +
  Default: `false`.
  batchSize::
    The number of messages to buffer when batching is enabled.
  +
  Default: `100`.
  batchBufferLimit::
    Default: `10000`.
  batchTimeout::
    Default: `5000`.
  compress::
    Whether data should be compressed when sent.
  +
  Default: `false`.
  deliveryMode::
    Delivery mode.
  +
  Default: `PERSISTENT`.
  prefix::
    A prefix to be added to the name of the `destination` exchange.
  +
  Default: "".
  requestHeaderPatterns::
    The request headers to be transported.
  +
  Default: `[STANDARD_REQUEST_HEADERS,'*']`.
  replyHeaderPatterns::
    The reply headers to be transported.
  +
  Default: `[STANDARD_REQUEST_HEADERS,'*']`.
  
  === Kafka-Specific Settings
  
  ==== Kafka Binder Properties
  
  spring.cloud.stream.kafka.binder.brokers::
    A list of brokers to which the Kafka binder will connect.
  +
  Default: `localhost`.
  spring.cloud.stream.kafka.binder.defaultBrokerPort::
    `brokers` allows hosts specified with or without port information (e.g., `host1,host2:port2`).
  This sets the default port when no port is configured in the broker list.
  +
  Default: `9092`.
  spring.cloud.stream.kafka.binder.zkNodes::
    A list of ZooKeeper nodes to which the Kafka binder can connect.
  +
  Default: `localhost`.
  spring.cloud.stream.kafka.binder.defaultZkPort::
    `zkNodes` allows hosts specified with or without port information (e.g., `host1,host2:port2`).
  This sets the default port when no port is configured in the node list.
  +
  Default: `2181`.
  spring.cloud.stream.kafka.binder.headers::
    The list of custom headers that will be transported by the binder.
  +
  Default: empty.
  spring.cloud.stream.kafka.binder.offsetUpdateTimeWindow::
    The frequency, in milliseconds, with which offsets are saved.
  Ignored if `0`.
  +
  Default: `10000`.
  spring.cloud.stream.kafka.binder.offsetUpdateCount::
    The frequency, in number of updates, which which consumed offsets are persisted.
  Ignored if `0`.
  Mutually exclusive with `offsetUpdateTimeWindow`.
  +
  Default: `0`.
  spring.cloud.stream.kafka.binder.requiredAcks::
    The number of required acks on the broker.
  +
  Default: `1`.
  spring.cloud.stream.kafka.binder.minPartitionCount::
    Effective only if `autoCreateTopics` or `autoAddPartitions` is set.
  The global minimum number of partitions that the binder will configure on topics on which it produces/consumes data.
  It can be superseded by the `partitionCount` setting of the producer or by the value of `instanceCount` * `concurrency` settings of the producer (if either is larger).
  +
  Default: `1`.
  spring.cloud.stream.kafka.binder.replicationFactor::
    The replication factor of auto-created topics if `autoCreateTopics` is active.
  +
  Default: `1`.
  spring.cloud.stream.kafka.binder.autoCreateTopics::
    If set to `true`, the binder will create new topics automatically.
  If set to `false`, the binder will rely on the topics being already configured.
  In the latter case, if the topics do not exist, the binder will fail to start.
  Of note, this setting is independent of the `auto.topic.create.enable` setting of the broker and it does not influence it: if the server is set to auto-create topics, they may be created as part of the metadata retrieval request, with default broker settings.
  +
  Default: `true`.
  spring.cloud.stream.kafka.binder.autoAddPartitions::
    If set to `true`, the binder will create add new partitions if required.
  If set to `false`, the binder will rely on the partition size of the topic being already configured.
  If the partition count of the target topic is smaller than the expected value, the binder will fail to start.
  +
  Default: `false`.
  
  
  ==== Kafka Consumer Properties
  
  The following properties are available for Kafka consumers only and
  must be prefixed with `spring.cloud.stream.kafka.bindings.<channelName>.consumer.`.
  
  autoCommitOffset::
    Whether to autocommit offsets when a message has been processed.
  If set to `false`, an `Acknowledgment` header will be available in the message headers for late acknowledgment.
  +
  Default: `true`.
  resetOffsets::
    Whether to reset offsets on the consumer to the value provided by `startOffset`.
  +
  Default: `false`.
  startOffset::
    The starting offset for new groups, or when `resetOffsets` is `true`.
  Allowed values: `earliest`, `latest`.
  +
  Default: null (equivalent to `earliest`).
  enableDlq::
    When set to true, it will send enable DLQ behavior for the consumer.
    Messages that result in errors will be forwarded to a topic named `error.<destination>.<group>`.
    This provides an alternative option to the more common Kafka replay scenario for the case when the number of errors is relatively small and replaying the entire original topic may be too cumbersome.
  +
  Default: `false`.
  
  ==== Kafka Producer Properties
  
  The following properties are available for Kafka producers only and
  must be prefixed with `spring.cloud.stream.kafka.bindings.<channelName>.producer.`.
  
  bufferSize::
    Upper limit, in bytes, of how much data the Kafka producer will attempt to batch before sending.
  +
  Default: `16384`.
  sync::
    Whether the producer is synchronous.
  +
  Default: `false`.
  batchTimeout::
    How long the producer will wait before sending in order to allow more messages to accumulate in the same batch.
  (Normally the producer does not wait at all, and simply sends all the messages that accumulated while the previous send was in progress.) A non-zero value may increase throughput at the expense of latency.
  +
  Default: `0`.
  
  [[contenttypemanagement]]
  == Content Type and Transformation
  
  