Data Transmission Service:Kafka を使用したサブスクリプションデータの消費

注意事項

• このトピックで紹介するデモを使用する場合、データ処理が完了する前にオフセットがコミットされてしまう可能性があり、自動コミットによりデータ損失が発生することがあります。これを防ぐため、手動コミットを推奨します。

  説明

  コミットに失敗した場合、クライアントは最後に記録された消費チェックポイントから再開します。これによりデータが重複する可能性があるため、アプリケーションロジックで対応する必要があります。

• データは Avro シリアル化を使用して保存されます。フォーマットの詳細については、Record.avsc ファイルをご参照ください。

  警告

  このトピックで提供されているもの以外の Kafka クライアントを使用する場合、逆シリアル化中にデータが誤って解析される可能性があります (DTS Avro 逆シリアル化の例)。必ずデータを検証してください。

• DTS では、offsetForTimes API は時間単位として秒を使用しますが、ネイティブの Kafka API はミリ秒を使用します。

• ディザスタリカバリなどのイベントにより、データサブスクリプションサーバーで一時的なネットワーク切断が発生する可能性があります。このトピックで提供されている Kafka クライアントを使用しない場合は、ネットワークリトライをサポートしていることを確認してください。

• ネイティブの Kafka クライアントをサブスクライブモードで使用する場合、DTS が増分データ収集モジュールを切り替えると、サーバー上に保存されている消費チェックポイントがクリアされる可能性があります。その場合、消費チェックポイントを手動で調整する必要があります。サブスクライブモードを使用する必要がある場合は、DTS 提供の SDK を使用してデータサブスクリプションデータを消費し、消費チェックポイントを自分で管理することを推奨します。詳細については、「SDK を使用したデータサブスクリプションデータの消費」および「消費チェックポイントの管理」をご参照ください。

Kafka クライアントランタイム

Kafka クライアントデモ をダウンロードしてください。使用方法については、デモの Readme ファイルをご参照ください。

操作手順

このトピックでは、IntelliJ IDEA で Kafka クライアントデモを実行して、データサブスクリプションチャンネルからデータを消費する方法について説明します。

1. データサブスクリプションチャンネルを作成します。詳細については、「サブスクリプションソリューションの概要」をご参照ください。

2. 1 つ以上のコンシューマーグループを作成します。詳細については、「コンシューマーグループの作成」をご参照ください。

3. Kafka クライアントデモ をダウンロードして解凍します。

   説明

   をクリックして、[Download ZIP] を選択し、ファイルをダウンロードします。

4. IntelliJ IDEA を開き、[Open] をクリックします。

5. ダイアログボックスで、Kafka クライアントデモを解凍したディレクトリに移動します。次の図のようにフォルダーを展開して、Project Object Model (POM) ファイル pom.xml を見つけます。

6. 表示されるダイアログボックスで、[Open as Project] をクリックします。

7. IntelliJ IDEA ウィンドウで、フォルダーを展開して、Kafka クライアントデモファイル NotifyDemoDB.java を見つけ、ダブルクリックします。

8. NotifyDemoDB.java ファイル内のパラメーター値を設定します。

   パラメーター	説明	取得方法
   USER_NAME	コンシューマーグループのユーザー名です。 警告 別のクライアントを使用する場合は、<コンシューマーグループアカウント>-<コンシューマーグループ ID> 形式でユーザー名を設定する必要があります。例：dtstest-dtsae******bpv。	DTS コンソールで、対象のサブスクリプションインスタンスの ID をクリックします。左側のナビゲーションウィンドウで、データ消費 をクリックします。表示されるページで、コンシューマーグループの コンシューマーグループ ID /名前 および アカウント 情報を確認できます。 説明 コンシューマーグループアカウントのパスワードは、コンシューマーグループ作成時に指定します。
   PASSWORD_NAME	コンシューマーグループアカウントのパスワードです。
   SID_NAME	コンシューマーグループの ID です。
   GROUP_NAME	この値はコンシューマーグループ ID と一致している必要があります。
   KAFKA_TOPIC	データサブスクリプションチャンネルのサブスクリプショントピックです。	DTS コンソールで、対象のサブスクリプションインスタンスの ID をクリックします。基本情報 ページで、トピック および ネットワーク 情報を確認できます。
   KAFKA_BROKER_URL_NAME	データサブスクリプションチャンネルのネットワークアドレスです。 説明 Kafka クライアントとデータサブスクリプションチャンネルが同じクラシックネットワークまたは VPC 内にある場合は、ネットワーク遅延を最小限に抑えるために内部 IP アドレスを使用してください。 ネットワークの不安定性の可能性があるため、パブリックエンドポイントの使用は推奨しません。
   INITIAL_CHECKPOINT_NAME	データ消費を開始する時刻を UNIX タイムスタンプで指定します。例：1592269238。 説明 タイムスタンプを保存することで、以下が可能になります。 中断後に最後に保存されたタイムスタンプから消費を再開し、データ損失を防ぎます。 特定の時点からデータの消費を開始します。 SUBSCRIBE_MODE_NAME が subscribe に設定されている場合、INITIAL_CHECKPOINT_NAME の値は、サブスクリプションクライアントを初めて起動する場合にのみ有効になります。	消費開始時刻は、サブスクリプションインスタンスのデータ範囲内に含まれている必要があり、UNIX タイムスタンプに変換する必要があります。 説明 サブスクリプションタスクリストの データ範囲 列で、サブスクリプションインスタンスのデータ範囲を確認できます。 検索エンジンを使用して、UNIX タイムスタンプコンバーターを見つけることができます。
   USE_CONFIG_CHECKPOINT_NAME	デフォルト値は true で、クライアントが指定された時刻から強制的に開始されます。これにより、受信済みだが未処理のデータが失われるのを防ぎます。	なし
   SUBSCRIBE_MODE_NAME	同じコンシューマーグループ内で 2 つ以上の Kafka クライアントを実行するには、すべてのクライアントでこのパラメーターを subscribe に設定します。 デフォルト値は assign です。このモードでは、この機能は無効になり、クライアントは 1 つだけデプロイすることを推奨します。	なし

9. 上部のメニューバーから、[Run] > [Run] を選択してクライアントを実行します。

   説明

   アプリケーションを初めて実行する際、IntelliJ IDEA は必要な依存パッケージを自動的にダウンロードおよびインストールするため、時間がかかる場合があります。

結果

クライアントがソースデータベースからの変更データを正常にサブスクライブしました。

詳細な変更データを表示するには、NotifyDemoDB.java ファイルの 25 行目の //log.info(ret); から // を削除して、クライアントを再度実行します。

よくある質問

• Q：クライアントの消費チェックポイントを手動で追跡する必要があるのはなぜですか？

  A：DTS が記録する消費チェックポイントは、DTS が Kafka クライアントからのコミットを受信した時点を示しており、アプリケーションがデータ処理を完了した時点を示すものではありません。アプリケーションまたは Kafka クライアントが予期せず停止した場合、独自に記録したチェックポイントを使用することで、正確に中断ポイントから消費を再開でき、データの重複とデータ損失の両方を防ぐことができます。

チェックポイントの管理

1. DTS データ収集モジュールのクラスター切り替えをサブスクリプションクライアントがリッスンするように設定します。

   これを行うには、コンシューマーに対して次のプロパティを設定して、ClusterSwitchListener をコンシューマーインターセプタとして登録します。

   properties.setProperty(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, ClusterSwitchListener.class.getName());

   次のコードは、ClusterSwitchListener のサンプル実装を示しています。

   public class ClusterSwitchListener implements ClusterResourceListener, ConsumerInterceptor {
       private final static Logger LOG = LoggerFactory.getLogger(ClusterSwitchListener.class);
       private ClusterResource originClusterResource = null;
       private ClusterResource currentClusterResource = null;
   
       public ConsumerRecords onConsume(ConsumerRecords records) {
           return records;
       }
   
   
       public void close() {
       }
   
       public void onCommit(Map offsets) {
       }
   
   
       public void onUpdate(ClusterResource clusterResource) {
           synchronized (this) {
               originClusterResource = currentClusterResource;
               currentClusterResource = clusterResource;
               if (null == originClusterResource) {
                   LOG.info("Cluster updated to " + currentClusterResource.clusterId());
               } else {
                   if (originClusterResource.clusterId().equals(currentClusterResource.clusterId())) {
                       LOG.info("Cluster not changed on update:" + clusterResource.clusterId());
                   } else {
                       LOG.error("Cluster changed");
                       throw new ClusterSwitchException("Cluster changed from " + originClusterResource.clusterId() + " to " + currentClusterResource.clusterId()
                               + ", consumer require restart");
                   }
               }
           }
       }
   
       public boolean isClusterResourceChanged() {
           if (null == originClusterResource) {
               return false;
           }
           if (originClusterResource.clusterId().equals(currentClusterResource.clusterId())) {
               return false;
           }
           return true;
       }
   
       public void configure(Map<String, ?> configs) {
       }
   
       public static class ClusterSwitchException extends KafkaException {
           public ClusterSwitchException(String message, Throwable cause) {
               super(message, cause);
           }
   
           public ClusterSwitchException(String message) {
               super(message);
           }
   
           public ClusterSwitchException(Throwable cause) {
               super(cause);
           }
   
           public ClusterSwitchException() {
               super();
           }
   
       }

2. DTS データ収集モジュールからのクラスター切り替えイベントを処理します。

   ClusterSwitchException がキャッチされた場合、次回のサブスクリプション用の初期チェックポイントを、最後に正常に消費されたレコードのタイムスタンプにリセットします。次のコードスニペットがその例です。

   try{
      //do some action
   } catch (ClusterSwitchListener.ClusterSwitchException e) {
      reset();
   }
   
   // Reset the checkpoint.
   public reset() {
     long offset = kafkaConsumer.offsetsForTimes(timestamp);
     kafkaConsumer.seek(tp,offset);
   }

   説明

   サンプル実装については、「KafkaRecordFetcher」をご参照ください。

データの型と dataTypeNumber のマッピング