Upgrade dependencies; prepare for release

Fixes #162

* Fix deprecations

Author: artembilan

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 {

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

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

@@ -64,6 +64,7 @@
 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 @@ protected void onInit() {
 					new KinesisClientLibConfiguration(this.consumerGroup,
 							this.stream,
 							null,
+							null,
 							this.streamInitialSequence,
 							this.kinesisProxyCredentialsProvider,
 							null,
@@ -294,8 +296,11 @@ protected void onInit() {
 							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();

src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java

@@ -577,7 +577,7 @@ private List<Shard> readShardList(String stream, int retryCount) {
 
 			try {
 				Thread.sleep(this.describeStreamBackoff);
-				readShardList(stream, retryCount++);
+				readShardList(stream, retryCount + 1);
 			}
 			catch (InterruptedException ex) {
 				Thread.currentThread().interrupt();
@@ -653,10 +653,10 @@ private List<Shard> detectShardsToConsume(String stream, int retry) {
 			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 @@ private void populateShardsForStream(final String stream, final CountDownLatch s
 					}
 				}
 			}
+			catch (Exception ex) {
+				logger.error("Error population shards for stream: " + stream, ex);
+			}
 			finally {
 				if (shardsGatherLatch != null) {
 					shardsGatherLatch.countDown();
@@ -866,6 +869,7 @@ public void run() {
 		public boolean isLongLived() {
 			return true;
 		}
+
 	}
 
 	private final class ShardConsumer {
@@ -1268,6 +1272,7 @@ private void checkpointIfPeriodicMode(@Nullable Record record) {
 		public String toString() {
 			return "ShardConsumer{" + "shardOffset=" + this.shardOffset + ", state=" + this.state + '}';
 		}
+
 	}
 
 	private final class ConsumerInvoker implements SchedulingAwareRunnable {
@@ -1346,6 +1351,7 @@ public void run() {
 		public boolean isLongLived() {
 			return true;
 		}
+
 	}
 
 	private final class ShardConsumerManager implements SchedulingAwareRunnable {
@@ -1445,5 +1451,7 @@ public void run() {
 		public boolean isLongLived() {
 			return true;
 		}
+
 	}
+
 }