Java微服务中如何检测Kafka消费组所有实例完成指定主题消息消费？

1. 标记当日消息批次边界 #

生产者在拆分文件发送Kafka消息时，需要为当日批次的消息添加唯一标识和边界标记：

• 为每条消息的消息头添加daily-batch-id，值为当日日期（如2024-05-20），用于区分不同日期的批次。
• 发送当日批次的第一条消息时，添加头batch-type: start；最后一条消息添加头batch-type: end，同时在消息体中携带当日批次对应的Kafka主题各分区的结束偏移量（可通过生产者的RecordMetadata获取发送后的偏移量汇总）。
• 将当日批次的daily-batch-id、各分区结束偏移量存储到数据库或OpenShift ConfigMap中，供后续检测使用。

2. 消费组偏移量检测逻辑 #

在Java微服务中实现定时或事件驱动的检测任务，通过Kafka AdminClient跟踪消费组的偏移量：

• 使用org.apache.kafka.clients.admin.AdminClient获取指定消费组的偏移量信息：

AdminClient adminClient = AdminClient.create(adminConfigs);
ConsumerGroupOffsets offsets = adminClient.listConsumerGroupOffsets("your-consumer-group-id").all().get();
// 遍历消费组订阅的所有分区，获取当前已提交的偏移量
for (Map.Entry<TopicPartition, OffsetAndMetadata> entry : offsets.partitionOffsetAndMetadata().entrySet()) {
    TopicPartition tp = entry.getKey();
    long committedOffset = entry.getValue().offset();
    // 从存储中获取该分区当日批次的结束偏移量
    long batchEndOffset = getBatchEndOffset(tp.topic(), tp.partition(), dailyBatchId);
    // 对比偏移量：已提交偏移量 >= 批次结束偏移量，则该分区处理完成
    boolean partitionCompleted = committedOffset >= batchEndOffset;
    // 汇总所有分区的完成状态
}

• 当所有分区的已提交偏移量都达到或超过当日批次的结束偏移量时，判定消费组已完成当日所有消息的消费。

3. 触发OpenShift Job #

通过Kubernetes Java客户端（如fabric8io的kubernetes-client）创建Job：

• 引入依赖（以Maven为例）：

<dependency>
    <groupId>io.fabric8</groupId>
    <artifactId>kubernetes-client</artifactId>
    <version>6.10.0</version>
</dependency>

• 编写创建Job的代码：

try (KubernetesClient client = new DefaultKubernetesClient()) {
    Job job = new JobBuilder()
        .withNewMetadata()
            .withName("daily-batch-job-" + dailyBatchId)
            .withNamespace("your-namespace")
            .addToLabels("batch-id", dailyBatchId)
        .endMetadata()
        .withNewSpec()
            .withNewTemplate()
                .withNewSpec()
                    .addNewContainer()
                        .withName("batch-processor")
                        .withImage("your-job-image:latest")
                        .addNewEnv()
                            .withName("BATCH_ID")
                            .withValue(dailyBatchId)
                        .endEnv()
                    .endContainer()
                    .withRestartPolicy("OnFailure")
                .endSpec()
            .endTemplate()
            .withBackoffLimit(3)
        .endSpec()
        .build();
    client.batch().jobs().inNamespace("your-namespace").create(job);
}

• 确保Java服务的ServiceAccount拥有创建Job的权限（通过OpenShift的Role/RoleBinding配置）。

4. 幂等性与异常处理 #

• 幂等性保障：在数据库中维护当日批次的触发状态（如pending/triggered/completed），仅当状态为pending时才触发Job，触发后更新状态为triggered，避免重复创建。
• 超时告警：设置超时阈值（如当日24:00后1小时），若到时间仍有分区未完成消费，触发告警通知运维排查。
• 失败消息处理：将消费失败的消息转发到死信队列（DLQ），并在检测逻辑中排除DLQ的偏移量，确保主主题的当日消息都被处理后再触发Job。

内容的提问来源于stack exchange，提问作者Swarup Saha