* Always call provided `AsyncHandler` from the internal instance
in the `KinesisMessageHandler`
* Make the `KinesisMessageHandler.obtainAsyncHandler()` as generic method
* Rename `sendFailureChannel` property just to the `failureChannel`
since the real operation is `put` not send
* Add `AwsHeaders.SERVICE_RESULT` to represent the service result, e.g.
in case of `PutRecordsRequest` in the `KinesisMessageHandler` to
send on success the whole `PutRecordsResult`
* Fix README to reflect the current reality of the code
* Fix `KinesisMessageHandlerTests` for provided `AsyncHandler` verification
This is groundwork to allow usage of a failure channel within the Kinesis binder per.
This implementation is intended to be backward-compatible with respect to the current handling
of `AsyncHandler`. Client code can still provide an `AsyncHandler`,
but doing so precludes the usage of channels for successful or unsuccessful sends.
Renaming to AwsRequestFailureException
generic getasynchandler method
always delegate or build handler
Added readme docs and using channel for tests
Since the listener may take some time for records processing,
there is a possibility that checkpoint will be stored too late
after the process and thus we are able to get the same records
in different channel adapter for the same shard, even if they are
in the same consumer group and use shared `MetadataStore`
This solution is some compromise for the current state of things and
has to be reconsidered in the future in favor of proper rebalance and
shard leader election solution
As a workaround for the duplicate records an additional
`@IdempotentReceiver` approach can be used
* Upgrade to Gradle 4.2.1, Checkstyle 8.3, AssertJ 3.8.0
* Fix race condition in the `KinesisMessageDrivenChannelAdapterTests`
Fixesspring-projects/spring-integration-aws#72
* Document Kinesis Channel Adapters
* Fix some inconsistency in the `KinesisMessageHandler`
* Add integration test against `KinesisLocalRunning` `@Rule`
* Document testing against Kinesalite
Fixesspring-projects/spring-integration-aws#64
* Add `DynamoDbRunning` for testing against locally ran DynamoDB
* Upgrade to Gradle 4.0.1, SI-4.3.11
* Switch on some Checkstyle rules for tests
Resolvesspring-projects/spring-integration-aws#75
Since we can build S3 entity key any deep path,
e.g. `my_bucket/foo/bar/baz/file.name` and S3 Object `key` representation
is exactly the whole path without bucket name, we should treat only bucket
as a remote dir; the key should be as a file name
* Change `S3InboundFileSynchronizer` logic to extract bucket name before
performing `copyFileToLocalDirectory()`
* Change `S3StreamingMessageSource` to override the `S3FileInfo.remoteDirectory`
to the only bucket name after `poll()`.
* Use for both cases a new `S3Session.normalizeBucketName()` method
* Upgrade to Gradle 4.0 and some other `build.gradle` polishing
* Fix whitespace in documentation headers
Many README headers are missing whitespace and so are not interpreted correctly by Github-flavored markdown.
* Fix code block completion
* Update whitespace for readability
Fixesspring-projects/spring-integration-aws#70
* Following the AWS Kinesis recommendation about some sleep interval in
between request to the Kinesis service, add `idleBetweenPolls` option
to the `KinesisMessageDrivenChannelAdapter` to sleep in the
`ConsumerDispatcher` main loop after each execution cycle
* Make default `recordLimit` as `10000` - to fetch as much as possible
data per one GetRecords request
* Change `Lock` and `Condition` in the `ConsumerInvoker` to the
`Semaphore`.
We really may release permit from the `ShardConsumer` to perform
the `ConsumerInvoker` cycle before stepping in that cycle
Fixesspring-projects/spring-integration-aws#70
* The `ConsumerDispatcher` is infinite, non-blocking loop:
add short `Thread.sleep(1)` to the loop to yield the CPU
* The `ConsumerInvoker` is infinite loop, too, but can be blocked:
add `Condition` to wait to the particular notification from the
`ShardConsumer` when it changes its state
Fixesspring-projects/spring-integration-aws#71
It isn't correct to call `converter.convert()` in the
`KinesisMessageHandler` if `payload` is already `ByteBuffer` or `byte[]`
On the other hand an application might be interested in the `byte[]`
payload on the consumer side.
The `KinesisMessageDrivenChannelAdapter` must be able to produce
`byte[]` `payload` without any conversion.
* Add logic into `KinesisMessageHandler` to check the `payload` type
before calling `converter.convert()`
* Allow to configure `converter` to `null` for the
`KinesisMessageDrivenChannelAdapter`.
The body of the consumed record is presented in the `payload` as is -
`byte[]`
* Add JavaDoc to the `KinesisMessageHandler.setConverter()`
Fixes: spring-projects/spring-integration-aws#67
* Use `org.springframework.util.StringUtils#trimLeadingCharacter()`
to remove any useless leading `/` in the bucket name
Fixesspring-projects/spring-integration-aws#63
The `CLOSED` shards have `endingSequenceNumber` value and they can't be
considered for consuming independently of the value in the `checkpoint`
* Introduce local `shardsToConsume` variable in the
`KinesisMessageDrivenChannelAdapter#populateShardsForStream()` to
store only `OPEN` shards for future consumption
* Fix the logic to determine the last `shardId` for a subsequent
`describeStream()` request based on the shards result exactly from
the `describeStreamResult.getStreamDescription().getShards()`, not
already filtered `shardsToConsume`
Fixes: spring-projects/spring-integration-aws#56
The upgrade to SC-AWS-1.2 RC1 resolves US_East_2 Region concern
since AWS SDK is polled in version `1.1.98`
Also fix deprecation warnings in S3 support components
When shard iterator is expired (`ExpiredIteratorException`),
we have to restart from the stored sequence number
When shard iterator is throttled (`ProvisionedThroughputExceededException`),
we have to "sleep" for backoff period
* Catch `ExpiredIteratorException` on the `.amazonKinesis.getRecords()` request
and move `ShardConsumer` to the `ConsumerState.EXPIRED`.
This state is treated as `NEW`, therefore a fresh `amazonKinesis.getShardIterator()`
is performed
* Catch `ProvisionedThroughputExceededException` on the `.amazonKinesis.getRecords()`
request and move `ShardConsumer` to the `ConsumerState.SLEEP`
for the `KinesisMessageDrivenChannelAdapter.this.consumerBackoff` period
Fixes: spring-projects/spring-integration-aws#61
With the `shutdownNow()` the currently ran task are marked with the
`Thread.interrupt()`.
The AWS SDK catch this situation and throws `AbortedException`
* To let currently ran task finish properly switch from the
`shutdownNow()` to the `shutdown()` for internal executors
The external Spring-based executors (`ExecutorConfigurationSupport`)
must be supplied with the `waitForTasksToCompleteOnShutdown` option
Fixes: spring-projects/spring-integration-aws#60
The attempts to extract `this.consumerInvokers.get(0)` when the `concurrency == 0` lead to the NPE
* Wrap `this.consumerInvokers.get(0)` block to the `if(concurrency != 0)`
`ConsumerInvoker` uses iterator over `consumers` property, which can be
concurrently modified in the `populateConsumer()`` method, thus triggering
`ConcurrentModificationException`:
```
Exception in thread "kinesisChannel-kinesis-consumer-1" java.util.ConcurrentModificationException
at java.util.ArrayList$Itr.checkForComodification(ArrayList.java:901)
at java.util.ArrayList$Itr.next(ArrayList.java:851)
at org.springframework.integration.aws.inbound.kinesis.KinesisMessageDrivenChannelAdapter$ConsumerInvoker.run(KinesisMessageDrivenChannelAdapter.java:866)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
```
* Use a `ConcurrentLinkedQueue` instead of `ArrayList` to fix a concurrent Modification problem.
Fixes GH-22 (https://github.com/spring-projects/spring-integration-aws/issues/22)
Rework logic to the `dispatching` and tasks
* Add `LimitExceededException` and configurable retry logic for the `describeStream`
* Skip `CLOSED` shards which has been read before according stored `checkpoint`. Will be useful for `after resharding` algorithm
* Add `adapt to resharding` logic
* Add `KinesisMessageDrivenChannelAdapterTests` based on mocks
* Add Thread Affinity for `ShardConsumer`s via `concurrency` option
* Introduce `concurrency` option and `ConsumerInvoker`.
`ShardConsumer`s are now distributed between `ConsumerInvoker`s if `concurrency > 0`
`ConsumerInvoker`s are scheduled to the `ConsumerExecutor` as a `isLongLived` tasks
The concurrency is adjusted if there are no more `ShardConsumer`s to process (`STOP` state because of closed shard).
At the same time newly populated `ShardConsumer`s (e.g. after resharding) is distributed evenly between existing `ConsumerExecutor`s if `maxConcurrency` is exceeded.
Otherwise fresh `ConsumerExecutor` is started for new `ShardConsumer`
Fixes GH-54 (https://github.com/spring-projects/spring-integration-aws/issues/54)
For consistency with other `AbstractRemoteFileStreamingMessageSource` implementations (e.g. `FtpStreamingMessageSource`) rename `S3InboundStreamingMessageSource` to the `S3StreamingMessageSource`
Since there was no any Milestones for the `1.1` version yet, it is safe to do this renaming breaking change
Fixes GH-22 (https://github.com/spring-projects/spring-integration-aws/issues/22)
* Add `KinesisMessageHandler` based on the `AmazonKinesisAsync.putRecordAsync()` operation.
The logic is fully similar to the `KafkaProducerMessageHandler` since both protocols pursue the same behavior
* Add some Kinesis specific `AwsHeaders`
* Upgrade to Gradle 2.14.1
To avoid some local work with files, add `S3PersistentAcceptOnceFileListFilter` to the `S3InboundFileSynchronizer` as a default one
* Upgrade to Gradle 3.1
* Upgrade to SI-4.3.2
* Some `build.gradle` polishing
* Downgrade to Checkstyle-6.16.1 because of `RequireThis` bug
