When KafkaStreamsStateStore annotation is used on a method with multiple input bindings,
it throws an exception. The reason is that each successive input binding after the first one
is trying to recreate the store that is already created. Fixing this issue.
Resolves#551
Kafka binder metrics is broken with a multi-binder configuraiton.
Fixing the issues by propagating the MeterRegistry bean into the
binder context from parent.
Adding test to verify.
Removing the formatter plugin from the parent pom.
Resolves#546Resolves#549
Fixes https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/521
Previously, all pollable message sources got the client id `message.source`.
MBean registration failed with a warning when multiple pollable sources were present.
Use the binding name as the client id by default, overridable using the `client.id`
consumer property.
**cherry-pick to 2.0.x**
Resolves#523
* Handle errors for non-existing topics
When topic creation is disabled both on the binder and the broker,
the binder currently throws an NPE. Catching this situation and
throw a more graceful error to the user.
Adding tests to verify.
Resolves#513
* Addressing PR review comments
Temporarily provide a custom HeaderMaper as part of the binder that is
copied from Spring Kafka so that we can preserve backward compatibility
with older producers. When older producers send non String types,
the header mapper in Spring Kafka treats that as MimeType. This change will
use a HeaderMppaer that reinstates the MimeTypeJsonDeserializer.
When we can consume the Spring Kafka version that provides the HeaderMapper
with this fix in it, we will remove this custom version.
Resolves#509
When Kafka Streams binder is used in multi binder environments, the properties defined under environment
is not propagated to the auto configuration class. The environment processing only takes place when the
actual binder configuration is instantiated (for example, KStreamConfiguration), and therefore the environment
properties are unavailable during the earlier autoconfiguration. This change makes the environment properties
availble during auto configuration.
Resolves#504
Remove the usage of `ImportBeanDefinitionRegistrar` in Kafka Streams binder
components since the regular use of getBean from the outer context is safe to do so.
Unignore tests
Resolves#501
* Addressing PR review comments
Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/502
Before applying the fix for,
https://github.com/spring-cloud/spring-cloud-stream/issues/1531
failed with:
```
org.springframework.messaging.MessageDeliveryException: failed to send Message to channel 'test.output'; nested exception is java.lang.IllegalArgumentException: Partition key cannot be null, failedMessage=GenericMessage [payload=byte[3], headers={kafka_partitionId=5, id=3350a823-c876-f7a9-f98b-fdbd2aaa4c12, timestamp=1542823925354}]
...
Caused by: java.lang.IllegalArgumentException: Partition key cannot be null
at org.springframework.util.Assert.notNull(Assert.java:198)
at org.springframework.cloud.stream.binder.PartitionHandler.extractKey(PartitionHandler.java:112)
at org.springframework.cloud.stream.binder.PartitionHandler.determinePartition(PartitionHandler.java:93)
at org.springframework.cloud.stream.binding.MessageConverterConfigurer$PartitioningInterceptor.preSend(MessageConverterConfigurer.java:381)
at org.springframework.integration.channel.AbstractMessageChannel$ChannelInterceptorList.preSend(AbstractMessageChannel.java:589)
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:435)
... 31 more
```
Resolves#503
Instead of checking for the text TopicExistsException in the exception
message, use strong type check for TopicExistsException through instanceof on
the cause of the exception.
Adding test to verify.
Resolves#209
Address checkstyle errors and warnings in spring-cloud-stream-binder-kafka-core.
Remove a duplicate dependency declaration from parent pom.
Resolves#483Resolves#484
