Spring Cloud Stream 功能手册致谢 - KafkaHeaders.ACKNOWLEDGMENT 不可用答案

【问题标题】：Spring Cloud Stream Functional Manual Acknowledgement - KafkaHeaders.ACKNOWLEDGMENT not availableSpring Cloud Stream 功能手册致谢 - KafkaHeaders.ACKNOWLEDGMENT 不可用
【发布时间】：2020-08-05 01:49:52
【问题描述】：

我正在使用 Spring Cloud Webflux 和 Spring Cloud Streams 功能接口来处理我的 kafka 处理。

如果我按顺序进行处理并且如果我终止应用程序，它会返回消息进行处理，因为没有消息丢失，所以按预期工作。但是，如果我尝试并行处理，它似乎是在承认 Kafka，这是可以理解的，因为它现在是一个单独的线程，因此想要转向手动确认。

我的代码：

application.yml（相关部分）

spring:
  cloud:
    stream:
      kafka:
        binder:
          brokers: localhost:9092
          autoAddPartitions: true
          minPartitionCount: 2         
      bindings:
          receiver-in-0:
               binder: kafka
              destination: topic-1
              content-type: text/plain;charset=UTF-8
              group: input-group-1
              consumer:
                  autoCommitOffset: false
spring.cloud.stream.function.definition: receiver

收货人代码

public Consumer<Flux<Message<String>>> receiver() throws IOException {
        return (sink -> {
            sink
            .onBackpressureBuffer()
            .parallel(4)
            .runOn(Schedulers.parallel())
            .subscribe((record)->{  
                Flux<Action> executor = new 
                           //Internal code which does transformation and provides a flux for execution (names changed)
 IncomingMessage().process(record);

                if(executor != null) {
                    Disposable disposable=null;
                    disposable= executor.subscribe(
                            (action)->{

                                try {
                                   //Process execute does the processing on the modified data (names changed)
                                    Process.execute(action);
                                    Acknowledgment acknowledgment = record.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class);
                                    if(acknowledgment !=null) {
                                        acknowledgment.acknowledge();
                                    }
                                }
                                catch(Exception e) {

                                log.fatal(e.getMsg());
                                }
                            },
                            (e)->{

                                log.fatal(e.getMsg());
                                }
                            });
                    if(disposable != null) {
                        disposable.dispose();
                    }

                }

            });

        });
    }

这里record.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class); 行总是给出null，所以我假设autoCommitOffset: false 不起作用，我尝试将以下配置也放在绑定部分但无济于事。

 receiver-in-0:
               binder: kafka
              destination: topic-1
              content-type: text/plain;charset=UTF-8
              group: input-group-1
              autoCommitOffset: false

我的要求是，如果我终止应用程序，即使在并行情况下，它也应该继续从第一个未确认消息中读取消息。

【问题讨论】：

请不要在 gitter 和 SO 之间交叉发布相同的问题。这是在浪费大家的时间。
@Oleg Zhurakousky：我没有。你可以在gitter中分享链接吗？如果它正在发布，我会遵循。
Kafka 的异步处理 (Flux) 充满了困难 - Kafka 不承认离散记录；它只为消费者组维护一个分区偏移量。如果他记录@2先完成，而在处理@1之前服务器出现故障；你将失去记录@1。
所以如果我理解，我不应该对记录进行并行处理，虽然异步仍然是顺序的，但 Flux？

标签： kafka-consumer-api spring-webflux spring-cloud-stream

【解决方案1】：

没有得到确认头的问题是由于标签的位置不正确。因为它纯粹是 kafka binder 属性。添加了以下属性

spring:
  cloud:
    stream:
      kafka:
        bindings:
           receiver-in-0:
              consumer:
                   autoCommitOffset: false

通道名称应与函数调用中的通道名称相同。或者默认，可以设置。

 spring:
    cloud:
      stream:
        kafka:
           default:
            consumer:
               autoCommitOffset: false

但是，并行处理传入的通量可能不是一个好主意，因为消息可能会被丢弃，因为稍后的消息可能会被确认。它需要更多的逻辑来确认，而不仅仅是设置参数确认。

【讨论】：