Quakus/Smallrye反应性kafka-来自消息的endpoint成功/失败响应

Question

我希望使用动态接受主题作为查询参数的成功/失败响应来响应RESTendpoint。在带有小型反应式消息传递的Quakus中，代码看起来就像下面用OutgoingKafkaRecordMetadata包装有效负载一样

即https：//myendpoint/PublishToKafka？主题=myDynamicTopic

@Channel("test")
Emitter<byte []> kafkaEmitter;

@POST
@Path("/publishToKafka")
public CompletionStage<Void> publishRecord(@QueryParam("topic") String topic, byte [] payload){

    kafkaEmitter.send(Message.of(payload).addMetadata(OutgoingKafkaRecordMetadata.<String>builder()
            .withKey("my-key")
            .withTopic("myDynamicTopic")
            .build()));
    
}

从Quarkus doco“如果endpoint没有返回CompletionStage，HTTP响应可能会在消息发送到Kafka之前写入，因此不会向用户报告失败。”这里的示例描述了当您直接发送有效负载（即发射器）时的这个过程。send（有效负载）返回CompletionStage但不返回发射器。send（message）返回void），但这需要提前配置主题。是否可以使用消息指定元数据，并且仍然使用成功/失败响应来响应调用客户端？（我不介意是使用Emitter和CompletionStage，还是使用CommunityEmitter和Uni）。

如有任何建议，将不胜感激。

Answer 1

因为你使用了一条消息（因为你需要指定主题），所以你需要一些更复杂的东西：

@Channel("test")
Emitter<byte []> kafkaEmitter;

@POST
@Path("/publishToKafka")
public CompletionStage<Void> publishRecord(@QueryParam("topic") String topic, byte [] payload){
    CompletableFuture<Void> future = new CompletableFuture<>();
    Message<byte[]> message = Message.of(payload).addMetadata(OutgoingKafkaRecordMetadata. 
           <String>builder()
            .withKey("my-key")
            .withTopic("myDynamicTopic")
            .build()));
    message = message.withAck(() -> {
         future.complete(null));
         return CompleteableFuture.completedFuture(null);
    }
     .withNack(t -> {
       future.completeExceptionnaly(t));
       return CompleteableFuture.completedFuture(null);
    });
    kafkaEmitter.send(message);
    return future;    
}

在这个片段中，我还附加了在消息被确认（被代理接受）或被拒绝（发生了错误）时调用的ack和nack处理程序。

这些回调将报告给future，这是在方法中创建的一个完整的future。这是要返回的对象，因为它将执行您想要的操作：指示结果。

我知道回调有点复杂。这主要是由于规范：我们必须返回CompleteableFuture.completedFuture（...）；以确认nack进程成功。如果我们返回未来；而不是（我们已将其设置为future.completeExceptionnaly（t））；），这将被解释为nack进程期间的失败。这基本上相当于命令式世界中catch-块中的抛出。

幸运的是，一个更简单的版本很快就会出现（别担心，我们不会崩溃）。