- 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
- bind Spring Integration `errorChannel` as a message producer when the property `spring.cloud.stream.bindings.error.destination` is set
- The error channel will use this property value as the destination name
- Add test
This resolves#329
Rename NoopBindable -> BindableAdapter
Return error channel name for getOutputs
Fix review comments
- Modified integration test to use `Sink` interface
- Add unit test
- 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
So you can add a parent for instance, like this:
new AggregateApplicationBuilder().from(TwitterStreamApplication.class)
.args(new String[] { "--language=en" }).to(StreamTransformer.class)
.parent(
SpringApplication.run(ModuleApplication.class, args))
.run();
Moved partition calculation from AbstractBinder to
its own class PartitionHandler.
Updated per review feedback
Refactored bean creation
Added synchronization around bean creation
- Uses builder pattern to create aggregate application that supports specifying
- args
- profiles
- config name for each child app and parent aggregate app separately.
An example config would look like this:
```
@SpringBootApplication
public class DoubleApplication {
public static void main(String[] args) {
new AggregateApplicationBuilder().
from(SourceApplication.class).args("--fixedDelay=5000")
.to(SinkApplication.class).args("--debug=true").run("--spring.application.name=aggregate-test");
}
```
This resolves#362
- Add property `spring.cloud.stream.bindings.<channelName>.trackHistoryProperties` with comma separated string values that specify the exact
property names in `ChannelBindingServiceProperties` to be added in the message tracker header
- Add tests
This resolves#252
Add test for empty trackedHistories
- rename `trackHistoryProperties` to `trackedProperties`
- updated Javadoc
- 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
- Add `AetherProxyProperties` as a ConfigurationProperties for the proxy settings needed for the ProxySelector in Aether system
- Update proxy settings in two places:
1) Where the remote repositories are setup
2) WHere the repository system session is created to resolve the artifacts
This resolves#272
Set authentication only for proxy settings
Set port type to `int`
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
- Make SubjectMessageHandler a SmartLifecycle and assign it to phase 0, allowing it to be started after the outputs are bound and before the inputs are bound;
- Setup Rx in the start() method;
- Set the ServiceActivator associated with the handler to phase 0 as well;
- 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
- The setter method `setBindersHealthIndicator(CompositeHealthIndicator bindersHealthIndicator)` is sufficient enough to autowire the candidate bean with the name `bindersHealthIndicator` based on the hint at the method argument.
This gets interesting when the health indicator is disabled via `management.health.binders.enabled` and thereby the `bindersHealthIndicator` bean is not instantiated. In this case, if there are any matching beans of the type
`CompositeHealthIndicator` then that competes for the autowiring in the setter.
- Add `Qualifier` with the name `bindersHealthIndicator` which makes the autowiring happen only with the bean named `bindersHealthIndicator` and thereby no `type` matching
- Add test that demonstrates this uses case
This resolves#339
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
In the context of trying to configure Rabbit, the docs mention that the Rabbit config properties be added under `rabbit`:
```
...
binders:
rabbit1:
type: rabbit
environment:
spring:
rabbit:
host: <host1>
...
```
however, the properties should be configured under `rabbitmq` as reference in the [Spring Boot Appendix A](https://docs.spring.io/spring-boot/docs/current/reference/html/common-application-properties.html)
