* 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
- Move KafkaBinderConfigurationProperties out of the configuration class
- Remove defaultProperties bean creation and use `locations` in `ConfigurationProperties` to load the default properties
Move properties to @PropertySource and add tests
Fixesspring-cloud/spring-cloud-stream#314
Give more time to redis on travis
Make multipliers more versatile, apply to kafka
Extract receive() method. Default is now 1s
bump multiplier for kafka on travis
Upgrade scala/SIK versions in mvn pom
Migrage to KafkaNativeOffsetManager from KafkaTopicOffsetManager
Introduce LoggingProducerListener for errors - See GH #151
Remove properties specific to the Topic offset manager
- Instead of escaping the topic name, perform validation that throws `RuntimeException` when the given kafka topic name doesn't meet the criteria set by Kafka.
This resolves#217
It actually still has to be managed using a property in
spring-cloud-stream-binders-parent, but this change is for users.
Any user that wants an app using stream and rabbit needs to use the
spring-cloud-dependencies BOM. A gradle user has to do nothing else
with this change. A maven user still has to add an explicit
<spring-aqmp.version/> to the application POM.
- Adds a generic 'bindersHealthIndicator' bean controlled by `management.health.binders.enabled` that tallies results from binders
- Binders are expected to expose one or more health indicators based on the status of the middleware connection. If no health indicators are exposed, the status is deemed to be 'UNKNOWN'
- Add support for Redis and Rabbit based on existing health indicators
Moved binder health indicator to autoconfiguration
Also updated copyright
