Fixes#509
- prevent autoconfiguration from reducing the existing partition count and/or replication factor
Split `autoConfigureTopics` in two separate settings, one for `autoCreateTopics`
and one for `autoAddPartitions` (to existing topics)
Renamed `defaultMinPartitionCount` to just `minPartitionCount` on the Kafka binder
Add a flag for controlling topic autoconfiguration by the Kafka binder
Fixes#504
- introduces a new binder setting `autoConfigureTopics` that allows the user to disable the automatic creation of topics by the binder
Additional tests for partitioning
Adding missing documentation for `replicationFactor`
Fixes#495
Currently, `minPartitionCount` is available both as a binder default and as a consumer property.
It would be simpler if it only was a binder setting (seeing as it has effect only as a default).
Fixes#498
- Honour the retry settings from ConsumerProperties;
- Add an additional `enableDlq` option in KafkaConsumerProperties that enables forwarding failed messages to a DLQ topic.
Fix#500
- Initialize a fixed-size ExecutorService of the size of the concurrency setting of the binder
- Clean up the ExecutorService on unbind
- Remove unused `spyOn` support
- Both the producer and consumer properties have `HeaderMode`
- Handle the case of embeddedHeaders and raw for both the Sending/ReceivingHandlers in Redis binder
This resolves#408
Move message values extraction to superclass
This resolves#425
Address review comments
- Update AssertTrue conditional logic
- Modify the message
Changes after rebase
- Changes after extendedProperties addition
* Add parameterized ExtendedConsumerProperties and ExtendedProducerProperties;
* Added ExtendedPropertiesBinder with the ability of managing per-binding property extensions and interacting the core service;
* Add extensions for Kafka and Rabbit
Addressing PR comments
Changed binder-specific binding prefix to `spring.cloud.stream.binderType.bindings`
Removing kafka-binder.properties
- Change the Binder interface to support ConsumerProperties/ProducerProperties beans and subclasses
- Binders can subclass the property beans to add new supported properties that will be automatically populated
- Spring Cloud Stream will infer the target type and populate the beans from the environment based on a `spring.cloud.stream.bindings..<bindingName>` prefix
- Remove binder defaults and retain only general binder configurations
TODO: a) decide on instanceIndex/partitionIndex alignment (we do not need both)
b) support `defaultProducer`/`defaultConsumer` properties
c) add leniency control on binding (fail/ignore for unknown properties)
d) add a `requiredProperties` configuration for consumer/producer properties to finely tune the mandatory properties expected to be supported by a bound application
Changes made during review:
- Add support for consumer and producer defaults
- Remove partitionIndex, keeping only instanceIndex
- Fix default properties for Kafka binder
- Move batching properties to Rabbit only
- When deserialising the message, replace `contentType` with the `originalContentType` only if the `originalContentType` isn't null which means the original content type was set by the producer before serialising.
- Add and fix tests
This resolves#404
- When deserializing the payload at the consumer endpoint, the non-byte stream payload type requires to use `String` object when the underlying message content-type is of any `text` type contentType (text/plain, text/xml and text/html).
- This fix is only needed to support any non-SCSt applications that will have the 'text/*` contentType of the message that is being deserialized
- Add test
This resolves#403
- Since the `originalContentType` header is preserved across the wire while the serialization/de-serialization processes take place using the deduced `contentType`,
it is convenient to set the `String` value of the `originalContentType` instead of setting the actual `MimeType` object in the header.
- Once the `de-serialization` has taken place, the deduced `contentType` is always replaced with the `originalContentType` as the `contentType` in the message header and this preserved contentType can be resolved to its corresponding MimeType using `ContentTypeResolver` anytime. Hence, using the `String` object for `originalContentType` doesn't do any harm.
- Fix the issue while setting the deduced contentType based on the `payload` when the `originalContentType` is `application/json`
- Currently, the contentType is set to `text/plain` which seems incorrect
- Also, add the `application/json` contentType to return `String` object during de-serialization process.
This resolves#383
Use `text/plain` contentType when the payload is of type `String`
- This is for the intermediate contentType that is used for serialization/de-serialization
- subclasses should implement the new onInit() method instead of overriding
- avoids the potential for a NPE in case an overriding subclass did not call the superclass method
- The `spring-cloud-stream-tuple` is now split into `spring-tuple`, `spring-integration-tuple` along with `spring-batch-tuple`
- Update dependencies to use the new ones
This resolves#326
Moved partition calculation from AbstractBinder to
its own class PartitionHandler.
Updated per review feedback
Refactored bean creation
Added synchronization around bean creation
- Add a binding property `spring.cloud.stream.bindings.<outputChannelName>.syncProducer`
When set to `true`, the Kafka producer metadata is set to use `sync` producer
- Update DefaultPropertiesAccessor for Kafka to use the `syncProducer` property
- Add test
This resolves#343
Move syncProducer as a binder level property
- Add a property `syncProducer` in Kafka Binder configuration properties (prefix: spring.cloud.stream.binder.kafka)
- Set Kafka ProducerMeatadata `sync` based on the value set in this property
- Update test
Resolves#317
Remove the `durable` binder configuration property
Make subscriber groups durable by default
Introduce `requiredGroups` property
Kafka groups (non-anonymous) now start by default at EARLIEST, which is more appropriate for new stream consumers
Addressing PR comments
Resolves#330
- removes `unbind` from the binder and moves it to the `Binding` instance itself;
- `Binding` is now an interface with a default implementation provided by SCS
- Removed all methods from Binder except for unbind()
- Removed old and unused code
- Removed circular reference between default Binder and Binding implementations
- Removed Binder type
- Health indicator fetches broker addresses using ZK configuration and matches it with the leaders of the paritions being used in the binder
This resolves#297
Use Set instead of List to store error messages per broker
Exception handling when connecting to ZK
Add ZK connect/session timeout values as configuration properties
- Set it in the Kafka binder so that the same can be used for health indicator as well
Rename ZK properties and move them to binder configuration properties
