本文介绍如何在 spring kafka 中精确测量 acknowledgment.acknowledge() 调用到 commitcallback 执行之间的延迟，通过 offsetandmetadataprovider 在提交上下文中注入时间戳元数据，实现端到端的 offset 提交耗时追踪。

本文介绍如何在 spring kafka 中精确测量 acknowledgment.acknowledge() 调用到 commitcallback 执行之间的延迟，通过 offsetandmetadataprovider 在提交上下文中注入时间戳元数据，实现端到端的 offset 提交耗时追踪。

在 Spring Kafka 批处理 + 异步手动提交（MANUAL_IMMEDIATE）场景下，开发者常需观测「业务处理完成 → 偏移量成功提交」这一关键链路的延迟，用于性能调优与 SLA 监控。但原生 Acknowledgment.acknowledge() 不支持透传上下文数据，CommitCallback 也无法直接关联发起方的起始时间——这正是 OffsetAndMetadataProvider 的设计初衷。

Spring Kafka 2.8.5+ 引入了 OffsetAndMetadataProvider 接口，允许你在每次构建提交请求的 OffsetAndMetadata 对象时注入自定义元数据（如 startTimeMs），该元数据将随 offset 一同被 Kafka 客户端传递至 OffsetCommitCallback，从而打通“发起”与“完成”的时间闭环。

✅ 实现步骤

1. 定义时间戳元数据容器（推荐使用 ByteBuffer 或 Base64 编码字符串，避免 Kafka 元数据长度限制）：

   public class CommitTiming {
       public static final String METADATA_KEY_START_TIME = "start_ts";
       public static byte[] encodeStartTime(long startTimeMs) {
           return ByteBuffer.allocate(Long.BYTES)
                   .putLong(startTimeMs)
                   .array();
       }
       public static long decodeStartTime(byte[] metadata) {
           if (metadata == null || metadata.length < Long.BYTES) return -1;
           return ByteBuffer.wrap(metadata).getLong();
       }
   }

2. 实现 OffsetAndMetadataProvider，记录提交发起时刻：

   @Component
   public class TimingOffsetProvider implements OffsetAndMetadataProvider {
       @Override
       public OffsetAndMetadata provide(ListenerMetadata listenerMetadata, long offset) {
           // 在 acknowledge() 被调用时（即 commit 请求生成时）记录时间
           long startTime = System.currentTimeMillis();
           byte[] metadata = CommitTiming.encodeStartTime(startTime);
           return new OffsetAndMetadata(offset, metadata);
       }
   }

3. 注册 Provider 并配置 ContainerFactory：

   @Bean
   public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory(
           TimingOffsetProvider timingProvider) {
       ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
       factory.setConsumerFactory(consumerFactory());
       factory.setBatchListener(true);
       factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
       factory.getContainerProperties().setSyncCommits(false);
       factory.getContainerProperties().setOffsetAndMetadataProvider(timingProvider); // ? 关键注册
   
       factory.getContainerProperties().setCommitCallback((offsets, ex) -> {
           // 遍历每个 partition 的 offset 提交结果
           offsets.forEach((tp, offsetAndMetadata) -> {
               long startTime = CommitTiming.decodeStartTime(offsetAndMetadata.metadata());
               if (startTime > 0) {
                   long durationMs = System.currentTimeMillis() - startTime;
                   // 上报指标：例如 Micrometer Timer、Prometheus Histogram 或日志
                   commitTimer.record(durationMs, TimeUnit.MILLISECONDS);
                   log.debug("Offset commit for {} took {}ms", tp, durationMs);
               }
           });
   
           if (ex == null) {
               log.info("All offsets committed successfully: {}", offsets);
           } else {
               log.error("Offset commit failed", ex);
           }
       });
   
       return factory;
   }

4. 在监听器中保持原有逻辑不变（acknowledge() 自动携带时间戳）：

   @KafkaListener(topics = "my_topic")
   public void consume(@Payload List<String> messages,
                       @Header(KafkaHeaders.ACKNOWLEDGMENT) Acknowledgment acknowledgment) {
       try {
           for (String message : messages) {
               processMessage(message);
           }
           acknowledgment.acknowledge(); // ✅ 此刻触发 TimingOffsetProvider 提供带时间戳的 OffsetAndMetadata
       } catch (Exception e) {
           log.error("Processing failed", e);
           throw e; // 触发重试或 DLQ，避免错误提交
       }
   }

⚠️ 注意事项

• OffsetAndMetadataProvider 仅在 MANUAL_IMMEDIATE 或 BATCH 模式下生效；MANUAL（延迟提交）不适用。
• Kafka 官方对 metadata 字节数组长度限制为 4096 字节，务必精简元数据（如仅存 long 时间戳）。
• 若监听器启用重试（@RetryableTopic），需确保 OffsetAndMetadataProvider 不在重试过程中重复注入新时间戳（当前机制天然满足：每次 acknowledge() 独立触发 provider）。
• 生产环境建议结合 Micrometer 的 Timer 或 DistributionSummary 上报 commitLatency 指标，并设置告警阈值（如 P99 > 500ms）。

通过该方案，你无需侵入 Kafka 客户端底层，即可在 Spring Kafka 抽象层实现高精度、可观测的异步提交延迟追踪，为消息系统稳定性与性能优化提供可靠数据支撑。