title: TPL .Net Concurrenty Issue with Azure EventHub Producer
body: <p>I am working with Azure Event Hub producer client and reading messages off of a kafka stream then pass it along to deserialize/map, then pass to Event Hub.  I have the consume loop which is creating a task for each consume and then two methods to do processing(this seems to have greatly improved the speed from the kafka lag perspective.  However, Event hub makes you create an event batch which I don't necessarily want to use.  I just want to send the data one message at a time for now.  In order to create a new batch I have to call Dispose().  I am running into an issue where there's another call to the function by the the time I call Dispose() and I get an error saying the object is being used by event hub.</p>
  <p>I've also tried using the overload for eventHubProducerClient.SendAsync that allows you to pass in a IEnumerable but i'm running into the same issue with that.</p>
  <p>So I believe this to be a synchronization issue, or maybe I need to do a lock somewhere?</p>
  <p>Any help would be appreciated.</p>
  <pre><code>       public void Execute()
                  {
                      using (_consumer)
                      {
                          try
                          {
                              _consumer.Subscribe(_streamConsumerSettings.Topic);
                              while (true)
                              {
                                  var result = _consumer.Consume(1000);
          
                                  if (result == null)
                                  {
                                      continue;
                                  }
                                  var process = Task.Factory.StartNew(() =&gt; ProcessMessage(result?.Message?.Value));
                                  var send = process.ContinueWith(t =&gt; SendMessage(process.Result));                        
                              }
          
                          }
                          catch (ConsumeException e)
                          {
                              _logger.LogError(e, e.StackTrace ?? e.Message);
                              _cancelConsume = true;
                              _consumer.Close();
                              RestartConsumer();
                          }
                      }
                  }
      
           public static EquipmentJson ProcessMessage(byte[] result)
           {
                      var json = _messageProcessor.DeserializeAndMap(result);
                      return json;
           }
          
           public static void SendMessage(EquipmentJson message)
           {
                      try 
                      {   
          
                          _eventHubClient.AddToBatch(message);             
                          
                      }
                      catch (Exception e)
                      {
                          _logger.LogError(e, e.StackTrace ?? e.Message);
                      }
            }
      
       
      
          public async Task AddToBatch(EquipmentJson message)
                  {
                      if 
        (!string.IsNullOrEmpty(message.EquipmentLocation))
                      {
                          try
                          {
                              var batch = await _equipmentLocClient.CreateBatchAsync();
                              batch.TryAdd(new EventData(Encoding.UTF8.GetBytes(message.EquipmentLocation)));
                              await _eventHubProducerClient.SendAsync(batch);
                              batch.Dispose();
                              _logger.LogInformation($&quot;Data sent {DateTimeOffset.UtcNow}&quot;);
                          }
                          catch (Exception e)
                          {
                              _logger.LogError(e, e.StackTrace ?? e.Message);
                          }
                      }
                  }
  
   public class EventHubClient : IEventHubClient
      {
          private readonly ILoggerAdapter&lt;EventHubClient&gt; _logger;
          private readonly EventHubClientSettings _eventHubClientSettings;
          private IMapper _mapper;
  
          
          private static EventHubProducerClient _equipmentLocClient;
  
  
          public EventHubClient(ILoggerAdapter&lt;EventHubClient&gt; logger, EventHubClientSettings eventHubClientSettings, IMapper mapper)
          {
              _logger = logger;
              _eventHubClientSettings = eventHubClientSettings;
              _mapper = mapper;
              _equipmentLocClient = new EventHubProducerClient(_eventHubClientSettings.ConnectionString, _eventHubClientSettings.EquipmentLocation);
  
          }
      }
  }
  </code></pre>