anchor.diff:  
   ```
   
  -
   === State Cleanup
   
   By default, the `Kafkastreams.cleanup()` method is called when the binding is stopped.
  
positive.code: 
  After that, you must set all the binding level properties on these new binding names.
  
  Please keep in mind that with the functional programming model described above, sticking with the default binding names make sense in most situations.
  The only reason you may still want to do this overriding is when you have larger number of configuration properties and you want to map the bindings to something more domain friendly.
  
  ==== Setting up bootstrap server configuration
  
negative.code:  Zookeeper URL
  +
  Default: `localhost`
  
  deserializationExceptionHandler::
   Deserialization error handler type.
   This handler is applied at the binder level and thus applied against all input binding in the application.
   There is a way to control it in a more fine-grained way at the consumer binding level.
   Possible values are - `logAndContinue`, `logAndFail` or `sendToDlq`
  +
  Default: `logAndFail`
  
  applicationId::
   Convenient way to set the application.id for the Kafka Streams application globally at the binder level.
   If the application contains multiple functions or `StreamListener` methods, then the application id should be set differently.
   See above where setting the application id is discussed in detail.
  +
  Default: application will generate a static application ID. See the application ID section for more details.
  
  stateStoreRetry.maxAttempts::
    Max attempts for trying to connect to a state store.
  +
  Default: 1
  
  stateStoreRetry.backoffPeriod::
    Backoff period when trying to connect to a state store on a retry.
  +
  Default: 1000 ms
  
  ==== Kafka Streams Producer Properties
  
  The following properties are _only_ available for Kafka Streams producers and must be prefixed with `spring.cloud.stream.kafka.streams.bindings.<binding name>.producer.`
  For convenience, if there are multiple output bindings and they all require a common value, that can be configured by using the prefix `spring.cloud.stream.kafka.streams.default.producer.`.
  
  keySerde::
    key serde to use
  +
  Default: See the above discussion on message de/serialization
  
  valueSerde::
    value serde to use
  +
  Default: See the above discussion on message de/serialization
  
  useNativeEncoding::
    flag to enable/disable native encoding
  +
  Default: `true`.
  
  streamPartitionerBeanName:
   Custom outbound partitioner bean name to be used at the consumer.
   Applications can provide custom `StreamPartitioner` as a Spring bean and the name of this bean can be provided to the producer to use instead of the default one.
  +
  Default: See the discussion above on outbound partition support.
  
  ==== Kafka Streams Consumer Properties
  
  The following properties are available for Kafka Streams consumers and must be prefixed with `spring.cloud.stream.kafka.streams.bindings.<binding-name>.consumer.`
  For convenience, if there are multiple input bindings and they all require a common value, that can be configured by using the prefix `spring.cloud.stream.kafka.streams.default.consumer.`.
  
  applicationId::
   Setting application.id per input binding. This is only preferred for `StreamListener` based processors, for function based processors see other approaches outlined above.
  +
  Default: See above.
  
  keySerde::
    key serde to use
  +
  Default: See the above discussion on message de/serialization
  
  valueSerde::
    value serde to use
  +
  Default: See the above discussion on message de/serialization
  
  materializedAs::
    state store to materialize when using incoming KTable types
  +
  Default: `none`.
  
  useNativeDecoding::
    flag to enable/disable native decoding
  +
  Default: `true`.
  
  dlqName::
    DLQ topic name.
  +
  Default: See above on the discussion of error handling and DLQ.
  
  startOffset::
    Offset to start from if there is no committed offset to consume from.
    This is mostly used when the consumer is consuming from a topic for the first time.
    Kafka Streams uses `earliest` as the default strategy and the binder uses the same default.
    This can be overridden to `latest` using this property.
  +
  Default: `earliest`.
  
  Note: Using `resetOffsets` on the consumer does not have any effect on Kafka Streams binder.
  Unlike the message channel based binder, Kafka Streams binder does not seek to beginning or end on demand.
  
  deserializationExceptionHandler::
   Deserialization error handler type.
   This handler is applied per consumer binding as opposed to the binder level property described before.
   Possible values are - `logAndContinue`, `logAndFail` or `sendToDlq`
  +
  Default: `logAndFail`
  
  timestampExtractorBeanName::
   Specific time stamp extractor bean name to be used at the consumer.
   Applications can provide `TimestampExtractor` as a Spring bean and the name of this bean can be provided to the consumer to use instead of the default one.
  +
  Default: See the discussion above on timestamp extractors.
  
  ==== Special note on concurrency
  
  In Kafka Streams, you can control of the number of threads a processor can create using the `num.stream.threads` property.
  This, you can do using the various `configuration` options described above under binder, functions, producer or consumer level.
  You can also use the `concurrency` property that core Spring Cloud Stream provides for this purpose.
  When using this, you need to use it on the consumer.
  When you have more than one input bindings either in a function or `StreamListener`, set this on the first input binding.
  For e.g. when setting `spring.cloud.stream.bindings.process-in-0.consumer.concurrency`, it will be transalted as `num.stream.threads` by the binder.
  