Почему обработка Azure EventHub видит при остановке (EventProcessorClient.stopprocessingAsync)? ⇐ C#

[Anonymous]

Я использую приложение .net 8 Web API для чтения событий из Azure EventHub. Служба, которая читает события, находится внутри хостинга. Если API не работает после всех попыток повторения, я пытаюсь остановить обработку EventHub в течение нескольких минут. Тем не менее, приложение висит бесконечно при остановке EventHub. Он не терпит неудачу и не бросает никаких исключений - он просто застрял. > В результате этот метод
будет выполнять сетевой ввод -вывод и может потребоваться ждать раздела
, которые были активны для завершения. < /P>
< /blockquote>
sumary: < /p>

Я читаю события из Azure EventHub. Механизм. < /li>
Если Полли исчерпает все повторения, и API остается недоступным, я останавливаюсь
eventhub. EventHub. < /li>
< /ul>
Вызов StopprocessingAsync (), по -видимому, заставляет процессор EventHub повесить бесконечно. Если да, то как я могу правильно прекратить обработку и возобновить позже?
Что является рекомендуемым подходом для прекращения обработки EventHub? < /Strong> < /p>
public class EventHubServiceTask : IEventHubServiceTask
{

private CancellationToken _startCancellationToken;

public EventHubServiceTask()
{
var storageClient = new BlobContainerClient(_blobStorageSettings.ConnectionString, _blobStorageSettings.ContainerName);
_eventProcessorClient = new EventProcessorClient(storageClient, EventHubConsumerClient.DefaultConsumerGroupName, _eventHubSettings.ConnectionString, _eventHubSettings.EventHubName);
_eventProcessorClient.ProcessEventAsync += ProcessEventHandler;
}

public async Task StartAsync(CancellationToken cancellationToken)
{
_startCancellationToken = cancellationToken;
await _eventProcessorClient.StartProcessingAsync(cancellationToken);
}

private async Task ProcessEventHandler(ProcessEventArgs eventArgs)
{
MyEvent? myevent= JsonConvert.DeserializeObject(eventArgs.Data.EventBody.ToString()) ?? new MyEvent();

if (await ServiceHealthCheckNotAvailable(myevent.CorrelationId, eventArgs.CancellationToken))
{
await HandleHealthCheckFailureAsync(myevent.CorrelationId, eventArgs.CancellationToken);
}

await eventArgs.UpdateCheckpointAsync(eventArgs.CancellationToken);

}

private async Task ServiceHealthCheckNotAvailable(Guid correlationId, CancellationToken cancellationToken)
{
return !await CheckHealthAsync(correlationId, cancellationToken);
}

private async Task HandleHealthCheckFailureAsync(Guid correlationId, CancellationToken cancellationToken)
{
try
{
await _eventProcessorClient.StopProcessingAsync(_startCancellationToken);
await Task.Delay(TimeSpan.FromSeconds(_retryPolicySettings.EventHubSleepTimeSeconds), cancellationToken);
await _eventProcessorClient.StartProcessingAsync(cancellationToken);
}
catch (Exception e)
{
throw;
}
}

private async Task CheckHealthAsync(Guid correlationId, CancellationToken cancellationToken)
{
var policy = Policy
.HandleResult(result => !result)
.WaitAndRetryAsync(_retryPolicySettings.RetryCount, retryAttempt => TimeSpan.FromSeconds(retryAttempt));

return await policy.ExecuteAsync(async () => await _service.IsHealthyAsync(correlationId, cancellationToken));

}


Подробнее здесь: https://stackoverflow.com/questions/794 ... lient-stop