There are issues when a bean declared as a function in the Kafka Streams
application tries to autowire a bean through method parameter injection.
Addressing these concerns.
Resolves#726
Use BeanFactoryUtils.beanNamesForTypeIncludingAncestors instead of
getBean from BeanFactory which forces the bean creation inside the
function detector condition. There was a race condition in which
applications were unable to autowire beans and use them in functions
while the detector condition was creating the beans. This change will
delay the creation of the function bean until it is needed.
When using the Kafka Streams binder, if the application chooses to
provide bootstrap server configuration through Kafka binder broker
property, then allow that. This way either type of broker config
works in Kafka Streams binder.
Resolves#401Resolves#720
Generate a random application id for Kafka Streams binder if the user
doesn't set one for the application. This is useful for development
purposes, as it avoids the creation of an explicit application id.
For production workloads, it is highly recommended to explicitly provide
and application id.
The gnerated application id follows a patter where it uses the function bean
name followed by a random UUID string which is followed by the literal appplicaitonId.
In the case of StreamListener, instead of function bean name, it uses the containing class +
StreamListener method name.
If the binder generates the application id, that information will be logged on the console
at startup.
Resolves#718Resolves#719
When joining two input KStreams, the binder throws an exceptin.
Fixing this issue by passing the wrapped target KStream from the
proxy object to the adapted StreamListener method.
Resolves#701
* Handle Deserialization errors when there is a key
Handle DLQ sending on deserialization errors when there is a key in the record.
Resolves#635
* Adding keySerde information in the functional bindings
* Introduce the ability to provide BiFunction beans as Kafka Streams processors.
* Bypass Spring Cloud Function catalogue for bean lookup by directly querying the bean factory.
* Add test to verify BiFunction behavior.
Resolves#690
This PR introduces some fundamental changes in the way functional model of Kafka Streams applications
are supported in the binder. For the most part, this PR is comprised of the following changes.
* Remove the usage of EnableBinding in Kafka Streams based Spring Cloud Stream applications where
a bean of type java.util.function.Function or java.util.function.Consumer is provides (instead of
a StreamListener). For StreamListener based Kafka Streams applications, EnableBinding is still
necessary and required.
* Target types (KStream, KTable, GlobalKTable) will be inferred and bound by the binder through
a new binder specific bindable proxy factory.
* By deault input bindings are named as <functionName>-input for functions with single input
and <functionName>-input-0...<functionName>-input-n for functions with n-1 number of inputs.
* By deault output bindings are named as <functionName>-output for functions with single output
and <functionName>-output-0...<functionName>-output-n for functions with n-1 number of outputs.
* If applications prefer custom input binding names, the defaults can be overridden through
spring.cloud.stream.function.inputBindings.<funcationName>. Similarly if custom outpub binding names
are needed, then that can be done through spring.cloud.streamfunction.outputBindings.<functionName>.
* Test changes
* Refactoring and polishing
Resolves#688
* Upgrade Kafaka versions used in tests.
* EmbeddedKafkaRule changes in tests.
* KafaTransactionTests changes (createProducer expectations in mockito).
* Kafka Streams bootstrap server now return an ArrayList instead of String.
Making the necessary changes in the binder to accommodate this.
* Kafka Streams state store tests - retention window changes.
* Topic provisioning for consumers ignores topic.properties settings if binding type is KTable or GlobalKTable
* Modify tests to verify the behavior during KTable/GlobalKTable bindings
* Polishing
Resolves#687
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
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
