Spring/spring-integration-aws
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Upgrade to AWS SDK 2.20.35

Browse Source 
* Simplify some code in the `KclMessageDrivenChannelAdapter`
* Tried to upgrade to LocalStack `2.0` - failed on S3 `createBucket`
with `Length Required` error.
Sounds like there was no LocalStack and AWS SDK integration testing
This commit is contained in:
abilan
2023-03-31 13:34:13 -04:00
parent ba6116428f
commit 75cd725401
2 changed files with 15 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
build.gradle
View File

@@ -32,7 +32,7 @@ repositories {
ext {
assertjVersion = '3.24.2'
awaitilityVersion = '4.2.0'
awsSdkVersion = '2.20.32'
awsSdkVersion = '2.20.35'
jacksonVersion = '2.14.2'
junitVersion = '5.9.2'
log4jVersion = '2.19.0'

21
src/main/java/org/springframework/integration/aws/inbound/kinesis/KclMessageDrivenChannelAdapter.java
View File

@@ -52,6 +52,7 @@ import software.amazon.kinesis.processor.RecordProcessorCheckpointer;
import software.amazon.kinesis.processor.ShardRecordProcessor;
import software.amazon.kinesis.processor.ShardRecordProcessorFactory;
import software.amazon.kinesis.processor.SingleStreamTracker;
import software.amazon.kinesis.processor.StreamTracker;
import software.amazon.kinesis.retrieval.KinesisClientRecord;
import software.amazon.kinesis.retrieval.RetrievalConfig;
@@ -239,17 +240,23 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport {
@Override
protected void onInit() {
super.onInit();
this.config =
new ConfigsBuilder(buildStreamTracker(),
this.consumerGroup,
this.kinesisClient,
this.dynamoDBClient,
this.cloudWatchClient,
this.workerId,
this.recordProcessorFactory);
}
private StreamTracker buildStreamTracker() {
if (this.streams.length == 1) {
this.config = new ConfigsBuilder(
new SingleStreamTracker(
StreamIdentifier.singleStreamInstance(this.streams[0]), this.streamInitialSequence),
this.consumerGroup, this.kinesisClient, this.dynamoDBClient, this.cloudWatchClient, this.workerId,
this.recordProcessorFactory);
return new SingleStreamTracker(StreamIdentifier.singleStreamInstance(this.streams[0]),
this.streamInitialSequence);
}
else {
this.config = new ConfigsBuilder(new StreamsTracker(), this.consumerGroup, this.kinesisClient,
this.dynamoDBClient, this.cloudWatchClient, this.workerId, this.recordProcessorFactory);
return new StreamsTracker();
}
}