发送消息 Kafka Streams Topology

Question

我想知道是否有办法从流拓扑中删除记录/消息？

我有如下设置：

 builder.stream("my-source-topic")
                .map(CustomMapper)
                .mapValues(CustomValueMapper)
                .filterNot(CustomFilter)
                .transformValues(CustomValueTransformer)
                .toStream()

每个 CustomMapper/CustomFilter 等都会覆盖它们各自的应用/转换方法，它们可能如下所示，如上所述，错误可能无法恢复，这是一个好的解决方案，这些消息将被手动处理并写入相应的日志。 假设在第一个映射期间发生不可恢复的错误，我现在如何防止后续阶段甚至处理记录，我想停止处理此记录并移至下一条记录。

@Override
    public V transform(K readOnlyKey, V value) {
        try {
        // do some logic
        } catch(Exception e){
            // process error - this might be unrecoverable.
            
            dropRecord(); // this is what i would be looking for if possible
        }
    }

我可以杀死线程并让 customUncaughtExceptionHandler 重新调度不会提交偏移量的线程，因此尝试再次处理错误记录。

为传递的对象创建包装器需要在每个处理步骤中添加检查以查看记录是否仍然有效。

在每个处理步骤之前添加 .branch() 也需要大量返工。

Answer 1

您只需返回null 即可在Transformer 中发送消息。请参阅Transformer#transform 的 Javadoc。 所以你的例子是：

    @Override
    public V transform(K readOnlyKey, V value) {
        try {
        // do some logic
        } catch(Exception e){
            // process error - this might be unrecoverable.
            
            return null;
        }
    }

请注意，您目前只能在 Transformer 中执行此操作，但不能在 ValueTransformer 中执行此操作。