SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Update Spring Boot from 3.0.0-M4 -> 3.0.0-SNAPSHOT

Browse Source 
Adapts to the following changes in upstreams libs:

- Spring Kafka removed ListenableFuture
- Spring AOT changed generator API

Fixes #2473 #2465

Checkstyle fixes
This commit is contained in:
Chris Bono
2022-08-08 23:34:44 -05:00
committed by Soby Chacko
parent b33c76ad84
commit f94da9d311
8 changed files with 81 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
binders/kafka-binder/pom.xml
View File

@@ -35,6 +35,12 @@
<dependencyManagement>
<dependencies>
<!-- TODO Remove this dep. block when SB moves to SI 6.0.0-SNAPSHOT -->
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-kafka</artifactId>
<version>6.0.0-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-streams</artifactId>

40
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderHealthIndicatorTests.java
View File

@@ -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<Integer, String> template = new KafkaTemplate<>(pf, true);
CountDownLatch latch = new CountDownLatch(records.size());
for (ProducerRecord<Integer, String> record : records) {
ListenableFuture<SendResult<Integer, String>> future = template
.send(record);
future.addCallback(
new ListenableFutureCallback<SendResult<Integer, String>>() {
@Override
public void onFailure(Throwable ex) {
Assert.fail();
}
@Override
public void onSuccess(SendResult<Integer, String> result) {
latch.countDown();
}
});
CompletableFuture<SendResult<Integer, String>> 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));
};
}

22
binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
View File

@@ -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<SendResult<K, V>> sentDlq = null;
CompletableFuture<SendResult<K, V>> sentDlq = null;
try {
sentDlq = this.kafkaTemplate.send(producerRecord);
sentDlq.addCallback(new ListenableFutureCallback<SendResult<K, V>>() {
@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<K, V> 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());
}
}
});

19
binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java
View File

@@ -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<SendResult> send(String topic,
Object payload) {
public CompletableFuture<SendResult> send(String topic, Object payload) {
sent.set(payload);
SettableListenableFuture<SendResult> future = new SettableListenableFuture<>();
future.setException(fooException);
CompletableFuture<SendResult> future = new CompletableFuture<>();
future.completeExceptionally(fooException);
return future;
}
@Override // SIK 2.3+
public ListenableFuture send(ProducerRecord record) {
public CompletableFuture<SendResult> send(ProducerRecord record) {
sent.set(record.value());
SettableListenableFuture<SendResult> future = new SettableListenableFuture<>();
future.setException(fooException);
CompletableFuture<SendResult> future = new CompletableFuture<>();
future.completeExceptionally(fooException);
return future;
}

38
binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitStreamMessageHandler.java
View File

@@ -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<Boolean> future;
CompletableFuture<Boolean> 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<Boolean> future) {
future.addCallback(bool -> this.successCallback.onSuccess(message),
ex -> this.failureCallback.failure(message, ex));
private void handleConfirms(Message<?> message, CompletableFuture<Boolean> 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<T> {
/**
* 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);
}
}

2
bom/spring-cloud-starter-parent/pom.xml
View File

@@ -6,7 +6,7 @@
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>3.0.0-M4</version>
<version>3.0.0-SNAPSHOT</version>
<relativePath/>
</parent>
<groupId>org.springframework.cloud</groupId>

30
core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderChildContextInitializer.java
View File

@@ -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<String, $T<? extends $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<String, $T<? extends $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<? extends $T>" + name + "Initializer = new $L()", ApplicationContextInitializer.class,
ClassName initializerClassName = aotGenerator.processAheadOfTime(context, childGenerationContext);
method.addStatement("$T<? extends $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()));

2
samples/pom.xml
View File

@@ -11,7 +11,7 @@
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>3.0.0-M4</version>
<version>3.0.0-SNAPSHOT</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>