Uso del calculador de la fuente de cambios

Como modelo de inserción para las notificaciones automáticas

Al igual que el procesador de la fuente de cambios, el estimador de la fuente de cambios funciona como un modelo de inserción. El estimador mide la diferencia entre el último elemento procesado (definido por el estado del contenedor de concesiones) y el último cambio en el contenedor e inserta este valor en un delegado. El intervalo en el que se toma la medida también puede personalizarse con un valor predeterminado de 5 segundos.

Por ejemplo, si el procesador de fuente de cambios usa el modo de versión más reciente y se define de esta manera:

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

La manera correcta de inicializar un calculador para medir ese procesador sería usar GetChangeFeedEstimatorBuilder como se explica a continuación:

ChangeFeedProcessor changeFeedEstimator = monitoredContainer
    .GetChangeFeedEstimatorBuilder("changeFeedEstimator", Program.HandleEstimationAsync, TimeSpan.FromMilliseconds(1000))
    .WithLeaseContainer(leaseContainer)
    .Build();

Donde tanto el procesador como el calculador comparten el mismo leaseContainer y el mismo nombre.

Los otros dos parámetros son el delegado, que recibe un número que representa cuántos cambios están pendientes de ser leídos por el procesador y el intervalo de tiempo en el que desea que se tome esta medida.

Un ejemplo de un delegado que recibe la estimación es:

static async Task HandleEstimationAsync(long estimation, CancellationToken cancellationToken)
{
    if (estimation > 0)
    {
        Console.WriteLine($"\tEstimator detected {estimation} items pending to be read by the Processor.");
    }

    await Task.Delay(0);
}

Puede enviarla a su solución de supervisión y usarla para saber cómo se comporta el progreso con el tiempo.

Como una estimación detallada a petición

A diferencia del modelo de inserción, hay una alternativa que le permite obtener la estimación a petición. Este modelo también proporciona información más detallada:

• Retardo estimado por concesión.
• Instancia que posee y procesa cada concesión, de modo que puede identificar si hay algún problema en una instancia.

Si el procesador de la fuente de cambios se define de la siguiente manera:

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

Puede crear el estimador con la misma configuración de concesión:

ChangeFeedEstimator changeFeedEstimator = monitoredContainer
    .GetChangeFeedEstimator("changeFeedEstimator", leaseContainer);

Y siempre que quiera, con la frecuencia que necesite, puede obtener la estimación detallada:

Console.WriteLine("Checking estimation...");
using FeedIterator<ChangeFeedProcessorState> estimatorIterator = changeFeedEstimator.GetCurrentStateIterator();
while (estimatorIterator.HasMoreResults)
{
    FeedResponse<ChangeFeedProcessorState> states = await estimatorIterator.ReadNextAsync();
    foreach (ChangeFeedProcessorState leaseState in states)
    {
        string host = leaseState.InstanceName == null ? $"not owned by any host currently" : $"owned by host {leaseState.InstanceName}";
        Console.WriteLine($"Lease [{leaseState.LeaseToken}] {host} reports {leaseState.EstimatedLag} as estimated lag.");
    }
}

Cada ChangeFeedProcessorState contiene la información de concesión y retraso, y también quién es la instancia actual que lo posee.

Implementación del estimador

No es necesario implementar el estimador de fuente de cambios como parte del procesador de fuente de cambios ni formar parte del mismo proyecto. Se recomienda implementar el estimador en una instancia independiente de los procesadores. Una sola instancia del estimador puede realizar un seguimiento del progreso de todas las concesiones e instancias de la implementación del procesador de fuente de cambios.

Cada estimación consume unidades de solicitud de los contenedores supervisados y de concesión. Una frecuencia de 1 minuto entre sí es un buen punto de partida; cuanto menor sea la frecuencia, mayor será la cantidad de unidades de solicitud consumidas.

En este ejemplo se usa el procesador de fuente de cambios en el modo de versión más reciente con la estimación del retraso en el procesamiento de eventos de fuente de cambios. En la aplicación: los documentos se insertan en un contenedor (el "contenedor de fuentes") y, mientras tanto, otra aplicación de trabajo o subproceso de trabajo extrae documentos insertados de la fuente de cambios del contenedor de fuente y funciona en ellos de alguna manera.

El procesador de fuente de cambios se compila y se inicia de la siguiente manera:

ChangeFeedProcessor changeFeedProcessorMainInstance = new ChangeFeedProcessorBuilder()
    .hostName("SampleHost_1")
    .feedContainer(feedContainer)
    .leaseContainer(leaseContainer)
    .handleChanges(handleChangesWithLag())
    .buildChangeFeedProcessor();

try {
    changeFeedProcessorMainInstance
        .start()
        .subscribeOn(Schedulers.boundedElastic())
        .timeout(Duration.ofSeconds(10))
        .then(Mono.just(changeFeedProcessorMainInstance)
                  .delayElement(Duration.ofSeconds(10))
                  .flatMap(value -> changeFeedProcessorMainInstance
                      .stop()
                      .subscribeOn(Schedulers.boundedElastic())
                      .timeout(Duration.ofSeconds(10))))
        .subscribe();
} catch (Exception ex) {
    logger.error("Change feed processor did not start and stopped in the expected time", ex);
    throw ex;
}

La comprobación del retardo del procesador de fuente de cambios se puede realizar en una aplicación independiente (como un monitor de estado) siempre que se usen los mismos contenedores de entrada (feedContainer y leaseContainer) y el mismo prefijo de concesión (CONTAINER_NAME + "-lease"). El código estimador requiere que el procesador de fuente de cambios tenga la oportunidad de inicializar completamente los documentos de leaseContainer.

El estimador calcula el retraso acumulado recuperando el estado actual del procesador de fuente de cambios y sumando los valores de retraso estimados para cada evento que se está procesando:

Mono<List<ChangeFeedProcessorState>> currentState = changeFeedProcessorMainInstance.getCurrentState();
currentState.map(state -> {
    for (ChangeFeedProcessorState changeFeedProcessorState : state) {
        totalLag.addAndGet(changeFeedProcessorState.getEstimatedLag());
    }
    return state;
}).subscribe();

// Initially totalLag should be zero
logger.info("Initially total lag is : {}", totalLag.get());

totalLag.set(0);
currentState = changeFeedProcessorSideCart.getCurrentState();
currentState.map(state -> {
    for (ChangeFeedProcessorState changeFeedProcessorState : state) {
        totalLag.addAndGet(changeFeedProcessorState.getEstimatedLag());
    }
    return state;
}).subscribe();

// Initially totalLag should be zero
logger.info("Initially total lag is : {}", totalLag.get());

El retraso total inicialmente debe ser cero y finalmente debe ser mayor o igual que el número de documentos creados. El valor total de retraso se puede registrar o usar para su posterior análisis, lo que permite supervisar el rendimiento del procesamiento de la fuente de cambios e identificar posibles cuellos de botella o retrasos en el sistema:

totalLag.set(0);
changeFeedProcessorSideCart.start().block();
currentState = changeFeedProcessorSideCart.getCurrentState();
currentState.map(state -> {
    for (ChangeFeedProcessorState changeFeedProcessorState : state) {
        totalLag.addAndGet(changeFeedProcessorState.getEstimatedLag());
    }
    return state;
}).block();

// Finally, totalLag should be greater or equal to the number of documents created
logger.info("Finally total lag is : {}", totalLag.get());

Un ejemplo de delegado que recibe cambios y los controla con retraso es:

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

        try {
            Thread.sleep(Duration.ofSeconds(5).toMillis());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        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();
            }
        }
        logger.info("End handleChangesWithLag()");
    };
}