AmqpPartitionReceiver.ReceivePumpAsync hangs when IPartitionReceiveHandler.ProcessErrorAsync throws

[Rattenkrieg]

There is no protection in AmqpPartitionReceiver.ReceivePumpAsync against exceptions which may be throwed from IPartitionReceiveHandle.ProcessErrorAsync
https://github.com/Azure/azure-event-hubs-dotnet/blob/dev/src/Microsoft.Azure.EventHubs/Amqp/AmqpPartitionReceiver.cs#L303
When such occurs receivePumpTask completes in Failed state and no progress being made by AmqpPartitionReceiver instance.
In EventProcessorHost scenario PartitionReceiveHandler : IPartitionReceiveHandler is just wrapper around IEventProcessor which first defensively tries Processor.ProcessEventsAsync then simply calls Processor.ProcessErrorAsync
If Processor.ProcessEventsAsync throws its exception propagates to AmqpPartitionReceiver.ReceivePumpAsync and goes one more time to Processor.ProcessErrorAsync.
If Processor.ProcessErrorAsync fails again on second call AmqpPartitionReceiver will end up hanged with its receivePumpTask being in Failed state.
Here is demonstration of such behavior.
I'm using IPartitionReceiveHandler since there is no way in current design to test directly on IEventProcessor, so in such test scenario only one exception from IPartitionReceiveHandler.ProcessErrorAsync is enough.

Proposed solution is to guard this line with try-catch OR to await/ContinueWith(..., TaskContinuationOptions.OnlyOnFaulted) on receivePumpTask and shut-down receiver instance (for example by setting PartitionPumpStatus.Errored)

[serkantkaraca]

It is by design, PartitionReceiveHandler will shutdown on any error. It is host's responsibility to re-initiate receive handler when this happens. In EventProcessorHost case we do fail the pump on any error from PartitionReceiveHandler and then let PartitionManager detect the failed pump and recover it. You can see how we handle the errors and fail the pump in PartitionReceiveHandler class in EventHubPartitionPump.cs in given code snipped below.

`public async Task ProcessErrorAsync(Exception error)
{
  if (error == null)
  {
    error = new Exception("No error info supplied by EventHub client");
  }               if (error is ReceiverDisconnectedException)
  {
    ProcessorEventSource.Log.PartitionPumpWarning(
    this.eventHubPartitionPump.Host.Id, this.eventHubPartitionPump.PartitionContext.PartitionId,
    "EventHub client disconnected, probably another host took the partition");
  }
  else
  {
    ProcessorEventSource.Log.PartitionPumpError(
    this.eventHubPartitionPump.Host.Id, this.eventHubPartitionPump.PartitionContext.PartitionId, "EventHub client error:", error.ToString());
  }

  // We would like to deliver all errors in the pump to error handler.
  await this.eventHubPartitionPump.ProcessErrorAsync(error).ConfigureAwait(false);

  this.eventHubPartitionPump.PumpStatus = PartitionPumpStatus.Errored;
}`

I will double check for the given scenario where https://github.com/Azure/azure-event-hubs-dotnet/blob/dev/src/Microsoft.Azure.EventHubs/Amqp/AmqpPartitionReceiver.cs#L303 throws an exception. Host should be able to recover any error. I will update the thread about my results.

[Rattenkrieg]

@serkantkaraca regarding your snippet from PartitionReceiveHandler.ProcessErrorAsync(Exception error)
PumpStatus will be set to Errored only if call to ProcessErrorAsync on previous line doesn't throw. PumpStatus should be set in finally block to guarantee state transition.

[serkantkaraca]

Ok, I totally understand your point now. I am marking this one as a bug. I will provide a fix today. Thx for reporting!

[serkantkaraca]

Fix merged into #117