第九章：完整示例 —— 订单服务发事件，库存服务消费并手动提交位点

bin/kafka-topics.sh \
  --create \
  --topic order-events \
  --bootstrap-server localhost:9092 \
  --partitions 3 \
  --replication-factor 1

import java.util.Properties;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

public class OrderCreatedProducer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        // 生产环境常见基线：更强确认 + 幂等生产
        props.put("acks", "all");
        props.put("enable.idempotence", "true");

        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
            String orderId = "order-1001";
            String event = "{\"eventType\":\"OrderCreated\",\"orderId\":\"order-1001\",\"userId\":\"u-9\",\"amount\":199}";

            ProducerRecord<String, String> record =
                new ProducerRecord<>("order-events", orderId, event);

            producer.send(record, (metadata, exception) -> {
                if (exception != null) {
                    exception.printStackTrace();
                    return;
                }

                System.out.println(
                    "sent to topic=" + metadata.topic() +
                    ", partition=" + metadata.partition() +
                    ", offset=" + metadata.offset()
                );
            });

            producer.flush();
        }
    }
}

import java.time.Duration;
import java.util.Collections;
import java.util.Map;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;

public class InventoryConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "inventory-service");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        // 关闭自动提交，自己控制“处理成功后再提交”
        props.put("enable.auto.commit", "false");

        // 新 group 第一次启动时，从最早位置开始读
        props.put("auto.offset.reset", "earliest");

        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
            consumer.subscribe(Collections.singletonList("order-events"));

            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));

                for (ConsumerRecord<String, String> record : records) {
                    try {
                        System.out.println(
                            "receive topic=" + record.topic() +
                            ", partition=" + record.partition() +
                            ", offset=" + record.offset() +
                            ", key=" + record.key() +
                            ", value=" + record.value()
                        );

                        // 这里模拟库存预留逻辑
                        reserveInventory(record.key(), record.value());

                        // 处理成功后，提交“下一条要读的 offset”
                        TopicPartition tp = new TopicPartition(record.topic(), record.partition());
                        OffsetAndMetadata next = new OffsetAndMetadata(record.offset() + 1);
                        consumer.commitSync(Map.of(tp, next));
                    } catch (Exception e) {
                        System.err.println("process failed, will be retried after restart: " + e.getMessage());
                    }
                }
            }
        }
    }

    private static void reserveInventory(String orderId, String eventJson) {
        // 示例里只打印；真实系统这里应该做幂等控制
        System.out.println("reserve inventory for " + orderId + " with " + eventJson);
    }
}

bin/kafka-consumer-groups.sh \
  --bootstrap-server localhost:9092 \
  --describe \
  --group inventory-service