BlockingCollection with ConcurrentQueue still remain object after GetConsumingEnumerable

Question

I have 2 threads, one is put data to IBlockingColection, and the second read it then send to kafka. My application is asp.net core api.

public class ConcurrentQueueForEvents
{
    private readonly BlockingCollection<IDomainEvent>  _queue;
    public ConcurrentQueueForEvents()
    {
        _queue = new BlockingCollection<IDomainEvent>(new ConcurrentQueue<IDomainEvent>());
    }
    public void Enqueue(IDomainEvent item)
    {
        _queue.Add(item);
        OnItemEnqueued();
    }

    public bool TryDequeue(out IDomainEvent result)
    {
        result = _queue.Take();
        return true;
    }
   public IEnumerable<IDomainEvent> GetConsumingEnumerable()
    {
        return _queue.GetConsumingEnumerable();
    }
public event EventHandler? ItemEnqueued;

    void OnItemEnqueued()
    {
        ItemEnqueued?.Invoke(this, EventArgs.Empty);
    }
}

at the controller, when user act an api, i send event to it, it was register as singleton on StartUp class

    public class PushEventToKafkaHandler 
{
    private readonly ConcurrentQueueForEvents _queue;
    private readonly MessageBrokerFactory _messageBrokerFactory;
    private readonly IIntegrateEventMapper _mapper;
    private readonly AsyncDuplicateLock _locker = new ();

    
    public PushEventToKafkaHandler(ConcurrentQueueForEvents queue, 
        MessageBrokerFactory messageBrokerFactory, IIntegrateEventMapper mapper)
    {
        _queue = queue;
        _messageBrokerFactory = messageBrokerFactory;
        _mapper = mapper;
    }
    
    public async Task Delivery()
    {
        foreach (var value in _queue.GetConsumingEnumerable())
        {
            var targetMessage = _mapper.GetIntegrateEventFromEvent(value);
            var s = _messageBrokerFactory.GetSender(targetMessage.MessageType.Name);
            //await Task.Run(() =>
            //{
             //   s.SendAsync(targetMessage);
            //});
            Console.WriteLine($"{Thread.CurrentThread.ManagedThreadId} ---- {_queue.GetHashCode()}");
        }
    }

    public void DeliveryByHook()
    {
        _queue.ItemEnqueued += async (sender, args) =>
        {
            try
            {
                _queue.TryDequeue(out var message);
                if (message is IIntegrateEvent @event)
                {
                    var targetMessage = _mapper.GetIntegrateEventFromEvent(message);
                    var s = _messageBrokerFactory.GetSender(targetMessage.MessageType.Name);
                    await s.SendAsync(targetMessage);
                }
            }
            catch (Exception e)
            {
                Console.WriteLine(e);
            }
            
        };
    }
}

this is the pusher that push event to kafka.

public void Configure(IApplicationBuilder app, IWebHostEnvironment env){
       var transporter = serviceProvider.GetService<PushEventToKafkaHandler>()!;
        Task.Run(() =>
        {
            transporter.Delivery();
        });
}

at StartUp I run it on other thread by Task.Run but, when i call api to send event, the transporter keep remain item on IBlockingCollection.

So when i send 5 times event, instead of sending 5 message, he send total 15 (1+2+3+4+5) messages.

What i wrong here?

update:

after adding log the result is:

] [User.Api] [::1] [1f4f4db8-b260-4cc6-bf87-5ab8d68ecc0a] Request finished HTTP/1.1 POST https://localhost:7162/bet application/json 248 - 200 - application/json;+charset=utf-8 89.7223ms
14 ---- 6555496
14 ---- 6555496
14 ---- 6555496
14 ---- 6555496
14 ---- 6555496

at the 5th of api hitting

Answer 1

after try with sample code, it is my mistake on code. the caller of Enqueue method is register as singleton, and not flush event after enqueue. so it keep message, then increase it each api hitting