建立 MirrorMaker 2.0 查核點連接器

MirrorMaker 2.0 查核點連接器會將某個 Kafka 叢集的消費者偏移量，複製到另一個 Kafka 叢集。Kafka 取用端會使用取用端偏移量，記錄分區中最後一則成功取用的訊息。複製偏移量可讓消費者切換至目標叢集，並從相同位置繼續處理。

這個連接器可用於：

  • 確保從來源叢集切換至目標叢集時，停機時間降至最低。

  • 在叢集之間提供一致的消費者狀態，實現順暢的容錯移轉。

  • 將資料移至目標叢集時，保留消費者進度。

Checkpoint 連接器的運作方式

查核點連接器會與 MirrorMaker 2.0 來源連接器搭配使用。Checkpoint 連接器會按照下列方式複製偏移：

  1. 針對每個複製的分區，MirrorMaker 2.0 來源連接器會將複製的記錄寫入目標叢集。

  2. 寫入目標叢集的記錄通常與對應的來源記錄有不同的偏移量。如要啟用從來源偏移量到目標偏移量的翻譯，MirrorMaker 2.0 來源連接器會將偏移量組合寫入來源叢集上名為 mm2-offset-syncs.target.internal 的主題。這些偏移量組合會將來源偏移量對應至相應的目標偏移量。

  3. 取用端會從來源叢集讀取記錄。為了記錄進度，消費者會定期提交每個分區最新處理的訊息。Kafka 會將已提交的偏移量 (也稱為「檢查點」) 儲存在名為 __consumer_offsets 的主題中。檢查點記錄包含消費者的消費者群組 ID。

  4. MirrorMaker 2.0 查核點連接器會將查核點記錄複製到目標叢集中名為 source.checkpoints.internal 的主題。

  5. 如果為 sync.group.offsets.enabled=true，查核點連接器會定期將查核點從 source.checkpoints.internal 複製到目標叢集上的 __consumer_offsets。這樣一來，消費者就能在容錯移轉後，從最新提交的偏移量重新啟動。如要繼續處理，消費者會在容錯移轉後使用相同的消費者群組 ID。

    sync.group.offsets.interval.seconds 設定可控管連接器將檢查點複製到 __consumer_offsets 的頻率。

建立 MirrorMaker 2.0 查核點連接器

控制台

  1. 前往 Google Cloud 控制台的「Connect Clusters」(連結叢集) 頁面。

    前往「Connect Clusters」(連結叢集)

  2. 按一下要建立連接器的 Connect 叢集。

    「Connect cluster details」(連結叢集詳細資料) 頁面隨即顯示。

  3. 按一下「Create Connector」(建立連接器)。

  4. 在「Connector name」(連接器名稱) 欄位中，輸入連接器名稱。

  5. 在「Connector plugin」(連接器外掛程式) 清單中，選取「MirrorMaker 2.0 Checkpoint」。

  6. 在「來源叢集」部分，選取下列任一選項：

    • Managed Service for Apache Kafka 叢集：在「Kafka cluster」(Kafka 叢集) 清單中，選取 Managed Service for Apache Kafka 叢集。
    • 自行管理或外部 Kafka 叢集：在「Cluster bootstrap server」(叢集啟動伺服器) 欄位中，輸入 Kafka 叢集的啟動位址，格式為 HOSTNAME:PORT_NUMBER。

  7. 選用：在「Configurations」(設定) 方塊中，新增設定屬性或編輯預設屬性。詳情請參閱「設定連接器」。

  8. 在「Group names or Group regex」(群組名稱或群組規則運算式) 欄位中，輸入要複製的消費者群組清單 (以半形逗號分隔)。群組名稱可使用規則運算式。

  9. 選取「Task restart policy」(任務重新啟動政策)。詳情請參閱「工作重新啟動政策」。

  10. 點選「建立」。

gcloud

執行 gcloud managed-kafka connectors create 指令：

gcloud managed-kafka connectors create CONNECTOR_ID \
    --connect-cluster=CONNECT_CLUSTER_ID \
    --location=LOCATION \
    --config-file=CONFIG_FILE

更改下列內容：

  • CONNECTOR_ID：連接器的 ID 或名稱。
  • CONNECT_CLUSTER_ID：建立連接器 的 Connect 叢集 ID。
  • LOCATION：Connect 叢集的位置。
  • CONFIG_FILE：本機 YAML 或 JSON 設定檔的路徑。

以下是 MirrorMaker 2.0 檢查點連接器的設定檔範例：

connector.class: "org.apache.kafka.connect.mirror.MirrorCheckpointConnector"
emit.checkpoints.interval.seconds: "CHECKPOINT_INTERVAL"
groups: "CONSUMER_GROUP_LIST"
groups.exclude: "console-consumer-.*,connect-.*,__.*"
source.cluster.alias: "source"
source.cluster.bootstrap.servers: "SOURCE_CLUSTER_BOOTSTRAP_ADDRESS"
sync.group.offsets.enabled: "true"
sync.group.offsets.interval.seconds: "OFFSETS_SYNC_INTERVAL"
target.cluster.alias: "target"
target.cluster.bootstrap.servers: "TARGET_CLUSTER_BOOTSTRAP_ADDRESS"
tasks.max: "3"

更改下列內容：

  • CHECKPOINT_INTERVAL：寫入檢查點的頻率，以秒為單位。

  • CONSUMER_GROUP_LIST：以半形逗號分隔的消費者群組清單，這些群組的偏移量會複製。清單可以包含規則運算式。

  • SOURCE_CLUSTER_BOOTSTRAP_ADDRESS：來源 Kafka 叢集的啟動位址。

  • OFFSETS_SYNC_INTERVAL：將檢查點複製到 __consumer_offsets 主題的頻率，以秒為單位。詳情請參閱「檢查點同步」。

  • TARGET_CLUSTER_BOOTSTRAP_ADDRESS：目標 Kafka 叢集的啟動位址。

設定連接器

本節說明您可以在連接器上設定的部分設定屬性。

如需這個連接器專屬屬性的完整清單，請參閱 Apache Kafka 說明文件中的「MirrorMaker 查核點設定」。

複製的用戶群組

如要指定要複製的使用者群組，請設定 groups 設定屬性。您可以指定個別名稱或使用規則運算式。例如："group1,group2,customer-*"。預設值為 .*。

如要從複製作業中排除消費者群組，請設定 groups.exclude 屬性。

查核點同步

下列設定可控管連接器將檢查點寫入目標叢集的方式。

  • sync.group.offsets.enabled：是否要將偏移量從 source.checkpoints.internal 複製到目標叢集上的 __consumer_offsets。預設值為 false。如要讓目標叢集的消費者在容錯移轉後順暢重新啟動，請將這項屬性設為 true。

  • emit.checkpoints.interval.seconds：將檢查點寫入內部檢查點主題的頻率。預設值為 60 秒。設定這項屬性時，請考量追蹤進度與盡量減少寫入檢查點的負擔之間的平衡。10 秒是合理的值。

  • sync.group.offsets.interval.seconds：將檢查點從內部檢查點主題複製到 __consumer_offsets 的頻率。預設值為 60 秒。只有在 sync.group.offsets.enabled 為 true 時，這項設定才會生效。

後續步驟

Apache Kafka® 是 Apache Software Foundation 或其關係企業在美國和/或其他國家/地區的註冊商標。