本文介绍在 Spring Kafka 中通过 OffsetAndMetadataProvider 为每次 offset 提交注入时间戳元数据，结合自定义 CommitCallback 实现从调用 acknowledge() 到实际提交完成的端到端延迟测量。
本文介绍在 spring kafka 中通过 `offsetandmetadataprovider` 为每次 offset 提交注入时间戳元数据，结合自定义 `commitcallback` 实现从调用 `acknowledge()` 到实际提交完成的端到端延迟测量。
在 Spring Kafka 中使用异步手动提交（MANUAL_IMMEDIATE + syncCommits=false）时，Acknowledgment.acknowledge() 调用立即返回，而真实提交由后台线程异步执行，并最终触发 CommitCallback。由于 acknowledge() 方法本身不支持传参，无法直接将上下文（如开始时间）透传至回调中——但 Spring Kafka 自 2.8.5 起提供了 OffsetAndMetadataProvider 扩展点，完美解决了这一问题。
核心思路：利用 OffsetAndMetadataProvider 注入时间戳元数据
OffsetAndMetadataProvider 在每次构建 OffsetAndMetadata 对象时被调用，它接收 ListenerMetadata（含监听器 ID、topic/partition 等）和待提交的 offset 值，允许你返回一个携带自定义 metadata 字节数组的 OffsetAndMetadata 实例。我们可以将纳秒级时间戳（如 System.nanoTime()）序列化为 byte[] 存入 metadata，再在 CommitCallback 中反序列化并计算耗时。
✅ 完整实现示例
@Configuration
public class KafkaConfig {
@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory());
factory.setBatchListener(true);
factory.getContainerProperties().setSyncCommits(false);
factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
// ? 关键：注册 OffsetAndMetadataProvider，记录 acknowledge 调用时刻
factory.getContainerProperties().setOffsetAndMetadataProvider((listenerMetadata, offset) -> {
long startTimeNs = System.nanoTime();
// 将时间戳编码为 8 字节 long（需确保字节序一致）
byte[] metadata = ByteBuffer.allocate(8).putLong(startTimeNs).array();
return new OffsetAndMetadata(offset, metadata);
});
// ? 关键：在 CommitCallback 中提取并计算耗时
factory.getContainerProperties().setCommitCallback((offsets, ex) -> {
if (ex == null) {
offsets.forEach((tp, offsetAndMetadata) -> {
byte[] metadata = offsetAndMetadata.metadata();
if (metadata != null && metadata.length == 8) {
long startTimeNs = ByteBuffer.wrap(metadata).getLong();
long endTimeNs = System.nanoTime();
double durationMs = (endTimeNs - startTimeNs) / 1_000_000.0;
// ✅ 发布监控指标（如 Micrometer Timer）
Timer.builder("kafka.offset.commit.duration")
.tag("topic", tp.topic())
.tag("partition", String.valueOf(tp.partition()))
.register(Metrics.globalRegistry)
.record(durationMs, TimeUnit.MILLISECONDS);
log.debug("Offset commit for {} took {:.2f} ms", tp, durationMs);
}
});
log.info("Successfully committed offsets: {}", offsets);
} else {
log.error("Failed to commit offsets: {}", offsets, ex);
}
});
return factory;
}
@KafkaListener(topics = "my_topic")
public void consume(@Payload List<String> messages,
@Header(KafkaHeaders.ACKNOWLEDGMENT) Acknowledgment acknowledgment) {
long startProcessNs = System.nanoTime(); // 可选：记录业务处理开始时间
try {
for (String message : messages) {
// ✅ 业务逻辑处理（建议添加异常防护）
processMessage(message);
}
} finally {
// ⚠️ 必须在此处调用 acknowledge() —— 此刻即“计时起点”
acknowledgment.acknowledge();
// 注意：不要在此处 sleep 或阻塞，否则影响吞吐
}
}
private void processMessage(String msg) {
// your business logic
}
}⚠️ 注意事项与最佳实践
- 线程安全：OffsetAndMetadataProvider 和 CommitCallback 均在 Kafka 客户端线程中执行，请避免在其中执行耗时或阻塞操作（如远程调用、同步 I/O）。指标采集应使用非阻塞方式（如 Micrometer 的 Timer.record()）。
- metadata 大小限制：Kafka 协议对 metadata 字段长度有限制（默认 ≤ 4KB），仅建议存储轻量上下文（如时间戳、trace ID），避免序列化大对象。
- 精度选择：System.nanoTime() 提供高精度单调时间，适合测量间隔；若需跨服务对齐，可改用 System.currentTimeMillis() 并注意时钟漂移。
- 错误处理：CommitCallback 中 ex != null 表示提交失败（如网络超时、Broker 不可用），此时应记录告警并考虑重试策略（Spring Kafka 默认自动重试，可通过 setCommitRetries() 配置）。
- 批处理语义：MANUAL_IMMEDIATE 模式下，每个 acknowledge() 触发一次提交请求，对应 CommitCallback 中的一个 Map<TopicPartition, OffsetAndMetadata>。若需按 partition 统计，可遍历该 map。
通过该方案，你不仅能精确测量 offset 提交延迟，还可扩展支持 trace 上下文透传、业务链路埋点等高级可观测性需求，是 Spring Kafka 生产环境性能调优与稳定性保障的关键实践之一。
