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

GH-2295: No Resolvers with ConsumerRecordMetadata

Browse Source 
Resolves https://github.com/spring-projects/spring-kafka/issues/2295

A subclass of `InvocableHandlerMethod` was used to signal that a method
needs a `ConsumerRecordMetadata` to be constructed.

However, the constructor that takes an existing IHM, does not fully clone
the argument passed into it.

Then, the payload resolver was not invoked to resolve that parameter.

Use a map of booleans instead, to signal the creation of the metadata.

**cherry-pick to 2.9.x, 2.8.x, 2.7.x**
This commit is contained in:
Gary Russell
2022-06-06 15:34:08 -04:00
committed by Artem Bilan
parent 6fd78f25c5
commit 6416235c32
2 changed files with 20 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/DelegatingInvocableHandler.java
View File

@@ -40,7 +40,6 @@ import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.messaging.handler.HandlerMethod;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.messaging.handler.annotation.support.PayloadMethodArgumentResolver;
@@ -75,6 +74,8 @@ public class DelegatingInvocableHandler {
private final Map<InvocableHandlerMethod, Boolean> handlerReturnsMessage = new ConcurrentHashMap<>();
private final Map<InvocableHandlerMethod, Boolean> handlerMetadataAware = new ConcurrentHashMap<>();
private final Object bean;
private final BeanExpressionResolver resolver;
@@ -137,11 +138,12 @@ public class DelegatingInvocableHandler {
@Nullable BeanExpressionContext beanExpressionContext,
@Nullable BeanFactory beanFactory, @Nullable Validator validator) {
this.handlers = new ArrayList<>();
this.handlers = new ArrayList<>(handlers);
for (InvocableHandlerMethod handler : handlers) {
this.handlers.add(wrapIfNecessary(handler));
checkSpecial(handler);
}
this.defaultHandler = wrapIfNecessary(defaultHandler);
this.defaultHandler = defaultHandler;
checkSpecial(defaultHandler);
this.bean = bean;
this.resolver = beanExpressionResolver;
this.beanExpressionContext = beanExpressionContext;
@@ -151,18 +153,17 @@ public class DelegatingInvocableHandler {
this.validator = validator == null ? null : new PayloadValidator(validator);
}
@Nullable
private InvocableHandlerMethod wrapIfNecessary(@Nullable InvocableHandlerMethod handler) {
private void checkSpecial(@Nullable InvocableHandlerMethod handler) {
if (handler == null) {
return null;
return;
}
Parameter[] parameters = handler.getMethod().getParameters();
for (Parameter parameter : parameters) {
if (parameter.getType().equals(ConsumerRecordMetadata.class)) {
return new DelegatingInvocableHandler.MetadataAwareInvocableHandlerMethod(handler);
this.handlerMetadataAware.put(handler, true);
return;
}
}
return handler;
}
/**
@@ -191,7 +192,7 @@ public class DelegatingInvocableHandler {
}
}
Object result;
if (handler instanceof MetadataAwareInvocableHandlerMethod) {
if (Boolean.TRUE.equals(this.handlerMetadataAware.get(handler))) {
Object[] args = new Object[providedArgs.length + 1];
args[0] = AdapterUtils.buildConsumerRecordMetadataFromArray(providedArgs);
System.arraycopy(providedArgs, 0, args, 1, providedArgs.length);
@@ -350,19 +351,6 @@ public class DelegatingInvocableHandler {
return this.defaultHandler != null;
}
/**
* A handler method that is aware of {@link ConsumerRecordMetadata}.
*
* @since 2.5
*/
private static final class MetadataAwareInvocableHandlerMethod extends InvocableHandlerMethod {
MetadataAwareInvocableHandlerMethod(HandlerMethod handlerMethod) {
super(handlerMethod);
}
}
private static final class PayloadValidator extends PayloadMethodArgumentResolver {
PayloadValidator(Validator validator) {

13
spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java
View File

@@ -107,6 +107,7 @@ import org.springframework.kafka.listener.KafkaListenerErrorHandler;
import org.springframework.kafka.listener.KafkaMessageListenerContainer;
import org.springframework.kafka.listener.ListenerExecutionFailedException;
import org.springframework.kafka.listener.MessageListenerContainer;
import org.springframework.kafka.listener.adapter.ConsumerRecordMetadata;
import org.springframework.kafka.listener.adapter.FilteringMessageListenerAdapter;
import org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter;
import org.springframework.kafka.listener.adapter.RecordFilterStrategy;
@@ -431,6 +432,7 @@ public class EnableKafkaIntegrationTests {
template.send("annotated8", 0, 1, "junk");
assertThat(this.multiListener.errorLatch.await(60, TimeUnit.SECONDS)).isTrue();
assertThat(this.multiListener.meta).isNotNull();
}
@Test
@@ -2213,18 +2215,21 @@ public class EnableKafkaIntegrationTests {
@KafkaListener(id = "multi", topics = "annotated8", errorHandler = "consumeMultiMethodException")
static class MultiListenerBean {
private final CountDownLatch latch1 = new CountDownLatch(1);
final CountDownLatch latch1 = new CountDownLatch(1);
private final CountDownLatch latch2 = new CountDownLatch(1);
final CountDownLatch latch2 = new CountDownLatch(1);
private final CountDownLatch errorLatch = new CountDownLatch(1);
final CountDownLatch errorLatch = new CountDownLatch(1);
volatile ConsumerRecordMetadata meta;
@KafkaHandler
public void bar(@NonNull String bar) {
public void bar(@NonNull String bar, @Header(KafkaHeaders.RECORD_METADATA) ConsumerRecordMetadata meta) {
if ("junk".equals(bar)) {
throw new RuntimeException("intentional");
}
else {
this.meta = meta;
this.latch1.countDown();
}
}