From 10d4652c21976d3cf0d4296fd143342fd34c4486 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 29 Oct 2020 11:45:34 -0400 Subject: [PATCH] Upgrade dependencies; prepare for release Fixes https://github.com/spring-projects/spring-integration-aws/issues/162 * Fix deprecations --- build.gradle | 14 +++++++------- gradle/wrapper/gradle-wrapper.properties | 2 +- .../kinesis/KclMessageDrivenChannelAdapter.java | 9 +++++++-- .../KinesisMessageDrivenChannelAdapter.java | 14 +++++++++++--- 4 files changed, 26 insertions(+), 13 deletions(-) diff --git a/build.gradle b/build.gradle index 00838ed..8deef77 100644 --- a/build.gradle +++ b/build.gradle @@ -5,9 +5,9 @@ plugins { id 'jacoco' id 'org.sonarqube' version '3.0' id 'checkstyle' - id 'org.ajoberstar.grgit' version '4.0.2' + id 'org.ajoberstar.grgit' version '4.1.0' id "io.spring.dependency-management" version '1.0.10.RELEASE' - id 'com.jfrog.artifactory' version '4.17.0' + id 'com.jfrog.artifactory' version '4.17.2' } description = 'Spring Integration AWS Support' @@ -23,17 +23,17 @@ repositories { } ext { - assertjVersion = '3.16.1' + assertjVersion = '3.18.0' awaitilityVersion = '4.0.3' dynamodbLockClientVersion = '1.1.0' - jacksonVersion = '2.11.2' + jacksonVersion = '2.11.3' junitVersion = '5.6.2' servletApiVersion = '4.0.1' localstackVersion = '0.1.22' log4jVersion = '2.13.3' springCloudAwsVersion = '2.2.4.RELEASE' - springIntegrationVersion = '5.2.8.RELEASE' - kinesisClientVersion = '1.13.3' + springIntegrationVersion = '5.3.3.RELEASE' + kinesisClientVersion = '1.14.0' kinesisProducerVersion = '0.14.1' idPrefix = 'aws' @@ -95,7 +95,7 @@ jacoco { checkstyle { configDirectory.set(rootProject.file("src/checkstyle")) - toolVersion = '8.35' + toolVersion = '8.36.2' } dependencies { diff --git a/gradle/wrapper/gradle-wrapper.properties b/gradle/wrapper/gradle-wrapper.properties index 6c9a224..be52383 100644 --- a/gradle/wrapper/gradle-wrapper.properties +++ b/gradle/wrapper/gradle-wrapper.properties @@ -1,5 +1,5 @@ distributionBase=GRADLE_USER_HOME distributionPath=wrapper/dists -distributionUrl=https\://services.gradle.org/distributions/gradle-6.6-bin.zip +distributionUrl=https\://services.gradle.org/distributions/gradle-6.7-bin.zip zipStoreBase=GRADLE_USER_HOME zipStorePath=wrapper/dists diff --git a/src/main/java/org/springframework/integration/aws/inbound/kinesis/KclMessageDrivenChannelAdapter.java b/src/main/java/org/springframework/integration/aws/inbound/kinesis/KclMessageDrivenChannelAdapter.java index ad6a4d4..e407570 100644 --- a/src/main/java/org/springframework/integration/aws/inbound/kinesis/KclMessageDrivenChannelAdapter.java +++ b/src/main/java/org/springframework/integration/aws/inbound/kinesis/KclMessageDrivenChannelAdapter.java @@ -64,6 +64,7 @@ import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorF import com.amazonaws.services.kinesis.clientlibrary.lib.worker.InitialPositionInStream; import com.amazonaws.services.kinesis.clientlibrary.lib.worker.KinesisClientLibConfiguration; import com.amazonaws.services.kinesis.clientlibrary.lib.worker.ShutdownReason; +import com.amazonaws.services.kinesis.clientlibrary.lib.worker.SimpleRecordsFetcherFactory; import com.amazonaws.services.kinesis.clientlibrary.lib.worker.Worker; import com.amazonaws.services.kinesis.model.Record; @@ -273,6 +274,7 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport { new KinesisClientLibConfiguration(this.consumerGroup, this.stream, null, + null, this.streamInitialSequence, this.kinesisProxyCredentialsProvider, null, @@ -294,8 +296,11 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport { KinesisClientLibConfiguration.DEFAULT_VALIDATE_SEQUENCE_NUMBER_BEFORE_CHECKPOINTING, null, KinesisClientLibConfiguration.DEFAULT_SHUTDOWN_GRACE_MILLIS, - KinesisClientLibConfiguration.DEFAULT_DDB_BILLING_MODE - ); + KinesisClientLibConfiguration.DEFAULT_DDB_BILLING_MODE, + new SimpleRecordsFetcherFactory(), + 0, + 0, + 0); } this.consumerGroup = this.config.getApplicationName(); diff --git a/src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java b/src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java index 43561a0..f386713 100644 --- a/src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java +++ b/src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java @@ -577,7 +577,7 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport try { Thread.sleep(this.describeStreamBackoff); - readShardList(stream, retryCount++); + readShardList(stream, retryCount + 1); } catch (InterruptedException ex) { Thread.currentThread().interrupt(); @@ -653,10 +653,10 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport String exceptionMessage = "Got an exception when processing shards in stream [" + stream + "]"; logger.info(exceptionMessage + ".\n Retrying... ", e); if (retry > 5) { - throw new IllegalStateException("Error processing shards in stream [\" + stream + \"].", e); + throw new IllegalStateException(exceptionMessage, e); } //Retry - detectShardsToConsume(stream, retry++); + detectShardsToConsume(stream, retry + 1); sleep(this.describeStreamBackoff, new IllegalStateException(exceptionMessage), false); } @@ -694,6 +694,9 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport } } } + catch (Exception ex) { + logger.error("Error population shards for stream: " + stream, ex); + } finally { if (shardsGatherLatch != null) { shardsGatherLatch.countDown(); @@ -866,6 +869,7 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport public boolean isLongLived() { return true; } + } private final class ShardConsumer { @@ -1268,6 +1272,7 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport public String toString() { return "ShardConsumer{" + "shardOffset=" + this.shardOffset + ", state=" + this.state + '}'; } + } private final class ConsumerInvoker implements SchedulingAwareRunnable { @@ -1346,6 +1351,7 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport public boolean isLongLived() { return true; } + } private final class ShardConsumerManager implements SchedulingAwareRunnable { @@ -1445,5 +1451,7 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport public boolean isLongLived() { return true; } + } + }