[TOC]

## 前言
我在 Kafka 貢獻將近一年，但一直未有機會參與性能優化相關議題。對於像 Kafka 這樣的高流量軟體而言，即便是小幅度的優化，當流量上升時，影響都會非常顯著，而且社群對這類問題格外關注。這次有機會參與記憶體優化，特此記錄整個優化過程。

本次改善的核心目標是：「**用更少的記憶體且優化吞吐量**」。優化的程式碼路徑自 2016 年功能完成後便未曾更動。修改這類長期未變動且屬於讀取熱路徑的程式碼時，社群會特別關注測試覆蓋率，以確保修正不會引入新問題。因此，完整的測試覆蓋對此次變更至關重要。

- Jira: [KAFKA-19898](https://issues.apache.org/jira/browse/KAFKA-18989)
- Github PR: [KAFKA-18989 Optimize FileRecord#searchForOffsetWithSize](https://github.com/apache/kafka/pull/19214)

## 此優化的價值
可以分成兩個部分來看：
1. **減少讀取流程的記憶體用量**：
在 Kafka producer 傳輸資料時，可以使用 `batch.size` [1] 參數控制每個批次（Batch Record）最多可以送出多少資料，藉此減少傳輸請求，提升效能。
Kafka server 收到這些批次資料後，會將資料記錄在實體檔案（Segment）中。每個批次資料都有一個 header [2] 記錄當前批次的元數據，包括：基礎偏移量（baseOffset）、producer id、記錄數量（Records Count）和時間戳記（base timestamp）等資訊。
當需要查找特定偏移量（offset）的資料時，系統必須讀取相關實體檔案。Kafka 先透過索引文件（.index）找到大致搜尋位置，然後在該區域進行線性掃描，遍歷每個批次資料，檢查偏移量範圍，確認目標偏移量是否在某批次的初始偏移量（baseOffset）與最後偏移量（lastOffset）之間。

2. **減輕 Kafka CI 的記憶體壓力**：
對 Kafka 社群而言，CI 目前運行在 GitHub Actions 上。雖然 GitHub Actions 為社群提供免費額度，但記憶體和 CPU 資源仍有限制。社群發現某些涉及大量資料傳輸的測試經常導致 OOM[3]，進而影響後續測試的穩定性。此 PR 合併後，將有助於提升 Kafka CI 的穩定性，減少因資源受限而導致的測試失敗。

- [1] [batch.size](https://kafka.apache.org/documentation/#producerconfigs_batch.size)
- [2] [DefaultRecordBatch.java](https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/common/record/DefaultRecordBatch.java)
- [3] [Problem CI](https://github.com/apache/kafka/pull/19031#issuecomment-2725678289)

## 優化的原理
### Zero-Copy

Kafka 採用零拷貝（Zero-Copy）技術，提高了資料傳輸的效率。透過記憶體映射（Memory-Mapped Files），作業系統可避免不必要的資料複製。這使得資料能夠直接從生產者的緩衝區讀取並寫入網路介面，或從網路介面讀取後直接存入消費者的緩衝區，而無需在多個緩衝區間重複拷貝。因此 Kafka 在熱路徑下讀取資料時，都會盡可能不去破壞零拷貝（Zero-Copy）這個機制。

### 基礎知識
前面提到，要找到特定的批次資料時，需要透過 `baseOffset` 和 `lastOffset` 來判斷它是否包含目標偏移量（offset）。那麼，這兩個 Offset 有什麼差別呢？

- **baseOffset**：從以下程式碼可以看出，它是直接從 buffer 中提取 8-bytes 的數據並載入記憶體，成本相對較低。

  protected final long offset;
  // skip...
  @Override
  public long baseOffset() {
    return offset;
  }

- **lastOffset**：需載入完整的 Batch Header（固定大小 61 Bytes），會破壞 Zero-Copy 機制，導致更多數據複製到 Java 堆上。

  @Override
  public long lastOffset() {
    return loadBatchHeader().lastOffset();
  }

  protected RecordBatch loadBatchHeader() {
    if (fullBatch != null)
      return fullBatch;

    if (batchHeader == null)
      batchHeader = loadBatchWithSize(headerSize(), "record batch header");

    return batchHeader;
  }

因此，若能優先使用 `baseOffset` 而非 `lastOffset` 進行搜尋，將能有效降低記憶體消耗並減少 GC 負擔。

### 舊的搜尋邏輯
了解 `baseOffset` 和 `lastOffset` 的差異後，來看舊的搜尋邏輯如何運作：

/**
 * Search forward for the file position of the message batch whose last offset is greater
 * than or equal to the target offset. If no such batch is found, return null.
 *
 * @param targetOffset The offset to search for.
 * @param startingPosition The starting position in the file to begin searching from.
 * @return the batch's base offset, its physical position, and its size (including log overhead)
 */
public LogOffsetPosition searchForOffsetWithSize(long targetOffset, int startingPosition) {
    for (FileChannelRecordBatch batch : batchesFrom(startingPosition)) {
        long offset = batch.lastOffset();
        if (offset >= targetOffset)
            return new LogOffsetPosition(batch.baseOffset(), batch.position(), batch.sizeInBytes());
    }
    return null;
}

從這段程式碼可以看出，每次遍歷 `BatchRecord` 時，系統都會調用 `lastOffset()` 來與目標 `offset` 進行比較。而當 `lastOffset` 的計算成本較高時，就會增加記憶體消耗，影響效能。

### 優化方向

優化目標是盡可能**減少 `lastOffset` 的使用，改為使用 `baseOffset` 進行比對**，以降低記憶體使用量。

新的邏輯確保不論 `BatchRecord` 內部有多少筆資料，只需要**調用一次 `lastOffset()` 方法**，就能夠確定目標位置，進一步減少不必要的記憶體消耗。

我畫了一張簡單的示意圖來呈現這中間的差異，這張圖只是簡略地描述新舊邏輯的差異，使用 baseOffset 會有一些新的案例需要特別的處理，這邊就不展開，如果想要看完整的處理邏輯，可以看 [FileRecords.java
](https://github.com/apache/kafka/pull/19214/files#diff-665df224fa48442d04cc7ae7e16593d0980dae6601c0237bef00d5d313e54a36)
## 驗證

### 初步估算

> **每條消息大小**：100 bytes
> **批次大小**：4096 bytes
> **每批次最大消息數**：4096 ÷ 100 ≈ 40

在舊邏輯中，**最差**的情況下每批消息都會調用 `lastOffset`，而 `lastOffset` 每次調用載入 61 bytes 的資料，因此記憶體使用量為：

**40 × 61 = 2440 bytes**

在新邏輯下，每批次僅調用 1 次 `lastOffset`，記憶體使用量為：

**1 × 61 = 61 bytes**

相比之下，記憶體占用從 **2440 bytes** 降至 **61 bytes**，減少約 **97.5%**。

### 實際測試

因為這次改的路徑只是讀取路徑的其中一環，為了避免其他因素的影響，所以在測試方面，使用了 JMH 進行測試，這樣就可以專注在這個方法本身的優化。測試的參數為：

> 資料量：1000000000 筆
> 每個批次的數據量：40 筆
> 每筆資料大小：大約 10 bytes
> 熱身 5 輪，測試 10 輪

由測試結果可知：

- 執行時間減少 50.4%，這意味著在高頻率的調用下，有可能吞吐量可以翻倍
- 標準差減少了 67.3%，更低的標準差意味著更一致系統執行時間，不會有過大的波動
- 極端情況的改善：最壞情況下的性能（最大值）從88.159毫秒改善到35.284毫秒，提升了60%，可以由此推斷性能尖峰資源的使用量也明線降低

性能提升和波動減少共同表示系統資源（CPU、記憶體、I/O）的使用更加高效，較低的平均延遲通常意味著更少的堆積請求和更低的系統資源佔用

| 指標 | 修改後 | 修改前 | 差異百分比 |
|------|-------|-------|------------|
| 平均執行時間 | 27.717 ms/op | 55.864 ms/op | -50.4% |
| 最小值 | 21.291 ms | 42.250 ms | -49.6% |
| 最大值 | 35.284 ms | 88.159 ms | -60.0% |
| 標準差 | 3.942 | 12.072 | -67.3% |

## 結論
這次的優化理論與解法並不算複雜，真正的挑戰在於如何精確觀察問題並找出根本原因。這需要對 Kafka 有一定的理解，同時也要掌握如何透過效能監控工具分析並定位瓶頸。這次的經驗是一個難得的學習機會，也讓我更有信心在未來的貢獻中，能夠獨立發現問題並提出有效的解決方案。

最後還是要推廣一下，如果對開源貢獻有興趣的人，歡迎加入[源來適你](https://www.facebook.com/opensource4you)。