Skip to content

GH-3114: DeserializationException propagation #3115

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

Merged
merged 3 commits into from
Mar 14, 2024
Merged

GH-3114: DeserializationException propagation #3115

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
a7690ab
GH-3114: DeserializationException propagation
sobychacko Mar 12, 2024
982c65e
Addressing PR review
sobychacko Mar 13, 2024
e8ac7fe
Addressing PR review
sobychacko Mar 14, 2024
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
7 changes: 7 additions & 0 deletions spring-kafka/src/main/java/org/springframework/kafka/support/DefaultKafkaHeaderMapper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.header.internals.RecordHeader;

import org.springframework.kafka.support.serializer.SerializationUtils;
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We need to be sure that there is no package tangle . Usually sub-packages uses their parents.

import org.springframework.messaging.MessageHeaders;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
Expand All @@ -48,6 +49,7 @@
*
* @author Gary Russell
* @author Artem Bilan
* @author Soby Chacko
*
* @since 1.3
*
Expand Down Expand Up @@ -314,6 +316,11 @@ public void toHeaders(Headers source, final Map<String, Object> headers) {
else if (headerName.equals(KafkaHeaders.LISTENER_INFO) && matchesForInbound(headerName)) {
headers.put(headerName, new String(header.value(), getCharset()));
}
else if ((headerName.equals(SerializationUtils.KEY_DESERIALIZER_EXCEPTION_HEADER) ||
headerName.equals(SerializationUtils.VALUE_DESERIALIZER_EXCEPTION_HEADER))
&& matchesForInbound(headerName)) {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I believe this header has to be mapped unconditionally. It is not some what end-user provides.

headers.put(headerName, header);
}
else if (!(headerName.equals(JSON_TYPES)) && matchesForInbound(headerName)) {
if (jsonTypes.containsKey(headerName)) {
String requestedType = jsonTypes.get(headerName);
Expand Down
42 changes: 9 additions & 33 deletions .../src/test/java/org/springframework/kafka/listener/DeadLetterPublishingRecovererTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
/*
* Copyright 2020-2023 the original author or authors.
* Copyright 2020-2024 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.
Expand Down Expand Up @@ -33,10 +33,6 @@
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectOutputStream;
import java.io.UncheckedIOException;
import java.time.Duration;
import java.util.Collections;
import java.util.HashMap;
Expand Down Expand Up @@ -80,6 +76,7 @@
/**
* @author Gary Russell
* @author Tomaz Fernandes
* @author Soby Chacko
* @since 2.4.3
*
*/
Expand Down Expand Up @@ -174,9 +171,9 @@ void valueHeaderStripped() {
DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(template);
Headers headers = new RecordHeaders();
headers.add(SerializationTestUtils.deserializationHeader(SerializationUtils.VALUE_DESERIALIZER_EXCEPTION_HEADER,
header(false)));
SerializationTestUtils.header(false)));
headers.add(SerializationTestUtils.deserializationHeader(SerializationUtils.KEY_DESERIALIZER_EXCEPTION_HEADER,
header(true)));
SerializationTestUtils.header(true)));
Headers custom = new RecordHeaders();
custom.add(new RecordHeader("foo", "bar".getBytes()));
recoverer.setHeadersFunction((rec, ex) -> custom);
Expand Down Expand Up @@ -206,7 +203,7 @@ void keyHeaderStripped() {
DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(template);
Headers headers = new RecordHeaders();
headers.add(SerializationTestUtils.deserializationHeader(SerializationUtils.KEY_DESERIALIZER_EXCEPTION_HEADER,
header(true)));
SerializationTestUtils.header(true)));
CompletableFuture future = new CompletableFuture();
future.complete(new Object());
willReturn(future).given(template).send(any(ProducerRecord.class));
Expand All @@ -225,9 +222,9 @@ void keyDeserOnly() {
KafkaOperations<?, ?> template = mock(KafkaOperations.class);
DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(template);
Headers headers = new RecordHeaders();
DeserializationException deserEx = createDeserEx(true);
DeserializationException deserEx = SerializationTestUtils.createDeserEx(true);
headers.add(SerializationTestUtils.deserializationHeader(SerializationUtils.KEY_DESERIALIZER_EXCEPTION_HEADER,
header(true, deserEx)));
SerializationTestUtils.header(deserEx)));
CompletableFuture future = new CompletableFuture();
future.complete(new Object());
willReturn(future).given(template).send(any(ProducerRecord.class));
Expand All @@ -250,9 +247,9 @@ void headersNotStripped() {
recoverer.setRetainExceptionHeader(true);
Headers headers = new RecordHeaders();
headers.add(SerializationTestUtils.deserializationHeader(SerializationUtils.VALUE_DESERIALIZER_EXCEPTION_HEADER,
header(false)));
SerializationTestUtils.header(false)));
headers.add(SerializationTestUtils.deserializationHeader(SerializationUtils.KEY_DESERIALIZER_EXCEPTION_HEADER,
header(true)));
SerializationTestUtils.header(true)));
CompletableFuture future = new CompletableFuture();
future.complete(new Object());
willReturn(future).given(template).send(any(ProducerRecord.class));
Expand Down Expand Up @@ -302,27 +299,6 @@ void tombstoneWithMultiTemplatesExplicit() {
verify(template2).send(any(ProducerRecord.class));
}

private byte[] header(boolean isKey) {
return header(isKey, createDeserEx(isKey));
}

private DeserializationException createDeserEx(boolean isKey) {
return new DeserializationException(
isKey ? "testK" : "testV",
isKey ? "key".getBytes() : "value".getBytes(), isKey, null);
}

private byte[] header(boolean isKey, DeserializationException deserEx) {
ByteArrayOutputStream baos = new ByteArrayOutputStream();
try {
new ObjectOutputStream(baos).writeObject(deserEx);
}
catch (IOException e) {
throw new UncheckedIOException(e);
}
return baos.toByteArray();
}

@SuppressWarnings({"unchecked", "rawtypes"})
@Test
void allOriginalHeaders() {
Expand Down
39 changes: 38 additions & 1 deletion ...-kafka/src/test/java/org/springframework/kafka/support/DefaultKafkaHeaderMapperTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
/*
* Copyright 2017-2022 the original author or authors.
* Copyright 2017-2024 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.
Expand Down Expand Up @@ -34,7 +34,11 @@
import org.apache.kafka.common.header.internals.RecordHeaders;
import org.junit.jupiter.api.Test;

import org.springframework.core.log.LogAccessor;
import org.springframework.kafka.support.DefaultKafkaHeaderMapper.NonTrustedHeaderType;
import org.springframework.kafka.support.serializer.DeserializationException;
import org.springframework.kafka.support.serializer.SerializationTestUtils;
import org.springframework.kafka.support.serializer.SerializationUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.ExecutorSubscribableChannel;
Expand All @@ -46,6 +50,7 @@
/**
* @author Gary Russell
* @author Artem Bilan
* @author Soby Chacko
*
* @since 1.3
*
Expand Down Expand Up @@ -321,6 +326,38 @@ void inboundJson() {
.containsKey("baz");
}

@Test
void deserializationExceptionHeadersAreMappedAsNonByteArray() {
DefaultKafkaHeaderMapper mapper = new DefaultKafkaHeaderMapper();

byte[] keyDeserExceptionBytes = SerializationTestUtils.header(true);
Header keyHeader = SerializationTestUtils.deserializationHeader(SerializationUtils.KEY_DESERIALIZER_EXCEPTION_HEADER,
keyDeserExceptionBytes);
byte[] valueDeserExceptionBytes = SerializationTestUtils.header(false);
Header valueHeader = SerializationTestUtils.deserializationHeader(SerializationUtils.VALUE_DESERIALIZER_EXCEPTION_HEADER,
valueDeserExceptionBytes);
Headers headers = new RecordHeaders(
new Header[] { keyHeader, valueHeader });
Map<String, Object> springHeaders = new HashMap<>();
mapper.toHeaders(headers, springHeaders);
assertThat(springHeaders.get(SerializationUtils.KEY_DESERIALIZER_EXCEPTION_HEADER)).isEqualTo(keyHeader);
assertThat(springHeaders.get(SerializationUtils.VALUE_DESERIALIZER_EXCEPTION_HEADER)).isEqualTo(valueHeader);

LogAccessor logger = new LogAccessor(this.getClass());

DeserializationException keyDeserializationException = SerializationUtils.byteArrayToDeserializationException(logger, keyHeader);
assertThat(keyDeserExceptionBytes).containsExactly(SerializationTestUtils.header(keyDeserializationException));

DeserializationException valueDeserializationException =
SerializationUtils.byteArrayToDeserializationException(logger, valueHeader);
assertThat(valueDeserExceptionBytes).containsExactly(SerializationTestUtils.header(valueDeserializationException));

headers = new RecordHeaders();
mapper.fromHeaders(new MessageHeaders(springHeaders), headers);
assertThat(headers.lastHeader(SerializationUtils.KEY_DESERIALIZER_EXCEPTION_HEADER)).isNull();
assertThat(headers.lastHeader(SerializationUtils.VALUE_DESERIALIZER_EXCEPTION_HEADER)).isNull();
}

public static final class Foo {

private String bar = "bar";
Expand Down
29 changes: 28 additions & 1 deletion ...ka/src/test/java/org/springframework/kafka/support/serializer/SerializationTestUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
/*
* Copyright 2023 the original author or authors.
* Copyright 2023-2024 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.
Expand All @@ -16,10 +16,16 @@

package org.springframework.kafka.support.serializer;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectOutputStream;
import java.io.UncheckedIOException;

import org.apache.kafka.common.header.Header;

/**
* @author Gary Russell
* @author Soby Chacko
* @since 2.9.11
*
*/
Expand All @@ -32,4 +38,25 @@ public static Header deserializationHeader(String key, byte[] value) {
return new DeserializationExceptionHeader(key, value);
}

public static byte[] header(boolean isKey) {
return header(createDeserEx(isKey));
}

public static DeserializationException createDeserEx(boolean isKey) {
return new DeserializationException(
isKey ? "testK" : "testV",
isKey ? "key".getBytes() : "value".getBytes(), isKey, null);
}

public static byte[] header(DeserializationException deserEx) {
ByteArrayOutputStream baos = new ByteArrayOutputStream();
try {
new ObjectOutputStream(baos).writeObject(deserEx);
}
catch (IOException e) {
throw new UncheckedIOException(e);
}
return baos.toByteArray();
}

}