本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。

# 使用自訂位移儲存主題
<a name="msk-connect-set-offset-storage-topic"></a>

若要在來源連接器之間提供偏移連續性，您可以使用所選偏移儲存主題，而非預設主題。指定偏移儲存主題有助您完成任務，例如，建立來源連接器從上一個連接器的最後一個偏移恢復讀取。

若要指定偏移儲存主題，請在建立連接器之前，在工作程序組態中提供 `offset.storage.topic` 屬性值。若您要重複使用偏移儲存主題來使用先前建立之連接器的偏移，您必須為新連接器指定與舊連接器相同的名稱。若您建立自訂偏移儲存主題，則必須在主題組態中將 [https://kafka.apache.org/27/documentation.html#topicconfigs_cleanup.policy](https://kafka.apache.org/27/documentation.html#topicconfigs_cleanup.policy) 設定為 `compact`。

**注意**  
若您在建立*目的地*連接器時指定偏移儲存主題，則若該主題尚不存在，MSK Connect 將會建立該主題。然而，該主題將不會用於儲存連接器偏移。  
反而是使用 Kafka 取用者群組通訊協定來管理目的地連接器偏移。每個目的地連接器都會建立名為 `connect-{CONNECTOR_NAME}` 的群組。只要取用者群組存在，您使用相同 `CONNECTOR_NAME` 值建立的任何連續目的地連接器都會從上次遞交的偏移繼續執行。

**重要**  
如果您想要在維持位移連續性的同時更新現有的連接器組態，請使用 UpdateConnector API。如需詳細資訊，請參閱[更新連接器](mkc-update-connector.md)。

**Example ：在重新建立來源連接器時指定位移儲存主題**  
如果您需要刪除並重新建立連接器，同時保持偏移連續性，您可以在工作者組態中指定偏移儲存主題。例如，假設您有一個變更資料擷取 (CDC) 連接器，並且想要重新建立它，而不會遺失您在 CDC 串流中的位置。以下步驟會說明如何完成這項任務。  

1. 在用戶端電腦上，執行以下命令以尋找連接器的偏移儲存主題名稱。使用您叢集的引導代理程式字串來取代 `{{<bootstrapBrokerString>}}`。如需有關取得引導代理程式字串的指示，請參閱 [取得 Amazon MSK 叢集的引導代理程式](msk-get-bootstrap-brokers.md)。

   ```
   {{<path-to-your-kafka-installation>}}/bin/kafka-topics.sh --list --bootstrap-server {{<bootstrapBrokerString>}}
   ```

   以下輸出顯示所有叢集主題的清單，包括任何預設內部連接器主題。在此範例中，現有 CDC 連接器會使用 MSK Connect 建立的[預設偏移儲存主題](msk-connect-default-offset-storage-topic.md)。此即為偏移儲存主題被稱為 `__amazon_msk_connect_offsets_my-mskc-connector_12345678-09e7-4abc-8be8-c657f7e4ff32-2` 之原因。

   ```
   __consumer_offsets
   __amazon_msk_canary
   __amazon_msk_connect_configs_my-mskc-connector_12345678-09e7-4abc-8be8-c657f7e4ff32-2
   __amazon_msk_connect_offsets_my-mskc-connector_12345678-09e7-4abc-8be8-c657f7e4ff32-2
   __amazon_msk_connect_status_my-mskc-connector_12345678-09e7-4abc-8be8-c657f7e4ff32-2
   my-msk-topic-1
   my-msk-topic-2
   ```

1. 開啟位於 [https://console.aws.amazon.com/msk/](https://console.aws.amazon.com/msk) 的 Amazon MSK 主控台。

1. 從**連接器**清單中選擇您的連接器。複製並儲存**連接器組態**欄位的內容，以便修改並使用該內容來建立新的連接器。

1. 選擇**刪除**以刪除連接器。文字輸入在欄位中輸入連接器名稱以確認刪除。

1. 使用符合您案例的值來建立自訂工作程序組態。如需說明，請參閱[建立自訂工作者組態](msk-connect-create-custom-worker-config.md)。

   在工作程序組態中，您必須將之前擷取之偏移儲存主題的名稱指定為 `offset.storage.topic` 的值，如以下組態所示。

   ```
   config.providers.secretManager.param.aws.region=eu-west-3
   key.converter=<org.apache.kafka.connect.storage.StringConverter>
   value.converter=<org.apache.kafka.connect.storage.StringConverter>
   config.providers.secretManager.class=com.github.jcustenborder.kafka.config.aws.SecretsManagerConfigProvider
   config.providers=secretManager
   offset.storage.topic={{__amazon_msk_connect_offsets_my-mskc-connector_12345678-09e7-4abc-8be8-c657f7e4ff32-2}}
   ```

1. 
**重要**  
您必須為新連接器取名為與舊連接器相同的名稱。

   使用您在之前步驟中設定的工作程序組態來建立新的連接器。如需說明，請參閱[建立 連接器](mkc-create-connector-intro.md)。