CODEWITHSUNDEEP
azureazure-cosmosdb
Tdawg90
Tdawg90

Reputation: 117

Process Cosmosdb ChangeFeedProcesser onces per day at 11pmUTC

I have an Azure Service Fabric app that has a worker role that instantiates and monitors a CosmosDB Change feed. I'm trying to get it to run once per day, at the end of the day.

Everything works except for getting it to run only once, and at any specific time.

I'm currently leveraging the FeedPollDelay set at 1 day, but this will slowly crawl as the delay is based off of the completion of the last scan. This also doesn't address the start time.

I've also played with the Start Time option, but regardless of the time I set, it immediately starts processing.

        DateTime dt = DateTime.Parse("2019-07-25 23:00");
        DateTime dtt = DateTime.SpecifyKind(dt, DateTimeKind.Utc);

        ChangeFeedProcessorOptions options = new ChangeFeedProcessorOptions()
        {
            StartFromBeginning = true,
            StartTime = dtt,
            FeedPollDelay = new TimeSpan(1, 0, 0, 0),
        };

Upvotes: 0

Views: 107

Answers (1)

Matias Quaranta
Matias Quaranta

Reputation: 15603

StartTime and StartFromBeginning only work if your leases collection is empty, as per the official docs.

Let's say you want to run at at 11PM UTC, you can easily achieve that with some worker thread or process that calls StartAsync at 11PM UTC and when you want to stop it, just call StopAsync.

If you want to stop it once it has drained all the pending changes, you can use the Estimator to measure what is the current delta between the current state of the collection and your Change Feed Processor. You could use the Estimator to check, every X amount of time, how many pending changes are there, and when it reaches 0, just call StopAsync on the Processor.

Something along these lines:

public async Task StartProcessorAsync()
{
    ChangeFeedProcessorBuilder builder = new ChangeFeedProcessorBuilder();
    //setup builder

    IChangeFeedProcessor processor = await builder.BuildAsync();
    await processor.StartAsync();

    await MeasureAndStopAsync(processor);
}


public async Task MeasureAndStopAsync(IChangeFeedProcessor processor)
{
    ChangeFeedProcessorBuilder builderForEstimator = new ChangeFeedProcessorBuilder();
    //setup builder just like for processor, same lease collection configuration

    IRemainingWorkEstimator estimator = await builderForEstimator.BuildEstimatorAsync();
    do
    {
        await Task.Delay(60000); // 1 minute
        long pendingChanges = await estimator.GetEstimatedRemainingWork();            
    }
    while(pendingChanges > 0);

    // Job's done
    await processor.StopAsync();
}

Upvotes: 1

Related Questions