翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。

KCL 2.x から KCL 3.x に移行する

このトピックでは、KCL 2.x のコンシューマーを KCL 3.x へ移行するための手順を段階的に説明します。KCL 3.x は、KCL 2.x コンシューマーのインプレース移行をサポートしています。ワーカーをローリング方式で順次移行している間も、Kinesis データストリームからのデータ取得を継続できます。

ステップ 1: 前提条件

KCL 3.x の使用を始める前に、以下のものが揃っていることを確認してください。

ステップ 2: 依存関係を追加する

Maven を使用している場合は、以下の依存関係を pom.xml ファイルに追加します。3.x.x を最新の KCL バージョンに置き換えていることを確認してください。

<dependency>
    <groupId>software.amazon.kinesis</groupId>
    <artifactId>amazon-kinesis-client</artifactId>
    <version>3.x.x</version> <!-- Use the latest version -->
</dependency>

Gradle を使用している場合は、以下を build.gradle ファイルに追加します。3.x.x を最新の KCL バージョンに置き換えていることを確認してください。

implementation 'software.amazon.kinesis:amazon-kinesis-client:3.x.x'

最新の KCL バージョンは、Maven Central Repository で確認できます。

ステップ 3: 移行関連の設定をセットアップする

KCL 2.x から KCL 3.x に移行するには、次の設定パラメータを設定する必要があります。

configsBuilder.coordiantorConfig().clientVersionConfig(ClientVersionConfig.CLIENT_VERSION_CONFIG_COMPLATIBLE_WITH_2X)

以下は、KCL 2.x から 3.x へ移行する際に、CoordinatorConfig.clientVersionConfig を設定する方法の例です。必要に応じて、その他の設定も要件に合わせて調整できます。

Scheduler scheduler = new Scheduler(
    configsBuilder.checkpointConfig(),
    configsBuilder.coordiantorConfig().clientVersionConfig(ClientVersionConfig.CLIENT_VERSION_CONFIG_COMPLATIBLE_WITH_2X),
    configsBuilder.leaseManagementConfig(),
    configsBuilder.lifecycleConfig(),
    configsBuilder.metricsConfig(),
    configsBuilder.processorConfig(),
    configsBuilder.retrievalConfig()
);

KCL 2.x と KCL 3.x では負荷分散アルゴリズムが異なるため、コンシューマーアプリケーション内のすべてのワーカーが、同じ時点では同一の負荷分散アルゴリズムを使用していることが重要です。異なる負荷分散アルゴリズムを使用するワーカーが同時に動作していると、両アルゴリズムが独立して動作するため、負荷分散が最適化されない可能性があります。

この KCL 2.x 互換設定により、KCL 3.x アプリケーションは KCL 2.x と互換性のあるモードで動作し、コンシューマーアプリケーション内のすべてのワーカーが KCL 3.x にアップグレードされるまで、KCL 2.x の負荷分散アルゴリズムを利用できます。移行が完了すると、KCL は自動的に完全な KCL 3.x の機能モードへ切り替わり、稼働中のすべてのワーカーで新しい KCL 3.x の負荷分散アルゴリズムを使用し始めます。

ステップ 4: shutdownRequested() メソッド実装のベストプラクティスに従う

KCL 3.x では、リース再割り当て時にリースが別のワーカーへ引き継がれる際のデータ再処理を最小限に抑えるため、スムーズなリース引き継ぎと呼ばれる機能が導入されています。これは、リースが引き継がれる前に、処理済みの最新シーケンス番号をリーステーブルにチェックポイントすることで実現されます。スムーズなリース引き継ぎを正しく機能させるためには、RecordProcessor クラスの shutdownRequested メソッド内で、checkpointer オブジェクトを必ず呼び出すようにする必要があります。shutdownRequested メソッド内で checkpointer オブジェクトを呼び出さない場合は、次の例に示すように実装できます。

/**
 * Invoked when either Scheduler has been requested to gracefully shutdown
 * or lease ownership is being transferred gracefully so the current owner
 * gets one last chance to checkpoint.
 *
 * Checkpoints and logs the data a final time.
 *
 * @param shutdownRequestedInput Provides access to a checkpointer, allowing a record processor to checkpoint
 *                               before the shutdown is completed.
 */
public void shutdownRequested(ShutdownRequestedInput shutdownRequestedInput) {
    try {
       // Ensure that all delivered records are processed 
       // and has been successfully flushed to the downstream before calling 
       // checkpoint
       // If you are performing any asynchronous processing or flushing to
       // downstream, you must wait for its completion before invoking
       // the below checkpoint method.
        log.info("Scheduler is shutting down, checkpointing.");
        shutdownRequestedInput.checkpointer().checkpoint();
    } catch (ShutdownException | InvalidStateException e) {
        log.error("Exception while checkpointing at requested shutdown. Giving up.", e);
    } 
}

ステップ 5: KCL 3.x でワーカーメトリクスを収集するための前提条件を確認する

KCL 3.x は、ワーカー間で負荷を均等に分散するため、CPU 使用率などの CPU 利用メトリクスを収集します。コンシューマーアプリケーションのワーカーは、Amazon EC2、Amazon ECS、Amazon EKS、または AWS Fargate 上で実行できます。KCL 3.x がワーカーから CPU 使用率メトリクスを収集できるのは、以下の前提条件が満たされている場合のみです。

Amazon Elastic Compute Cloud(Amazon EC2)

Amazon Elastic Container Service (Amazon ECS) on Amazon EC2

での Amazon ECSAWS Fargate

Amazon Elastic Kubernetes Service (Amazon EKS) on Amazon EC2

での Amazon EKSAWS Fargate

ステップ 6: KCL 3.x 用に IAM アクセス許可を更新する

KCL 3.x コンシューマーアプリケーションに関連付けられている IAM ロールまたは IAM ポリシーに、次の許可を追加する必要があります。これは、KCL アプリケーションで使用中の既存の IAM ポリシーを更新する作業を伴います。詳細については、「KCL コンシューマーアプリケーションに必要な IAM アクセス許可」を参照してください。

ステップ 7: KCL 3.x のコードをワーカーにデプロイする

移行に必要な設定を行い、これまでの移行チェックリストをすべて完了したら、コードをビルドしてワーカーへデプロイできます。

ステップ 8: 移行を完了する

KCL 3.x のコードをデプロイしている間、KCL は引き続き KCL 2.x のリース割り当てアルゴリズムを使用します。すべてのワーカーに KCL 3.x のコードを正常にデプロイすると、KCL がそれを自動的に検知し、ワーカーのリソース使用状況に基づく新しいリース割り当てアルゴリズムに切り替わります。新しいリース割り当てアルゴリズムの詳細については、KCL がワーカーにリースを割り当て、負荷を分散する方法 を参照してください。

デプロイ中は、CloudWatch に出力される次のメトリクスを使用して移行状況をモニタリングできます。これらのメトリクスは Migration オペレーションの下でモニタリングできます。すべてのメトリクスは KCL アプリケーション単位のメトリクスであり、メトリクスレベルは SUMMARY に設定されています。CurrentState:3xWorker メトリクスの Sum 統計値が、KCL アプリケーション内のワーカー合計数と一致した場合、KCL 3.x への移行が正常に完了したことを示します。

KCL は、移行中に 2.x 互換モードへロールバックできる機能を提供しています。KCL 3.x への移行が完了し、ロールバックが不要になった場合は、CLIENT_VERSION_CONFIG_COMPATIBLE_WITH_2X の CoordinatorConfig.clientVersionConfig 設定を削除することをお勧めします。この設定を削除すると、KCL アプリケーションから移行関連のメトリクスが出力されなくなります。