反序列化错误后如何提交偏移量？SpringKafka

Question

我能够使用ErrorDesrializationHandler成功处理反序列化错误，但当我重新启动我的消费者时，它再次开始重新处理由于反序列化而导致的所有失败消息。
由于反序列化异常无法到达Kafka Listener，如何确认并提交偏移量？
谢谢。

我正在使用的自定义错误处理程序：

 class KafkaErrorHandler implements ConsumerAwareErrorHandler{

private static final Logger LOG = LoggerFactory.getLogger(KafkaErrorHandler.class);

@Override
public void handle(Exception thrownException, List<ConsumerRecord<?, ?>> records, Consumer<?, ?> consumer, MessageListenerContainer container) {
    if (!records.isEmpty()) {
        ConsumerRecord<?, ?> record = records.get(0);
        String topic = record.topic();
        long offset = record.offset();
        int partition = record.partition();
        if (thrownException.getClass().equals(DeserializationException.class)) {
            DeserializationException exception = (DeserializationException) thrownException;
            String malformedMessage = new String(exception.getData());
            LOG.info("Skipping message with topic {} and offset {} " +
                    "- malformed message: {} , exception: {}", topic, offset, malformedMessage, exception.getLocalizedMessage());
        } else {
            LOG.info("Skipping message with topic {} - offset {} - partition {} - exception {}", topic, offset, partition, thrownException);
        }
    } else {
        LOG.info("Consumer exception - cause: {}", thrownException.getMessage());
    }
}

}

Answer 1

使用<代码>确认模式。手动_立即，请参阅

/**
 * Set to true to commit the offset for a recovered record.
 * The container must be configured with
 * {@link org.springframework.kafka.listener.ContainerProperties.AckMode#MANUAL_IMMEDIATE}.
 * Whether or not the commit is sync or async depends on the container's syncCommits
 * property.
 * @param commitRecovered true to commit.
 */
@Override
public void setCommitRecovered(boolean commitRecovered) { // NOSONAR enhanced javadoc
    super.setCommitRecovered(commitRecovered);
}

在DefaultErrorHandler上。

在确认模式下不可能。手册；对于其他AckModes，默认情况下，容器将提交恢复记录的偏移量（基于isAckAfterHandle（））。

编辑

提交偏移量的逻辑位于SeekUtils.seekOrRecover（在查找之后）。

if (commitRecovered) {
    if (container.getContainerProperties().getAckMode().equals(AckMode.MANUAL_IMMEDIATE)) {
        ConsumerRecord<?, ?> record = records.get(0);
        Map<TopicPartition, OffsetAndMetadata> offsetToCommit = Collections.singletonMap(
                new TopicPartition(record.topic(), record.partition()),
                ListenerUtils.createOffsetAndMetadata(container, record.offset() + 1));
        if (container.getContainerProperties().isSyncCommits()) {
            consumer.commitSync(offsetToCommit, container.getContainerProperties().getSyncCommitTimeout());
        }
        else {
            OffsetCommitCallback commitCallback = container.getContainerProperties().getCommitCallback();
            if (commitCallback == null) {
                commitCallback = LOGGING_COMMIT_CALLBACK;
            }
            consumer.commitAsync(offsetToCommit, commitCallback);
        }
    }
    else {
        logger.debug(() -> "'commitRecovered' ignored, container AckMode must be MANUAL_IMMEDIATE, not "
                + container.getContainerProperties().getAckMode());
    }
}