Processador do feed de alterações no Azure Cosmos DB

O processador de feed de alterações no .NET está disponível para o modo de versão mais recente e modo de todas as versões e exclusões. Todos os modos de versões e exclusões estão em versão prévia e têm suporte para o processador do feed de alterações a partir da versão 3.40.0-preview.0. O ponto de entrada para ambos os modos é sempre o contêiner monitorado.

Para ler usando o modo de versão mais recente, em uma instância Container, você chama 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;
}

Para ler usando o modo de todas as versões e exclusões, chame GetChangeFeedProcessorBuilderWithAllVersionsAndDeletes da instância 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();

Para ambos os modos, o primeiro parâmetro é um nome distinto que descreve a meta deste processador. O segundo nome é a implementação do delegado que manipula as alterações.

Aqui está um exemplo de um delegado para o modo de versão mais 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.");
}

Aqui está um exemplo de um delegado para o modo de todas as versões e exclusões:

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 with id {item.Previous.id}.");
        }
        else
        {
            Console.WriteLine($"\tDetected {item.Metadata.OperationType} operation for item with id {item.Current.id}.");
        }
        // Simulate work
        await Task.Delay(1);
    }
}

Posteriormente, você define o nome da instância de computação ou o identificador exclusivo usando WithInstanceName. O nome da instância de computação deve ser exclusivo e diferente para cada instância de computação que você está implantando. Você define o contêiner para manter o estado de concessão usando WithLeaseContainer.

Chamar Build fornecerá a instância do processador que você pode iniciar chamando StartAsync.

Ciclo de vida do processamento

O ciclo de vida normal de uma instância de host é:

1. Ler o feed de alterações.
2. Se não houver alterações, suspenda por um período predefinido (personalizável usando WithPollInterval no Construtor) e acesse o nº 1.
3. Se houver alterações, envie-as para o delegado.
4. Quando o delegado terminar de processar as alterações com êxito, atualize o repositório de concessão com o último ponto processado no tempo e vá para 1.

Tratamento de erros

O processador do feed de alterações é resiliente aos erros de código do usuário. Se a implementação do seu representante tiver uma exceção não tratada (Etapa nº 4), o thread que está processando esse lote específico de alterações será interrompido e um novo thread será criado. O novo thread verifica o ponto no tempo mais recente que o armazenamento de concessão salvou para esse intervalo de valores de chave de partição. O novo thread é reiniciado a partir daí, enviando efetivamente o mesmo lote de alterações para o representante. Esse comportamento continuará até que seu representante processe as alterações corretamente e, por isso, o processador do feed de alterações tem uma garantia de "pelo menos uma vez".

Para impedir que o processador do feed de alterações fique "preso" repetindo continuamente o mesmo lote de alterações, adicione lógica ao código delegado para gravar documentos, mediante exceção, em uma fila de mensagens com erros. Esse design garante o controle das alterações não processadas enquanto ainda pode continuar a processar alterações futuras. A fila de mensagens com erros pode ser outro contêiner do Azure Cosmos DB. O armazenamento de dados exato não importa. Você só quer que as alterações não processadas sejam mantidas.

Você também pode usar o avaliador do feed de alterações para monitorar o progresso das instâncias do seu processador do feed de alterações à medida que elas leem o feed de alterações, ou você pode usar as notificações do ciclo de vida para detectar falhas adjacentes.

Notificações do ciclo de vida

Você pode conectar o processador do feed de alterações a qualquer evento relevante no seu ciclo de vida. Você pode optar por ser notificado sobre um ou todos eles. A recomendação é registrar pelo menos a notificação de erro:

• Registre um manipulador do WithLeaseAcquireNotification para ser notificado quando o host atual adquirir uma concessão para começar a processá-la.
• Registre um manipulador do WithLeaseReleaseNotification para ser notificado quando o host atual liberar uma concessão e parar de processá-la.
• Registre um manipulador de WithErrorNotification para ser notificado quando o host atual encontrar uma exceção durante o processamento. Você deve conseguir distinguir se a origem é o representante do usuário (uma exceção não tratada) ou um erro que o processador encontra ao tentar acessar o contêiner monitorado (por exemplo, problemas de rede).

As notificações do ciclo de vida estão disponíveis em ambos os modos do feed de alterações. Aqui está um exemplo de notificações do ciclo de vida no modo de versão mais 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();

Unidade de implantação

Uma única unidade de implantação do processador do feed de alterações consiste de uma ou mais instâncias de computação que têm o mesmo valor de processorName e a mesma configuração de contêiner de concessão, mas cada uma com nomes de instância diferentes. Você pode ter várias unidades de implantação em que cada uma tem um fluxo de negócios diferente para as alterações e cada unidade de implantação que consiste em uma ou mais instâncias.

Por exemplo, você pode ter uma unidade de implantação que aciona uma API externa sempre que houver uma alteração no seu contêiner. Outra unidade de implantação pode mover dados em tempo real cada vez que houver uma alteração. Quando ocorre uma alteração no seu contêiner monitorado, todas as suas unidades de implantação serão notificadas.

Escala dinâmica

Conforme mencionado anteriormente, você pode ter uma ou mais instâncias de computação em uma unidade de implementação. Para aproveitar a distribuição de computação na unidade de implantação, os únicos requisitos principais são:

• Todas as instâncias devem ter a mesma configuração do contêiner de concessão.
• Todas as instâncias devem ter o mesmo valor de processorName.
• Cada instância precisa ter um nome de instância diferente (WithInstanceName).

Se essas três condições se aplicarem, o processador do feed de alterações distribuirá todas as concessões que estão no contêiner de concessões entre todas as instâncias executadas dessa unidade de implantação e paralelizará a computação usando um algoritmo de distribuição igual. Uma concessão só pode pertencer a uma instância em um determinado momento, por isso, o número máximo de instâncias não deve ser maior que o número de concessões.

O número de instâncias pode aumentar e diminuir. O processador do feed de alterações ajusta dinamicamente a carga redistribuindo de acordo.

Além disso, o processador do feed de alterações pode ajustar dinamicamente a escala de um contêiner se a taxa de transferência ou o armazenamento do contêiner aumentar. Quando seu contêiner se expande, o processador do feed de alterações lida com o cenário de forma transparente, aumentando dinamicamente as concessões e distribuindo as novas concessões entre as instâncias existentes.

Hora de início

Por padrão, quando um processador do feed de alterações fore iniciado pela primeira vez, ele inicializará o contêiner de concessão e iniciará seu ciclo de vida do processamento. As alterações ocorridas no contêiner monitorado antes de o processador do feed de alterações ter sido inicializado pela primeira vez não serão detectadas.

Lendo de uma data e hora anteriores

É possível inicializar o processador do feed de alterações para ler as alterações de uma data e hora específicas, passando uma instância de DateTime para a extensão do construtor 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();

O processador do feed de alterações é inicializado para essa data e hora específicas e começa a ler as alterações que ocorreram depois.

Lendo desde o início

Em outros cenários, como nas migrações de dados ou se você estiver analisando o histórico de um contêiner, será necessário ler o feed de alterações desde o início do tempo de vida desse contêiner. Você pode usar WithStartTime na extensão do construtor, mas passar DateTime.MinValue.ToUniversalTime(), o que gera a representação UTC do valor mínimo de DateTime como no exemplo a seguir:

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

O processador do feed de alterações é inicializado e começa a ler as alterações desde o início do tempo de vida do contêiner.

Para obter exemplos completos de trabalho, consulte aqui. Um exemplo de implementação de representante ao ler o feed de alterações no modo de versão mais 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);

A implementação do representante para ler o feed de alterações no modo de todas as versões e exclusões é semelhante, mas em vez de chamar .handleChanges(), chame .handleAllVersionsAndDeletesChanges(). O modo de todas as versões e exclusões está em versão prévia e disponível na versão do SDK do Java > = 4.42.0.

Veja um exemplo:

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

Em qualquer modo do feed de alterações, você pode atribuí-lo a changeFeedProcessorInstance e passar os parâmetros do nome da instância de computação (hostName), o contêiner monitorado (chamado de feedContainer aqui ) e o leaseContainer. Em seguida, inicie o processador do feed de alterações:

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 de vida do processamento

O ciclo de vida normal de uma instância de host é:

1. Ler o feed de alterações.
2. Se não houver alterações, suspenda por um período predefinido (personalizável com options.setFeedPollDelay no construtor) e acesse o número 1.
3. Se houver alterações, envie-as para o delegado.
4. Quando o representante terminar de processar as alterações com sucesso, atualize o repositório de concessão usando o ponto processado mais recente no tempo e acesse o nº 1.

Tratamento de erros

O processador do feed de alterações é resiliente aos erros de código do usuário. Se a implementação do seu representante tiver uma exceção sem tratamento (Etapa 4), o thread que está processando esse lote específico de alterações será interrompido e um novo thread será criado. O novo thread verifica o último ponto no tempo que o repositório de concessões salvou em relação a esse intervalo de valores de chave de partição e reinicia a partir daí, enviando com eficácia o mesmo lote de alterações para o representante. Esse comportamento continua até que seu representante processe as alterações corretamente. É por isso que o processador do feed de alterações tem uma garantia de "pelo menos uma vez".

Para impedir que o processador do feed de alterações fique "preso" repetindo continuamente o mesmo lote de alterações, adicione lógica ao código delegado para gravar documentos, mediante exceção, em uma mensagem com erros. Esse design garante o controle das alterações não processadas enquanto ainda pode continuar a processar alterações futuras. A mensagem com erros pode ser outro contêiner do Azure Cosmos DB. O armazenamento de dados exato não importa. Você só quer que as alterações não processadas sejam mantidas.

Você também pode o avaliador do feed de alterações para monitorar o progresso das instâncias do seu processador do feed de alterações à medida que elas leem o feed de alterações.

Unidade de implantação

Uma unidade de implantação do processador do feed de alterações consiste de uma ou mais instâncias de computação que tenham a mesma configuração de contêiner de concessão e o mesmo leasePrefix, mas valores de hostName diferentes. Você pode ter várias unidades de implantação nas quais cada uma tem um fluxo de negócios diferente para as alterações e cada unidade de implantação consiste de uma ou mais instâncias.

Por exemplo, você pode ter uma unidade de implantação que aciona uma API externa sempre que houver uma alteração no seu contêiner. Outra unidade de implantação pode mover dados em tempo real cada vez que houver uma alteração. Quando ocorre uma alteração no seu contêiner monitorado, todas as suas unidades de implantação serão notificadas.

Escala dinâmica

Conforme mencionado anteriormente, você pode ter uma ou mais instâncias de computação em uma unidade de implementação. Para aproveitar a distribuição de computação na unidade de implantação, os únicos requisitos principais são:

• Todas as instâncias devem ter a mesma configuração do contêiner de concessão.
• Todas as instâncias devem ter o mesmo valor definido em options.setLeasePrefix (ou nenhum valor definido).
• Cada instância precisa ter um hostName diferente.

Se essas três condições se aplicarem, o processador do feed de alterações distribuirá todas as concessões no contêiner de concessões entre todas as instâncias em execução dessa unidade de implantação e paralelizará a computação usando um algoritmo de distribuição igual. Uma concessão só pode pertencer a uma instância em um determinado momento, por isso, o número máximo de instâncias não deve ser maior que o número de concessões.

O número de instâncias pode aumentar e diminuir. O processador do feed de alterações ajusta dinamicamente a carga redistribuindo de acordo. As unidades de implantação podem compartilhar o mesmo contêiner de concessão, mas cada uma deve ter um valor de leasePrefix diferente.

Além disso, o processador do feed de alterações pode ajustar dinamicamente a escala de um contêiner se a taxa de transferência ou o armazenamento do contêiner aumentar. Quando seu contêiner se expande, o processador do feed de alterações lida com o cenário de forma transparente, aumentando dinamicamente as concessões e distribuindo as novas concessões entre as instâncias existentes.

Hora de início

Por padrão, quando um processador do feed de alterações fore iniciado pela primeira vez, ele inicializará o contêiner de concessão e iniciará seu ciclo de vida do processamento. As alterações ocorridas no contêiner monitorado antes de o processador do feed de alterações ter sido inicializado pela primeira vez não serão detectadas.

Lendo de uma data e hora anteriores

É possível inicializar o processador do feed de alterações para ler as alterações começando em uma data e hora específicas definindo setStartTime em options. O processador do feed de alterações é inicializado nessa data e hora específicas e começa a ler as alterações que ocorreram depois.

Lendo desde o início

No exemplo, setStartFromBeginning é definido como false, que é o mesmo que o valor padrão. Em outros cenários, como nas migrações de dados ou se você estiver analisando o histórico de um contêiner, será necessário ler o feed de alterações desde o início do tempo de vida desse contêiner. Para fazer isso, você pode definir setStartFromBeginning como true. O processador do feed de alterações é inicializado e começa a ler as alterações desde o início do tempo de vida do contêiner.