Processor voor wijzigingenfeed in Azure Cosmos DB

De processor van de wijzigingenfeed in .NET is beschikbaar voor de nieuwste versiemodus en alle versies en verwijdermodus. Alle versies en verwijderingsmodus is in preview en wordt ondersteund voor de wijzigingenfeedprocessor vanaf versie 3.40.0-preview.0. Het toegangspunt voor beide modi is altijd de bewaakte container.

Als u wilt lezen met de nieuwste versiemodus, roept Containeru in een GetChangeFeedProcessorBuilder exemplaar het volgende aan:

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

Als u wilt lezen in alle versies en in de verwijdermodus, roep GetChangeFeedProcessorBuilderWithAllVersionsAndDeletes aan vanuit het Container exemplaar.

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

Voor beide modi is de eerste parameter een afzonderlijke naam die het doel van deze processor beschrijft. De tweede naam is de gedelegeerde-implementatie die wijzigingen verwerkt.

Hier volgt een voorbeeld van een gemachtigde voor de nieuwste versiemodus:

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

Hier volgt een voorbeeld van een gemachtigde voor alle versies en de verwijdermodus:

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

Daarna definieert u de naam van het rekenproces of de unieke id met behulp van WithInstanceName. De naam van het rekenproces moet uniek en verschillend zijn voor elk rekenproces dat u implementeert. U stelt de container in om de leasestatus te behouden met behulp van WithLeaseContainer.

Het aanroepen van Build geeft u het processorexemplaar dat u kunt starten door StartAsync aan te roepen.

Verwerkingslevenscyclus

De normale levenscyclus van een host-exemplaar is:

1. Lees de wijzigingenfeed.
2. Als er geen wijzigingen zijn, wacht een vooraf gedefinieerde hoeveelheid tijd (aanpasbaar met behulp van WithPollInterval in de Builder) en ga naar stap 1.
3. Als er wijzigingen zijn, stuurt u deze naar de gemachtigde.
4. Wanneer de gemachtigde succesvol klaar is met het verwerken van de wijzigingen, werkt men het leaseregister bij met het laatst verwerkte tijdstip en gaat men naar stap 1.

Foutafhandeling

De wijzigingenfeedprocessor is bestand tegen fouten in gebruikerscode. Als uw gedelegeerde-implementatie een niet-verwerkte uitzondering heeft (stap 4), wordt de thread die die bepaalde batch wijzigingen verwerkt, gestopt en wordt er uiteindelijk een nieuwe thread gemaakt. De nieuwe thread controleert het nieuwste tijdstip dat de verhuurwinkel heeft opgeslagen voor dat bereik van partitiesleutelwaarden. De nieuwe thread start daar opnieuw op, waarmee hetzelfde pakket wijzigingen naar de gemachtigde wordt verzonden. Dit gedrag wordt voortgezet totdat de gedelegeerde de wijzigingen correct verwerkt. Dit is de reden waarom de wijzigingenfeedverwerker ten minste één keer een garantie heeft.

Als u wilt voorkomen dat uw wijzigingenfeedprocessor voortdurend blijft hangen bij het proberen opnieuw te verwerken van dezelfde batch wijzigingen, moet u logica toevoegen aan de code die gedelegeerd is om bij uitzondering documenten naar een wachtrij met foutberichten te schrijven. Dit ontwerp zorgt ervoor dat u niet-verwerkte wijzigingen kunt bijhouden terwijl u toekomstige wijzigingen nog steeds kunt blijven verwerken. De wachtrij met foutberichten kan een andere Azure Cosmos DB-container zijn. Het exacte gegevensarchief maakt niet uit. U wilt gewoon dat de niet-verwerkte wijzigingen behouden blijven.

U kunt de wijzigingsfeedschatter ook gebruiken om de voortgang van de exemplaren van de wijzigingenfeedprocessor te controleren terwijl ze de wijzigingenfeed lezen, of u kunt levenscyclusmeldingen gebruiken om onderliggende fouten te detecteren.

Zorg ervoor dat de time-out van de clientnetwerkaanvraag langer is dan de time-out aan de serverzijde om te voorkomen dat time-outs niet overeenkomen die kunnen leiden tot verwerkingsfouten. De standaard time-out aan de serverzijde is vijf seconden. Houd fouten gerelateerd aan time-outs in de gaten en pas de configuraties dienovereenkomstig aan.

Meldingen over levenscyclus

U kunt de wijzigingenfeedprocessor verbinden met elke relevante gebeurtenis in de levenscyclus. U kunt ervoor kiezen om een of meer meldingen te ontvangen. De aanbeveling is om ten minste de foutmelding te registreren:

• Registreer een handler voor WithLeaseAcquireNotification om een melding te krijgen wanneer de huidige host een lease verkrijgt voor verwerking.
• Registreer een handler om WithLeaseReleaseNotification op de hoogte te worden gebracht wanneer de huidige host een lease vrijgeeft en stopt met verwerken ervan.
• Registreer een handler voor WithErrorNotification om op de hoogte te worden gebracht wanneer de huidige host tijdens de verwerking een uitzonderingssituatie tegenkomt. U moet kunnen onderscheiden of de bron delegeren van de gebruiker is (een niet-verwerkte uitzondering) of een fout die de processor tegenkomt wanneer deze toegang probeert te krijgen tot de bewaakte container (bijvoorbeeld netwerkproblemen).

Meldingen over levenscyclus zijn beschikbaar in beide modi voor wijzigingenfeeds. Hier volgt een voorbeeld van levenscyclusmeldingen in de nieuwste versiemodus:

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

Implementatie-eenheid

Eén implementatie-eenheid voor de wijzigingsfeedprocessor bestaat uit een of meer rekeninstanties met dezelfde waarde voor processorName en dezelfde leasecontainerconfiguratie, maar verschillende exemplaarnamen. U kunt veel implementatie-eenheden hebben waarin elke eenheid een andere bedrijfsstroom heeft voor de wijzigingen en elke implementatie-eenheid bestaat uit een of meer exemplaren.

U hebt bijvoorbeeld één implementatie-eenheid die een externe API activeert telkens wanneer er een wijziging in uw container is. Een andere implementatie-eenheid kan gegevens in realtime verplaatsen telkens wanneer er een wijziging is. Wanneer er een wijziging in uw bewaakte container plaatsvindt, worden al uw implementatie-eenheden op de hoogte gebracht.

Dynamische schaalbaarheid

Zoals eerder vermeld, kunt u binnen een implementatie-eenheid een of meer rekeninstanties hebben. Als u wilt profiteren van de rekendistributie binnen de implementatie-eenheid, zijn de enige belangrijke vereisten:

• Alle instanties moeten dezelfde configuratie voor de leasecontainer hebben.
• Alle exemplaren moeten dezelfde waarde hebben voor processorName.
• Elk exemplaar moet een andere naam hebben (WithInstanceName).

Als deze drie voorwaarden van toepassing zijn, distribueert de wijzigingenfeed-processor alle leases die zich in de leasecontainer bevinden over alle actieve exemplaren van die implementatie-eenheid, en wordt de berekening geparalleliseerd met behulp van een algoritme voor gelijke distributie. Een lease is op elk moment eigendom van één instantie, dus het aantal instanties mag niet groter zijn dan het aantal leases.

Het aantal gevallen kan toenemen en afnemen. De wijzigingenfeedprocessor past de belasting dynamisch aan door deze dienovereenkomstig te distribueren.

Bovendien kan de wijzigingenfeedprocessor dynamisch de schaal van een container aanpassen als de doorvoer of opslag van de container toeneemt. Wanneer uw container groeit, behandelt de change feed-verwerker het scenario transparant door de leases dynamisch uit te breiden en de nieuwe leases tussen bestaande instanties te verdelen.

Begintijd

Standaard wordt, wanneer een verwerker van een wijzigingenfeed voor het eerst wordt gestart, de leasecontainer geïnitialiseerd en begint de levenscyclus van de verwerking. Wijzigingen die zijn aangebracht in de bewaakte container voordat de verwerker van de wijzigingenfeed voor het eerst wordt geïnitialiseerd, worden niet gedetecteerd.

Lezen vanaf een vorige datum en tijd

Het is mogelijk om de change feed-processor te initialiseren om veranderingen te lezen vanaf een specifieke datum en tijd door een exemplaar van DateTime aan de WithStartTime builder-extensie door te geven.

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

De wijzigingenfeedverwerker wordt geïnitialiseerd voor die specifieke datum en tijd en begint de wijzigingen te lezen die daarna zijn aangebracht.

Lezen vanaf het begin

In andere scenario's, zoals in gegevensmigraties of als u de hele geschiedenis van een container analyseert, moet u de wijzigingenfeed lezen vanaf het begin van de levensduur van die container. U kunt de builder extensie gebruiken WithStartTime, maar DateTime.MinValue.ToUniversalTime() doorgeven, waarmee de UTC-weergave van de minimale DateTime waarde wordt gegenereerd, zoals in dit voorbeeld:

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

De processor van de wijzigingenfeed wordt geïnitialiseerd en begint met het lezen van wijzigingen vanaf het begin van de levensduur van de container.

Zie deze GitHub-opslagplaats voor volledige werkvoorbeelden. Een voorbeeld van een gedelegeerde implementatie bij het lezen van de wijzigingenfeed in de nieuwste versiemodus is:

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

De implementatie van gedelegeerden voor het lezen van de wijzigingenfeed in alle versies en verwijderingsmodus is vergelijkbaar, maar in plaats van .handleChanges() aan te roepen, roep .handleAllVersionsAndDeletesChanges() aan. De modus Alle versies en verwijderingen is als preview beschikbaar in de Java SDK-versie >= 4.42.0.

Hier volgt een voorbeeld:

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 beide wijzigingsfeedmodi kunt u het toewijzen aan changeFeedProcessorInstance en de parameters van de naam van de berekeningsinstantie (hostName), de bewaakte container (hier genoemd feedContainer), en de leaseContainer doorgeven. Start vervolgens de verwerker van de wijzigingenfeed:

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

Verwerkingslevenscyclus

De normale levenscyclus van een host-exemplaar is:

1. Lees de wijzigingenfeed.
2. Als er geen wijzigingen zijn, slaap voor een vooraf gedefinieerde hoeveelheid tijd (aanpasbaar met options.setFeedPollDelay in de builder) en ga naar stap 1.
3. Als er wijzigingen zijn, stuurt u deze naar de gemachtigde.
4. Wanneer de verantwoordelijke de wijzigingen succesvol heeft verwerkt, werkt u de leaseopslagplaats bij met behulp van het meest recente verwerkte tijdstip en gaat u naar stap 1.

Foutafhandeling

De wijzigingenfeedprocessor is bestand tegen fouten in gebruikerscode. Als uw gedelegeerde-implementatie een niet-verwerkte uitzondering (stap 4) heeft, wordt de thread die een bepaalde batch met wijzigingen verwerkt, gestopt en wordt er een nieuwe thread gemaakt. De nieuwe thread controleert het meest recente tijdstip waarop het leasearchief is opgeslagen voor dat bereik van partitiesleutelwaarden, waarna deze opnieuw wordt opgestart, waardoor dezelfde batch wijzigingen naar de gemachtigde worden verzonden. Dit gedrag gaat door totdat de gedelegeerde de wijzigingen correct verwerkt. Dit is de reden waarom de wijzigingenfeedprocessor ten minste eenmaal een garantie heeft.

Als u wilt voorkomen dat uw wijzigingenfeedprocessor vastloopt door continu dezelfde batch wijzigingen opnieuw te proberen, moet u logica toevoegen aan de gedelegeerde code om documenten bij een uitzondering naar een foutbericht te schrijven. Dit ontwerp zorgt ervoor dat u niet-verwerkte wijzigingen kunt bijhouden terwijl u toekomstige wijzigingen nog steeds kunt blijven verwerken. Het foutbericht kan een andere Azure Cosmos DB-container zijn. Het exacte gegevensarchief maakt niet uit. U wilt gewoon dat de niet-verwerkte wijzigingen behouden blijven.

U kunt ook de wijzigingenfeed-estimator gebruiken om de voortgang van uw wijzigingenfeedprocessor-exemplaren te controleren terwijl ze de wijzigingenfeed lezen.

Zorg ervoor dat de time-out van de clientnetwerkaanvraag langer is dan de time-out aan de serverzijde om te voorkomen dat time-outs niet overeenkomen die kunnen leiden tot verwerkingsfouten. De standaard time-out aan de serverzijde is vijf seconden. Houd fouten gerelateerd aan time-outs in de gaten en pas de configuraties dienovereenkomstig aan.

Implementatie-eenheid

Eén implementatie-eenheid voor de wijzigingenfeedprocessor bestaat uit een of meer rekeninstanties met dezelfde leasecontainerconfiguratie en dezelfde leasePrefix, maar verschillende hostName waarden. U kunt veel implementatie-eenheden hebben waarin elke eenheid een andere bedrijfsstroom heeft voor de wijzigingen en elke implementatie-eenheid bestaat uit een of meer exemplaren.

U hebt bijvoorbeeld één implementatie-eenheid die een externe API activeert telkens wanneer er een wijziging in uw container is. Een andere implementatie-eenheid kan gegevens in realtime verplaatsen telkens wanneer er een wijziging is. Wanneer er een wijziging in uw bewaakte container plaatsvindt, worden al uw implementatie-eenheden op de hoogte gebracht.

Dynamische schaalbaarheid

Zoals eerder vermeld, kunt u binnen een implementatie-eenheid een of meer rekeninstanties hebben. Als u wilt profiteren van de rekendistributie binnen de implementatie-eenheid, zijn de enige belangrijke vereisten:

• Alle exemplaren moeten dezelfde configuratie van de leasecontainer hebben
• Alle exemplaren moeten dezelfde waarde hebben ingesteld options.setLeasePrefix in (of helemaal geen set)
• Elk exemplaar moet verschillend zijn hostName

Als deze drie voorwaarden van toepassing zijn, distribueert de wijzigingsstroomprocessor alle leases in de leasecontainer over alle actieve exemplaren van die deploymenteenheid, en wordt de berekening geparallelliseerd met behulp van een algoritme voor gelijke distributie. Een lease is op elk moment eigendom van één instantie, dus het aantal instanties mag niet groter zijn dan het aantal leases.

Het aantal gevallen kan toenemen en afnemen. De wijzigingenfeedprocessor past de belasting dynamisch aan door deze dienovereenkomstig te distribueren. Implementatie-eenheden kunnen dezelfde leasecontainer delen, maar ze moeten elk een andere leasePrefix waarde hebben.

Bovendien kan de wijzigingenfeedprocessor dynamisch de schaal van een container aanpassen als de doorvoer of opslag van de container toeneemt. Wanneer uw container groeit, behandelt de change feed-verwerker het scenario transparant door de leases dynamisch uit te breiden en de nieuwe leases tussen bestaande instanties te verdelen.

Begintijd

Standaard wordt, wanneer een verwerker van een wijzigingenfeed voor het eerst wordt gestart, de leasecontainer geïnitialiseerd en begint de levenscyclus van de verwerking. Wijzigingen die zijn aangebracht in de bewaakte container voordat de verwerker van de wijzigingenfeed voor het eerst is geïnitialiseerd, worden niet gedetecteerd.

Lezen vanaf een vorige datum en tijd

Het is mogelijk om de verwerker van de wijzigingenfeed te initialiseren om wijzigingen te lezen vanaf een specifieke datum en tijd door setStartTime in te stellenoptions. De verwerker van de wijzigingenfeed wordt geïnitialiseerd voor die specifieke datum en tijd en begint met het lezen van de wijzigingen die daarna zijn aangebracht.

Lezen vanaf het begin

In het voorbeeld setStartFromBeginning is ingesteld op false, wat gelijk is aan de standaardwaarde. In andere scenario's, zoals in gegevensmigraties of als u de hele geschiedenis van een container analyseert, moet u de wijzigingenfeed lezen vanaf het begin van de levensduur van die container. Hiervoor kunt u instellen setStartFromBeginning op true. De processor van de wijzigingenfeed wordt geïnitialiseerd en begint met het lezen van wijzigingen vanaf het begin van de levensduur van de container.