Skip to content

GH-3555: Improve docs for KafkaHeaders.CONVERSION_FAILURES #3611

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 2 commits into from
Nov 11, 2024
Merged

GH-3555: Improve docs for KafkaHeaders.CONVERSION_FAILURES #3611

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
4b8e8a4
GH-3555: Improve docs for `KafkaHeaders.CONVERSION_FAILURES`
artembilan Nov 1, 2024
3b7cf6c
Fix Checkstyle violations
artembilan Nov 1, 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
18 changes: 18 additions & 0 deletions spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/serdes.adoc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -724,6 +724,24 @@ public void listen(List<Message<Foo>> fooMessages) {
}
----

If record in the batch cannot be converted, its payload is set as `null` into the target `payloads` list.
The conversion exception is logged as warning for this record and also stored into a `KafkaHeaders.CONVERSION_FAILURES` header as an item of the `List<ConversionException>`.
The target `@KafkaListener` method may perform Java `Stream` API to filter out those `null` values from the payload list or do something with the conversion exceptions header:

[source, java]
----
@KafkaListener(id = "foo", topics = "foo", autoStartup = "false")
public void listen(List<Foo> list,
@Header(KafkaHeaders.CONVERSION_FAILURES) List<ConversionException> conversionFailures) {

for (int i = 0; i < list.size(); i++) {
if (conversionFailures.get(i) != null) {
throw new BatchListenerFailedException("Conversion Failed", conversionFailures.get(i), i);
}
}
}
----

[[conversionservice-customization]]
== `ConversionService` Customization

Expand Down
65 changes: 39 additions & 26 deletions ...main/java/org/springframework/kafka/support/converter/BatchMessagingMessageConverter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,14 +24,15 @@
import java.util.List;
import java.util.Map;

import org.apache.commons.logging.LogFactory;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.record.TimestampType;
import org.apache.kafka.common.utils.Bytes;

import org.springframework.core.log.LogAccessor;
import org.springframework.core.log.LogMessage;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.DefaultKafkaHeaderMapper;
import org.springframework.kafka.support.JacksonPresent;
Expand All @@ -54,7 +55,7 @@
* <p>
* If a {@link RecordMessageConverter} is provided, and the batch type is a {@link ParameterizedType}
* with a single generic type parameter, each record will be passed to the converter, thus supporting
* a method signature {@code List<Foo> foos}.
* a method signature {@code List<MyType> myObjects}.
*
* @author Marius Bogoevici
* @author Gary Russell
Expand All @@ -63,11 +64,13 @@
* @author Sanghyeok An
* @author Hope Kim
* @author Borahm Lee
* @author Artem Bilan
*
* @since 1.1
*/
public class BatchMessagingMessageConverter implements BatchMessageConverter {

protected final LogAccessor logger = new LogAccessor(LogFactory.getLog(getClass())); // NOSONAR
protected final LogAccessor logger = new LogAccessor(getClass()); // NOSONAR

@Nullable
private final RecordMessageConverter recordConverter;
Expand Down Expand Up @@ -102,7 +105,7 @@ public BatchMessagingMessageConverter(@Nullable RecordMessageConverter recordCon

/**
* Generate {@link Message} {@code ids} for produced messages. If set to {@code false},
* will try to use a default value. By default set to {@code false}.
* will try to use a default value. By default, set to {@code false}.
* @param generateMessageId true if a message id should be generated
*/
public void setGenerateMessageId(boolean generateMessageId) {
Expand All @@ -111,7 +114,7 @@ public void setGenerateMessageId(boolean generateMessageId) {

/**
* Generate {@code timestamp} for produced messages. If set to {@code false}, -1 is
* used instead. By default set to {@code false}.
* used instead. By default, set to {@code false}.
* @param generateTimestamp true if a timestamp should be generated
*/
public void setGenerateTimestamp(boolean generateTimestamp) {
Expand Down Expand Up @@ -147,8 +150,8 @@ public void setRawRecordHeader(boolean rawRecordHeader) {
public Message<?> toMessage(List<ConsumerRecord<?, ?>> records, @Nullable Acknowledgment acknowledgment,
Consumer<?, ?> consumer, Type type) {

KafkaMessageHeaders kafkaMessageHeaders = new KafkaMessageHeaders(this.generateMessageId,
this.generateTimestamp);
KafkaMessageHeaders kafkaMessageHeaders =
new KafkaMessageHeaders(this.generateMessageId, this.generateTimestamp);

Map<String, Object> rawHeaders = kafkaMessageHeaders.getRawHeaders();
List<Object> payloads = new ArrayList<>();
Expand All @@ -169,16 +172,18 @@ public Message<?> toMessage(List<ConsumerRecord<?, ?>> records, @Nullable Acknow

String listenerInfo = null;
for (ConsumerRecord<?, ?> record : records) {
addRecordInfo(record, type, payloads, keys, topics, partitions, offsets, timestampTypes, timestamps, conversionFailures);
if (this.headerMapper != null && record.headers() != null) {
Map<String, Object> converted = convertHeaders(record.headers(), convertedHeaders);
addRecordInfo(record, type, payloads, keys, topics, partitions, offsets, timestampTypes, timestamps,
conversionFailures);
Headers recordHeaders = record.headers();
if (this.headerMapper != null && recordHeaders != null) {
Map<String, Object> converted = convertHeaders(recordHeaders, convertedHeaders);
Object obj = converted.get(KafkaHeaders.LISTENER_INFO);
if (obj instanceof String) {
listenerInfo = (String) obj;
if (obj instanceof String info) {
listenerInfo = info;
}
}
else {
natives.add(record.headers());
natives.add(recordHeaders);
}
if (this.rawRecordHeader) {
raws.add(record);
Expand All @@ -198,6 +203,7 @@ public Message<?> toMessage(List<ConsumerRecord<?, ?>> records, @Nullable Acknow

private void addToRawHeaders(Map<String, Object> rawHeaders, List<Map<String, Object>> convertedHeaders,
List<Headers> natives, List<ConsumerRecord<?, ?>> raws, List<ConversionException> conversionFailures) {

if (this.headerMapper != null) {
rawHeaders.put(KafkaHeaders.BATCH_CONVERTED_HEADERS, convertedHeaders);
}
Expand All @@ -211,16 +217,18 @@ private void addToRawHeaders(Map<String, Object> rawHeaders, List<Map<String, Ob
}

private void addRecordInfo(ConsumerRecord<?, ?> record, Type type, List<Object> payloads, List<Object> keys,
List<String> topics, List<Integer> partitions, List<Long> offsets, List<String> timestampTypes,
List<Long> timestamps, List<ConversionException> conversionFailures) {
List<String> topics, List<Integer> partitions, List<Long> offsets, List<String> timestampTypes,
List<Long> timestamps, List<ConversionException> conversionFailures) {

payloads.add(obtainPayload(type, record, conversionFailures));
keys.add(record.key());
topics.add(record.topic());
partitions.add(record.partition());
offsets.add(record.offset());
timestamps.add(record.timestamp());
if (record.timestampType() != null) {
timestampTypes.add(record.timestampType().name());
TimestampType timestampType = record.timestampType();
if (timestampType != null) {
timestampTypes.add(timestampType.name());
}
}

Expand Down Expand Up @@ -264,24 +272,29 @@ protected Object extractAndConvertValue(ConsumerRecord<?, ?> record, Type type)
protected Object convert(ConsumerRecord<?, ?> record, Type type, List<ConversionException> conversionFailures) {
try {
Object payload = this.recordConverter
.toMessage(record, null, null, ((ParameterizedType) type).getActualTypeArguments()[0]).getPayload();
.toMessage(record, null, null, ((ParameterizedType) type).getActualTypeArguments()[0]).getPayload();
conversionFailures.add(null);
return payload;
}
catch (ConversionException ex) {
byte[] original = null;
if (record.value() instanceof byte[]) {
original = (byte[]) record.value();
if (record.value() instanceof byte[] bytes) {
original = bytes;
}
else if (record.value() instanceof Bytes) {
original = ((Bytes) record.value()).get();
else if (record.value() instanceof Bytes bytes) {
original = bytes.get();
}
else if (record.value() instanceof String) {
original = ((String) record.value()).getBytes(StandardCharsets.UTF_8);
else if (record.value() instanceof String string) {
original = string.getBytes(StandardCharsets.UTF_8);
}
if (original != null) {
SerializationUtils.deserializationException(record.headers(), original, ex, false);
conversionFailures.add(ex);
this.logger.warn(ex,
LogMessage.format("Could not convert message for topic=%s, partition=%d, offset=%d",
record.topic(),
record.partition(),
record.offset()));
return null;
}
throw new ConversionException("The batch converter can only report conversion failures to the listener "
Expand All @@ -296,8 +309,8 @@ else if (record.value() instanceof String) {
* @return true if the conditions are met.
*/
private boolean containerType(Type type) {
return type instanceof ParameterizedType
&& ((ParameterizedType) type).getActualTypeArguments().length == 1;
return type instanceof ParameterizedType parameterizedType
&& parameterizedType.getActualTypeArguments().length == 1;
}

}