Обрабатывать Cosmosdb ChangeFeedProcesser в день в 23:00 UTC

Question

У меня есть приложение Azure Service Fabric, в котором есть рабочая роль, которая создает и отслеживает фид CosmosDB Change. Я пытаюсь заставить его работать один раз в день, в конце дня.

Все работает, за исключением того, что он запускается только один раз и в любое конкретное время.

В настоящее время я использую FeedPollDelay, установленный на 1 день, но он будет медленно сканироваться, поскольку задержка основана на завершении последнего сканирования. Это также не относится ко времени начала.

Я также играл с опцией Start Time, но независимо от установленного времени, он немедленно начинает обработку.

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

Answer 1

StartTime и StartFromBeginning работают только в том случае, если ваша коллекция аренды пуста, согласно официальным документам .

Допустим, вы хотите работать в 11:00 UTC, вы можете легко добиться этого с помощью некоторого рабочего потока или процесса, который вызывает StartAsync в 11:00 UTC, и когда вы хотите остановить его, просто вызовите StopAsync.

Если вы хотите остановить его после того, как он исчерпал все ожидающие изменения, вы можете использовать Estimator , чтобы измерить текущую дельту между текущим состоянием коллекции и вашим процессором подачи изменений. Вы можете использовать Estimator для проверки, каждые X раз, сколько есть ожидающих изменений, и когда оно достигает 0, просто вызовите StopAsync на Процессоре.

Что-то вроде этого:

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