apache-kafka - Quarkus 应用程序中 Kafka 反序列化器中的 CDI 上下文

Question

我有一个基于 Kafka 的带有 Smallrye 响应式(Reactive)消息传递的 Quarkus 项目。因为我想使用“复杂的 pojo”，所以我需要一个自定义的反序列化器。
我想制作这两个类 CDI bean，这样我就可以注入(inject)和使用我的自定义记录器，它是一个 CDI bean。有没有办法做到这一点？

现在我注入(inject)的记录器对象只是空的:

import org.apache.kafka.common.serialization.Serializer;
import javax.enterprise.context.ApplicationScoped;
import javax.inject.Inject;

@ApplicationScoped
public class MySerializer implements Serializer<MyDto>
{
    @Inject MyLogger logger;

    @Override public byte[] serialize(String topicName, MyDto myDto)
    {
        // this causes a java.lang.NullPointerException
        logger.info("serializing");

        ...
    }
}

Answer 1

据我所知，你只能用kafka注册一个类名，它会在内部创建那个类，即。不使用 CDI。
可能的解决方法:使注册的对象成为 CDI bean 周围的瘦包装器，并将工作委托(delegate)给 bean:

public class MySerializer implements Serializer<MyDto> {
    private MySerializerCdi delegate;

    public MySerializer() {
        delegate = CDI.current().select(MySerializerCdi.class).get();
    }

    @Override public byte[] serialize(String topicName, MyDto myDto) {
        return delegate.serialize(topicName, myDto);
    }
    ...
}

...并相应地重命名您的原始 CDI 类。