Fixes https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/677
The logic for resetting offsets only on the initial assignment used a simple
boolean; this is insufficient when concurrency is > 1.
Use a concurrent set instead to determine whether or not a particular topic/partition
has been sought.
Also, change the `initial` argument on `KafkaBindingRebalanceListener.onPartitionsAssigned()`
to be derived from a `ThreadLocal` and add javadocs about retaining the state by partition.
**backport to all supported versions** (Except `KafkaBindingRebalanceListener` which did
not exist before 2.1.x)
polishing
When using native de/serializaion (which is now the default), the binder should infer the common Serde's
used on input and output. The Serdes inferred are - Integer, Long, Short, Double, Float, String, byte[] and
Spring Kafka provided JsonSerde.
Resolves#368
Address PR review comments
Addressing PR review comments
Resolves#672
In Kafka Streams binder, use the native Serde mechanism as the default instead of the framework
provided message conversion on the input and output bindings. This is to align applications
written using Kafka Streams binder more compatible with native concepts and mechanisms.
Users can still disable native encoding and decoding through configuration.
Amend tests to accommodate the flipped configuration.
Resolves#651
Fixing a bug around when we have muliple beans defined as functions/csonumers in the new
Kafka Streams binder functional support.
Polishing
Resolves#636
Introducing the ability to provide custom state stores as regular
Spring beans and use them in the functional model of Kafka Streams binding.
Resolves#642
Fixing a bug around when we have muliple beans defined as functions/csonumers in the new
Kafka Streams binder functional support.
Polishing
Resolves#636
Filter in only Kafka streams function beans
It appears we have refactored to rename the class from `KStreamProcessor` to `KafkaStreamsProcessor` [see a5344655cb (diff-4a8582ee2d07e268f77a89c0633e42f5)], but the docs weren't updated. This commit does exactly that.
With multiplex topics, topics can be an array. Sine message listener container
bean name is created by calling a toString() on the topics, when it is an array
it creates an unusual bean name. Fixing the issue by creating the bean name by
using destination string directly.
When two processors with same name are present in the same application,
there is a bean creation conflict. Fixing that issue.
Add test to verify.
Modify existing tests.
Resolves#589
