Procesadores de fuente de cambios de Azure Cosmos DB

El procesador de fuente de cambios de .NET está disponible para modo de versión más reciente y todas las versiones y elimina el modo. Todas las versiones y el modo de eliminaciones están en versión preliminar y se admiten para el procesador de fuente de cambios a partir de la versión 3.40.0-preview.0. El punto de entrada para ambos modos siempre es el contenedor supervisado.

Para leer con el modo de versión más reciente, en una instancia Container, llame a 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 leer con todas las versiones y el modo de eliminaciones, llame a GetChangeFeedProcessorBuilderWithAllVersionsAndDeletes desde la instancia 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();

En ambos modos, el primer parámetro es un nombre distinto que describe el objetivo de este procesador. El segundo nombre es la implementación del delegado que controla los cambios.

Este es un ejemplo de un delegado para el modo de versión más reciente:

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

Este es un ejemplo de un delegado para todas las versiones y el modo de eliminaciones:

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

Después, se define el nombre de la instancia de proceso o el identificador único mediante WithInstanceName. El nombre de la instancia de proceso debe ser único y diferente para cada instancia de proceso que vaya a implementar. Debe establecer el contenedor para mantener el estado de concesión mediante WithLeaseContainer.

La llamada a Build le proporciona la instancia del procesador que puede iniciar mediante una llamada a StartAsync.

Ciclo de vida de procesamiento

El ciclo de vida normal de una instancia de host es:

1. Leer la fuente de cambios.
2. Si no hay ningún cambio, se mantiene en suspensión durante un periodo de tiempo predefinido (personalizable con WithPollInterval en el generador) y se vuelve al primer paso.
3. Si hay cambios, enviarlos al delegado.
4. Cuando el delegado termina de procesar los cambios correctamente, actualizar el almacén de concesión con el último punto en el tiempo e ir al nº 1.

Control de errores

El procesador de fuente de cambios es resistente a los errores de código de usuario. Si la implementación del delegado tiene una excepción no controlada (paso nº 4), el subproceso que está procesando ese lote específico de cambios se detiene y finalmente se crea un nuevo subproceso. El nuevo subproceso comprueba el último punto en el tiempo que el almacén de concesiones ha guardado para ese intervalo de valores de clave de partición. El nuevo subproceso se reiniciará desde allí, enviando de forma eficaz el mismo lote de cambios al delegado. Este comportamiento continuará hasta que el delegado procese los cambios correctamente, y es el motivo por el que el procesador de fuente de cambios tiene una garantía de "al menos una vez".

Para evitar que el procesador de fuente de cambios se "atasque" al reintentar continuamente el mismo lote de cambios, debe agregar lógica al código de delegado para escribir documentos, en caso de excepción, en una cola de mensajes fallidos. Este diseño garantiza que se pueda realizar un seguimiento de los cambios sin procesar a la vez que se siguen procesando cambios futuros. La cola de mensajes fallidos podría ser otro contenedor de Azure Cosmos DB. El almacén de datos exacto no importa, sino simplemente que se conserven los cambios sin procesar.

Además, puede usar el estimador de la fuente de cambios para supervisar el progreso de las instancias del procesador de la fuente de cambios a medida que leen la fuente de cambios, o bien puede usar la notificación del ciclo de vida para detectar errores subyacentes.

Notificaciones de ciclo de vida

Puede conectar el procesador de fuente de cambios a cualquier evento relevante en su ciclo de vida. Puede optar por recibir notificaciones de uno o de todos ellos. La recomendación es registrar al menos la notificación del error:

• Registre un controlador para que WithLeaseAcquireNotification reciba una notificación cuando el host actual adquiera una concesión para empezar a procesarlo.
• Registre un controlador para que WithLeaseReleaseNotification reciba una notificación cuando el host actual libere una concesión y deje de procesarla.
• Registre un controlador de WithErrorNotification para recibir una notificación cuando el host actual encuentre una excepción durante el procesamiento. Debe poder distinguir si el origen es el delegado de usuario (una excepción no controlada) o un error que el procesador encuentra cuando intenta acceder al contenedor supervisado (por ejemplo, problemas de red).

Las notificaciones del ciclo de vida están disponibles en ambos modos de fuente de cambios. Este es un ejemplo de notificaciones del ciclo de vida en el modo de versión más reciente:

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

Unidad de implementación

Una sola unidad de implementación de procesador de fuente de cambios está formada por una o varias instancias de proceso con el mismo valor de processorName y la misma configuración de contenedor de concesión, pero con un nombre de instancia diferente en cada caso. Puede tener muchas unidades de implementación donde cada una tiene un flujo de negocio diferente para los cambios y cada unidad de implementación que consta de una o varias instancias.

Por ejemplo, podría tener una unidad de implementación que desencadene una API externa cada vez que se produzca un cambio en el contenedor. Otra unidad de implementación podría trasladar datos, en tiempo real cada vez que se produzca un cambio. Cuando se produzca un cambio en el contenedor supervisado, todas las unidades de implementación recibirán una notificación.

Escalado dinámico

Como se mencionó antes, en una unidad de implementación puede tener una o más instancias de proceso. Para beneficiarse de la distribución de proceso dentro de la unidad de implementación, los únicos requisitos clave son los siguientes:

• Todas las instancias deben tener la misma configuración de contenedor de concesión.
• Todas las instancias deben tener el mismo valor de processorName.
• Cada instancia tiene que tener un nombre de instancia diferente (WithInstanceName).

Si estas tres condiciones son aplicables, el procesador de fuente de cambios, mediante el uso de un algoritmo de distribución equitativa, distribuirá todas las concesiones en el contenedor de concesiones en todas las instancias en ejecución de esa unidad de implementación y paralizará el proceso. Una concesión es propiedad de una instancia en un momento dado, por lo que el número máximo de instancias no debería ser superior al número de concesiones.

El número de instancias puede crecer y reducirse. El procesador de fuente de cambios ajusta dinámicamente la carga redistribuyéndola en consecuencia.

Además, el procesador de fuente de cambios puede ajustarse de forma dinámica a la escala del contenedor si aumenta el rendimiento o el almacenamiento de este. Cuando el contenedor crece, el procesador de fuente de cambios controla de forma transparente estos escenarios al aumentar dinámicamente las concesiones y distribuir las nuevas concesiones entre las instancias existentes.

Hora de inicio

De manera predeterminada, cuando se inicie un procesador de fuente de cambios por primera vez, inicializará el contenedor de concesiones e iniciará su ciclo de vida de procesamiento. No se detectó ningún cambio ocurrido en el contenedor supervisado antes de que se inicializara por primera vez el procesador de fuente de cambios.

Lectura desde una fecha y hora anteriores

Es posible inicializar el procesador de fuente de cambios para leer los cambios a partir de una fecha y hora específicas; para ello, pase una instancia de DateTime a la extensión del generador 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();

El procesador de fuente de cambios se inicializará para esa fecha y hora específicas y comenzará a leer los cambios que se produzcan después.

Lectura desde el principio

En otros escenarios, como las migraciones de datos o el análisis de todo el historial de un contenedor, es necesario leer la fuente de cambios desde el principio de la vigencia de ese contenedor. Para ello, podemos usar WithStartTime en la extensión del generador, pero pasando DateTime.MinValue.ToUniversalTime(), de forma que se genere la representación UTC del valor DateTime mínimo, tal como se muestra en el siguiente ejemplo:

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

El procesador de fuente de cambios se inicializará y comenzará a leer los cambios desde el principio de la vigencia del contenedor.

Para obtener ejemplos de trabajo completos, consulte aquí. Un ejemplo de una implementación de delegado al leer la fuente de cambios en el modo de versión más reciente es el siguiente:

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

La implementación del delegado para leer la fuente de cambios en el modo de todas las versiones y eliminaciones es similar, pero en lugar de llamar a .handleChanges() se llama a .handleAllVersionsAndDeletesChanges(). El modo de Todas las versiones y Eliminaciones está en versión preliminar y está disponible en la versión > = 4.42.0 del SDK de Java.

Este es un ejemplo:

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

En cualquiera de los modos de la fuente de cambios, puede asignar esto a changeFeedProcessorInstance y pasar los parámetros del nombre de instancia de proceso (hostName), el contenedor supervisado (aquí denominado feedContainer) y leaseContainer. A continuación, iniciamos el procesador de fuente de cambios:

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 de procesamiento

El ciclo de vida normal de una instancia de host es:

1. Leer la fuente de cambios.
2. Si no hay ningún cambio, se mantiene en suspensión durante un periodo de tiempo predefinido (personalizable con options.setFeedPollDelay en el generador) y se vuelve al primer paso.
3. Si hay cambios, enviarlos al delegado.
4. Cuando el delegado termina de procesar los cambios correctamente, se actualiza el almacén de concesión con el último punto en el tiempo procesado y se vuelve al primer paso.

Control de errores

El procesador de fuente de cambios es resistente a los errores de código de usuario. Esto significa que si la implementación del delegado tuviera una excepción no controlada (paso nº 4), el subproceso que está procesando ese lote específico de cambios se detendrá y se creará un nuevo subproceso. El nuevo subproceso comprueba el último punto en el tiempo que el almacén de concesiones ha guardado para ese intervalo de valores de clave de partición, y se reinicia desde allí, enviando de manera eficaz el mismo lote de cambios al delegado. Este comportamiento continúa hasta que el delegado procesa los cambios correctamente. Es la razón por la que el procesador de fuente de cambios tiene una garantía de "al menos una vez".

Para evitar que el procesador de fuente de cambios se "atasque" al reintentar continuamente el mismo lote de cambios, debe agregar lógica al código del delegado para escribir documentos, en caso de excepción, en una cola de mensajes fallidos. Este diseño garantiza que se pueda realizar un seguimiento de los cambios sin procesar a la vez que se siguen procesando cambios futuros. La cola de mensajes fallidos podría ser otro contenedor de Azure Cosmos DB. El almacén de datos exacto no importa, sino simplemente que se conserven los cambios sin procesar.

También puede usar el estimador de fuente de cambios para supervisar el progreso de las instancias del procesador de fuente de cambios a medida que leen la fuente de cambios.

Unidad de implementación

Una sola unidad de implementación de procesador de fuente de cambios está formada por una o varias instancias de proceso que tienen el mismo leasePrefix y la misma configuración de contenedor de concesión, pero con diferentes valores de hostName. Puede tener muchas unidades de implementación donde cada una tiene un flujo de negocio diferente para los cambios y cada unidad de implementación que consta de una o varias instancias.

Por ejemplo, podría tener una unidad de implementación que desencadene una API externa cada vez que se produzca un cambio en el contenedor. Otra unidad de implementación podría trasladar datos, en tiempo real cada vez que se produzca un cambio. Cuando se produzca un cambio en el contenedor supervisado, todas las unidades de implementación recibirán una notificación.

Escalado dinámico

Como se mencionó antes, en una unidad de implementación puede tener una o más instancias de proceso. Para beneficiarse de la distribución de proceso dentro de la unidad de implementación, los únicos requisitos clave son los siguientes:

• Todas las instancias deben tener la misma configuración de contenedor de concesión.
• Todas las instancias deben tener el mismo valor establecido en options.setLeasePrefix (o ninguno establecido en absoluto).
• Cada instancia tiene que tener un hostName diferente.

Si estas tres condiciones son aplicables, el procesador de fuente de cambios, mediante el uso de un algoritmo de distribución equitativa, distribuirá todas las concesiones en el contenedor de concesiones en todas las instancias en ejecución de esa unidad de implementación y paralizará el proceso. Una concesión es propiedad de una instancia en un momento dado, por lo que el número máximo de instancias no debería ser superior al número de concesiones.

El número de instancias puede crecer y reducirse. El procesador de fuente de cambios ajusta dinámicamente la carga redistribuyéndola en consecuencia. Las unidades de implementación pueden compartir el mismo contenedor de concesión, pero deben tener un valor de leasePrefix diferente.

Además, el procesador de fuente de cambios puede ajustarse de forma dinámica a la escala del contenedor si aumenta el rendimiento o el almacenamiento de este. Cuando el contenedor crece, el procesador de fuente de cambios controla de forma transparente estos escenarios al aumentar dinámicamente las concesiones y distribuir las nuevas concesiones entre las instancias existentes.

Hora de inicio

De manera predeterminada, cuando se inicie un procesador de fuente de cambios por primera vez, inicializará el contenedor de concesiones e iniciará su ciclo de vida de procesamiento. No se detectó ningún cambio ocurrido en el contenedor supervisado antes de que se inicializara por primera vez el procesador de fuente de cambios.

Lectura desde una fecha y hora anteriores

Es posible inicializar el procesador de fuente de cambios para leer los cambios a partir de una fecha y hora específicas, estableciendo setStartTime en options. El procesador de fuente de cambios se inicializará para esa fecha y hora específicas y comenzará a leer los cambios que se produzcan después.

Lectura desde el principio

En el ejemplo, se establece setStartFromBeginning en false, que es el mismo que el valor predeterminado. En otros escenarios, como las migraciones de datos o el análisis de todo el historial de un contenedor, es necesario leer la fuente de cambios desde el principio de la vigencia de ese contenedor. Para ello, puede establecer setStartFromBeginning en true. El procesador de fuente de cambios se inicializará y comenzará a leer los cambios desde el principio de la vigencia del contenedor.