From c157b9b09481696702a90ed143038356eafc7fb2 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Sat, 12 Aug 2023 10:27:48 -0400 Subject: [PATCH] Upgrade some deps; fix deprecations --- build.gradle | 28 ++++++++-------- gradle/docs.gradle | 17 ++++++---- settings.gradle | 4 +-- .../aggregator/AggregatorTests.java | 30 +++++++++-------- .../ConnectionFactoryShutDownTests.java | 7 ++-- .../kafka/inbound/KafkaMessageSource.java | 32 ++++++++++--------- 6 files changed, 64 insertions(+), 54 deletions(-) diff --git a/build.gradle b/build.gradle index a5581fd5a2..fcb8f0e60c 100644 --- a/build.gradle +++ b/build.gradle @@ -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 { diff --git a/gradle/docs.gradle b/gradle/docs.gradle index a12aa834d8..36bd06a1ff 100644 --- a/gradle/docs.gradle +++ b/gradle/docs.gradle @@ -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 ] } ) diff --git a/settings.gradle b/settings.gradle index eecaca0099..64279744c9 100644 --- a/settings.gradle +++ b/settings.gradle @@ -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' diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java index a2d6e8163a..c2df32b457 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java @@ -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 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> messages = new ArrayList>(60000); + private final Collection> 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 diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionFactoryShutDownTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionFactoryShutDownTests.java index f54693d56b..c135336321 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionFactoryShutDownTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionFactoryShutDownTests.java @@ -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(); } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java index 40b08c7028..597e8bcf94 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java @@ -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 extends AbstractMessageSource this.consumerProperties.setClientId(getComponentName()); } - Map props = this.consumerFactory.getConfigurationProperties(); Properties kafkaConsumerProperties = this.consumerProperties.getKafkaConsumerProperties(); this.checkNullKeyForExceptions = this.consumerProperties.isCheckDeserExWhenKeyNull() || @@ -390,16 +388,7 @@ public class KafkaMessageSource extends AbstractMessageSource 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 configs = new HashMap<>(suppliedConsumerFactory.getConfigurationProperties()); - configs.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1); - DefaultKafkaConsumerFactory 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 extends AbstractMessageSource 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 extends AbstractMessageSource } } - private void checkDeserializationException(ConsumerRecord cRecord, String headerName) { - DeserializationException exception = ListenerUtils.getExceptionFromHeader(cRecord, headerName, this.logger); + private void checkDeserializationException(ConsumerRecord record, String headerName) { + DeserializationException exception = SerializationUtils.getExceptionFromHeader(record, headerName, this.logger); if (exception != null) { throw exception; } @@ -707,6 +696,19 @@ public class KafkaMessageSource extends AbstractMessageSource } } + private static ConsumerFactory fixConsumerFactory(ConsumerFactory suppliedConsumerFactory) { + Map configs = new HashMap<>(suppliedConsumerFactory.getConfigurationProperties()); + configs.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1); + DefaultKafkaConsumerFactory 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;