Azure Cosmos DB の変更フィード プロセッサ

.NET の変更フィード プロセッサは、最新バージョン モードと、すべてのバージョンと削除モードで使用できます。 すべてのバージョンと削除モードはプレビュー段階であり、バージョン 3.40.0-preview.0 以降の変更フィード プロセッサでサポートされています。 どちらのモードでも、エントリ ポイントは常に監視対象のコンテナーです。

最新バージョン モードを使用して読み取る場合は、Container インスタンスで GetChangeFeedProcessorBuilder を呼び出します。

/// <summary>
/// Start the Change Feed Processor to listen for changes and process them with the HandleChangesAsync implementation.
/// </summary>
private static async Task<ChangeFeedProcessor> StartChangeFeedProcessorAsync(
    CosmosClient cosmosClient,
    IConfiguration configuration)
{
    string databaseName = configuration["SourceDatabaseName"];
    string sourceContainerName = configuration["SourceContainerName"];
    string leaseContainerName = configuration["LeasesContainerName"];

    Container leaseContainer = cosmosClient.GetContainer(databaseName, leaseContainerName);
    ChangeFeedProcessor changeFeedProcessor = cosmosClient.GetContainer(databaseName, sourceContainerName)
        .GetChangeFeedProcessorBuilder<ToDoItem>(processorName: "changeFeedSample", onChangesDelegate: HandleChangesAsync)
            .WithInstanceName("consoleHost")
            .WithLeaseContainer(leaseContainer)
            .Build();

    Console.WriteLine("Starting Change Feed Processor...");
    await changeFeedProcessor.StartAsync();
    Console.WriteLine("Change Feed Processor started.");
    return changeFeedProcessor;
}

すべてのバージョンと削除モードを使用して読み取る場合は、GetChangeFeedProcessorBuilderWithAllVersionsAndDeletes インスタンスから Container を呼び出します。

Container leaseContainer = client.GetContainer(Program.databaseName, Program.leasesContainer);
Container monitoredContainer = client.GetContainer(Program.databaseName, containerName);
ChangeFeedProcessor changeFeedProcessor = monitoredContainer
    .GetChangeFeedProcessorBuilderWithAllVersionsAndDeletes<ToDoItem>(processorName: "changeFeedBasic", onChangesDelegate: Program.HandleChangesAsync)
        .WithInstanceName("consoleHost")
        .WithLeaseContainer(leaseContainer)
        .Build();

どちらのモードでも、最初のパラメーターは、このプロセッサの目標を記述する個別の名前です。 2 番目の名前は、変更を処理するデリゲートの実装です。

最新バージョン モードのデリゲートの例を次に示します。

/// <summary>
/// The delegate receives batches of changes as they are generated in the change feed and can process them.
/// </summary>
static async Task HandleChangesAsync(
    ChangeFeedProcessorContext context,
    IReadOnlyCollection<ToDoItem> changes,
    CancellationToken cancellationToken)
{
    Console.WriteLine($"Started handling changes for lease {context.LeaseToken}...");
    Console.WriteLine($"Change Feed request consumed {context.Headers.RequestCharge} RU.");
    // SessionToken if needed to enforce Session consistency on another client instance
    Console.WriteLine($"SessionToken ${context.Headers.Session}");

    // We may want to track any operation's Diagnostics that took longer than some threshold
    if (context.Diagnostics.GetClientElapsedTime() > TimeSpan.FromSeconds(1))
    {
        Console.WriteLine($"Change Feed request took longer than expected. Diagnostics:" + context.Diagnostics.ToString());
    }

    foreach (ToDoItem item in changes)
    {
        Console.WriteLine($"Detected operation for item with id {item.id}, created at {item.creationTime}.");
        // Simulate some asynchronous operation
        await Task.Delay(10);
    }

    Console.WriteLine("Finished handling changes.");
}

すべてのバージョンと削除モードのデリゲートの例を次に示します。

static async Task HandleChangesAsync(ChangeFeedProcessorContext context, IReadOnlyCollection<ChangeFeedItem<ToDoItem>> changes, CancellationToken cancellationToken)
{
    Console.WriteLine($"Started handling changes for lease {context.LeaseToken}...");
    Console.WriteLine($"Change Feed request consumed {context.Headers.RequestCharge} RU.");
    // SessionToken if needed to enforce Session consistency on another client instance
    Console.WriteLine($"SessionToken ${context.Headers.Session}");

    // We may want to track any operation's Diagnostics that took longer than some threshold
    if (context.Diagnostics.GetClientElapsedTime() > TimeSpan.FromSeconds(1))
    {
        Console.WriteLine($"Change Feed request took longer than expected. Diagnostics:" + context.Diagnostics.ToString());
    }

    foreach (ChangeFeedItem<ToDoItem> item in changes)
    {
        if (item.Metadata.OperationType == ChangeFeedOperationType.Delete)
        {
            Console.WriteLine($"\tDetected {item.Metadata.OperationType} operation for item.");
        }
        else
        {
            Console.WriteLine($"\tDetected {item.Metadata.OperationType} operation for item with id {item.Current.id}.");
        }
        // Simulate work
        await Task.Delay(1);
    }
}

その後、WithInstanceName を使用してコンピューティング インスタンス名または一意識別子を定義します。 コンピューティング インスタンス名は一意であり、デプロイするコンピューティング インスタンスごとに異なる必要があります。 WithLeaseContainer を使用して、リース状態を維持するようにコンテナーを設定します。

Build を呼び出すとプロセッサ インスタンスが提供され、StartAsync を呼び出すことによってそれを開始できます。

処理のライフ サイクル

ホスト インスタンスの通常のライフ サイクルは次のとおりです。

1. 変更フィードを読み取ります。
2. 変更がない場合は、事前に定義した時間 (ビルダーの WithPollInterval を使用してカスタマイズ可能) だけスリープし、1 に戻ります。
3. 変更がある場合は、それらをデリゲートに送信します。
4. デリゲートによる変更の処理が正常に完了すると、最後に処理された時点でリース ストアを更新し、#1 に移ります。

エラー処理

変更フィード プロセッサには、ユーザー コード エラーに対する回復性があります。 デリゲートの実装にハンドルされない例外がある場合 (ステップ 4)、その特定の変更バッチを処理しているスレッドは停止し、最終的には新しいスレッドが作成されます。 新しいスレッドは、リース ストアにその範囲のパーティション キー値が保存された最新の時点をチェックします。 新しいスレッドはそこから再開し、実質的に同じ変更のバッチをデリゲートに送信します。 デリゲートによって変更が正しく処理されるまで、この動作が続けられます。これにより、変更フィード プロセッサでは "少なくとも 1 回" が保証されます。

変更フィード プロセッサが同じバッチの変更を継続的に再試行して "行き詰まる" ことがないように、例外が発生した場合に、ドキュメントに書き込むためのデリゲート コードのロジックをエラー メッセージ キューに追加する必要があります。 このように設計することで、今後の変更の処理を継続しながら、未処理の変更を追跡できます。 エラー メッセージ キューは、別の Azure Cosmos DB コンテナーである可能性があります。 具体的なデータ ストアは重要ではありません。 未処理の変更が永続化されることのみが必要です。

変更フィード推定機能を使用して、変更フィードを読み取る変更フィード プロセッサ インスタンスの進行状況を監視することもできます。または、ライフ サイクル通知を使用して基になっている障害を検出できます。

ライフサイクルに関する通知

変更フィード プロセッサは、そのライフ サイクル内の関連するイベントに接続できます。 1 つまたはすべての通知を受け取るように選択できます。 少なくとも、エラー通知を登録することをお勧めします。

• WithLeaseAcquireNotification のハンドラーを登録して、現在のホストがリースを取得して処理を開始したときに通知されるようにします。
• WithLeaseReleaseNotification のハンドラーを登録して、現在のホストがリースをリリースして処理を停止したときに通知されるようにします。
• WithErrorNotification のハンドラーを登録すると、現在のホストが処理中に例外を検出したときに通知を受け取ります。 ソースがユーザー デリゲート (ハンドルされない例外) であるか、監視対象のコンテナーにアクセスしようとしたときにプロセッサで発生したエラー (ネットワークの問題など) であるかを区別できる必要があります。

ライフサイクル通知は、どちらの変更フィード モードでも使用できます。 最新バージョン モードでのライフサイクル通知の例を次に示します。

Container.ChangeFeedMonitorLeaseAcquireDelegate onLeaseAcquiredAsync = (string leaseToken) =>
{
    Console.WriteLine($"Lease {leaseToken} is acquired and will start processing");
    return Task.CompletedTask;
};

Container.ChangeFeedMonitorLeaseReleaseDelegate onLeaseReleaseAsync = (string leaseToken) =>
{
    Console.WriteLine($"Lease {leaseToken} is released and processing is stopped");
    return Task.CompletedTask;
};

Container.ChangeFeedMonitorErrorDelegate onErrorAsync = (string LeaseToken, Exception exception) =>
{
    if (exception is ChangeFeedProcessorUserException userException)
    {
        Console.WriteLine($"Lease {LeaseToken} processing failed with unhandled exception from user delegate {userException.InnerException}");
    }
    else
    {
        Console.WriteLine($"Lease {LeaseToken} failed with {exception}");
    }

    return Task.CompletedTask;
};

ChangeFeedProcessor changeFeedProcessor = monitoredContainer
    .GetChangeFeedProcessorBuilder<ToDoItem>("changeFeedNotifications", handleChanges)
        .WithLeaseAcquireNotification(onLeaseAcquiredAsync)
        .WithLeaseReleaseNotification(onLeaseReleaseAsync)
        .WithErrorNotification(onErrorAsync)
        .WithInstanceName("consoleHost")
        .WithLeaseContainer(leaseContainer)
        .Build();

展開ユニット

変更フィード プロセッサの 1 つのデプロイ ユニットは、processorName の値が同じでリース コンテナー構成も同じであるが、インスタンス名が異なる 1 つ以上のコンピューティング インスタンスで構成されます。 多数のデプロイ ユニットを使用することができ、各ユニットは変更に対して異なるビジネス フローを備え、各デプロイ ユニットは 1 つまたは複数のインスタンスで構成されます。

たとえば、あるデプロイ ユニットは、コンテナーに変更が発生するたびに外部 API をトリガーします。 別のデプロイ ユニットでは、変更が発生するたびに、リアルタイムでデータを移動することができます。 監視対象コンテナーで変更が発生すると、すべてのデプロイ ユニットが通知を受け取ります。

動的スケーリング

前述したように、1 つのデプロイ ユニット内には、1 つまたは複数のコンピューティング インスタンスを使用できます。 デプロイ ユニット内でコンピューティングの分散を利用するには、次の要件のみが重要になります。

• すべてのインスタンスのリース コンテナーの構成が同じである必要があります。
• すべてのインスタンスの processorName の値が同じである必要があります。
• 各インスタンスには、異なるインスタンス名が設定されている必要があります (WithInstanceName)。

これら 3 つの条件に当てはまる場合、変更フィード プロセッサでは、リース コンテナー内のすべてのリースが、そのデプロイ ユニットの実行中の全インスタンスに分散され、均等分散アルゴリズムを使ってコンピューティングが並列化されます。 1 つのリースは特定の時点で 1 つのインスタンスによって所有されるため、インスタンスの数をリースの数より大きくすることはできません。

インスタンス数は増減する可能性があります。 変更フィード プロセッサは、負荷をそれに応じて再分散することで動的に調整します。

さらに、コンテナーのスループットまたはストレージの増加に応じて、変更フィード プロセッサでコンテナーのスケールを動的に調整できます。 コンテナーが拡張されると、変更フィード プロセッサでは、リースを動的に増やし、既存のインスタンス間で新しいリースを分散することによって、このシナリオが透過的に処理されます。

開始時刻

既定では、変更フィード プロセッサは、初めて開始したときに、リース コンテナーを初期化し、その処理のライフ サイクルを開始します。 変更フィード プロセッサが初めて初期化される前に監視対象コンテナー内で発生した変更は検出されません。

以前の日時からの読み取り

のインスタンスを DateTime ビルダー拡張機能に渡すことで、"特定の日時" 以降の変更を読み取るよう変更フィード プロセッサを初期化することができます。WithStartTime

Container leaseContainer = client.GetContainer(databaseId, Program.leasesContainer);
Container monitoredContainer = client.GetContainer(databaseId, Program.monitoredContainer);
ChangeFeedProcessor changeFeedProcessor = monitoredContainer
    .GetChangeFeedProcessorBuilder<ToDoItem>("changeFeedTime", Program.HandleChangesAsync)
        .WithInstanceName("consoleHost")
        .WithLeaseContainer(leaseContainer)
        .WithStartTime(particularPointInTime)
        .Build();

変更フィード プロセッサは、その特定の日時に対して初期化され、それ以降に発生した変更の読み取りを開始します。

最初からの読み取り

データの移行やコンテナーの履歴全体の分析など、他のシナリオでは、"そのコンテナーの有効期間の最初" から変更フィードを読み取る必要があります。 ビルダー拡張機能で WithStartTime を使用できますが、DateTime.MinValue.ToUniversalTime() を渡すと、次の例のように DateTime の最小値の UTC 表現が生成されます。

Container leaseContainer = client.GetContainer(databaseId, Program.leasesContainer);
Container monitoredContainer = client.GetContainer(databaseId, Program.monitoredContainer);
ChangeFeedProcessor changeFeedProcessor = monitoredContainer
    .GetChangeFeedProcessorBuilder<ToDoItem>("changeFeedBeginning", Program.HandleChangesAsync)
        .WithInstanceName("consoleHost")
        .WithLeaseContainer(leaseContainer)
        .WithStartTime(DateTime.MinValue.ToUniversalTime())
        .Build();

変更フィード プロセッサは初期化され、コンテナーの有効期間の最初から変更の読み取りを開始します。

完全な作業サンプルについては、こちらを参照してください。 最新バージョン モードで変更フィードを読み取る場合のデリゲート実装の例を次に示します。

private static Consumer<List<JsonNode>> handleChanges() {
    return (List<JsonNode> docs) -> {
        logger.info("Start handleChanges()");

        for (JsonNode document : docs) {
            try {
                //Change Feed hands the document to you in the form of a JsonNode
                //As a developer you have two options for handling the JsonNode document provided to you by Change Feed
                //One option is to operate on the document in the form of a JsonNode, as shown below. This is great
                //especially if you do not have a single uniform data model for all documents.
                logger.info("Document received: " + OBJECT_MAPPER.writerWithDefaultPrettyPrinter()
                        .writeValueAsString(document));

                //You can also transform the JsonNode to a POJO having the same structure as the JsonNode,
                //as shown below. Then you can operate on the POJO.
                CustomPOJO2 pojo_doc = OBJECT_MAPPER.treeToValue(document, CustomPOJO2.class);
                logger.info("id: " + pojo_doc.getId());

            } catch (JsonProcessingException e) {
                e.printStackTrace();
            }
        }
        isWorkCompleted = true;
        logger.info("End handleChanges()");

    };
}

options = new ChangeFeedProcessorOptions();
options.setStartFromBeginning(false);
options.setLeasePrefix("myChangeFeedDeploymentUnit");
options.setFeedPollDelay(Duration.ofSeconds(5));
options.setFeedPollThroughputControlConfig(throughputControlGroupConfig);

すべてのバージョンと削除モードで変更フィードを読み取るためのデリゲート実装は似ていますが、.handleChanges() を呼び出す代わりに .handleAllVersionsAndDeletesChanges() を呼び出します。 すべてのバージョンと削除モードはプレビュー段階であり、Java SDK バージョン >= 4.42.0 で使用できます。

次に例を示します。

public static ChangeFeedProcessor getChangeFeedProcessorForAllVersionsAndDeletesMode(String hostName, CosmosAsyncContainer feedContainer, CosmosAsyncContainer leaseContainer) {
    return new ChangeFeedProcessorBuilder()
            .hostName(hostName)
            .options(options)
            .feedContainer(feedContainer)
            .leaseContainer(leaseContainer)
            .handleAllVersionsAndDeletesChanges((List<ChangeFeedProcessorItem> changeFeedProcessorItems) -> {
                logger.info("--->handleAllVersionsAndDeletesChanges() START");

                for (ChangeFeedProcessorItem item : changeFeedProcessorItems) {
                    try {
                        // AllVersionsAndDeletes Change Feed hands the document to you in the form of ChangeFeedProcessorItem
                        // As a developer you have two options for handling the ChangeFeedProcessorItem provided to you by Change Feed
                        // One option is to operate on the item as it is and call the different getters for different states, as shown below.
                        logger.info("---->DOCUMENT RECEIVED: {}", OBJECT_MAPPER.writerWithDefaultPrettyPrinter()
                                .writeValueAsString(item));

                        logger.info("---->CURRENT RECEIVED: {}", item.getCurrent());
                        logger.info("---->PREVIOUS RECEIVED: {}", item.getPrevious());
                        logger.info("---->METADATA RECEIVED: {}", item.getChangeFeedMetaData());

                        // You can also transform the ChangeFeedProcessorItem to JsonNode and work on the generic json structure.
                        // This is great especially if you do not have a single uniform data model for all documents.
                        logger.info("----=>JsonNode received: " + item.toJsonNode());

                    } catch (JsonProcessingException e) {
                        e.printStackTrace();
                    }
                }
                logger.info("--->handleAllVersionsAndDeletesChanges() END");

            })
            .buildChangeFeedProcessor();
}

どちらの変更フィード モードでも、これを changeFeedProcessorInstance に割り当てて、コンピューティング インスタンス名 (hostName)、監視対象コンテナー (ここでは feedContainer という名前)、そして leaseContainer のパラメーターを渡すことができます。 次に、変更フィード プロセッサを開始します。

logger.info("Start Change Feed Processor on worker (handles changes asynchronously)");
ChangeFeedProcessor changeFeedProcessorInstance = new ChangeFeedProcessorBuilder()
    .hostName("SampleHost_1")
    .feedContainer(feedContainer)
    .leaseContainer(leaseContainer)
    .handleChanges(handleChanges())
    .options(options)
    .buildChangeFeedProcessor();
changeFeedProcessorInstance.start()
                           .subscribeOn(Schedulers.boundedElastic())
                           .subscribe();

処理のライフ サイクル

ホスト インスタンスの通常のライフ サイクルは次のとおりです。

1. 変更フィードを読み取ります。
2. 変更がない場合は、事前に定義した時間 (ビルダーの options.setFeedPollDelay でカスタマイズ可能) だけスリープし、#1 に移ります
3. 変更がある場合は、それらをデリゲートに送信します。
4. デリゲートによる変更の処理が "正常" に完了したら、最後に処理された時点でリース ストアを更新し、#1 に移ります。

エラー処理

変更フィード プロセッサには、ユーザー コード エラーに対する回復性があります。 デリゲートの実装にハンドルされない例外がある場合 (ステップ #4)、その特定の変更バッチを処理しているスレッドは停止し、新しいスレッドが作成されます。 新しいスレッドでは、そのパーティション キー値の範囲に対してリース ストアが保存した最新の時点が確認され、そこから処理が再開されて、事実上、同じ変更バッチがデリゲートに送信されます。 この動作は、デリゲートが変更を正しく処理するまで続きます。 このため、変更フィード プロセッサでは "少なくとも 1 回" が保証されます。

変更フィード プロセッサが同じバッチの変更を継続的に再試行して "行き詰まる" ことがないように、例外が発生した場合に、ドキュメントに書き込むためのデリゲート コードのロジックをエラー メッセージに追加する必要があります。 このように設計することで、今後の変更の処理を継続しながら、未処理の変更を追跡できます。 エラー メッセージは、別の Azure Cosmos DB コンテナーである可能性があります。 具体的なデータ ストアは重要ではありません。 未処理の変更が永続化されることのみが必要です。

変更フィード推定機能を使用して、変更フィードを読み取る変更フィード プロセッサ インスタンスの進行状況を監視することもできます。

展開ユニット

変更フィード プロセッサの 1 つのデプロイ ユニットは、同じリース コンテナーの構成と同じ leasePrefix を持つが、それぞれ hostName の値が異なる 1 つまたは複数のインスタンスで構成されます。 多数のデプロイ ユニットを使用することができ、それぞれが変更に対して異なるビジネス フローを備え、各デプロイ ユニットは 1 つまたは複数のインスタンスで構成されます。

たとえば、あるデプロイ ユニットは、コンテナーに変更が発生するたびに外部 API をトリガーします。 別のデプロイ ユニットでは、変更が発生するたびに、リアルタイムでデータを移動することができます。 監視対象コンテナーで変更が発生すると、すべてのデプロイ ユニットが通知を受け取ります。

動的スケーリング

前述したように、1 つのデプロイ ユニット内には、1 つまたは複数のコンピューティング インスタンスを使用できます。 デプロイ ユニット内でコンピューティングの分散を利用するには、次の要件のみが重要になります。

• すべてのインスタンスのリース コンテナーの構成が同じである必要があります。
• すべてのインスタンスで options.setLeasePrefix に同じ値が設定されている (または、何も設定されていない) ことが必要です。
• 各インスタンスには、異なる hostName が設定されている必要があります。

これら 3 つの条件に当てはまる場合、変更フィード プロセッサでは、リース コンテナー内のすべてのリースが、該当するデプロイ ユニットの実行中の全インスタンスに分散され、均等分散アルゴリズムを使ってコンピューティングが並列化されます。 1 つのリースは特定の時点で 1 つのインスタンスによって所有されるため、インスタンスの数をリースの数より大きくすることはできません。

インスタンス数は増減する可能性があります。 変更フィード プロセッサは、負荷をそれに応じて再分散することで動的に調整します。 デプロイ ユニットは同じリース コンテナーを共有できますが、leasePrefix の値はそれぞれ異なっている必要があります。

さらに、コンテナーのスループットまたはストレージの増加に応じて、変更フィード プロセッサでコンテナーのスケールを動的に調整できます。 コンテナーが拡張されると、変更フィード プロセッサでは、リースを動的に増やし、既存のインスタンス間で新しいリースを分散することによって、このシナリオが透過的に処理されます。

開始時刻

既定では、変更フィード プロセッサは、初めて開始したときに、リース コンテナーを初期化し、その処理のライフ サイクルを開始します。 変更フィード プロセッサが初めて初期化される前に監視対象コンテナー内で発生した変更は検出されません。

以前の日時からの読み取り

で setStartTime を設定することで、options以降の変更を読み取るよう変更フィード プロセッサを初期化することができます。 変更フィード プロセッサは、その特定の日時に対して初期化され、それ以降に発生した変更の読み取りを開始します。

最初からの読み取り

このサンプルでは setStartFromBeginning が false に設定されていますが、これは既定値と同じです。 データの移行やコンテナーの履歴全体の分析など、他のシナリオでは、"そのコンテナーの有効期間の最初" から変更フィードを読み取る必要があります。 これを行うには、setStartFromBeginning を true に設定します。 変更フィード プロセッサは初期化され、コンテナーの有効期間の最初から変更の読み取りを開始します。