写够副本才算数，处理完了再挪书签

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      acks: all                     # ISR 里的副本都写到才算成功
      properties:
        enable.idempotence: true    # 重试不会写出重复
    consumer:
      auto-offset-reset: earliest   # 新消费组从头读，不漏上线前的消息
      enable-auto-commit: false
    listener:
      ack-mode: manual              # 业务处理完，手动提交 offset

package demo;

import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.TopicBuilder;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RestController;

@SpringBootApplication
@RestController
public class OrderApp {

    private static final Logger log = LoggerFactory.getLogger(OrderApp.class);
    private final KafkaTemplate<String, String> kafka;

    public OrderApp(KafkaTemplate<String, String> kafka) {
        this.kafka = kafka;
    }

    public static void main(String[] args) {
        SpringApplication.run(OrderApp.class, args);
    }

    /** 启动时建 topic：3 个分区。单机只能 1 个副本，生产环境是 3 */
    @Bean
    NewTopic orderCreated() {
        return TopicBuilder.name("order-created").partitions(3).replicas(1).build();
    }

    /** 生产者：订单号当 key，同一个订单永远进同一个分区 */
    @PostMapping("/orders/{id}")
    public String create(@PathVariable String id) {
        kafka.send("order-created", id, "{\"orderId\":\"" + id + "\"}")
             .whenComplete((r, ex) -> {
                 if (ex != null) {
                     log.error("发送失败 order={}", id, ex);          // 第⑥步：落库，稍后重发
                 } else {
                     log.info("已写入 order={} 分区={} offset={}", id,
                             r.getRecordMetadata().partition(), r.getRecordMetadata().offset());
                 }
             });
        return "ok\n";
    }

    /** 消费组 inventory：开几个实例，3 个分区就在它们之间分 */
    @KafkaListener(topics = "order-created", groupId = "inventory")
    public void deductStock(ConsumerRecord<String, String> rec, Acknowledgment ack) {
        log.info("[库存] 扣减 order={} 分区={} offset={}", rec.key(), rec.partition(), rec.offset());
        ack.acknowledge();                                           // 处理完再提交
    }

    /** 消费组 points：另一个组，每条消息再收一份 */
    @KafkaListener(topics = "order-created", groupId = "points")
    public void addPoints(ConsumerRecord<String, String> rec, Acknowledgment ack) {
        log.info("[积分] 加积分 order={} 分区={}", rec.key(), rec.partition());
        ack.acknowledge();
    }
}