The `KinesisMessageDrivenChannelAdapter` may exit its `active` state,
but `consumerInvokers` won't be cleared in the main consumer loop
**Cherry-pick to master**
Fixes https://github.com/spring-projects/spring-integration-aws/issues/109
The plain `Converter` is not able to deal with `MessageHeaders` as a
context for target payload conversion.
* Use `MessageConverter` abstraction to get a request message context
for target conversion
* Deprecate a `setConverter()` in favor of a new `setMessageConverter()`
* Use delegating `ConvertingFromMessageConverter` for default serialization
behavior and for backward compatibility
* Add `flushDuration` option into `KplMessageHandler` for calling a
`kinesisProducer.flush()` periodically
* Call `kinesisProducer.flush()` immediately after `addUserRecord()`
`if (this.flushDuration.toMillis() <= 0) {`
Fixes https://github.com/spring-projects/spring-integration-aws/issues/153
Right now we ignore an `embeddedHeadersMapper` result when we also have a `converter`.
* Move `converter` logic for `batch` mode into `else` of the
`if (KclMessageDrivenChannelAdapter.this.embeddedHeadersMapper != null) {`.
This way we cover all the possible use-case, and don't override results
of each other and also don't perform extra logic for nothing
Fixes https://github.com/spring-projects/spring-integration-aws/issues/152
`map()` into `build()`` the result of a stream for the
`embeddedHeadersMapper` in the `processMultipleRecords()`.
Otherwise our `payload` is going to have `AbstractIntegrationMessageBuilder` instances.
The idea is to allow to process a batch of Messages.
Fixes https://github.com/spring-projects/spring-integration-aws/issues/108
* Add `CheckpointMode.periodic`
* Implement `CheckpointMode.periodic` in the `KinesisMessageDrivenChannelAdapter`
* Support all the `CheckpointMode`s in the `KclMessageDrivenChannelAdapter`
* Add `@author`, polishing and Copyrights
* Since internal `KinesisProxy` instance is fully based on the
`CredentialsProvider` provided in the `KinesisClientLibConfiguration`,
we don't have choice unless expose such an option for the
`KclMessageDrivenChannelAdapter`
* Make it as `DefaultAWSCredentialsProviderChain` for those ctors which
are based on the default client and add one more ctor argument for
externally provided clients
* Move an `executor` option to the setter and rely on the
`SimpleAsyncTaskExecutor` by default
* Do not copy `record.getData()` to a new array since
`record.getData().array()` is fully enough for the current version
relates to spring-cloud/spring-cloud-stream-binder-aws-kinesis#65, spring-cloud/spring-cloud-stream-binder-aws-kinesis#53, #99
Add UUID consumer generation
fixed author tag
upgraded kcl version to 2.0.5
fixed source file encoding
- added setter for checkpoint interval
- change application name (concatenate consumer group and stream name)
fixed javadoc
take into account requested changes from spring-integration
fixed code style issues
take into account requested changes from spring-integration
fixed code style issues
take into account requested changes from spring-integration
Fixesspring-projects/spring-integration-aws#106
It is really OK to have a `null` returned for the
`S3MessageHandler.objectAclExpression` with the meaning do not set any
ACL on the S3 object.
* Allow `null` for `objectAclExpression` evaluation result
* Optimize asserts skipping string concatenation until the `Supplier`
invocation
* Introduce a `S3MessageHandler.MessageS3ProgressListener` to have a
request message context in the `progressChanged` event case
Fixes https://github.com/spring-projects/spring-integration-aws/issues/97
This fixes an issue where `additionalTimeToWait` has a negative value
that is subtracted later by the Amazon DynamoDB client to
`leaseDuration`, making it impossible to acquire the lock if expired.
Also allow to use the existing `refreshPeriod` properties in this case.
* Use default refresh period in tryLock()
The 'refreshPeriod' value might not satisfy both tryLock() and lock()
requirements and should probably be used exclusively in the second case.
We can make the value used for tryLock() configurable as well in another
pull request.
* Rollback previous commit and add author name
There is a race condition when real consumers are started later in
different thread.
* Add `eventually()` matcher to check the `Map<KinesisShardOffset, ?> shardConsumers`
for particular keys to appear eventually
Since we don't have choice and wait at minimum `leaseDuration` to be
able to iterate in the Amazon LockClient at least two times and that is
going to be just a contract of this client, so we need to add up a
`timeout` for the `tryLock` to that `leaseDuration`
https://stackoverflow.com/questions/51428196/spring-aws-kinesis-binder-acquiring-and-releasing-lock-issues-in-dynamo-db-while
According the logic in the `AmazonDynamoDBLockClient.acquireLock()` we
need to let the loop to iterate at least twice to really get access to
the existing lock and check its expiration status
* Fix the `DynamoDbLockRegistry.tryLock()` to set a proper
`.withRefreshPeriod(0L)` and `.withRefreshPeriod(0L)` to let it pull
the DB until we reach a `leaseDuration` limit on the lock or its
expiration status
