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

GH-1066: Protect against null Headers

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

Certain clients (e.g. mapR) that emulate the Kafka clients do not properly
populate the `ConsumerRecord.headers()` field.

Check for null before mapping; also check `timestampType`.

**cherry-pick to 2.2.x, 2.1.x**

# Conflicts:
#	spring-kafka/src/main/java/org/springframework/kafka/support/converter/MessageConverter.java
#	spring-kafka/src/main/java/org/springframework/kafka/support/converter/MessagingMessageConverter.java
This commit is contained in:
Gary Russell
2019-04-12 16:54:33 -04:00
committed by Artem Bilan
parent 04f4db38e2
commit 40b21c1cd2
5 changed files with 99 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -125,6 +125,7 @@ public class BatchMessagingMessageConverter implements BatchMessageConverter {
@Override
public Message<?> toMessage(List<ConsumerRecord<?, ?>> records, Acknowledgment acknowledgment,
Consumer<?, ?> consumer, Type type) {
KafkaMessageHeaders kafkaMessageHeaders = new KafkaMessageHeaders(this.generateMessageId,
this.generateTimestamp);
@@ -167,9 +168,11 @@ public class BatchMessagingMessageConverter implements BatchMessageConverter {
topics.add(record.topic());
partitions.add(record.partition());
offsets.add(record.offset());
timestampTypes.add(record.timestampType().name());
if (record.timestampType() != null) {
timestampTypes.add(record.timestampType().name());
}
timestamps.add(record.timestamp());
if (this.headerMapper != null) {
if (this.headerMapper != null && record.headers() != null) {
Map<String, Object> converted = new HashMap<>();
this.headerMapper.toHeaders(record.headers(), converted);
convertedHeaders.add(converted);

27
spring-kafka/src/main/java/org/springframework/kafka/support/converter/MessageConverter.java
View File

@@ -16,6 +16,14 @@
package org.springframework.kafka.support.converter;
import java.util.Map;
import org.apache.kafka.clients.consumer.Consumer;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.lang.Nullable;
/**
* A top level interface for message converters.
*
@@ -25,4 +33,23 @@ package org.springframework.kafka.support.converter;
*/
public interface MessageConverter {
default void commonHeaders(Acknowledgment acknowledgment, Consumer<?, ?> consumer, Map<String, Object> rawHeaders,
Object theKey, Object topic, Object partition, Object offset,
@Nullable Object timestampType, Object timestamp) {
rawHeaders.put(KafkaHeaders.RECEIVED_MESSAGE_KEY, theKey);
rawHeaders.put(KafkaHeaders.RECEIVED_TOPIC, topic);
rawHeaders.put(KafkaHeaders.RECEIVED_PARTITION_ID, partition);
rawHeaders.put(KafkaHeaders.OFFSET, offset);
rawHeaders.put(KafkaHeaders.TIMESTAMP_TYPE, timestampType);
rawHeaders.put(KafkaHeaders.RECEIVED_TIMESTAMP, timestamp);
if (acknowledgment != null) {
rawHeaders.put(KafkaHeaders.ACKNOWLEDGMENT, acknowledgment);
}
if (consumer != null) {
rawHeaders.put(KafkaHeaders.CONSUMER, consumer);
}
}
}

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

@@ -101,11 +101,12 @@ public class MessagingMessageConverter implements RecordMessageConverter {
@Override
public Message<?> toMessage(ConsumerRecord<?, ?> record, Acknowledgment acknowledgment, Consumer<?, ?> consumer,
Type type) {
KafkaMessageHeaders kafkaMessageHeaders = new KafkaMessageHeaders(this.generateMessageId,
this.generateTimestamp);
Map<String, Object> rawHeaders = kafkaMessageHeaders.getRawHeaders();
if (this.headerMapper != null) {
if (this.headerMapper != null && record.headers() != null) {
this.headerMapper.toHeaders(record.headers(), rawHeaders);
}
else {
@@ -117,12 +118,9 @@ public class MessagingMessageConverter implements RecordMessageConverter {
}
rawHeaders.put(KafkaHeaders.NATIVE_HEADERS, record.headers());
}
rawHeaders.put(KafkaHeaders.RECEIVED_MESSAGE_KEY, record.key());
rawHeaders.put(KafkaHeaders.RECEIVED_TOPIC, record.topic());
rawHeaders.put(KafkaHeaders.RECEIVED_PARTITION_ID, record.partition());
rawHeaders.put(KafkaHeaders.OFFSET, record.offset());
rawHeaders.put(KafkaHeaders.TIMESTAMP_TYPE, record.timestampType().name());
rawHeaders.put(KafkaHeaders.RECEIVED_TIMESTAMP, record.timestamp());
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 (acknowledgment != null) {
rawHeaders.put(KafkaHeaders.ACKNOWLEDGMENT, acknowledgment);

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

@@ -21,6 +21,7 @@ import static org.mockito.Mockito.mock;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
@@ -112,4 +113,18 @@ public class BatchMessageConverterTests {
return headers;
}
@SuppressWarnings("unchecked")
@Test
public void missingHeaders() {
BatchMessageConverter converter = new BatchMessagingMessageConverter();
Headers nullHeaders = null;
ConsumerRecord<String, String> record = new ConsumerRecord<>("foo", 1, 42, -1L, null, 0L, 0, 0, "bar", "baz",
nullHeaders);
List<ConsumerRecord<?, ?>> records = Collections.singletonList(record);
Message<?> message = converter.toMessage(records, null, null, null);
assertThat(((List<String>) message.getPayload())).contains("baz");
assertThat(message.getHeaders().get(KafkaHeaders.RECEIVED_TOPIC, List.class)).contains("foo");
assertThat(message.getHeaders().get(KafkaHeaders.RECEIVED_MESSAGE_KEY, List.class)).contains("bar");
}
}

47
spring-kafka/src/test/java/org/springframework/kafka/support/converter/MessagingMessageConverterTests.java Normal file
View File

@@ -0,0 +1,47 @@
/*
* Copyright 2019 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.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.kafka.support.converter;
import static org.assertj.core.api.Assertions.assertThat;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.header.Headers;
import org.junit.jupiter.api.Test;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.Message;
/**
* @author Gary Russell
* @since 2.1.13
*
*/
public class MessagingMessageConverterTests {
@Test
void missingHeaders() {
MessagingMessageConverter converter = new MessagingMessageConverter();
Headers nullHeaders = null;
ConsumerRecord<String, String> record = new ConsumerRecord<>("foo", 1, 42, -1L, null, 0L, 0, 0, "bar", "baz",
nullHeaders);
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");
}
}