一個 Kafka 叢集通常由 3 個以上的 Broker 組成

建議配置：
- Broker 數量 ≥ 3
- Replication Factor = 3
- Min ISR = 2
- acks = all（或 -1）

可容忍故障數 = Replication Factor - Min ISR
例如：RF=3, Min ISR=2 → 可容忍 1 台 Broker 故障

Kafka 預設使用以下 Port：
- 9092：Broker 對外服務（PLAINTEXT）
- 9093：Broker 對外服務（SSL）
- 9094：Broker 間通訊
- 2181：ZooKeeper（若使用）

# 下載 Kafka
cd /opt
wget https://downloads.apache.org/kafka/3.7.0/kafka_2.13-3.7.0.tgz

# 解壓縮
tar -xzf kafka_2.13-3.7.0.tgz
ln -s kafka_2.13-3.7.0 kafka

# 設定環境變數
echo 'export KAFKA_HOME=/opt/kafka' >> ~/.bashrc
echo 'export PATH=$PATH:$KAFKA_HOME/bin' >> ~/.bashrc
source ~/.bashrc

# 產生唯一的 Cluster ID
KAFKA_CLUSTER_ID="$(kafka-storage.sh random-uuid)"
echo $KAFKA_CLUSTER_ID
# 輸出範例：MkU3OEVBNTcwNTJENDM2Qk

# 節點 ID（每個節點必須唯一）
node.id=1

# 節點角色：broker, controller, 或兩者
process.roles=broker,controller

# Controller 選舉投票者
controller.quorum.voters=1@localhost:9093

# 監聽位址
listeners=PLAINTEXT://:9092,CONTROLLER://:9093
advertised.listeners=PLAINTEXT://localhost:9092

# 日誌目錄
log.dirs=/var/kafka-logs

# Controller 監聽設定
controller.listener.names=CONTROLLER

# 預設 Partition 數量
num.partitions=3

# 預設副本因子
default.replication.factor=1

# 日誌保留時間（168 小時 = 7 天）
log.retention.hours=168

# 建立日誌目錄
sudo mkdir -p /var/kafka-logs
sudo chown -R $USER:$USER /var/kafka-logs

# 格式化儲存目錄
kafka-storage.sh format -t $KAFKA_CLUSTER_ID \
  -c /opt/kafka/config/kraft/server.properties

# 前景執行（除錯用）
kafka-server-start.sh /opt/kafka/config/kraft/server.properties

# 背景執行
kafka-server-start.sh -daemon /opt/kafka/config/kraft/server.properties

# 檢查是否啟動成功
jps -l | grep kafka
# 應該看到：kafka.Kafka

# 建立測試 Topic
kafka-topics.sh --create \
  --topic test-topic \
  --bootstrap-server localhost:9092 \
  --partitions 3 \
  --replication-factor 1

# 列出 Topic
kafka-topics.sh --list --bootstrap-server localhost:9092

# 發送測試訊息
echo "Hello Kafka" | kafka-console-producer.sh \
  --topic test-topic \
  --bootstrap-server localhost:9092

# 消費測試訊息
kafka-console-consumer.sh \
  --topic test-topic \
  --from-beginning \
  --bootstrap-server localhost:9092

node.id=1
process.roles=broker,controller
controller.quorum.voters=1@192.168.1.101:9093,2@192.168.1.102:9093,3@192.168.1.103:9093

listeners=PLAINTEXT://:9092,CONTROLLER://:9093
advertised.listeners=PLAINTEXT://192.168.1.101:9092

controller.listener.names=CONTROLLER
inter.broker.listener.name=PLAINTEXT

log.dirs=/var/kafka-logs

num.partitions=6
default.replication.factor=3
min.insync.replicas=2

log.retention.hours=168
log.segment.bytes=1073741824

node.id=2
advertised.listeners=PLAINTEXT://192.168.1.102:9092

node.id=3
advertised.listeners=PLAINTEXT://192.168.1.103:9092

# 1. 在任一節點產生 Cluster ID
KAFKA_CLUSTER_ID="$(kafka-storage.sh random-uuid)"
echo $KAFKA_CLUSTER_ID
# 記下此 ID，所有節點使用相同的 ID

# 2. 在每個節點執行格式化（使用相同的 CLUSTER_ID）
kafka-storage.sh format -t $KAFKA_CLUSTER_ID \
  -c /opt/kafka/config/kraft/server.properties

# 3. 在每個節點啟動 Kafka
kafka-server-start.sh -daemon /opt/kafka/config/kraft/server.properties

# 4. 驗證叢集狀態
kafka-metadata.sh --snapshot /var/kafka-logs/__cluster_metadata-0/00000000000000000000.log \
  --command "describe"

############################# Server Basics #############################

# Broker 唯一識別碼
broker.id=1

# 節點角色（KRaft 模式）
process.roles=broker,controller

# 日誌儲存目錄（可設定多個，以逗號分隔）
log.dirs=/var/kafka-logs,/var/kafka-logs2

# 監聽位址
listeners=PLAINTEXT://0.0.0.0:9092
advertised.listeners=PLAINTEXT://kafka-broker-1:9092

############################# Performance #############################

# 網路請求處理執行緒數
num.network.threads=8

# I/O 處理執行緒數
num.io.threads=16

# Socket 緩衝區大小
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400

# 請求最大大小
socket.request.max.bytes=104857600

# 批次接收的訊息數量
queued.max.requests=500

############################# Log Configuration #############################

# 預設 Partition 數量
num.partitions=6

# 預設副本因子
default.replication.factor=3

# 最小同步副本數
min.insync.replicas=2

# 日誌區段大小（1 GB）
log.segment.bytes=1073741824

# 日誌保留時間（7 天）
log.retention.hours=168

# 日誌保留大小（-1 表示無限制）
log.retention.bytes=-1

# 日誌清理檢查間隔（5 分鐘）
log.retention.check.interval.ms=300000

# 日誌清理策略：delete 或 compact
log.cleanup.policy=delete

Partition 數量 ≥ max(預期吞吐量 / 單一 Partition 吞吐量, Consumer 數量)

# 建議設定
default.replication.factor=3
min.insync.replicas=2

Properties props = new Properties();

// 基本連線設定
props.put("bootstrap.servers", "kafka-1:9092,kafka-2:9092,kafka-3:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

// acks 設定：訊息確認機制
// 0: 不等待確認（最快，可能遺失）
// 1: Leader 確認（平衡）
// all/-1: 所有 ISR 確認（最安全）
props.put("acks", "all");

// 重試設定
props.put("retries", 3);
props.put("retry.backoff.ms", 1000);

// 批次設定
props.put("batch.size", 16384);           // 批次大小（bytes）
props.put("linger.ms", 5);                // 批次等待時間（ms）
props.put("buffer.memory", 33554432);     // 緩衝區大小（32 MB）

// 冪等性設定（避免重複）
props.put("enable.idempotence", true);

// 壓縮設定
props.put("compression.type", "snappy");  // none, gzip, snappy, lz4, zstd

Properties props = new Properties();

// 基本連線設定
props.put("bootstrap.servers", "kafka-1:9092,kafka-2:9092,kafka-3:9092");
props.put("group.id", "order-processing-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

// Offset 重置策略
// earliest: 從最早的訊息開始
// latest: 從最新的訊息開始
// none: 如果沒有 offset 則拋出例外
props.put("auto.offset.reset", "earliest");

// 自動提交設定
props.put("enable.auto.commit", false);   // 建議手動提交
props.put("auto.commit.interval.ms", 5000);

// 拉取設定
props.put("max.poll.records", 500);       // 單次拉取最大筆數
props.put("max.poll.interval.ms", 300000); // 兩次 poll 最大間隔
props.put("fetch.min.bytes", 1);          // 最小拉取資料量
props.put("fetch.max.wait.ms", 500);      // 最大等待時間

// Session 設定
props.put("session.timeout.ms", 45000);   // Session 超時
props.put("heartbeat.interval.ms", 15000); // 心跳間隔

// 隔離級別（配合 Exactly Once）
props.put("isolation.level", "read_committed");

############################# Retention Policy #############################

# 依時間保留（預設 7 天）
log.retention.hours=168
# 或更精確的設定
log.retention.ms=604800000

# 依大小保留（-1 表示無限制）
log.retention.bytes=-1

# 日誌區段大小（觸發 Rolling）
log.segment.bytes=1073741824

# 日誌區段時間（7 天）
log.roll.hours=168

# 清理策略
# delete: 刪除過期資料
# compact: 壓縮（保留每個 Key 的最新值）
# compact,delete: 先壓縮後刪除
log.cleanup.policy=delete

# 基本建立
kafka-topics.sh --create \
  --topic order-events \
  --bootstrap-server localhost:9092 \
  --partitions 6 \
  --replication-factor 3

# 建立時指定額外設定
kafka-topics.sh --create \
  --topic user-activities \
  --bootstrap-server localhost:9092 \
  --partitions 12 \
  --replication-factor 3 \
  --config retention.ms=259200000 \
  --config cleanup.policy=compact

# 列出所有 Topic
kafka-topics.sh --list --bootstrap-server localhost:9092

# 查看 Topic 詳細資訊
kafka-topics.sh --describe \
  --topic order-events \
  --bootstrap-server localhost:9092

# 輸出範例：
# Topic: order-events	TopicId: ABC123...
# PartitionCount: 6	ReplicationFactor: 3
# Configs: retention.ms=604800000
#   Partition: 0	Leader: 1	Replicas: 1,2,3	Isr: 1,2,3
#   Partition: 1	Leader: 2	Replicas: 2,3,1	Isr: 2,3,1
#   ...

# 增加 Partition（只能增加，不能減少）
kafka-topics.sh --alter \
  --topic order-events \
  --partitions 12 \
  --bootstrap-server localhost:9092

# 修改 Topic 設定
kafka-configs.sh --alter \
  --entity-type topics \
  --entity-name order-events \
  --add-config retention.ms=172800000 \
  --bootstrap-server localhost:9092

# 查看 Topic 設定
kafka-configs.sh --describe \
  --entity-type topics \
  --entity-name order-events \
  --bootstrap-server localhost:9092

# 刪除 Topic
kafka-topics.sh --delete \
  --topic test-topic \
  --bootstrap-server localhost:9092

# 注意：確保 delete.topic.enable=true（預設為 true）

# 基本發送
kafka-console-producer.sh \
  --topic order-events \
  --bootstrap-server localhost:9092

# 帶 Key 發送
kafka-console-producer.sh \
  --topic order-events \
  --bootstrap-server localhost:9092 \
  --property "parse.key=true" \
  --property "key.separator=:"

# 輸入格式：key:value
# order-001:{"orderId":"001","amount":100}

import org.apache.kafka.clients.producer.*;
import java.util.Properties;

public class OrderProducer {
    
    public static void main(String[] args) {
        // 設定 Producer 參數
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, 
                  "kafka-1:9092,kafka-2:9092,kafka-3:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, 
                  "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, 
                  "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        props.put(ProducerConfig.RETRIES_CONFIG, 3);
        
        // 建立 Producer
        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
            
            // 建立訊息
            String topic = "order-events";
            String key = "order-001";
            String value = "{\"orderId\":\"001\",\"amount\":100}";
            
            ProducerRecord<String, String> record = 
                new ProducerRecord<>(topic, key, value);
            
            // 同步發送
            RecordMetadata metadata = producer.send(record).get();
            System.out.printf("Sent to partition %d, offset %d%n", 
                              metadata.partition(), metadata.offset());
            
            // 非同步發送（建議）
            producer.send(record, (meta, exception) -> {
                if (exception != null) {
                    exception.printStackTrace();
                } else {
                    System.out.printf("Sent to partition %d, offset %d%n", 
                                      meta.partition(), meta.offset());
                }
            });
            
            // 確保所有訊息發送完成
            producer.flush();
        }
    }
}

# 從最新訊息開始消費
kafka-console-consumer.sh \
  --topic order-events \
  --bootstrap-server localhost:9092

# 從最早訊息開始消費
kafka-console-consumer.sh \
  --topic order-events \
  --from-beginning \
  --bootstrap-server localhost:9092

# 指定 Consumer Group
kafka-console-consumer.sh \
  --topic order-events \
  --group order-processor \
  --bootstrap-server localhost:9092

# 顯示 Key 和 Timestamp
kafka-console-consumer.sh \
  --topic order-events \
  --bootstrap-server localhost:9092 \
  --property print.key=true \
  --property print.timestamp=true

import org.apache.kafka.clients.consumer.*;
import java.time.Duration;
import java.util.*;

public class OrderConsumer {
    
    public static void main(String[] args) {
        // 設定 Consumer 參數
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, 
                  "kafka-1:9092,kafka-2:9092,kafka-3:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-processing-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, 
                  "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, 
                  "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100);
        
        // 建立 Consumer
        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
            
            // 訂閱 Topic
            consumer.subscribe(Collections.singletonList("order-events"));
            
            // 持續消費
            while (true) {
                ConsumerRecords<String, String> records = 
                    consumer.poll(Duration.ofMillis(1000));
                
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("Partition: %d, Offset: %d, Key: %s, Value: %s%n",
                                      record.partition(), 
                                      record.offset(), 
                                      record.key(), 
                                      record.value());
                    
                    // 處理業務邏輯
                    processOrder(record.value());
                }
                
                // 手動提交 Offset
                consumer.commitSync();
            }
        }
    }
    
    private static void processOrder(String orderJson) {
        // 業務處理邏輯
        System.out.println("Processing: " + orderJson);
    }
}

# 列出所有 Consumer Group
kafka-consumer-groups.sh --list --bootstrap-server localhost:9092

# 查看特定 Group 的 Offset 狀態
kafka-consumer-groups.sh --describe \
  --group order-processing-group \
  --bootstrap-server localhost:9092

# 輸出範例：
# GROUP                  TOPIC          PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
# order-processing-group order-events   0          1000            1050            50
# order-processing-group order-events   1          980             980             0
# order-processing-group order-events   2          1100            1200            100

# 重置到最早（需先停止 Consumer）
kafka-consumer-groups.sh --reset-offsets \
  --group order-processing-group \
  --topic order-events \
  --to-earliest \
  --execute \
  --bootstrap-server localhost:9092

# 重置到最新
kafka-consumer-groups.sh --reset-offsets \
  --group order-processing-group \
  --topic order-events \
  --to-latest \
  --execute \
  --bootstrap-server localhost:9092

# 重置到特定 Offset
kafka-consumer-groups.sh --reset-offsets \
  --group order-processing-group \
  --topic order-events:0 \
  --to-offset 500 \
  --execute \
  --bootstrap-server localhost:9092

# 重置到特定時間
kafka-consumer-groups.sh --reset-offsets \
  --group order-processing-group \
  --topic order-events \
  --to-datetime "2024-01-01T00:00:00.000" \
  --execute \
  --bootstrap-server localhost:9092

// 確保相同訂單的訊息順序
String orderId = "order-001";
ProducerRecord<String, String> record = new ProducerRecord<>(
    "order-events",
    orderId,  // 使用 orderId 作為 Key
    "{\"status\":\"created\"}"
);

// 方案 1：冪等處理（推薦）
public void processOrder(ConsumerRecord<String, String> record) {
    String messageId = record.topic() + "-" + record.partition() + "-" + record.offset();
    
    // 檢查是否已處理過
    if (processedMessages.contains(messageId)) {
        return; // 跳過已處理的訊息
    }
    
    // 處理業務邏輯
    doBusinessLogic(record.value());
    
    // 記錄已處理
    processedMessages.add(messageId);
}

// 方案 2：使用業務 Key 去重
public void processOrderWithDedup(String orderId, String orderData) {
    // 使用 Redis 或資料庫檢查
    if (redis.setnx("processed:" + orderId, "1") == 0) {
        return; // 已處理過
    }
    redis.expire("processed:" + orderId, 86400); // 設定過期時間
    
    doBusinessLogic(orderData);
}

<dependencies>
    <!-- Spring Kafka -->
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
        <version>3.1.0</version>
    </dependency>
    
    <!-- JSON 序列化 -->
    <dependency>
        <groupId>com.fasterxml.jackson.core</groupId>
        <artifactId>jackson-databind</artifactId>
    </dependency>
</dependencies>

spring:
  kafka:
    bootstrap-servers: kafka-1:9092,kafka-2:9092,kafka-3:9092
    
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      acks: all
      retries: 3
      properties:
        enable.idempotence: true
        max.in.flight.requests.per.connection: 5
    
    consumer:
      group-id: order-service
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      auto-offset-reset: earliest
      enable-auto-commit: false
      properties:
        spring.json.trusted.packages: "com.example.model"
    
    listener:
      ack-mode: manual
      concurrency: 3

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;

@Service
public class OrderEventProducer {
    
    private final KafkaTemplate<String, OrderEvent> kafkaTemplate;
    
    public OrderEventProducer(KafkaTemplate<String, OrderEvent> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }
    
    // 同步發送
    public void sendSync(String topic, String key, OrderEvent event) {
        try {
            SendResult<String, OrderEvent> result = 
                kafkaTemplate.send(topic, key, event).get();
            
            log.info("Sent message to partition {} with offset {}",
                     result.getRecordMetadata().partition(),
                     result.getRecordMetadata().offset());
        } catch (Exception e) {
            log.error("Failed to send message", e);
            throw new RuntimeException(e);
        }
    }
    
    // 非同步發送（推薦）
    public void sendAsync(String topic, String key, OrderEvent event) {
        kafkaTemplate.send(topic, key, event)
            .whenComplete((result, ex) -> {
                if (ex != null) {
                    log.error("Failed to send message", ex);
                } else {
                    log.info("Sent message to partition {} with offset {}",
                             result.getRecordMetadata().partition(),
                             result.getRecordMetadata().offset());
                }
            });
    }
}

// 訊息模型
@Data
public class OrderEvent {
    private String orderId;
    private String eventType;
    private BigDecimal amount;
    private LocalDateTime timestamp;
}

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;

@Service
public class OrderEventConsumer {
    
    @KafkaListener(
        topics = "order-events",
        groupId = "order-service",
        containerFactory = "kafkaListenerContainerFactory"
    )
    public void handleOrderEvent(
            @Payload OrderEvent event,
            @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
            @Header(KafkaHeaders.OFFSET) long offset,
            Acknowledgment ack) {
        
        log.info("Received event: {} from partition {} offset {}",
                 event, partition, offset);
        
        try {
            // 處理業務邏輯
            processOrderEvent(event);
            
            // 手動確認
            ack.acknowledge();
        } catch (Exception e) {
            log.error("Failed to process event", e);
            // 不確認，訊息將被重新消費
            throw e;
        }
    }
    
    // 批次消費
    @KafkaListener(
        topics = "order-events",
        groupId = "order-batch-service",
        batch = "true"
    )
    public void handleBatchEvents(
            List<OrderEvent> events,
            Acknowledgment ack) {
        
        log.info("Received {} events", events.size());
        
        for (OrderEvent event : events) {
            processOrderEvent(event);
        }
        
        ack.acknowledge();
    }
    
    private void processOrderEvent(OrderEvent event) {
        // 業務處理
    }
}

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.*;
import org.springframework.kafka.listener.ContainerProperties.AckMode;

@Configuration
public class KafkaConfig {
    
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, OrderEvent> 
            kafkaListenerContainerFactory(
                ConsumerFactory<String, OrderEvent> consumerFactory) {
        
        ConcurrentKafkaListenerContainerFactory<String, OrderEvent> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
        
        factory.setConsumerFactory(consumerFactory);
        factory.setConcurrency(3); // 並行消費數
        factory.getContainerProperties().setAckMode(AckMode.MANUAL);
        
        // 錯誤處理
        factory.setCommonErrorHandler(new DefaultErrorHandler(
            new FixedBackOff(1000L, 3) // 重試 3 次，間隔 1 秒
        ));
        
        return factory;
    }
}

# 使用 kafka-consumer-groups.sh
kafka-consumer-groups.sh --describe \
  --group order-processing-group \
  --bootstrap-server localhost:9092

# 輸出範例：
# GROUP                  TOPIC          PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
# order-processing-group order-events   0          1000            1500            500
# order-processing-group order-events   1          800             800             0
# order-processing-group order-events   2          1200            2000            800

# 啟動 Kafka 時啟用 JMX
export KAFKA_JMX_OPTS="-Dcom.sun.management.jmxremote \
  -Dcom.sun.management.jmxremote.port=9999 \
  -Dcom.sun.management.jmxremote.authenticate=false \
  -Dcom.sun.management.jmxremote.ssl=false"

kafka-server-start.sh config/server.properties

# jmx_exporter_config.yml
lowercaseOutputName: true
lowercaseOutputLabelNames: true

rules:
  - pattern: kafka.server<type=(.+), name=(.+), clientId=(.+), topic=(.+), partition=(.*)><>Value
    name: kafka_server_$1_$2
    type: GAUGE
    labels:
      clientId: "$3"
      topic: "$4"
      partition: "$5"
  
  - pattern: kafka.server<type=(.+), name=(.+)><>Value
    name: kafka_server_$1_$2
    type: GAUGE
    
  - pattern: kafka.controller<type=(.+), name=(.+)><>Value
    name: kafka_controller_$1_$2
    type: GAUGE

# Broker 日誌位置
/opt/kafka/logs/server.log
/opt/kafka/logs/controller.log

# 常見錯誤關鍵字
grep -E "ERROR|WARN|Exception" /opt/kafka/logs/server.log

# 檢查 Controller 日誌
grep "controller" /opt/kafka/logs/controller.log

# 1. 檢查叢集狀態
kafka-metadata.sh --snapshot /var/kafka-logs/__cluster_metadata-0/*.log \
  --command "describe"

# 2. 設定 inter.broker.protocol.version（升級前）
# 保持與舊版本相容
inter.broker.protocol.version=3.5

# 3. 逐一停止 Broker
kafka-server-stop.sh

# 4. 備份配置檔
cp /opt/kafka/config/kraft/server.properties \
   /opt/kafka/config/kraft/server.properties.bak

# 5. 更新 Kafka
cd /opt
wget https://downloads.apache.org/kafka/3.7.0/kafka_2.13-3.7.0.tgz
tar -xzf kafka_2.13-3.7.0.tgz
rm kafka
ln -s kafka_2.13-3.7.0 kafka

# 6. 恢復配置檔
cp /opt/kafka/config/kraft/server.properties.bak \
   /opt/kafka/config/kraft/server.properties

# 7. 啟動 Broker
kafka-server-start.sh -daemon /opt/kafka/config/kraft/server.properties

# 8. 確認 Broker 加入叢集
kafka-metadata.sh --snapshot /var/kafka-logs/__cluster_metadata-0/*.log \
  --command "describe"

# 9. 所有 Broker 升級完成後，更新 protocol version
# 編輯 server.properties，移除或更新 inter.broker.protocol.version

# 10. 再次逐一重啟 Broker

# 1. 停止 Broker
kafka-server-stop.sh

# 2. 恢復舊版本
rm /opt/kafka
ln -s kafka_2.13-3.5.0 /opt/kafka

# 3. 恢復配置檔
cp /opt/kafka/config/kraft/server.properties.bak \
   /opt/kafka/config/kraft/server.properties

# 4. 啟動 Broker
kafka-server-start.sh -daemon /opt/kafka/config/kraft/server.properties

# 5. 驗證服務正常
kafka-topics.sh --list --bootstrap-server localhost:9092

# 1. 建立 CA
openssl req -new -x509 -keyout ca-key -out ca-cert -days 365 \
  -subj "/CN=Kafka-CA"

# 2. 建立 Broker Keystore
keytool -genkey -keystore kafka.server.keystore.jks -validity 365 \
  -storepass changeit -keypass changeit \
  -dname "CN=kafka-broker-1" -storetype pkcs12

# 3. 建立憑證簽署請求（CSR）
keytool -keystore kafka.server.keystore.jks -certreq \
  -file kafka-broker-1.csr -storepass changeit

# 4. CA 簽署憑證
openssl x509 -req -CA ca-cert -CAkey ca-key -in kafka-broker-1.csr \
  -out kafka-broker-1-signed.crt -days 365 -CAcreateserial

# 5. 匯入 CA 憑證
keytool -keystore kafka.server.keystore.jks -import -alias CARoot \
  -file ca-cert -storepass changeit -noprompt

# 6. 匯入已簽署的憑證
keytool -keystore kafka.server.keystore.jks -import -alias kafka-broker-1 \
  -file kafka-broker-1-signed.crt -storepass changeit -noprompt

# 7. 建立 Truststore
keytool -keystore kafka.server.truststore.jks -import -alias CARoot \
  -file ca-cert -storepass changeit -noprompt

# server.properties
listeners=SSL://0.0.0.0:9093
advertised.listeners=SSL://kafka-broker-1:9093

ssl.keystore.location=/opt/kafka/ssl/kafka.server.keystore.jks
ssl.keystore.password=changeit
ssl.key.password=changeit
ssl.truststore.location=/opt/kafka/ssl/kafka.server.truststore.jks
ssl.truststore.password=changeit

ssl.client.auth=required
ssl.enabled.protocols=TLSv1.2,TLSv1.3

# server.properties
listeners=SASL_SSL://0.0.0.0:9094
advertised.listeners=SASL_SSL://kafka-broker-1:9094

sasl.enabled.mechanisms=PLAIN
sasl.mechanism.inter.broker.protocol=PLAIN

# JAAS 設定
listener.name.sasl_ssl.plain.sasl.jaas.config=\
  org.apache.kafka.common.security.plain.PlainLoginModule required \
  username="admin" \
  password="admin-secret" \
  user_admin="admin-secret" \
  user_producer="producer-secret" \
  user_consumer="consumer-secret";

# client.properties
security.protocol=SASL_SSL
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
  username="producer" \
  password="producer-secret";

ssl.truststore.location=/opt/kafka/ssl/kafka.client.truststore.jks
ssl.truststore.password=changeit

# 允許 producer 用戶寫入 order-events topic
kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 \
  --add --allow-principal User:producer \
  --operation Write --topic order-events

# 允許 consumer 用戶從 order-events topic 讀取
kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 \
  --add --allow-principal User:consumer \
  --operation Read --topic order-events \
  --group order-processing-group

# 查看 ACL
kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 \
  --list --topic order-events

格式：<domain>.<entity>.<event-type>
範例：
- order.payment.completed
- user.profile.updated
- inventory.stock.depleted

建議公式：
Partition 數量 = max(T/P, C)

T = 預期吞吐量（MB/s）
P = 單一 Partition 吞吐量（約 10 MB/s）
C = Consumer 數量

// ❌ 錯誤：每個實例使用不同的 Group ID
String groupId = "consumer-" + UUID.randomUUID();

// ✅ 正確：同一服務的實例使用相同的 Group ID
String groupId = "order-processing-service";

// ❌ 錯誤：處理時間過長導致頻繁 Rebalance
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        slowProcess(record); // 處理時間 > max.poll.interval.ms
    }
}

// ✅ 正確：控制單次處理時間
props.put("max.poll.records", 100);  // 減少單次拉取數量
props.put("max.poll.interval.ms", 300000);  // 適當增加間隔

❌ 錯誤：
- 頻繁查詢特定訊息
- 依賴 Kafka 做隨機讀取
- 將 Kafka 作為唯一的資料儲存

✅ 正確：
- Kafka 適合順序讀取與寫入
- 需要查詢時，將資料同步至資料庫
- Kafka 作為事件傳輸管道，不是資料庫

// ❌ 錯誤：未處理重複消費
public void processOrder(OrderEvent event) {
    orderRepository.save(event.getOrder()); // 重複消費會產生重複資料
}

// ✅ 正確：實作冪等處理
public void processOrder(OrderEvent event) {
    if (orderRepository.existsById(event.getOrderId())) {
        return; // 已處理過
    }
    orderRepository.save(event.getOrder());
}

// ❌ 錯誤：在同步 API 中等待 Kafka 回應
@PostMapping("/orders")
public ResponseEntity<Order> createOrder(@RequestBody Order order) {
    producer.send(record).get(); // 阻塞等待，延遲高
    return ResponseEntity.ok(order);
}

// ✅ 正確：分離同步與非同步邏輯
@PostMapping("/orders")
public ResponseEntity<Order> createOrder(@RequestBody Order order) {
    Order savedOrder = orderRepository.save(order); // 同步儲存
    producer.send(record); // 非同步發送事件
    return ResponseEntity.ok(savedOrder);
}

❌ 錯誤：
- 傳輸 > 1MB 的訊息（如檔案、圖片）
- 未設定 max.request.size

✅ 正確：
- 大型資料存放在物件儲存（S3、Azure Blob）
- Kafka 只傳輸資料的參考（URL、ID）
- 必要時調整 max.message.bytes、max.request.size

# 列出 Topic
kafka-topics.sh --list --bootstrap-server localhost:9092

# 建立 Topic
kafka-topics.sh --create --topic <name> --partitions <n> \
  --replication-factor <r> --bootstrap-server localhost:9092

# 查看 Topic
kafka-topics.sh --describe --topic <name> --bootstrap-server localhost:9092

# 刪除 Topic
kafka-topics.sh --delete --topic <name> --bootstrap-server localhost:9092

# 列出 Consumer Group
kafka-consumer-groups.sh --list --bootstrap-server localhost:9092

# 查看 Offset 狀態
kafka-consumer-groups.sh --describe --group <group> \
  --bootstrap-server localhost:9092

# 重置 Offset
kafka-consumer-groups.sh --reset-offsets --group <group> \
  --topic <topic> --to-earliest --execute --bootstrap-server localhost:9092

# 發送訊息
kafka-console-producer.sh --topic <topic> --bootstrap-server localhost:9092

# 消費訊息
kafka-console-consumer.sh --topic <topic> --from-beginning \
  --bootstrap-server localhost:9092