[BUG] System.InvalidOperationException at Azure.Messaging.EventHubs.EventProcessorClient+<OnInitializingPartitionAsync>d__60.MoveNext

[petrformanek]

Description of the bug
Thread concurrency issue found in logs.

Expected behavior
No exceptions thrown.

Actual behavior (include Exception or Stack Trace)

{"method":"System.ThrowHelper.ThrowInvalidOperationException_ConcurrentOperationsNotSupported","assembly":"System.Private.CoreLib, Version=4.0.0.0, Culture=neutral, PublicKeyToken=7cec85d7bea7798e","level":0,"line":0}
{"method":"System.Collections.Generic.Dictionary`2.TryInsert","assembly":"System.Private.CoreLib, Version=4.0.0.0, Culture=neutral, PublicKeyToken=7cec85d7bea7798e","level":1,"line":0}
{"method":"Azure.Messaging.EventHubs.EventProcessorClient+<OnInitializingPartitionAsync>d__60.MoveNext","assembly":"Azure.Messaging.EventHubs.Processor, Version=5.1.0.0, Culture=neutral, PublicKeyToken=92742159e12e44c8","level":2,"line":0}
{"method":"System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw","assembly":"System.Private.CoreLib, Version=4.0.0.0, Culture=neutral, PublicKeyToken=7cec85d7bea7798e","level":3,"line":0}
{"method":"System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification","assembly":"System.Private.CoreLib, Version=4.0.0.0, Culture=neutral, PublicKeyToken=7cec85d7bea7798e","level":4,"line":0}
{"method":"Azure.Messaging.EventHubs.Primitives.EventProcessor`1+<TryStartProcessingPartitionAsync>d__75.MoveNext","assembly":"Azure.Messaging.EventHubs, Version=5.1.0.0, Culture=neutral, PublicKeyToken=92742159e12e44c8","level":5,"line":0}

To Reproduce
Hard to reproduce.

Environment:

• Azure.Messaging.EventHubs.Processor v 5.1.0
• Linux docker container running in AKS with .NET Core 3.1

[jsquire]

Hi @petrformanek. We're sorry that you're experiencing difficulties and appreciate you reaching out to let us know. This isn't one that we've seen before, but seems like a race condition at first glance. The good news is that this would trigger retry and recovery paths and should not have an impact on the overall heath of the processor. In the worst case, I would expect to see a small delay in this partition being processed.

I'd definitely like to take a deeper look at this, and would appreciate any additional context or insight that you're able to provide. Some things that may be helpful:

• Was this something that you saw regular occurrences of, in your logs?
• How many partitions does the Event Hub being processed have?
• How many EventProcessorClient instances are active against that Event Hub for the consumer group?
• Do you have a handler hooked up for the PartitionInitializing event and, if so, would you be willing to share the general structure/responsibility of it?

I'd also welcome any additional insight that you think may be relevant. Again, thank you for taking the time to report this and your help in gathering information for investigation.

[petrformanek]

Hi @jsquire. Thank you for a quick response.

As you wrote the processor health was ok, it was able to process events on other partitions, because they were already initialized and were processing events for some time.
When the processor tried to take ownership of partition 0 then it got stuck with the InvalidOperationException and it was calling my PartitionInitializing handler each 10 seconds for like 30 minutes until the ownership of partition was taken by another processing instance running in AKS. So there was 30 minutes delay in events processing.

Answers to your points:

• I've seen it for the first time after like 2 months of using v5 SDK
• In this case Event Hub has 8 partitions. We also have second channel with 32 partitions
• Minimum are 2 instances which can automatically scale up to 8 instances. In case of second channel to 32 instances
• Yes, it is very simple, see bellow

public virtual Task PartitionInitializingAsync(PartitionInitializingEventArgs arg)
{
    if (arg.CancellationToken.IsCancellationRequested)
    {
        return Task.CompletedTask;
    }
    arg.DefaultStartingPosition = EventPosition.Latest;
    _logger.LogInformation($"Initializing partition {arg.PartitionId}");

    return Task.CompletedTask;
}

[jsquire]

Thanks, @petrformanek. I appreciate the additional context. We'll start investigating.

[jsquire]

Apologies for the delay and, again, thanks for calling attention to this. We introduced a race condition by not using a concurrent set where we should have for tracking the partitions being processed. This was fixed by #12928 and will be included in our next release.

I'm going to close this out as resolved. Please feel free to reopen if you believe that further discussion is needed.