如何在Quarkus处理阶段创建多个KafaRecords？

问题描述

Kafka quickstart示例说明了如何使用Kafka在Quarkus中生成，处理，读取和写入消息。我想在已发布的消息中设置Kafka键，而我正在使用KafkaRecord进行设置。如果处理阶段生成单个KafkaRecord，但使用Multi创建多个记录时失败，则此方法有效。下面的代码显示了Quarkus的Kafka快速入门示例中的PriceGenerator类的简单扩展：

/**
 * A bean producing random prices every 5 seconds.
 * The prices are written to a Kafka topic (prices). The Kafka configuration is specified in the application configuration.
 */
@ApplicationScoped
public class PriceGenerator {

    private Random random = new Random();

    @Outgoing("convert-to-kafka")
    public Multi<Integer> generate() {
        return Multi.createFrom().ticks().every(Duration.ofSeconds(5))
                .onOverflow().drop()
                .map(tick -> random.nextInt(100));
    }

    @Incoming("convert-to-kafka")
    @Outgoing("generated-price")
    public Multi<KafkaRecord<String,Integer>> convert(int price) {
        return Multi.createFrom()
            .items(KafkaRecord.of("original",price),KafkaRecord.of("double",2*price));
    }
}

这不起作用，并生成以下类广播异常：

2020-08-21 08:36:05,284 ERROR [io.sma.rea.mes.provider] (executor-thread-1) SRMSG00200: The method org.acme.kafka.PriceGenerator#convert has thrown an exception: java.lang.ClassCastException: class org.eclipse.microprofile.reactive.messaging.Message$$Lambda$753/0x0000000800564840 cannot be cast to class java.lang.Integer (org.eclipse.microprofile.reactive.messaging.Message$$Lambda$753/0x0000000800564840 is in unnamed module of loader io.quarkus.bootstrap.classloading.QuarkusClassLoader @7c9bdee9; java.lang.Integer is in module java.base of loader 'bootstrap')
at org.acme.kafka.PriceGenerator_SmallRyeMessagingInvoker_convert_168e29259fd7b03bb28d7d0c9b052712b01b02a4.invoke(PriceGenerator_SmallRyeMessagingInvoker_convert_168e29259fd7b03bb28d7d0c9b052712b01b02a4.zig:46)

一种解决方法是将convert方法分为两个处理阶段，一个阶段使用Multi<Integer>作为返回类型来复制价格消息，然后将其转换为 进入KafkaRecords。如下所示：

@ApplicationScoped
public class PriceGenerator {

    private Random random = new Random();

    @Outgoing("duplicate")
    public Multi<Integer> generate() {
        return Multi.createFrom().ticks().every(Duration.ofSeconds(1))
                .onOverflow().drop()
                .map(tick -> random.nextInt(100));
    }

    @Incoming("duplicate")
    @Outgoing("convert-to-kafka")
    public Multi<Integer> duplicate(int price) {
        return Multi.createFrom()
            .items(price,2* price);
    }

    @Incoming("convert-to-kafka")
    @Outgoing("generated-price")
    public KafkaRecord<String,Integer> convert(int price) {
        return KafkaRecord.of("some key",price);
    }
}

我相信返回Multi<KafkaRecord<~>>类型的阶段也应该起作用，但是也许我错过了一些东西... 我已经使用Quarkus 1.7.0.Final运行以上示例。

解决方法

暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！

如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。

小编邮箱:dio#foxmail.com (将#修改为@）