用弹簧云流处理部分失败的批次

Question

当将Spring-Cloud-tream用于流式应用程序（功能样式）带有批处理时，是否有一种方法可以重试/DLQ失败的消息，但还可以处理（流）非拖放记录？

例如： 功能收到了10个记录的批次，并尝试将其转换为其他类型并返回新记录以生产。假设纪录8未能在映射时失败，是否可以完成记录0-7的生成，然后重试/DLQ记录8？

用索引将批处理列列列列德列德德式置换不会导致先前的消息发送。

---

Spring Kafka版本：2.8.0

代码：

        @Override
public List<Message<Context>> apply(Message<List<Context>> listMessage) {

    List<Message<Context>> output = new ArrayList<>();

    IntStream.range(0, listMessage.getPayload().size()).forEach(index -> {
        try {

            Record<Context> record = Record.fromBatch(listMessage, index);
            output.add(MessageBuilder.withPayload(record.getValue()).build());
            if (index == listMessage.getPayload().size() - 1) {
                throw new TransientError("offset " + record.getOffset() + "failed", new RuntimeException());
            }
        } catch (Exception e) {
          
            throw new BatchListenerFailedException("Tigger retry", e, index);
        }
    });

    return output;
}

定制器：

private CommonErrorHandler getCommonErrorHandler(String group) {
        DefaultErrorHandler errorHandler = new DefaultErrorHandler(getRecoverer(group), getBackOff());
        errorHandler.setLogLevel(KafkaException.Level.DEBUG);
        errorHandler.setAckAfterHandle(true);
        errorHandler.setClassifications(Map.of(
                        PermanentError.class, false,
                        TransientError.class, true,
                        SerializationException.class, properties.isRetryDesErrors()),
                false);
        errorHandler.setRetryListeners(getRetryListener());
        return errorHandler;
    }

    private ConsumerRecordRecoverer getRecoverer(String group) {
        KafkaOperations<?, ?> operations = new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(producerProperties()));
        DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(
                operations, getDestinationResolver(group));
        recoverer.setHeadersFunction(this::buildAdditionalHeaders);
        return recoverer;
    }

YAML：

spring:
  cloud:
    function:
      definition: batchFunc
    stream:
      default-binder: kafka-string-avro
      binders:
        kafka-string-avro:
          type: kafka
          environment.spring.cloud.stream.kafka.binder.consumerProperties:
            key.deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
            value.deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
            spring.deserializer.key.delegate.class: org.apache.kafka.common.serialization.StringDeserializer
            spring.deserializer.value.delegate.class: io.confluent.kafka.serializers.KafkaAvroDeserializer
            schema.registry.url: ${SCHEMA_REGISTRY_URL:http://localhost:8081}
            value.subject.name.strategy: io.confluent.kafka.serializers.subject.TopicNameStrategy
            specific.avro.reader: true
          environment.spring.cloud.stream.kafka.binder.producerProperties:
            key.serializer: org.apache.kafka.common.serialization.StringSerializer
            value.serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
            schema.registry.url: ${SCHEMA_REGISTRY_URL:http://localhost:8081}
            value.subject.name.strategy: io.confluent.kafka.serializers.subject.TopicNameStrategy
      bindings:
        batchFunc-in-0:
          binder: kafka-string-avro
          destination: records.publisher.output
          group: function2-in-group
          contentType: application/*+avro
          consumer:
            batchMode: true
        function2-out-0:
          binder: kafka-string-avro
          destination: reporter.output
          producer:
            useNativeEncoding: true
      kafka:
        binder:
          brokers: ${KAFKA_BROKER_ADDRESS:localhost:9092}
          autoCreateTopics: ${KAFKA_AUTOCREATE_TOPICS:false}
        default:
          consumer:
            startOffset: ${START_OFFSET:latest}
            enableDlq: false
      default:
        consumer:
          maxAttempts: 1
          defaultRetryable: false