This PR safe guards state store instances in case there are multiple KafkaStreams
instances present that have distinct application IDs but share State Store Names.
Change is backwards compatible: In case no KafkaStreams association of the thread
can be found, all local state stores are queried as before.
In case an associated KafkaStreams Instance is found, but required StateStore is
not found in this instance, a warning is issued but backwards compatibility is
preserved by looking up all state stores.
Store within KafkaStreams instance of thread is preferred over "foreign" store with same name.
Warning is issued if requested store is not found within KafkaStreams instance of thread.
The main benefit here is to get rid of randomly selecting stores across all KafkaStreams instances
in case a store is contained within multiple streams instances with same name.
Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/1161
* Disconnect spring-cloud-scheam-registry-client from Kafka Streams binder
that is used for testing
* Deprecate MessageConverterDelegateSerde
* Remove MessageConverterDelegateSerdeTest
The overview doc gets unncessarily copied to the GitHub
repository root README. This only needs to reside in the
reference manual docs. Cleaning the process that generates
the root README for the repository.
Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/1138
Don't report DOWN if a container is stopped normally.
This is a valid state when containers are not auto-startup or are stopped while
the app remains running.
Containers are stopped abnormally when
- a listener throws an `Error`
- a `CommonContainerStoppingErrorHandler` (or similar) is configured to stop the
container after an error.
When both regular Kafka and Kafka Streams functions are present,
the code that was added recently for function composition in
Kafka Streams binder was accidentally creating a binadable proxy
factory bean for non Kafka Streams functions. Resolving this issue.
Resovles https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/1156
* GH-1140: CommonErrorHandler per consumer binding
Setting CommonErrorHandler on consumer binding through its bean name.
If present, binder will resolve this bean and assign it on the listener
container.
Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/1140
* Addressing PR review
Coalesce the stream join tests in Kafka Streams binder around the functional model.
Remove duplicating the same join tests using the StreamListener model.
* Consumer/Producer prefix in Kafka Streams binder
Kafka Streams allows the applications to separate the consumer and producer
properties using consumer/producer prefixes. Add these prefixes automatically
if they are missing from the application.
Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/1065
* Addressing PR review comments
Start Kafka Streams bindgings only when they are not running.
Similarly, stop them only if they are running.
Without these guards in the bindings for KStream, KTable and GlobalKTable,
it may cause NPE's due to the backing concurrent collections in KafkaStreamsRegistry
not finding the proper KafkaStreams object, especially when the StreamsBuilderFactory
bean is already stopped through the binder provided manager.
* Honoring auto startup in Kafka Streams binder
When using Kafka Streams bineder, the processors are started
unconditionally, i.e. autoStartup is always true by default.
If spring.kafka.streams.auto-startup is set, then honor that
as the auto-startup flag in the binder.
Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/1126
* Addressing PR review comments
Auto startup flag is honored per individual consumer bindings as well.
* Addressing PR review comments
* Making KafkaStreamsRegistry#kafkaStreams set to use a concurrent Set.
Restore Kafka Streams error state behavior in the binder, equivalent
to Kafka Streams prior to 2.8. Starting with 2.8, users can customize
the way uncaught errors are interpreted via an UncaughtExceptionHandler
in the applicaiton. Binder now sets a default handler that shuts down
the Kafka Streams client if thre is an uncaught error.
Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/1110
At the moment, Kafka Streams binder only allows KStream bindings on the outbound.
There is a delegation mechanism in which we stil can use KStream for output binding
while allowing the applications to provide a KTable type as the function return type.
Update docs.
Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/1085Resolves#1105
