Processore del feed di modifiche in Azure Cosmos DB

Il processore del feed di modifiche in .NET è disponibile per la modalità Versione più recente e quella Tutte le versioni ed eliminazioni. La modalità Tutte le versioni ed eliminazioni è in anteprima ed è supportata per il processore del feed di modifiche a partire dalla versione 3.40.0-preview.0. Il punto di ingresso per entrambe le modalità è sempre il contenitore monitorato.

Per leggere usando la modalità Versione più recente, in un'istanza Container chiamare 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;
}

Per leggere usando tutte le versioni e la modalità di eliminazione, chiamare GetChangeFeedProcessorBuilderWithAllVersionsAndDeletes dall'istanza 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();

Per entrambe le modalità, il primo parametro è un nome distinto che descrive l'obiettivo di questo processore. Il secondo nome è l'implementazione del delegato che gestisce le modifiche.

Ecco un esempio di delegato per la modalità di versione più recente:

/// <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.");
}

Di seguito è riportato un esempio di delegato per tutte le versioni ed eliminazioni in modalità:

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);
    }
}

Successivamente, si definisce il nome dell'istanza di calcolo o l'identificatore univoco usando WithInstanceName. Il nome dell'istanza di calcolo deve essere univoco e diverso per ogni istanza di calcolo che viene distribuita. Impostare il contenitore per mantenere lo stato del lease usando WithLeaseContainer.

Chiamando Build, si otterrà l'istanza del processore che è possibile avviare chiamando StartAsync.

Ciclo di vita dell'elaborazione

Il normale ciclo di vita di un'istanza dell'host è:

1. Leggere il feed di modifiche.
2. Se non sono state apportate modifiche, sospendere per un periodo di tempo predefinito personalizzabile con WithPollInterval nel generatore e procedere al passaggio 1.
3. Se ci sono modifiche, inviarle al delegato.
4. Quando il delegato termina correttamente l'elaborazione delle modifiche, aggiornare l'archivio dei lease con l'ultimo punto nel tempo elaborato e passare a #1.

Gestione degli errori

Il processore dei feed di modifiche è resiliente agli errori del codice utente. Se l'implementazione del delegato presenta un'eccezione non gestita (passaggio 4), il thread che elabora tale batch di modifiche verrà arrestato e verrà creato un nuovo thread. Il nuovo thread controlla l'ultimo momento nel tempo in cui l'archivio di lease ha salvato per tale intervallo di valori di chiave di partizione. Il nuovo thread riparte da quel momento, inviando effettivamente lo stesso batch di modifiche al delegato. Questo comportamento continua fino a quando il delegato non elabora correttamente le modifiche ed è il motivo per cui il processore di feed di modifiche ha una garanzia di tipo "at least once".

Per evitare che il processore di feed di modifiche si blocchi continuamente durante la ripetizione dei tentativi per lo stesso batch di modifiche, è necessario aggiungere nel codice del delegato la logica per scrivere documenti, in caso di eccezione, in una coda di messaggi con errori. Questa progettazione garantisce la possibilità di tenere traccia delle modifiche non elaborate continuando comunque a elaborare le modifiche future. La coda di messaggi con errori può essere un altro contenitore di Azure Cosmos DB. L'archivio dati esatto non è importante. Si vuole semplicemente salvare in modo permanente le modifiche non elaborate.

È anche possibile usare lo strumento di stima del feed di modifiche per monitorare lo stato di avanzamento delle istanze del processore durante la lettura del feed di modifiche o usare le notifiche sul ciclo di vita per rilevare gli errori sottostanti.

Notifiche sul ciclo di vita

È possibile connettere il processore di feed di modifiche a qualsiasi evento pertinente nel ciclo di vita. È possibile scegliere di ricevere una notifica per uno o per tutti. È consigliabile registrare almeno la notifica di errore:

• Registrare un gestore per WithLeaseAcquireNotification per ricevere una notifica quando l'host corrente acquisisce un lease per avviarne l'elaborazione.
• Registrare un gestore per WithLeaseReleaseNotification per ricevere una notifica quando l'host corrente rilascia un lease per arrestarne l'elaborazione.
• Registrare un gestore per WithErrorNotification per ricevere una notifica quando l'host corrente riscontra un'eccezione durante l'elaborazione. È necessario essere in grado di distinguere se l'origine è il delegato utente (un'eccezione non gestita) o un errore rilevato dal processore quando tenta di accedere al contenitore monitorato (ad esempio, problemi di rete).

Le notifiche del ciclo di vita sono disponibili in entrambe le modalità del feed di modifiche. Ecco un esempio di notifiche del ciclo di vita nella modalità versione più recente:

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();

Unità di distribuzione

Una singola unità di distribuzione del processore di feed di modifiche è costituita da una o più istanze di calcolo con lo stesso valore per processorName e la stessa configurazione del contenitore di lease, ma nomi di istanza diversi. Possono essere presenti numerose unità di distribuzione in cui ciascuna di esse prevede un flusso aziendale diverso per le modifiche e ogni unità di distribuzione è costituita da una o più istanze.

Ad esempio, si potrebbe avere un'unità di distribuzione che attiva un'API esterna ogni volta che viene apportata una modifica al contenitore. Un'altra unità di distribuzione potrebbe spostare i dati in tempo reale ogni volta che viene apportata una modifica. Quando si verifica una modifica nel contenitore monitorato, tutte le unità di distribuzione riceveranno una notifica.

Scalabilità dinamica

Come indicato in precedenza, all'interno di un'unità di distribuzione possono essere presenti una o più istanze. Per sfruttare i vantaggi della distribuzione di calcolo all'interno dell'unità di distribuzione, gli unici requisiti fondamentali sono i seguenti:

• Tutte le istanze devono avere la stessa configurazione del contenitore di lease.
• Tutte le istanze devono avere lo stesso valore per processorName.
• Ogni istanza deve avere un nome di istanza diverso (WithInstanceName).

Se queste tre condizioni sono soddisfatte, il processore di feed di modifiche distribuirà tutti i lease nel contenitore di lease in tutte le istanze in esecuzione di tale unità di distribuzione e parallelizzerà il calcolo usando un algoritmo di distribuzione uguale. Un lease è sempre di proprietà di un'istanza, quindi il numero di istanze non deve essere maggiore del numero di lease.

Il numero di istanze può aumentare e diminuire. Il processore di feed di modifiche regola dinamicamente il carico ridistribuendolo di conseguenza.

Inoltre, il processore di feed di modifiche può adattarsi dinamicamente alle dimensioni dei contenitori se la velocità effettiva o lo spazio di archiviazione del contenitore aumentano. Quando il contenitore si espande, il processore di feed di modifiche gestisce in modo trasparente questi scenari aumentando dinamicamente i lease e distribuendo i nuovi lease tra le istanze esistenti.

Data/ora di avvio

Quando un processore di feed di modifiche viene avviato per la prima volta, inizializza il contenitore di lease e avvia il ciclo di vita di elaborazione. Eventuali modifiche apportate al contenitore monitorato precedentemente alla prima inizializzazione del processore dei feed di modifiche non verranno rilevate.

Lettura da una data e un'ora precedenti

È possibile inizializzare il processore di feed di modifiche in modo da leggere le modifiche a partire da una data e ora specifiche, passando un'istanza di DateTime all'estensione del generatore 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();

Il processore di feed di modifiche verrà inizializzato per la data e l'ora specifiche e inizierà a leggere le modifiche che si sono verificate in seguito.

Lettura dall'inizio

In altri scenari, ad esempio la migrazione dei dati o l'analisi dell'intera cronologia di un contenitore, è necessario leggere il feed di modifiche dall'inizio della durata del contenitore. È possibile usare WithStartTime sull'estensione del generatore, ma passando DateTime.MinValue.ToUniversalTime(), che genera la rappresentazione UTC del valore minimo DateTime, come nell'esempio seguente:

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();

Il processore di feed di modifiche viene inizializzato e inizia a leggere le modifiche dall'inizio della durata del contenitore.

Per esempi di lavoro completi, vedere qui. Un esempio di implementazione di un delegato durante la lettura del feed di modifiche in modalità versione più recente è:

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);

L'implementazione del delegato per la lettura del feed di modifiche in modalità tutte le versioni e le eliminazioni è simile, ma invece di chiamare .handleChanges(), chiamare .handleAllVersionsAndDeletesChanges(). La modalità Tutte le versioni e le eliminazioni è disponibile in anteprima ed nella versione di Java SDK >= 4.42.0.

Ecco un esempio:

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();
}

In entrambe le modalità del feed di modifiche è possibile assegnarlo a changeFeedProcessorInstance e passare i parametri del nome dell'istanza di calcolo (hostName), il contenitore monitorato (qui denominato feedContainer) e l'oggetto leaseContainer. Avviare quindi il processore di feed di modifiche:

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();

Ciclo di vita dell'elaborazione

Il normale ciclo di vita di un'istanza dell'host è:

1. Leggere il feed di modifiche.
2. Se non sono state apportate modifiche, sospendere per un periodo di tempo predefinito personalizzabile con options.setFeedPollDelay nel generatore e procedere al passaggio 1.
3. Se ci sono modifiche, inviarle al delegato.
4. Quando il delegato termina correttamente l'elaborazione delle modifiche, aggiornare l'archivio dei lease con l'ultimo momento nel tempo elaborato e procedere al passaggio 1.

Gestione degli errori

Il processore dei feed di modifiche è resiliente agli errori del codice utente. Se l'implementazione del delegato presenta un'eccezione non gestita (passaggio 4), il thread che elabora tale batch di modifiche verrà arrestato e verrà creato un nuovo thread. Il nuovo thread verifica quale sia l'ultimo punto nel tempo dell'archivio dei leasing salvato per tale intervallo di valori della chiave di partizione e viene riavviato da quel punto, inviando efficacemente lo stesso batch di modifiche al delegato. Questo comportamento continua fino a quando il delegato non elabora correttamente le modifiche. Si tratta del motivo per cui il processore del feed di modifiche ha una garanzia di tipo "at least once".

Per evitare che il processore di feed di modifiche si blocchi continuamente durante la ripetizione dei tentativi per lo stesso batch di modifiche, è necessario aggiungere nel codice del delegato la logica per scrivere documenti, in caso di eccezione, in una coda di messaggi con errori. Questa progettazione garantisce la possibilità di tenere traccia delle modifiche non elaborate continuando comunque a elaborare le modifiche future. La coda di messaggi con errori potrebbe essere un altro contenitore di Azure Cosmos DB. L'archivio dati esatto non è importante. Si vuole semplicemente salvare in modo permanente le modifiche non elaborate.

È anche possibile usare lo strumento di stima del feed di modifiche per monitorare lo stato di avanzamento delle istanze durante la lettura del feed di modifiche.

Unità di distribuzione

Una singola unità di distribuzione del processore di feed di modifiche è costituita da una o più istanze di calcolo con la stessa configurazione del contenitore di lease e lo stesso valore per leasePrefix, ma valori diversi per hostName. Possono essere presenti numerose unità di distribuzione in cui ciascuna di esse prevede un flusso aziendale diverso per le modifiche e ogni unità di distribuzione è costituita da una o più istanze.

Ad esempio, si potrebbe avere un'unità di distribuzione che attiva un'API esterna ogni volta che viene apportata una modifica al contenitore. Un'altra unità di distribuzione potrebbe spostare i dati in tempo reale ogni volta che viene apportata una modifica. Quando si verifica una modifica nel contenitore monitorato, tutte le unità di distribuzione riceveranno una notifica.

Scalabilità dinamica

Come indicato in precedenza, all'interno di un'unità di distribuzione possono essere presenti una o più istanze. Per sfruttare i vantaggi della distribuzione di calcolo all'interno dell'unità di distribuzione, gli unici requisiti fondamentali sono i seguenti:

• Tutte le istanze devono avere la stessa configurazione del contenitore di lease.
• Tutte le istanze devono avere lo stesso valore impostato in options.setLeasePrefix (o nessun valore impostato).
• Ogni istanza deve avere un nome di istanza hostName diverso.

Se queste tre condizioni sono soddisfatte, il processore di feed di modifiche distribuirà tutti i lease nel contenitore di lease in tutte le istanze in esecuzione di tale unità di distribuzione e parallelizzerà il calcolo usando un algoritmo di distribuzione uguale. Un lease è sempre di proprietà di un'istanza, quindi il numero di istanze non deve essere maggiore del numero di lease.

Il numero di istanze può aumentare e diminuire. Il processore di feed di modifiche regola dinamicamente il carico ridistribuendolo di conseguenza. Le unità di distribuzione possono condividere lo stesso contenitore di lease, ma ognuna deve avere un valore diverso per leasePrefix.

Inoltre, il processore di feed di modifiche può adattarsi dinamicamente alle dimensioni dei contenitori se la velocità effettiva o lo spazio di archiviazione del contenitore aumentano. Quando il contenitore si espande, il processore di feed di modifiche gestisce in modo trasparente questi scenari aumentando dinamicamente i lease e distribuendo i nuovi lease tra le istanze esistenti.

Data/ora di avvio

Quando un processore di feed di modifiche viene avviato per la prima volta, inizializza il contenitore di lease e avvia il ciclo di vita di elaborazione. Eventuali modifiche apportate al contenitore monitorato precedentemente alla prima inizializzazione del processore di feed di modifiche non verranno rilevate.

Lettura da una data e un'ora precedenti

È possibile inizializzare il processore di feed di modifiche in modo da leggere le modifiche a partire da una data e ora specifiche, impostando setStartTime in options. Il processore di feed di modifiche verrà inizializzato per la data e l'ora specifiche e inizierà a leggere le modifiche che si sono verificate in seguito.

Lettura dall'inizio

Nell'esempio setStartFromBeginning è impostato su false, che corrisponde al valore predefinito. In altri scenari, ad esempio la migrazione dei dati o l'analisi dell'intera cronologia di un contenitore, è necessario leggere il feed di modifiche dall'inizio della durata del contenitore. A tale scopo, è possibile impostare setStartFromBeginning su true. Il processore di feed di modifiche viene inizializzato e inizia a leggere le modifiche dall'inizio della durata del contenitore.