Skip to content

GH-3121: Fix NPE when acknowledging a DLQ record without an Acknowledgment header #3309

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Sanskar121543 wants to merge 1 commit into
base: main
Choose a base branch
from
+128 −1
Open

GH-3121: Fix NPE when acknowledging a DLQ record without an Acknowledgment header #3309

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 11 additions & 1 deletion ...rc/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1757,7 +1757,17 @@ void sendToDlq(ConsumerRecord<?, ?> consumerRecord, Headers headers,
finally {
if (ackMode == ContainerProperties.AckMode.MANUAL
|| ackMode == ContainerProperties.AckMode.MANUAL_IMMEDIATE) {
messageHeaders.get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class).acknowledge();
Acknowledgment acknowledgment = messageHeaders.get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class);
if (acknowledgment != null) {
acknowledgment.acknowledge();
}
else {
// Can happen when the error message does not carry the original
// message's headers (e.g. no original message available), so the
// Acknowledgment is not reachable here; log instead of throwing an NPE.
KafkaMessageChannelBinder.this.logger.warn("No Acknowledgment header available; "
+ "could not acknowledge the record sent to DLQ " + sb);
}
}
}
}
Expand Down
117 changes: 117 additions & 0 deletions ...fka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderUnitTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@

package org.springframework.cloud.stream.binder.kafka;

import java.lang.reflect.Constructor;
import java.lang.reflect.Method;
import java.time.Duration;
import java.util.ArrayList;
Expand All @@ -25,17 +26,20 @@
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.header.internals.RecordHeaders;
import org.apache.kafka.common.serialization.Deserializer;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
Expand All @@ -47,16 +51,24 @@
import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties;
import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties;
import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner;
import org.springframework.cloud.stream.binder.kafka.utils.DlqPartitionFunction;
import org.springframework.cloud.stream.config.ListenerContainerCustomizer;
import org.springframework.cloud.stream.provisioning.ConsumerDestination;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.AbstractMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.kafka.support.SendResult;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHeaders;

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatCode;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyInt;
Expand All @@ -65,6 +77,7 @@
import static org.mockito.BDDMockito.given;
import static org.mockito.BDDMockito.willAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;

/**
Expand Down Expand Up @@ -341,4 +354,108 @@ public Deserializer<byte[]> getValueDeserializer() {
messageChannelBinding.unbind();
}

/*
* GH-3121: when the message being sent to the DLQ carries no Acknowledgment header
* (e.g. an ErrorMessage without the original message's headers), sendToDlq must not
* blow up with a NullPointerException while trying to acknowledge in MANUAL /
* MANUAL_IMMEDIATE ack mode.
*/
@SuppressWarnings("unchecked")
@Test
void sendToDlqDoesNotThrowWhenAcknowledgmentHeaderIsMissing() throws Exception {
KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(
new KafkaBinderConfigurationProperties(new TestKafkaProperties(), mock(ObjectProvider.class)),
mock(KafkaTopicProvisioner.class));

KafkaTemplate<byte[], byte[]> kafkaTemplate = mock(KafkaTemplate.class);
SendResult<byte[], byte[]> sendResult = mock(SendResult.class);
given(kafkaTemplate.send(any(org.apache.kafka.clients.producer.ProducerRecord.class))).willReturn(CompletableFuture.completedFuture(sendResult));

Object dlqSender = newDlqSender(binder, kafkaTemplate, 10_000L);

ConsumerRecord<Object, Object> consumerRecord = new ConsumerRecord<>("topic", 0, 0L, "key", "value");
MessageHeaders headersWithoutAcknowledgment = new MessageHeaders(Collections.emptyMap());

assertThatCode(() -> invokeSendToDlq(dlqSender, consumerRecord, headersWithoutAcknowledgment,
ContainerProperties.AckMode.MANUAL_IMMEDIATE))
.doesNotThrowAnyException();
}

@SuppressWarnings("unchecked")
@Test
void sendToDlqStillAcknowledgesWhenAcknowledgmentHeaderIsPresent() throws Exception {
KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(
new KafkaBinderConfigurationProperties(new TestKafkaProperties(), mock(ObjectProvider.class)),
mock(KafkaTopicProvisioner.class));

KafkaTemplate<byte[], byte[]> kafkaTemplate = mock(KafkaTemplate.class);
SendResult<byte[], byte[]> sendResult = mock(SendResult.class);
given(kafkaTemplate.send(any(org.apache.kafka.clients.producer.ProducerRecord.class))).willReturn(CompletableFuture.completedFuture(sendResult));

Object dlqSender = newDlqSender(binder, kafkaTemplate, 10_000L);

ConsumerRecord<Object, Object> consumerRecord = new ConsumerRecord<>("topic", 0, 0L, "key", "value");
Acknowledgment acknowledgment = mock(Acknowledgment.class);
MessageHeaders headersWithAcknowledgment = new MessageHeaders(
Collections.singletonMap(KafkaHeaders.ACKNOWLEDGMENT, acknowledgment));

invokeSendToDlq(dlqSender, consumerRecord, headersWithAcknowledgment,
ContainerProperties.AckMode.MANUAL_IMMEDIATE);

verify(acknowledgment).acknowledge();
}

@SuppressWarnings("unchecked")
@Test
void sendToDlqDoesNotAcknowledgeWhenAckModeIsNotManual() throws Exception {
KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(
new KafkaBinderConfigurationProperties(new TestKafkaProperties(), mock(ObjectProvider.class)),
mock(KafkaTopicProvisioner.class));

KafkaTemplate<byte[], byte[]> kafkaTemplate = mock(KafkaTemplate.class);
SendResult<byte[], byte[]> sendResult = mock(SendResult.class);
given(kafkaTemplate.send(any(org.apache.kafka.clients.producer.ProducerRecord.class))).willReturn(CompletableFuture.completedFuture(sendResult));

Object dlqSender = newDlqSender(binder, kafkaTemplate, 10_000L);

ConsumerRecord<Object, Object> consumerRecord = new ConsumerRecord<>("topic", 0, 0L, "key", "value");
Acknowledgment acknowledgment = mock(Acknowledgment.class);
MessageHeaders headersWithAcknowledgment = new MessageHeaders(
Collections.singletonMap(KafkaHeaders.ACKNOWLEDGMENT, acknowledgment));

invokeSendToDlq(dlqSender, consumerRecord, headersWithAcknowledgment, ContainerProperties.AckMode.RECORD);

verify(acknowledgment, never()).acknowledge();
}

private Object newDlqSender(KafkaMessageChannelBinder binder, KafkaTemplate<?, ?> kafkaTemplate, long timeout)
throws Exception {

Class<?> dlqSenderType = Class.forName(
"org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder$DlqSender");
Constructor<?> constructor = dlqSenderType
.getDeclaredConstructor(KafkaMessageChannelBinder.class, KafkaTemplate.class, long.class);
constructor.setAccessible(true);
return constructor.newInstance(binder, kafkaTemplate, timeout);
}

private void invokeSendToDlq(Object dlqSender, ConsumerRecord<Object, Object> consumerRecord,
MessageHeaders messageHeaders, ContainerProperties.AckMode ackMode) throws Exception {

Method sendToDlq = dlqSender.getClass().getDeclaredMethod("sendToDlq", ConsumerRecord.class,
org.apache.kafka.common.header.Headers.class, String.class, String.class, Throwable.class,
DlqPartitionFunction.class, MessageHeaders.class, ContainerProperties.AckMode.class);
sendToDlq.setAccessible(true);
try {
sendToDlq.invoke(dlqSender, consumerRecord, new RecordHeaders(), "dlqTopic", "group",
new RuntimeException("boom"), DlqPartitionFunction.ORIGINAL_PARTITION, messageHeaders, ackMode);
}
catch (java.lang.reflect.InvocationTargetException ex) {
if (ex.getCause() instanceof RuntimeException runtimeException) {
throw runtimeException;
}
throw ex;
}
}

}