Spring Stream kafka活页夹测试自定义标题

Question

我试图弄清楚如何在Spring消息中包含自定义标头

我觉得我遗漏了一些东西，因为我似乎可以使用TestChannelBinder使其正常工作，例如。

import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

import java.util.function.Function;

@Component
@Slf4j
public class BaseStream implements Function<Message<String>, String> {
    @Override
    public String apply(Message<String> transactionMessage) {
        log.debug("Converted Message: {} ", transactionMessage);
        return transactionMessage.getPayload();
    }

}

带测试粘合剂的测试等级：

import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.cloud.stream.binder.test.InputDestination;
import org.springframework.cloud.stream.binder.test.OutputDestination;
import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
import org.springframework.context.annotation.Import;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.test.context.TestPropertySource;


@SpringBootTest
@TestPropertySource("classpath:testStream.properties")
@Import(TestChannelBinderConfiguration.class)
public class TestForStream {


    @Autowired
    InputDestination inputDestination;
    @Autowired
    OutputDestination outputDestination;

    @Test
    void contextLoads() {
        inputDestination.send(MessageBuilder
                .withPayload("Test Payload")
                .setHeader("customHeader", "headerSpecificData")
                .build());
    }
}

测试tream.properties

spring.cloud.function.definition=baseStream
spring.cloud.stream.bindings.baseStream-in-0.destination=test-in
spring.cloud.stream.bindings.baseStream-out-0.destination=test-out
spring.cloud.stream.bindings.baseStream-in-0.group=test-group-base

运行时记录：

Converted Message: GenericMessage [payload=Test Payload, headers={id=5c6d1082-c084-0b25-4afc-b5d97bf537f9, customHeader=headerSpecificData, contentType=application/json, timestamp=1639398696800, target-protocol=kafka}]

这就是我想要做的。但当我试着测试Kafka拜德时，它似乎包含了

我只是想知道，如果有人能看到我在测试中的错误，因为我已经尝试了各种方法来实现这一点，并且看到它与测试活页夹一起工作，我会假设它适用于Kafka活页夹。

Kafka活页夹测试的测试类：

import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.serializer.JsonSerializer;
import org.springframework.kafka.test.EmbeddedKafkaBroker;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.kafka.test.utils.KafkaTestUtils;
import org.springframework.test.context.TestPropertySource;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;


@EmbeddedKafka(partitions = 1, brokerProperties = { "listeners=PLAINTEXT://localhost:9092", "port=9092"})
@SpringBootTest
@TestPropertySource("classpath:testStream.properties")
public class TestForStream {

    public static CountDownLatch latch = new CountDownLatch(1);
    @Autowired
    public EmbeddedKafkaBroker broker;

    @Test
    void contextLoads() {
        sleep(5);//Included this as it takes some time to init>

        sendMessage("test-in", MessageBuilder
                .withPayload("Test Payload")
                .setHeader("customHeader", "headerSpecificData")
                .build());


    }

    public <T> ProducerFactory<String, T> createProducerFactory() {
        Map<String, Object> configs = new HashMap<>(KafkaTestUtils.producerProps(broker));
        configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        //Is JsonSerializer correct for a message?
        return new DefaultKafkaProducerFactory<>(configs);
    }

    public <T> void sendMessage(String topic, T listObj) {
        try {
            KafkaTemplate<String, T> kafkaTemplate = new KafkaTemplate<>(createProducerFactory());
            kafkaTemplate.send(new ProducerRecord<>(topic, listObj));
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    public void sleep(long time){
        try {
            latch.await(time, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}

消息的kafka活页夹测试日志：

Converted Message: GenericMessage [payload={"payload":"Test Payload","headers":{"customHeader":"headerSpecificData","id":"d540a3ca-28db-b137-fc86-c25cc4b7eb8b","timestamp":1639399810476}}, headers={deliveryAttempt=1, kafka_timestampType=CREATE_TIME, kafka_receivedTopic=test-in, target-protocol=kafka, kafka_offset=0, scst_nativeHeadersPresent=true, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@79580279, id=1cf2d382-df29-2672-4180-07da99e58244, kafka_receivedPartitionId=0, kafka_receivedTimestamp=1639399810526, contentType=application/json, __TypeId__=[B@24c79350, kafka_groupId=test-group-base, timestamp=1639399810651}]

因此，在这里，消息已包含在有效负载中，并且kafka标头已按预期包含在标头中。

我已经尝试了spring.cloud.stream.kafka.binder.headers和HeaderMode，看看他们是否会改变什么，但没有用。

编辑：

使用springCloudVersion=2020.0.3




Answer 1

我正在使用：

public <T> void sendMessage(String topic, T listObj) {
    try {
        KafkaTemplate<String, T> kafkaTemplate = new KafkaTemplate<>(createProducerFactory());
        kafkaTemplate.send(new ProducerRecord<>(topic, listObj));
    }catch (Exception e){
        e.printStackTrace();
    }
}

发送将消息作为值的消息。

我应该使用的内容：

public void sendMessage(String topic, Message<?> listObj) {
    try {
        KafkaTemplate<String, Message<?>> kafkaTemplate = new KafkaTemplate<>(createProducerFactory());
        kafkaTemplate.setDefaultTopic(topic);
        kafkaTemplate.send(listObj);
    }catch (Exception e){
        e.printStackTrace();
    }
}