From 259ff7ba2d84ebdf616a4868d86f4be267747504 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 10 Feb 2025 15:11:08 -0500 Subject: [PATCH] GH-252: Fix NPE in the `KinesisMessageDrivenChannelAdapter.` Fixes: https://github.com/spring-projects/spring-integration-aws/issues/252 * Adapt to Java `23` * Upgrade to Gradle `8.12.1` * Include `org.mockito` & `net.bytebuddy` dependencies explicitly since they are overridden by transitive dependencies to not compatible versions with Java `23` * Add `-parameters` compiler option to include method param names into bytecode for better discovery by reflection --- build.gradle | 13 ++++++------- gradle/wrapper/gradle-wrapper.properties | 4 ++-- gradlew | 3 +-- .../aws/inbound/S3InboundFileSynchronizer.java | 3 ++- .../aws/inbound/S3StreamingMessageSource.java | 3 ++- .../kinesis/KinesisMessageDrivenChannelAdapter.java | 12 +++++++----- .../aws/support/KplBackpressureException.java | 2 +- 7 files changed, 21 insertions(+), 19 deletions(-) diff --git a/build.gradle b/build.gradle index c3a76de..b53118a 100644 --- a/build.gradle +++ b/build.gradle @@ -26,7 +26,6 @@ repositories { } ext { - assertjVersion = '3.26.3' awaitilityVersion = '4.2.2' awsSdkVersion = '2.20.162' jacksonVersion = '2.15.4' @@ -126,10 +125,11 @@ dependencies { optionalApi "jakarta.servlet:jakarta.servlet-api:$servletApiVersion" - testImplementation('org.springframework.integration:spring-integration-test') { - exclude group: 'junit' - } - testImplementation "org.assertj:assertj-core:$assertjVersion" + testImplementation 'org.mockito:mockito-core:5.15.2' + testImplementation 'net.bytebuddy:byte-buddy:1.15.11' + testImplementation 'net.bytebuddy:byte-buddy-agent:1.15.11' + + testImplementation 'org.springframework.integration:spring-integration-test' testImplementation("org.awaitility:awaitility:$awaitilityVersion") { exclude group: 'org.hamcrest' } @@ -172,7 +172,7 @@ javadoc { // enable all compiler warnings; individual projects may customize further ext.xLintArg = '-Xlint:all,-options' -[compileJava, compileTestJava]*.options*.compilerArgs = [xLintArg] +[compileJava, compileTestJava]*.options*.compilerArgs = [xLintArg, '-parameters'] test { maxHeapSize = '1024m' @@ -186,7 +186,6 @@ check.dependsOn javadoc task updateCopyrights { onlyIf { !isCI } inputs.files(modifiedFiles) - outputs.dir('build/classes') doLast { def now = Calendar.instance.get(Calendar.YEAR) as String diff --git a/gradle/wrapper/gradle-wrapper.properties b/gradle/wrapper/gradle-wrapper.properties index 8e876e1..d710477 100644 --- a/gradle/wrapper/gradle-wrapper.properties +++ b/gradle/wrapper/gradle-wrapper.properties @@ -1,7 +1,7 @@ distributionBase=GRADLE_USER_HOME distributionPath=wrapper/dists -distributionSha256Sum=1541fa36599e12857140465f3c91a97409b4512501c26f9631fb113e392c5bd1 -distributionUrl=https\://services.gradle.org/distributions/gradle-8.10.1-bin.zip +distributionSha256Sum=8d97a97984f6cbd2b85fe4c60a743440a347544bf18818048e611f5288d46c94 +distributionUrl=https\://services.gradle.org/distributions/gradle-8.12.1-bin.zip networkTimeout=10000 validateDistributionUrl=true zipStoreBase=GRADLE_USER_HOME diff --git a/gradlew b/gradlew index f5feea6..f3b75f3 100755 --- a/gradlew +++ b/gradlew @@ -86,8 +86,7 @@ done # shellcheck disable=SC2034 APP_BASE_NAME=${0##*/} # Discard cd standard output in case $CDPATH is set (https://github.com/gradle/gradle/issues/25036) -APP_HOME=$( cd -P "${APP_HOME:-./}" > /dev/null && printf '%s -' "$PWD" ) || exit +APP_HOME=$( cd -P "${APP_HOME:-./}" > /dev/null && printf '%s\n' "$PWD" ) || exit # Use the maximum available, or set MAX_FD != -1 to use that value. MAX_FD=maximum diff --git a/src/main/java/org/springframework/integration/aws/inbound/S3InboundFileSynchronizer.java b/src/main/java/org/springframework/integration/aws/inbound/S3InboundFileSynchronizer.java index 2c75627..7030e69 100644 --- a/src/main/java/org/springframework/integration/aws/inbound/S3InboundFileSynchronizer.java +++ b/src/main/java/org/springframework/integration/aws/inbound/S3InboundFileSynchronizer.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2023 the original author or authors. + * Copyright 2016-2025 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -53,6 +53,7 @@ public class S3InboundFileSynchronizer extends AbstractInboundFileSynchronizer sessionFactory) { super(sessionFactory); doSetRemoteDirectoryExpression(new LiteralExpression(null)); diff --git a/src/main/java/org/springframework/integration/aws/inbound/S3StreamingMessageSource.java b/src/main/java/org/springframework/integration/aws/inbound/S3StreamingMessageSource.java index cd19d70..c79b8e5 100644 --- a/src/main/java/org/springframework/integration/aws/inbound/S3StreamingMessageSource.java +++ b/src/main/java/org/springframework/integration/aws/inbound/S3StreamingMessageSource.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2023 the original author or authors. + * Copyright 2016-2025 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -45,6 +45,7 @@ public class S3StreamingMessageSource extends AbstractRemoteFileStreamingMessage super(template, null); } + @SuppressWarnings("this-escape") public S3StreamingMessageSource(RemoteFileTemplate template, Comparator comparator) { super(template, comparator); doSetFilter(new S3PersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "s3StreamingMessageSource")); 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 473637d..2159087 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 @@ -1,5 +1,5 @@ /* - * Copyright 2017-2024 the original author or authors. + * Copyright 2017-2025 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -1150,7 +1150,7 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport }; } - private void rewindIteratorOnError(Exception ex, GetRecordsResponse result) { + private void rewindIteratorOnError(Exception ex, @Nullable GetRecordsResponse result) { String lastCheckpoint = this.checkpointer.getLastCheckpointValue(); String highestSequence = this.checkpointer.getHighestSequence(); @@ -1159,7 +1159,7 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport logger.info(ex, "getRecords request has thrown exception. " + "No checkpoints - re-request with the current shard iterator."); } - else if (highestSequence.equals(lastCheckpoint)) { + else if (highestSequence.equals(lastCheckpoint) && result != null) { logger.info(ex, "Record processor has thrown exception. " + "Ignore since the highest sequence in batch was check-pointed."); this.shardIterator = result.nextShardIterator(); @@ -1187,8 +1187,10 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport } } - private boolean reRequestCurrentShardIterator(@Nullable String lastCheckpoint, GetRecordsResponse result) { - if (lastCheckpoint == null) { + private boolean reRequestCurrentShardIterator(@Nullable String lastCheckpoint, + @Nullable GetRecordsResponse result) { + + if (lastCheckpoint == null || result == null) { return true; } List records = result.records(); diff --git a/src/main/java/org/springframework/integration/aws/support/KplBackpressureException.java b/src/main/java/org/springframework/integration/aws/support/KplBackpressureException.java index 28d314e..a5ee604 100644 --- a/src/main/java/org/springframework/integration/aws/support/KplBackpressureException.java +++ b/src/main/java/org/springframework/integration/aws/support/KplBackpressureException.java @@ -34,7 +34,7 @@ public class KplBackpressureException extends RuntimeException { @Serial private static final long serialVersionUID = 1L; - private final UserRecord userRecord; + private final transient UserRecord userRecord; public KplBackpressureException(String message, UserRecord userRecord) { super(message);