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
This commit is contained in:
13
build.gradle
13
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
|
||||
|
||||
4
gradle/wrapper/gradle-wrapper.properties
vendored
4
gradle/wrapper/gradle-wrapper.properties
vendored
@@ -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
|
||||
|
||||
3
gradlew
vendored
3
gradlew
vendored
@@ -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
|
||||
|
||||
@@ -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<S
|
||||
* {@link Session} instances.
|
||||
* @param sessionFactory The session factory.
|
||||
*/
|
||||
@SuppressWarnings("this-escape")
|
||||
public S3InboundFileSynchronizer(SessionFactory<S3Object> sessionFactory) {
|
||||
super(sessionFactory);
|
||||
doSetRemoteDirectoryExpression(new LiteralExpression(null));
|
||||
|
||||
@@ -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<S3Object> template, Comparator<S3Object> comparator) {
|
||||
super(template, comparator);
|
||||
doSetFilter(new S3PersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "s3StreamingMessageSource"));
|
||||
|
||||
@@ -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<Record> records = result.records();
|
||||
|
||||
@@ -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);
|
||||
|
||||
Reference in New Issue
Block a user