Spring kafka 嵌入式测试

Question

我正在使用嵌入式 kafka 编写一个 junit 测试用例。我们有一个管道，其中生产者 > 主题 > 消费者 > 工作（） > 生产。 我正在使用第三方模式注册表（通过提供虚假 url 来模拟我的测试）和与之相关的特定 serdes。在 kafka 用户组上讨论了这个之后，这样做的方法是使用 一个模拟注册表，用于手动序列化数据并将 byte[] 本身传递给生产者而不是 avro 记录。在这种情况下，我的消费者怎么会失败，因为它需要一个特定的记录有效负载。关于如何解决这个问题的任何想法？

//Listener method


    */
        @KafkaListener(topics = test1,id="tesId1")
        public void onMessage(@Payload Log log, 
                @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
                @Header(KafkaHeaders.OFFSET) Long offset) throws Exception
                 {

               }


    // test class
    @RunWith(SpringRunner.class)
    @SpringBootTest
    @DirtiesContext
    @EmbeddedKafka(topics = { "test1" })
    @TestPropertySource(properties = { "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}" })
    public class ConsumerTests {

    }

Answer 1

只需使用原始KafkaTemplate（无泛型）和字节数组序列化器。

例如，使用 JSON 和 StringSerializer：

@SpringBootApplication
public class So53179695Application {

    public static void main(String[] args) {
        SpringApplication.run(So53179695Application.class, args);
    }

    @Bean
    public RecordMessageConverter converter() {
        return new StringJsonMessageConverter();
    }

    @KafkaListener(id = "foo", topics = "foo")
    public void listen(Foo in) {
        System.out.println(in);
    }

    public static class Foo {

        private String bar;

        public Foo() {
            super();
        }

        Foo(String bar) {
            this.bar = bar;
        }

        public String getBar() {
            return this.bar;
        }

        public void setBar(String bar) {
            this.bar = bar;
        }

        @Override
        public String toString() {
            return "Foo [bar=" + this.bar + "]";
        }

    }

}

和

@RunWith(SpringRunner.class)
@SpringBootTest
public class So53179695ApplicationTests {

    @ClassRule
    public static EmbeddedKafkaRule embeddedKafka = 
        new EmbeddedKafkaRule(1, false, "foo");

    @BeforeClass
    public static void setup() {
        System.setProperty("spring.kafka.bootstrap-servers",
            embeddedKafka.getEmbeddedKafka().getBrokersAsString());
    }

    @Autowired
    public KafkaTemplate<String, Foo> template;

    @SuppressWarnings("rawtypes")
    @Autowired
    public KafkaTemplate rawTemplate;

    @SuppressWarnings("unchecked")
    @Test
    public void test() throws Exception {
//      template.send("foo", new Foo("bar"));
        rawTemplate.send("foo", "{\"bar\":\"baz\"}");
        Thread.sleep(10_000);
    }

}

和

Foo [bar=baz]

请注意，两个模板都指向同一个物理对象 - 由于 java 的类型擦除，在运行时并不重要。

这假设您仍在消费者端使用 Avro 反序列化器（或本示例中的 JSON）。

或者你可以在消费者端使用你的模拟反序列化器来创建一个Log。