From f94da9d3114920e25ecd2af2ad932d47a518a5e6 Mon Sep 17 00:00:00 2001 From: Chris Bono Date: Mon, 8 Aug 2022 23:34:44 -0500 Subject: [PATCH] Update Spring Boot from `3.0.0-M4` -> `3.0.0-SNAPSHOT` Adapts to the following changes in upstreams libs: - Spring Kafka removed ListenableFuture - Spring AOT changed generator API Fixes #2473 #2465 Checkstyle fixes --- binders/kafka-binder/pom.xml | 6 +++ ...afkaStreamsBinderHealthIndicatorTests.java | 40 +++++++------------ .../kafka/KafkaMessageChannelBinder.java | 22 ++++------ .../stream/binder/kafka/KafkaBinderTests.java | 19 +++++---- .../rabbit/RabbitStreamMessageHandler.java | 38 +++++++++++++----- bom/spring-cloud-starter-parent/pom.xml | 2 +- .../binder/BinderChildContextInitializer.java | 30 +++++++------- samples/pom.xml | 2 +- 8 files changed, 81 insertions(+), 78 deletions(-) diff --git a/binders/kafka-binder/pom.xml b/binders/kafka-binder/pom.xml index 28427e24b..605a509ea 100644 --- a/binders/kafka-binder/pom.xml +++ b/binders/kafka-binder/pom.xml @@ -35,6 +35,12 @@ + + + org.springframework.integration + spring-integration-kafka + 6.0.0-SNAPSHOT + org.apache.kafka kafka-streams diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderHealthIndicatorTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderHealthIndicatorTests.java index 69f32d64a..16308c673 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderHealthIndicatorTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderHealthIndicatorTests.java @@ -18,6 +18,7 @@ package org.springframework.cloud.stream.binder.kafka.streams.integration; import java.util.List; import java.util.Map; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.function.Function; @@ -29,7 +30,7 @@ import org.apache.kafka.streams.KafkaStreams; import org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler; import org.apache.kafka.streams.kstream.KStream; import org.assertj.core.util.Lists; -import org.junit.Assert; +import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; @@ -42,7 +43,6 @@ import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.cloud.stream.binder.kafka.streams.KafkaStreamsBinderHealthIndicator; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; -import org.springframework.kafka.config.KafkaStreamsCustomizer; import org.springframework.kafka.config.StreamsBuilderFactoryBeanConfigurer; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.core.DefaultKafkaProducerFactory; @@ -52,13 +52,12 @@ import org.springframework.kafka.test.EmbeddedKafkaBroker; import org.springframework.kafka.test.condition.EmbeddedKafkaCondition; import org.springframework.kafka.test.context.EmbeddedKafka; import org.springframework.kafka.test.utils.KafkaTestUtils; -import org.springframework.util.concurrent.ListenableFuture; -import org.springframework.util.concurrent.ListenableFutureCallback; import static org.assertj.core.api.Assertions.assertThat; /** * @author Arnaud Jardiné + * @author Chris Bono */ @EmbeddedKafka(topics = {"out", "out2"}) public class KafkaStreamsBinderHealthIndicatorTests { @@ -152,22 +151,16 @@ public class KafkaStreamsBinderHealthIndicatorTests { KafkaTemplate template = new KafkaTemplate<>(pf, true); CountDownLatch latch = new CountDownLatch(records.size()); for (ProducerRecord record : records) { - ListenableFuture> future = template - .send(record); - future.addCallback( - new ListenableFutureCallback>() { - @Override - public void onFailure(Throwable ex) { - Assert.fail(); - } - - @Override - public void onSuccess(SendResult result) { - latch.countDown(); - } - }); + CompletableFuture> future = template.send(record); + future.whenComplete((result, ex) -> { + if (ex != null) { + Assertions.fail(); + } + else { + latch.countDown(); + } + }); } - latch.await(5, TimeUnit.SECONDS); embeddedKafka.consumeFromEmbeddedTopics(consumer, topics); @@ -281,13 +274,8 @@ public class KafkaStreamsBinderHealthIndicatorTests { @Bean public StreamsBuilderFactoryBeanConfigurer customizer() { return factoryBean -> { - factoryBean.setKafkaStreamsCustomizer(new KafkaStreamsCustomizer() { - @Override - public void customize(KafkaStreams kafkaStreams) { - kafkaStreams.setUncaughtExceptionHandler(exception -> - StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_CLIENT); - } - }); + factoryBean.setKafkaStreamsCustomizer(kafkaStreams -> kafkaStreams.setUncaughtExceptionHandler(exception -> + StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_CLIENT)); }; } diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index eb961ae16..dd31ea5db 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java @@ -28,6 +28,7 @@ import java.util.List; import java.util.Map; import java.util.Set; import java.util.UUID; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; @@ -136,8 +137,6 @@ import org.springframework.util.StringUtils; import org.springframework.util.backoff.BackOff; import org.springframework.util.backoff.ExponentialBackOff; import org.springframework.util.backoff.FixedBackOff; -import org.springframework.util.concurrent.ListenableFuture; -import org.springframework.util.concurrent.ListenableFutureCallback; /** * A {@link org.springframework.cloud.stream.binder.Binder} that uses Kafka as the @@ -155,6 +154,7 @@ import org.springframework.util.concurrent.ListenableFutureCallback; * @author Lukasz Kaminski * @author Taras Danylchuk * @author Yi Liu + * @author Chris Bono */ public class KafkaMessageChannelBinder extends // @checkstyle:off @@ -1581,22 +1581,16 @@ public class KafkaMessageChannelBinder extends .append(keyOrValue(value)) .append("'").append(" received from ") .append(consumerRecord.partition()); - ListenableFuture> sentDlq = null; + CompletableFuture> sentDlq = null; try { sentDlq = this.kafkaTemplate.send(producerRecord); - sentDlq.addCallback(new ListenableFutureCallback>() { - - @Override - public void onFailure(Throwable ex) { - KafkaMessageChannelBinder.this.logger - .error("Error sending to DLQ " + sb.toString(), ex); + sentDlq.whenComplete((result, ex) -> { + if (ex != null) { + KafkaMessageChannelBinder.this.logger.error("Error sending to DLQ " + sb, ex); } - - @Override - public void onSuccess(SendResult result) { + else { if (KafkaMessageChannelBinder.this.logger.isDebugEnabled()) { - KafkaMessageChannelBinder.this.logger - .debug("Sent to DLQ " + sb.toString() + ": " + result.getRecordMetadata()); + KafkaMessageChannelBinder.this.logger.debug("Sent to DLQ " + sb + ": " + result.getRecordMetadata()); } } }); diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java index b08eb07e8..dc8fd1bfe 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2021 the original author or authors. + * Copyright 2016-2022 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. @@ -29,6 +29,7 @@ import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.UUID; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; @@ -149,8 +150,6 @@ import org.springframework.messaging.support.MessageBuilder; import org.springframework.util.Assert; import org.springframework.util.MimeTypeUtils; import org.springframework.util.backoff.FixedBackOff; -import org.springframework.util.concurrent.ListenableFuture; -import org.springframework.util.concurrent.SettableListenableFuture; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatExceptionOfType; @@ -162,6 +161,7 @@ import static org.mockito.Mockito.mock; * @author Ilayaperumal Gopinathan * @author Henryk Konsek * @author Gary Russell + * @author Chris Bono */ @EmbeddedKafka(count = 1, controlledShutdown = true, topics = "error.pollableDlq.group-pcWithDlq", brokerProperties = {"transaction.state.log.replication.factor=1", "transaction.state.log.min.isr=1"}) @@ -2474,19 +2474,18 @@ public class KafkaBinderTests extends new KafkaTemplate(mock(ProducerFactory.class)) { @Override // SIK < 2.3 - public ListenableFuture send(String topic, - Object payload) { + public CompletableFuture send(String topic, Object payload) { sent.set(payload); - SettableListenableFuture future = new SettableListenableFuture<>(); - future.setException(fooException); + CompletableFuture future = new CompletableFuture<>(); + future.completeExceptionally(fooException); return future; } @Override // SIK 2.3+ - public ListenableFuture send(ProducerRecord record) { + public CompletableFuture send(ProducerRecord record) { sent.set(record.value()); - SettableListenableFuture future = new SettableListenableFuture<>(); - future.setException(fooException); + CompletableFuture future = new CompletableFuture<>(); + future.completeExceptionally(fooException); return future; } diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitStreamMessageHandler.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitStreamMessageHandler.java index 6930d42be..36f587d73 100644 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitStreamMessageHandler.java +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitStreamMessageHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2021-2021 the original author or authors. + * Copyright 2021-2022 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. @@ -16,6 +16,7 @@ package org.springframework.cloud.stream.binder.rabbit; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; @@ -28,6 +29,7 @@ import org.springframework.context.Lifecycle; import org.springframework.integration.amqp.support.AmqpHeaderMapper; import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper; import org.springframework.integration.handler.AbstractMessageHandler; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessageHandlingException; @@ -36,8 +38,6 @@ import org.springframework.rabbit.stream.producer.RabbitStreamOperations; import org.springframework.rabbit.stream.support.StreamMessageProperties; import org.springframework.util.Assert; import org.springframework.util.MimeType; -import org.springframework.util.concurrent.ListenableFuture; -import org.springframework.util.concurrent.SuccessCallback; /** * {@link MessageHandler} based on {@link RabbitStreamOperations}. @@ -45,6 +45,7 @@ import org.springframework.util.concurrent.SuccessCallback; * TODO: This class will move to Spring Integration in 6.0. * * @author Gary Russell + * @author Chris Bono * @since 3.2 * */ @@ -148,7 +149,7 @@ public class RabbitStreamMessageHandler extends AbstractMessageHandler implement @Override protected void handleMessageInternal(Message requestMessage) { - ListenableFuture future; + CompletableFuture future; com.rabbitmq.stream.Message streamMessage; if (requestMessage.getPayload() instanceof com.rabbitmq.stream.Message) { streamMessage = (com.rabbitmq.stream.Message) requestMessage.getPayload(); @@ -163,9 +164,15 @@ public class RabbitStreamMessageHandler extends AbstractMessageHandler implement handleConfirms(requestMessage, future); } - private void handleConfirms(Message message, ListenableFuture future) { - future.addCallback(bool -> this.successCallback.onSuccess(message), - ex -> this.failureCallback.failure(message, ex)); + private void handleConfirms(Message message, CompletableFuture future) { + future.whenComplete((bool, ex) -> { + if (ex != null) { + this.failureCallback.failure(message, ex); + } + else { + this.successCallback.onSuccess(message); + } + }); if (this.sync) { try { future.get(this.confirmTimeout, TimeUnit.MILLISECONDS); @@ -242,18 +249,27 @@ public class RabbitStreamMessageHandler extends AbstractMessageHandler implement return true; } + /** + * Callback for when publishing succeeds. + */ + interface SuccessCallback { + /** + * Called when the future completes with success. + * Note that Exceptions raised by this method are ignored. + * @param result the result of the future + */ + void onSuccess(@Nullable T result); + } + /** * Callback for when publishing fails. */ - public interface FailureCallback { - + interface FailureCallback { /** * Message publish failure. * @param message the message. * @param throwable the throwable. */ void failure(Message message, Throwable throwable); - } - } diff --git a/bom/spring-cloud-starter-parent/pom.xml b/bom/spring-cloud-starter-parent/pom.xml index 3fa27238a..6fc1be6a5 100644 --- a/bom/spring-cloud-starter-parent/pom.xml +++ b/bom/spring-cloud-starter-parent/pom.xml @@ -6,7 +6,7 @@ org.springframework.boot spring-boot-starter-parent - 3.0.0-M4 + 3.0.0-SNAPSHOT org.springframework.cloud diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderChildContextInitializer.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderChildContextInitializer.java index e9bf1d53b..93c89d9f1 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderChildContextInitializer.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderChildContextInitializer.java @@ -30,7 +30,6 @@ import org.springframework.aot.generate.MethodReference; import org.springframework.beans.factory.aot.BeanRegistrationAotContribution; import org.springframework.beans.factory.aot.BeanRegistrationAotProcessor; import org.springframework.beans.factory.aot.BeanRegistrationCode; -import org.springframework.beans.factory.aot.BeanRegistrationExcludeFilter; import org.springframework.beans.factory.support.RegisteredBean; import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContextAware; @@ -46,7 +45,7 @@ import org.springframework.util.Assert; * @author Chris Bono * @since 4.0 */ -public class BinderChildContextInitializer implements ApplicationContextAware, BeanRegistrationAotProcessor, BeanRegistrationExcludeFilter { +public class BinderChildContextInitializer implements ApplicationContextAware, BeanRegistrationAotProcessor { private final LogAccessor logger = new LogAccessor(LogFactory.getLog(getClass())); private DefaultBinderFactory binderFactory; @@ -78,7 +77,7 @@ public class BinderChildContextInitializer implements ApplicationContextAware, B } @Override - public boolean isExcluded(RegisteredBean registeredBean) { + public boolean isBeanExcludedFromAotProcessing() { return false; } @@ -140,24 +139,25 @@ public class BinderChildContextInitializer implements ApplicationContextAware, B @Override public void applyTo(GenerationContext generationContext, BeanRegistrationCode beanRegistrationCode) { ApplicationContextAotGenerator aotGenerator = new ApplicationContextAotGenerator(); - GeneratedMethod postProcessorMethod = beanRegistrationCode.getMethodGenerator() - .generateMethod("addChildContextInitializers").using(builder -> { - builder.addJavadoc("Use AOT child context initialization"); - builder.addModifiers(Modifier.PRIVATE, Modifier.STATIC); - builder.addParameter(RegisteredBean.class, "registeredBean"); - builder.addParameter(BinderChildContextInitializer.class, "instance"); - builder.returns(BinderChildContextInitializer.class); - builder.addStatement("$T> initializers = new $T<>()", Map.class, + + GeneratedMethod postProcessorMethod = beanRegistrationCode.getMethods().add("addChildContextInitializers", + (method) -> { + method.addJavadoc("Use AOT child context initialization"); + method.addModifiers(Modifier.PRIVATE, Modifier.STATIC); + method.addParameter(RegisteredBean.class, "registeredBean"); + method.addParameter(BinderChildContextInitializer.class, "instance"); + method.returns(BinderChildContextInitializer.class); + method.addStatement("$T> initializers = new $T<>()", Map.class, ApplicationContextInitializer.class, ConfigurableApplicationContext.class, HashMap.class); this.childContexts.forEach((name, context) -> { this.logger.debug(() -> "Generating AOT child context initializer for " + name); GenerationContext childGenerationContext = generationContext.withName(name + "Binder"); - ClassName initializerClassName = aotGenerator.generateApplicationContext(context, childGenerationContext); - builder.addStatement("$T" + name + "Initializer = new $L()", ApplicationContextInitializer.class, + ClassName initializerClassName = aotGenerator.processAheadOfTime(context, childGenerationContext); + method.addStatement("$T" + name + "Initializer = new $L()", ApplicationContextInitializer.class, ConfigurableApplicationContext.class, initializerClassName); - builder.addStatement("initializers.put($S," + name + "Initializer)", name); + method.addStatement("initializers.put($S," + name + "Initializer)", name); }); - builder.addStatement("return instance.withChildContextInitializers(initializers)"); + method.addStatement("return instance.withChildContextInitializers(initializers)"); }); beanRegistrationCode.addInstancePostProcessor( MethodReference.ofStatic(beanRegistrationCode.getClassName(), postProcessorMethod.getName())); diff --git a/samples/pom.xml b/samples/pom.xml index 6578280c7..eafd1e7a6 100644 --- a/samples/pom.xml +++ b/samples/pom.xml @@ -11,7 +11,7 @@ org.springframework.boot spring-boot-starter-parent - 3.0.0-M4 + 3.0.0-SNAPSHOT