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

GH-1757: Option for Raw ConsumerRecord Header

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

* Add docs.
This commit is contained in:
Gary Russell
2021-04-07 14:47:18 -04:00
committed by GitHub
parent 290d11a319
commit 20397d95ea
6 changed files with 112 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
spring-kafka-docs/src/main/asciidoc/kafka.adoc
View File

@@ -4663,6 +4663,7 @@ TIP: See <<tip-assign-all-parts>>.
This section describes how to handle various exceptions that may arise when you use Spring for Apache Kafka.
[[listener-error-handlers]]
===== Listener Error Handlers
Starting with version 2.0, the `@KafkaListener` annotation has a new attribute: `errorHandler`.
@@ -4686,6 +4687,26 @@ You have access to the spring-messaging `Message<?>` object produced by the mess
The error handler can throw the original or a new exception, which is thrown to the container.
Anything returned by the error handler is ignored.
Starting with version 2.7, you can set the `rawRecordHeader` property on the `MessagingMessageConverter` and `BatchMessagingMessageConverter` which causes the raw `ConsumerRecord` to be added to the converted `Message<?>` in the `KafkaHeaders.RAW_DATA` header.
This is useful, for example, if you wish to use a `DeadLetterPublishingRecoverer` in a listener error handler.
It might be used in a request/reply scenario where you wish to send a failure result to the sender, after some number of retries, after capturing the failed record in a dead letter topic.
====
[source, java]
----
@Bean
KafkaListenerErrorHandler eh(DeadLetterPublishingRecoverer recoverer) {
return (msg, ex) -> {
if (msg.getHeaders().get(KafkaHeaders.DELIVERY_ATTEMPT, Integer.class) > 9) {
recoverer.accept(msg.getHeaders().get(KafkaHeaders.RAW_DATA, ConsumerRecord.class), ex);
return "FAILED";
}
throw ex;
};
}
----
====
It has a sub-interface (`ConsumerAwareListenerErrorHandler`) that has access to the consumer object, through the following method:
====

4
spring-kafka-docs/src/main/asciidoc/whats-new.adoc
View File

@@ -35,6 +35,10 @@ See <<message-listener-container>> for more information.
You can now validate the payload parameter of `@KafkaHandler` methods (class-level listeners).
See <<kafka-validation>> for more information.
You can now set the `rawRecordHeader` property on the `MessagingMessageConverter` and `BatchMessagingMessageConverter` which causes the raw `ConsumerRecord` to be added to the converted `Message<?>`.
This is useful, for example, if you wish to use a `DeadLetterPublishingRecoverer` in a listener error handler.
See <<listener-error-handlers>> for more information.
[[x27-dlt]]
==== `DeadLetterPublishingRecover` Changes

21
spring-kafka/src/main/java/org/springframework/kafka/support/converter/BatchMessagingMessageConverter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2019 the original author or authors.
* Copyright 2016-2021 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.
@@ -71,6 +71,8 @@ public class BatchMessagingMessageConverter implements BatchMessageConverter {
private KafkaHeaderMapper headerMapper;
private boolean rawRecordHeader;
/**
* Create an instance that does not convert the record values.
*/
@@ -123,6 +125,16 @@ public class BatchMessagingMessageConverter implements BatchMessageConverter {
return this.recordConverter;
}
/**
* Set to true to add the raw {@code List<ConsumerRecord<?, ?>>} as a header
* {@link KafkaHeaders#RAW_DATA}.
* @param rawRecordHeader true to add the header.
* @since 2.7
*/
public void setRawRecordHeader(boolean rawRecordHeader) {
this.rawRecordHeader = rawRecordHeader;
}
@Override
public Message<?> toMessage(List<ConsumerRecord<?, ?>> records, @Nullable Acknowledgment acknowledgment,
Consumer<?, ?> consumer, Type type) {
@@ -140,12 +152,16 @@ public class BatchMessagingMessageConverter implements BatchMessageConverter {
List<Long> timestamps = new ArrayList<>();
List<Map<String, Object>> convertedHeaders = new ArrayList<>();
List<Headers> natives = new ArrayList<>();
List<ConsumerRecord<?, ?>> raws = new ArrayList<>();
if (this.headerMapper != null) {
rawHeaders.put(KafkaHeaders.BATCH_CONVERTED_HEADERS, convertedHeaders);
}
else {
rawHeaders.put(KafkaHeaders.NATIVE_HEADERS, natives);
}
if (this.rawRecordHeader) {
rawHeaders.put(KafkaHeaders.RAW_DATA, raws);
}
commonHeaders(acknowledgment, consumer, rawHeaders, keys, topics, partitions, offsets, timestampTypes,
timestamps);
@@ -177,6 +193,9 @@ public class BatchMessagingMessageConverter implements BatchMessageConverter {
}
natives.add(record.headers());
}
if (this.rawRecordHeader) {
raws.add(record);
}
}
return MessageBuilder.createMessage(payloads, kafkaMessageHeaders);
}

18
spring-kafka/src/main/java/org/springframework/kafka/support/converter/MessagingMessageConverter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2019 the original author or authors.
* Copyright 2016-2021 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.
@@ -62,6 +62,8 @@ public class MessagingMessageConverter implements RecordMessageConverter {
private KafkaHeaderMapper headerMapper;
private boolean rawRecordHeader;
public MessagingMessageConverter() {
if (JacksonPresent.isJackson2Present()) {
this.headerMapper = new DefaultKafkaHeaderMapper();
@@ -98,6 +100,16 @@ public class MessagingMessageConverter implements RecordMessageConverter {
this.headerMapper = headerMapper;
}
/**
* Set to true to add the raw {@link ConsumerRecord} as a header
* {@link KafkaHeaders#RAW_DATA}.
* @param rawRecordHeader true to add the header.
* @since 2.7
*/
public void setRawRecordHeader(boolean rawRecordHeader) {
this.rawRecordHeader = rawRecordHeader;
}
@Override
public Message<?> toMessage(ConsumerRecord<?, ?> record, Acknowledgment acknowledgment, Consumer<?, ?> consumer,
Type type) {
@@ -119,7 +131,9 @@ public class MessagingMessageConverter implements RecordMessageConverter {
String ttName = record.timestampType() != null ? record.timestampType().name() : null;
commonHeaders(acknowledgment, consumer, rawHeaders, record.key(), record.topic(), record.partition(),
record.offset(), ttName, record.timestamp());
if (this.rawRecordHeader) {
rawHeaders.put(KafkaHeaders.RAW_DATA, record);
}
return MessageBuilder.createMessage(extractAndConvertValue(record, type), kafkaMessageHeaders);
}

49
spring-kafka/src/test/java/org/springframework/kafka/support/converter/BatchMessageConverterTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017-2019 the original author or authors.
* Copyright 2017-2021 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.
@@ -44,13 +44,14 @@ import org.springframework.messaging.MessageHeaders;
/**
* @author Biju Kunjummen
* @author Artem Bilan
* @author Gary Russell
*
* @since 1.3
*/
public class BatchMessageConverterTests {
@Test
public void testBatchConverters() {
void testBatchConverters() {
BatchMessageConverter batchMessageConverter = new BatchMessagingMessageConverter();
MessageHeaders headers = testGuts(batchMessageConverter);
@@ -61,10 +62,11 @@ public class BatchMessageConverterTests {
Map<String, Object> map = converted.get(0);
assertThat(map).hasSize(1);
assertThat(new String((byte[]) map.get("foo"))).isEqualTo("bar");
assertThat(headers.get(KafkaHeaders.RAW_DATA)).isNull();
}
@Test
public void testNoMapper() {
void testNoMapper() {
BatchMessagingMessageConverter batchMessageConverter = new BatchMessagingMessageConverter();
batchMessageConverter.setHeaderMapper(null);
@@ -79,16 +81,26 @@ public class BatchMessageConverterTests {
assertThat(new String(next.value())).isEqualTo("bar");
}
@Test
void raw() {
BatchMessagingMessageConverter batchMessageConverter = new BatchMessagingMessageConverter();
batchMessageConverter.setRawRecordHeader(true);
MessageHeaders headers = testGuts(batchMessageConverter);
@SuppressWarnings("unchecked")
List<Map<String, Object>> converted = (List<Map<String, Object>>) headers
.get(KafkaHeaders.BATCH_CONVERTED_HEADERS);
assertThat(converted).hasSize(3);
Map<String, Object> map = converted.get(0);
assertThat(map).hasSize(1);
assertThat(new String((byte[]) map.get("foo"))).isEqualTo("bar");
@SuppressWarnings("unchecked")
List<ConsumerRecord<?, ?>> rawHeader = headers.get(KafkaHeaders.RAW_DATA, List.class);
assertThat(rawHeader).extracting(rec -> (String) rec.value())
.containsExactly("value1", "value2", "value3");
}
private MessageHeaders testGuts(BatchMessageConverter batchMessageConverter) {
Header header = new RecordHeader("foo", "bar".getBytes());
Headers kHeaders = new RecordHeaders(new Header[] { header });
List<ConsumerRecord<?, ?>> consumerRecords = new ArrayList<>();
consumerRecords.add(new ConsumerRecord<>("topic1", 0, 1, 1487694048607L,
TimestampType.CREATE_TIME, 123L, 2, 3, "key1", "value1", kHeaders));
consumerRecords.add(new ConsumerRecord<>("topic1", 0, 2, 1487694048608L,
TimestampType.CREATE_TIME, 123L, 2, 3, "key2", "value2", kHeaders));
consumerRecords.add(new ConsumerRecord<>("topic1", 0, 3, 1487694048609L,
TimestampType.CREATE_TIME, 123L, 2, 3, "key3", "value3", kHeaders));
List<ConsumerRecord<?, ?>> consumerRecords = recordList();
Acknowledgment ack = mock(Acknowledgment.class);
@@ -118,6 +130,19 @@ public class BatchMessageConverterTests {
return headers;
}
private List<ConsumerRecord<?, ?>> recordList() {
Header header = new RecordHeader("foo", "bar".getBytes());
Headers kHeaders = new RecordHeaders(new Header[] { header });
List<ConsumerRecord<?, ?>> consumerRecords = new ArrayList<>();
consumerRecords.add(new ConsumerRecord<>("topic1", 0, 1, 1487694048607L,
TimestampType.CREATE_TIME, 123L, 2, 3, "key1", "value1", kHeaders));
consumerRecords.add(new ConsumerRecord<>("topic1", 0, 2, 1487694048608L,
TimestampType.CREATE_TIME, 123L, 2, 3, "key2", "value2", kHeaders));
consumerRecords.add(new ConsumerRecord<>("topic1", 0, 3, 1487694048609L,
TimestampType.CREATE_TIME, 123L, 2, 3, "key3", "value3", kHeaders));
return consumerRecords;
}
@SuppressWarnings("unchecked")
@Test
public void missingHeaders() {

15
spring-kafka/src/test/java/org/springframework/kafka/support/converter/MessagingMessageConverterTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2019 the original author or authors.
* Copyright 2019-2021 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.
@@ -41,6 +41,7 @@ public class MessagingMessageConverterTests {
assertThat(message.getPayload()).isEqualTo("baz");
assertThat(message.getHeaders().get(KafkaHeaders.RECEIVED_TOPIC)).isEqualTo("foo");
assertThat(message.getHeaders().get(KafkaHeaders.RECEIVED_MESSAGE_KEY)).isEqualTo("bar");
assertThat(message.getHeaders().get(KafkaHeaders.RAW_DATA)).isNull();
}
@Test
@@ -53,4 +54,16 @@ public class MessagingMessageConverterTests {
assertThat(message.getHeaders().containsKey(KafkaHeaders.RECEIVED_MESSAGE_KEY)).isFalse();
}
@Test
void raw() {
MessagingMessageConverter converter = new MessagingMessageConverter();
converter.setRawRecordHeader(true);
ConsumerRecord<String, String> record = new ConsumerRecord<>("foo", 1, 42, -1L, null, 0L, 0, 0, "bar", "baz");
Message<?> message = converter.toMessage(record, null, null, null);
assertThat(message.getPayload()).isEqualTo("baz");
assertThat(message.getHeaders().get(KafkaHeaders.RECEIVED_TOPIC)).isEqualTo("foo");
assertThat(message.getHeaders().get(KafkaHeaders.RECEIVED_MESSAGE_KEY)).isEqualTo("bar");
assertThat(message.getHeaders().get(KafkaHeaders.RAW_DATA)).isSameAs(record);
}
}