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

Upgrade some deps; fix deprecations

Browse Source
This commit is contained in:
Artem Bilan
2023-08-12 10:27:48 -04:00
parent 50bf00e6a1
commit c157b9b094
6 changed files with 64 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
build.gradle
View File

@@ -19,8 +19,8 @@ plugins {
id 'org.sonarqube' version '4.3.0.3225'
id 'io.spring.nohttp' version '0.0.11' apply false
id 'org.ajoberstar.grgit' version '4.1.1'
id 'io.spring.dependency-management' version '1.1.2'
id 'com.jfrog.artifactory' version '4.33.1' apply false
id 'io.spring.dependency-management' version '1.1.3'
id 'com.jfrog.artifactory' version '5.0.3' apply false
id 'org.jetbrains.dokka' version '1.8.20'
id 'org.antora' version '1.0.0'
id 'io.spring.antora.generate-antora-yml' version '0.0.1'
@@ -45,8 +45,8 @@ ext {
linkScmConnection = 'scm:git:git://github.com/spring-projects/spring-integration.git'
linkScmDevConnection = 'scm:git:ssh://git@github.com:spring-projects/spring-integration.git'
modifiedFiles = providers.provider {
files(grgit.status().unstaged.modified).filter { f -> f.name.endsWith('.java') || f.name.endsWith('.kt') } }
modifiedFiles =
files(grgit.status().unstaged.modified).filter { f -> f.name.endsWith('.java') || f.name.endsWith('.kt') }
apacheSshdVersion = '2.10.0'
artemisVersion = '2.29.0'
@@ -55,7 +55,7 @@ ext {
assertkVersion = '0.26.1'
avroVersion = '1.11.2'
awaitilityVersion = '4.2.0'
camelVersion = '4.0.0-RC1'
camelVersion = '4.0.0-RC2'
commonsDbcp2Version = '2.9.0'
commonsIoVersion = '2.13.0'
commonsNetVersion = '3.9.0'
@@ -81,35 +81,35 @@ ext {
jrubyVersion = '9.4.3.0'
jsonpathVersion = '2.8.0'
junit4Version = '4.13.2'
junitJupiterVersion = '5.10.0-RC1'
junitJupiterVersion = '5.10.0'
jythonVersion = '2.7.3'
kotlinCoroutinesVersion = '1.7.2'
kryoVersion = '5.5.0'
lettuceVersion = '6.2.4.RELEASE'
log4jVersion = '2.20.0'
mailVersion = '2.0.2'
micrometerTracingVersion = '1.2.0-M1'
micrometerVersion = '1.12.0-M1'
micrometerTracingVersion = '1.2.0-SNAPSHOT'
micrometerVersion = '1.12.0-SNAPSHOT'
mockitoVersion = '5.4.0'
mongoDriverVersion = '4.10.2'
mysqlVersion = '8.0.33'
pahoMqttClientVersion = '1.2.5'
postgresVersion = '42.6.0'
protobufVersion = '3.24.0-RC1'
protobufVersion = '3.24.0-RC2'
r2dbch2Version = '1.0.0.RELEASE'
reactorVersion = '2023.0.0-M1'
reactorVersion = '2023.0.0-SNAPSHOT'
resilience4jVersion = '2.1.0'
romeToolsVersion = '2.1.0'
rsocketVersion = '1.1.4'
servletApiVersion = '6.0.0'
smackVersion = '4.4.6'
springAmqpVersion = '3.0.6'
springDataVersion = '2023.1.0-M1'
springDataVersion = '2023.1.0-SNAPSHOT'
springGraphqlVersion = '1.2.2'
springKafkaVersion = '3.0.10-SNAPSHOT'
springRetryVersion = '2.0.2'
springSecurityVersion = '6.2.0-M1'
springVersion = '6.1.0-M2'
springSecurityVersion = '6.2.0-SNAPSHOT'
springVersion = '6.1.0-SNAPSHOT'
springWsVersion = '4.0.4'
testcontainersVersion = '1.18.3'
tomcatVersion = '10.1.11'
@@ -290,7 +290,7 @@ configure(javaProjects) { subproject ->
tasks.register('updateCopyrights') {
onlyIf { !isCI }
inputs.files(modifiedFiles.map(files -> files.filter { f -> f.path.contains(subproject.name) }))
inputs.files(modifiedFiles.filter { f -> f.path.contains(subproject.name) })
outputs.dir('build/classes')
doLast {

17
gradle/docs.gradle
View File

@@ -6,17 +6,22 @@ ext {
antora {
version = '3.2.0-alpha.2'
playbook = file('src/reference/antora/antora-playbook.yml')
options = ['to-dir' : project.layout.buildDirectory.dir('site').get().toString(), clean: true, fetch: !project.gradle.startParameter.offline, stacktrace: true]
options = [
'to-dir' : project.layout.buildDirectory.dir('site').get().toString(),
clean: true,
fetch: !project.gradle.startParameter.offline,
stacktrace: true
]
dependencies = [
'@antora/atlas-extension': '1.0.0-alpha.1',
'@antora/atlas-extension': '1.0.0-alpha.2',
'@antora/collector-extension': '1.0.0-alpha.3',
'@asciidoctor/tabs': '1.0.0-beta.3',
'@springio/antora-extensions': '1.4.2',
'@springio/asciidoctor-extensions': '1.0.0-alpha.8',
'@asciidoctor/tabs': '1.0.0-beta.6',
'@springio/antora-extensions': '1.5.0',
'@springio/asciidoctor-extensions': '1.0.0-alpha.9',
]
}
tasks.named("generateAntoraYml") {
tasks.named('generateAntoraYml') {
asciidocAttributes = project.provider( {
return ['project-version' : project.version ]
} )

4
settings.gradle
View File

@@ -1,6 +1,6 @@
plugins {
id 'com.gradle.enterprise' version '3.13.4'
id 'io.spring.ge.conventions' version '0.0.13'
id 'com.gradle.enterprise' version '3.14.1'
id 'io.spring.ge.conventions' version '0.0.14'
}
rootProject.name = 'spring-integration'

30
spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2022 the original author or authors.
* Copyright 2002-2023 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.
@@ -27,9 +27,9 @@ import java.util.concurrent.locks.ReentrantLock;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.integration.channel.DirectChannel;
@@ -50,6 +50,7 @@ import org.springframework.messaging.support.GenericMessage;
import org.springframework.util.StopWatch;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import static org.mockito.Mockito.mock;
/**
@@ -69,7 +70,7 @@ public class AggregatorTests {
private final List<MessageGroupExpiredEvent> expiryEvents = new ArrayList<>();
@Before
@BeforeEach
public void configureAggregator() {
this.aggregator = new AggregatingMessageHandler(new MultiplyingProcessor(), store);
this.aggregator.setBeanFactory(mock(BeanFactory.class));
@@ -80,7 +81,7 @@ public class AggregatorTests {
}
@Test
@Ignore
@Disabled
public void testAggPerf() throws InterruptedException, ExecutionException, TimeoutException {
AggregatingMessageHandler handler = new AggregatingMessageHandler(new DefaultAggregatingMessageGroupProcessor());
handler.setCorrelationStrategy(message -> "foo");
@@ -170,7 +171,7 @@ public class AggregatorTests {
}
stopwatch.stop();
logger.warn("Sent " + 120000 + " in " + stopwatch.getTotalTimeSeconds() +
" (10k in " + stopwatch.getLastTaskTimeMillis() + "ms)");
" (10k in " + stopwatch.lastTaskInfo().getTimeMillis() + "ms)");
Collection<?> result = resultFuture.get(10, TimeUnit.SECONDS);
assertThat(result).isNotNull();
@@ -186,7 +187,7 @@ public class AggregatorTests {
private final ReentrantLock lock = new ReentrantLock();
private final Collection<Message<?>> messages = new ArrayList<Message<?>>(60000);
private final Collection<Message<?>> messages = new ArrayList<>(60000);
private final MessageChannel outputChannel;
@@ -233,14 +234,14 @@ public class AggregatorTests {
if (i % 10000 == 0) {
stopwatch.stop();
logger.warn("Sent " + i + " in " + stopwatch.getTotalTimeSeconds() +
" (10k in " + stopwatch.getLastTaskTimeMillis() + "ms)");
" (10k in " + stopwatch.lastTaskInfo().getTimeMillis() + "ms)");
stopwatch.start();
}
handler.handleMessage(message);
}
stopwatch.stop();
logger.warn("Sent " + 120000 + " in " + stopwatch.getTotalTimeSeconds() +
" (10k in " + stopwatch.getLastTaskTimeMillis() + "ms)");
" (10k in " + stopwatch.lastTaskInfo().getTimeMillis() + "ms)");
Collection<?> result = resultFuture.get(10, TimeUnit.SECONDS);
assertThat(result).isNotNull();
@@ -465,7 +466,7 @@ public class AggregatorTests {
}
@Test
@Ignore
@Disabled
// dropped backwards compatibility for setting capacity limit (it's always Integer.MAX_VALUE)
public void testTrackedCorrelationIdsCapacityAtLimit() {
QueueChannel replyChannel = new QueueChannel();
@@ -484,7 +485,7 @@ public class AggregatorTests {
}
@Test
@Ignore
@Disabled
// dropped backwards compatibility for setting capacity limit (it's always Integer.MAX_VALUE)
public void testTrackedCorrelationIdsCapacityPassesLimit() {
QueueChannel replyChannel = new QueueChannel();
@@ -504,10 +505,11 @@ public class AggregatorTests {
assertThat(discardChannel.receive(0)).isNull();
}
@Test(expected = MessageHandlingException.class)
@Test
public void testExceptionThrownIfNoCorrelationId() throws InterruptedException {
Message<?> message = createMessage(3, null, 2, 1, new QueueChannel(), null);
this.aggregator.handleMessage(message);
assertThatExceptionOfType(MessageHandlingException.class)
.isThrownBy(() -> this.aggregator.handleMessage(message));
}
@Test

7
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionFactoryShutDownTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2022 the original author or authors.
* Copyright 2002-2023 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.
@@ -20,7 +20,7 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.junit.jupiter.api.Test;
import org.springframework.messaging.MessagingException;
import org.springframework.util.StopWatch;
@@ -70,7 +70,8 @@ public class ConnectionFactoryShutDownTests {
watch.start();
factory.stop();
watch.stop();
assertThat(watch.getLastTaskTimeMillis() < 10000).as("Expected < 10000, was: " + watch.getLastTaskTimeMillis())
assertThat(watch.lastTaskInfo().getTimeMillis() < 10000)
.as("Expected < 10000, was: " + watch.lastTaskInfo().getTimeMillis())
.isTrue();
assertThat(latch1.await(10, TimeUnit.SECONDS)).isTrue();
}

32
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java
View File

@@ -61,7 +61,6 @@ import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ConsumerAwareRebalanceListener;
import org.springframework.kafka.listener.ConsumerProperties;
import org.springframework.kafka.listener.ErrorHandlingUtils;
import org.springframework.kafka.listener.ListenerUtils;
import org.springframework.kafka.listener.LoggingCommitCallback;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.DefaultKafkaHeaderMapper;
@@ -281,7 +280,6 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object>
this.consumerProperties.setClientId(getComponentName());
}
Map<String, Object> props = this.consumerFactory.getConfigurationProperties();
Properties kafkaConsumerProperties = this.consumerProperties.getKafkaConsumerProperties();
this.checkNullKeyForExceptions =
this.consumerProperties.isCheckDeserExWhenKeyNull() ||
@@ -390,16 +388,7 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object>
this.logger.warn(() -> ConsumerConfig.MAX_POLL_RECORDS_CONFIG
+ "' has been forced from " + (maxPoll == null ? "unspecified" : maxPoll)
+ " to 1, to avoid having to seek after each record");
Map<String, Object> configs = new HashMap<>(suppliedConsumerFactory.getConfigurationProperties());
configs.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1);
DefaultKafkaConsumerFactory<K, V> fixedConsumerFactory = new DefaultKafkaConsumerFactory<>(configs);
if (suppliedConsumerFactory.getKeyDeserializer() != null) {
fixedConsumerFactory.setKeyDeserializer(suppliedConsumerFactory.getKeyDeserializer());
}
if (suppliedConsumerFactory.getValueDeserializer() != null) {
fixedConsumerFactory.setValueDeserializer(suppliedConsumerFactory.getValueDeserializer());
}
return fixedConsumerFactory;
return fixConsumerFactory(suppliedConsumerFactory);
}
else {
return suppliedConsumerFactory;
@@ -494,7 +483,7 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object>
createConsumer();
this.running = true;
}
if (this.pausing && !this.paused && this.assignedPartitions.size() > 0) {
if (this.pausing && !this.paused && !this.assignedPartitions.isEmpty()) {
this.consumer.pause(this.assignedPartitions);
this.paused = true;
}
@@ -675,8 +664,8 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object>
}
}
private void checkDeserializationException(ConsumerRecord<K, V> cRecord, String headerName) {
DeserializationException exception = ListenerUtils.getExceptionFromHeader(cRecord, headerName, this.logger);
private void checkDeserializationException(ConsumerRecord<K, V> record, String headerName) {
DeserializationException exception = SerializationUtils.getExceptionFromHeader(record, headerName, this.logger);
if (exception != null) {
throw exception;
}
@@ -707,6 +696,19 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object>
}
}
private static <K, V> ConsumerFactory<K, V> fixConsumerFactory(ConsumerFactory<K, V> suppliedConsumerFactory) {
Map<String, Object> configs = new HashMap<>(suppliedConsumerFactory.getConfigurationProperties());
configs.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1);
DefaultKafkaConsumerFactory<K, V> fixedConsumerFactory = new DefaultKafkaConsumerFactory<>(configs);
if (suppliedConsumerFactory.getKeyDeserializer() != null) {
fixedConsumerFactory.setKeyDeserializer(suppliedConsumerFactory.getKeyDeserializer());
}
if (suppliedConsumerFactory.getValueDeserializer() != null) {
fixedConsumerFactory.setValueDeserializer(suppliedConsumerFactory.getValueDeserializer());
}
return fixedConsumerFactory;
}
private class IntegrationConsumerRebalanceListener implements ConsumerRebalanceListener {
private final ConsumerRebalanceListener providedRebalanceListener;