使用Kafka与Micronaut

Question

是否有任何示例项目展示了如何将Kafka与Micronaut结合使用？我很难让它工作。

我有以下制片人：

@KafkaClient
interface AppClient {

@Topic("topic-name")
    void sendMessage(@KafkaKey String id, Event event)
}

听众：

@KafkaListener(
    groupId="group-id",
    offsetReset = OffsetReset.EARLIEST
)
class AppListener {

@Topic("topic-name")
    void onMessage(Event event) {
        // do stuff
    }
}

我的申请。yml包含：

kafka:
  bootstrap:
    servers: localhost:2181

以及应用测试。yml（这是正确的吗？它应该与application.yml位于同一目录中吗？还不确定嵌入式服务器应该如何使用）：

kafka:
  #  embedded:
  #    enabled: true
  #    topics: promo-api-promotions
  bootstrap:
    servers: localhost:9092

我的测试结果如下：

@MicronautTest
class AppSpec extends Specification {

@Shared
@AutoCleanup
EmbeddedServer server = ApplicationContext.run(EmbeddedServer)

@Shared
private AppClient appClient =
        server.applicationContext.getBean(AppClient)

def 'The upload endpoint is called'() {
  // test here
  appClient.sendMessage(id, event)
  // other test stuff
}

我面临的主要问题是：

>

我的消费者没有从我的主题中消费。我可以看到制作者在Kafka中创建了主题，并且创建了客户端组，但是偏移量保持在0。

当测试启动时，我遇到了问题，看起来好像创建了两个客户端实例，因此MBean注册失败（此外，如果我尝试使用嵌入式Kafka，我会收到另一条消息，关于端口9092已经在使用，因为它尝试启动服务器两次）：

javax.management.InstanceAlreadyExistsExc0019：kafka.consumer： type=app-info， id=app-kafka-client-app-listener atcom.sun.jmx.mbeanserver.Repository.addMBean（Repository.java:437）atcom.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerAnd Repository（DefaultMBeanServerInterceptor.java:1898

Answer 1

您应该添加嵌入式配置kafka。嵌入的已启用，并将其传递到ApplicationContext。运行方法。

Map<String, Object> config = Collections.
    unmodifiableMap(new HashMap<String, Object>() {
        {
            put(AbstractKafkaConfiguration.EMBEDDED, true);
            put(AbstractKafkaConfiguration.EMBEDDED_TOPICS, "test_topic");
    }
});

try (ApplicationContext ctx = ApplicationContext.run(config)) {

消费者在另一个线程中使用Kafka的内容，你必须等待一段时间，直到你的Applister赶上。你可以在KafkaProducerListenerTest中看到一个简短的例子

还记得Micronaut文档：嵌入Kafka中描述的Kafka依赖项吗

Answer 2

设法解决了第二个问题——传入侦听器的对象没有@JsonCreator。我发现这一点是通过在玩游戏时尝试使用Jackson对象映射器从JSON构建对象。

如果其他人有同样的问题——在继续前进之前，确保对象模型与杰克逊一起工作！