Merge pull request #11243 from abpframework/liangshiwei/inoutbox

Batch publish events from outbox to the event bus

framework/src/Volo.Abp.EntityFrameworkCore.PostgreSql/Volo/Abp/EntityFrameworkCore/DistributedEvents/PostgreSqlDbContextEventOutbox.cs

@@ -13,7 +13,7 @@ public class PostgreSqlDbContextEventOutbox<TDbContext> : DbContextEventOutbox<T
     }
 
     [UnitOfWork]
-    public override async Task DeleteAsync(Guid id)
+    public async override Task DeleteAsync(Guid id)
     {
         var dbContext = (IHasEventOutbox)await DbContextProvider.GetDbContextAsync();
         var tableName = dbContext.OutgoingEvents.EntityType.GetSchemaQualifiedTableName();

framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/DbContextEventOutbox.cs

@@ -56,4 +56,15 @@ public class DbContextEventOutbox<TDbContext> : IDbContextEventOutbox<TDbContext
             dbContext.Remove(outgoingEvent);
         }
     }
+
+    [UnitOfWork]
+    public virtual async Task DeleteManyAsync(IEnumerable<Guid> ids)
+    {
+        var dbContext = (IHasEventOutbox)await DbContextProvider.GetDbContextAsync();
+        var outgoingEvents = await dbContext.OutgoingEvents.Where(x => ids.Contains(x.Id)).ToListAsync();
+        if (outgoingEvents.Any())
+        {
+            dbContext.RemoveRange(outgoingEvents);
+        }
+    }
 }

framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs

@@ -87,12 +87,33 @@ public class AzureDistributedEventBus : DistributedEventBusBase, ISingletonDepen
         await TriggerHandlersAsync(eventType, eventData);
     }
 
-    public override async Task PublishFromOutboxAsync(OutgoingEventInfo outgoingEvent, OutboxConfig outboxConfig)
+    public async override Task PublishFromOutboxAsync(OutgoingEventInfo outgoingEvent, OutboxConfig outboxConfig)
     {
         await PublishAsync(outgoingEvent.EventName, outgoingEvent.EventData, outgoingEvent.Id);
     }
 
-    public override async Task ProcessFromInboxAsync(IncomingEventInfo incomingEvent, InboxConfig inboxConfig)
+    public async override Task PublishManyFromOutboxAsync(IEnumerable<OutgoingEventInfo> outgoingEvents, OutboxConfig outboxConfig)
+    {
+        var outgoingEventArray =  outgoingEvents.ToArray();
+
+        var publisher = await _publisherPool.GetAsync(
+            _options.TopicName,
+            _options.ConnectionName);
+
+        using var messageBatch = await publisher.CreateMessageBatchAsync();
+
+        foreach (var outgoingEvent in outgoingEventArray)
+        {
+            if (!messageBatch.TryAddMessage(new ServiceBusMessage(outgoingEvent.EventData) { Subject = outgoingEvent.EventName }))
+            {
+                throw new AbpException("The message is too large to fit in the batch. Set AbpEventBusBoxesOptions.OutboxWaitingEventMaxCount to reduce the number");
+            }
+        }
+
+        await publisher.SendMessagesAsync(messageBatch);
+    }
+
+    public async override Task ProcessFromInboxAsync(IncomingEventInfo incomingEvent, InboxConfig inboxConfig)
     {
         var eventType = _eventTypes.GetOrDefault(incomingEvent.EventName);
         if (eventType == null)
@@ -180,7 +201,7 @@ public class AzureDistributedEventBus : DistributedEventBusBase, ISingletonDepen
             .Locking(factories => factories.Clear());
     }
 
-    protected override async Task PublishToEventBusAsync(Type eventType, object eventData)
+    protected async override Task PublishToEventBusAsync(Type eventType, object eventData)
     {
         await PublishAsync(EventNameAttribute.GetNameOrDefault(eventType), eventData);
     }

framework/src/Volo.Abp.EventBus.Kafka/Volo/Abp/EventBus/Kafka/KafkaDistributedEventBus.cs

@@ -79,12 +79,7 @@ public class KafkaDistributedEventBus : DistributedEventBusBase, ISingletonDepen
             return;
         }
 
-        string messageId = null;
-
-        if (message.Headers.TryGetLastBytes("messageId", out var messageIdBytes))
-        {
-            messageId = System.Text.Encoding.UTF8.GetString(messageIdBytes);
-        }
+        var messageId = message.GetMessageId();
 
         if (await AddToInboxAsync(messageId, eventName, eventType, message.Value))
         {
@@ -164,7 +159,7 @@ public class KafkaDistributedEventBus : DistributedEventBusBase, ISingletonDepen
         GetOrCreateHandlerFactories(eventType).Locking(factories => factories.Clear());
     }
 
-    protected override async Task PublishToEventBusAsync(Type eventType, object eventData)
+    protected async override Task PublishToEventBusAsync(Type eventType, object eventData)
     {
         await PublishAsync(
             eventType,
@@ -198,7 +193,43 @@ public class KafkaDistributedEventBus : DistributedEventBusBase, ISingletonDepen
         );
     }
 
-    public override async Task ProcessFromInboxAsync(
+    public override Task PublishManyFromOutboxAsync(IEnumerable<OutgoingEventInfo> outgoingEvents, OutboxConfig outboxConfig)
+    {
+        var producer = ProducerPool.Get();
+        var outgoingEventArray = outgoingEvents.ToArray();
+        producer.BeginTransaction();
+        try
+        {
+            foreach (var outgoingEvent in outgoingEventArray)
+            {
+                var messageId = outgoingEvent.Id.ToString("N");
+                var headers = new Headers 
+                {
+                    { "messageId", System.Text.Encoding.UTF8.GetBytes(messageId)}
+                };
+
+                 producer.Produce(
+                    AbpKafkaEventBusOptions.TopicName,
+                    new Message<string, byte[]>
+                    {
+                        Key = outgoingEvent.EventName,
+                        Value = outgoingEvent.EventData,
+                        Headers = headers
+                    });
+            }
+            
+            producer.CommitTransaction();
+        }
+        catch (Exception e)
+        {
+            producer.AbortTransaction();
+            throw;
+        }
+        
+        return Task.CompletedTask;
+    }
+
+    public async override Task ProcessFromInboxAsync(
         IncomingEventInfo incomingEvent,
         InboxConfig inboxConfig)
     {
@@ -241,13 +272,29 @@ public class KafkaDistributedEventBus : DistributedEventBusBase, ISingletonDepen
         return PublishAsync(topicName, eventName, body, headers, headersArguments);
     }
 
-    private async Task PublishAsync(string topicName, string eventName, byte[] body, Headers headers, Dictionary<string, object> headersArguments)
+    private Task<DeliveryResult<string, byte[]>> PublishAsync(
+        string topicName, 
+        string eventName,
+        byte[] body, 
+        Headers headers,
+        Dictionary<string, object> headersArguments)
     {
         var producer = ProducerPool.Get(AbpKafkaEventBusOptions.ConnectionName);
 
+        return PublishAsync(producer, topicName, eventName, body, headers, headersArguments);
+    }
+
+    private Task<DeliveryResult<string, byte[]>> PublishAsync(
+        IProducer<string, byte[]> producer,
+        string topicName,
+        string eventName,
+        byte[] body,
+        Headers headers,
+        Dictionary<string, object> headersArguments)
+    {
         SetEventMessageHeaders(headers, headersArguments);
 
-        await producer.ProduceAsync(
+        return producer.ProduceAsync(
             topicName,
             new Message<string, byte[]>
             {

framework/src/Volo.Abp.EventBus.Kafka/Volo/Abp/EventBus/Kafka/MessageExtensions.cs

@@ -0,0 +1,18 @@
+using Confluent.Kafka;
+
+namespace Volo.Abp.EventBus.Kafka;
+
+public static class MessageExtensions
+{
+    public static string GetMessageId<TKey, TValue>(this Message<TKey, TValue> message)
+    {
+        string messageId = null;
+
+        if (message.Headers.TryGetLastBytes("messageId", out var messageIdBytes))
+        {
+            messageId = System.Text.Encoding.UTF8.GetString(messageIdBytes);
+        }
+
+        return messageId;
+    }
+}

framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqDistributedEventBus.cs

@@ -34,6 +34,8 @@ public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDe
     protected IRabbitMqMessageConsumerFactory MessageConsumerFactory { get; }
     protected IRabbitMqMessageConsumer Consumer { get; private set; }
 
+    private bool _exchangeCreated;
+
     public RabbitMqDistributedEventBus(
         IOptions<AbpRabbitMqEventBusOptions> options,
         IConnectionPool connectionPool,
@@ -180,7 +182,7 @@ public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDe
         GetOrCreateHandlerFactories(eventType).Locking(factories => factories.Clear());
     }
 
-    protected override async Task PublishToEventBusAsync(Type eventType, object eventData)
+    protected async override Task PublishToEventBusAsync(Type eventType, object eventData)
     {
         await PublishAsync(eventType, eventData, null);
     }
@@ -197,7 +199,30 @@ public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDe
         return PublishAsync(outgoingEvent.EventName, outgoingEvent.EventData, null, eventId: outgoingEvent.Id);
     }
 
-    public override async Task ProcessFromInboxAsync(
+    public async override Task PublishManyFromOutboxAsync(
+        IEnumerable<OutgoingEventInfo> outgoingEvents,
+        OutboxConfig outboxConfig)
+    {
+        using (var channel = ConnectionPool.Get(AbpRabbitMqEventBusOptions.ConnectionName).CreateModel())
+        {
+            var outgoingEventArray = outgoingEvents.ToArray();
+            channel.ConfirmSelect();
+
+            foreach (var outgoingEvent in outgoingEventArray)
+            {
+                await PublishAsync(
+                    channel,
+                    outgoingEvent.EventName, 
+                    outgoingEvent.EventData,  
+                    properties: null,
+                    eventId: outgoingEvent.Id);
+            }
+
+            channel.WaitForConfirmsOrDie();
+        }
+    }
+
+    public async override Task ProcessFromInboxAsync(
         IncomingEventInfo incomingEvent,
         InboxConfig inboxConfig)
     {
@@ -233,7 +258,7 @@ public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDe
         return PublishAsync(eventName, body, properties, headersArguments);
     }
 
-    protected Task PublishAsync(
+    protected virtual Task PublishAsync(
         string eventName,
         byte[] body,
         IBasicProperties properties,
@@ -242,37 +267,66 @@ public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDe
     {
         using (var channel = ConnectionPool.Get(AbpRabbitMqEventBusOptions.ConnectionName).CreateModel())
         {
-            channel.ExchangeDeclare(
-                AbpRabbitMqEventBusOptions.ExchangeName,
-                AbpRabbitMqEventBusOptions.GetExchangeTypeOrDefault(),
-                durable: true
-            );
-
-            if (properties == null)
-            {
-                properties = channel.CreateBasicProperties();
-                properties.DeliveryMode = RabbitMqConsts.DeliveryModes.Persistent;
-            }
+            return PublishAsync(channel, eventName, body, properties, headersArguments, eventId);
+        }
+    }
 
-            if (properties.MessageId.IsNullOrEmpty())
-            {
-                properties.MessageId = (eventId ?? GuidGenerator.Create()).ToString("N");
-            }
+    protected virtual Task PublishAsync(
+        IModel channel,
+        string eventName,
+        byte[] body,
+        IBasicProperties properties,
+        Dictionary<string, object> headersArguments = null,
+        Guid? eventId = null)
+    {
+        EnsureExchangeExists(channel);
 
-            SetEventMessageHeaders(properties, headersArguments);
+        if (properties == null)
+        {
+            properties = channel.CreateBasicProperties();
+            properties.DeliveryMode = RabbitMqConsts.DeliveryModes.Persistent;
+        }
 
-            channel.BasicPublish(
-                exchange: AbpRabbitMqEventBusOptions.ExchangeName,
-                routingKey: eventName,
-                mandatory: true,
-                basicProperties: properties,
-                body: body
-            );
+        if (properties.MessageId.IsNullOrEmpty())
+        {
+            properties.MessageId = (eventId ?? GuidGenerator.Create()).ToString("N");
         }
 
+        SetEventMessageHeaders(properties, headersArguments);
+
+        channel.BasicPublish(
+            exchange: AbpRabbitMqEventBusOptions.ExchangeName,
+            routingKey: eventName,
+            mandatory: true,
+            basicProperties: properties,
+            body: body
+        );
+
         return Task.CompletedTask;
     }
 
+    private void EnsureExchangeExists(IModel channel)
+    {
+        if (_exchangeCreated)
+        {
+            return;
+        }
+
+        try
+        {
+            channel.ExchangeDeclarePassive(AbpRabbitMqEventBusOptions.ExchangeName);
+        }
+        catch (Exception)
+        {
+            channel.ExchangeDeclare(
+                AbpRabbitMqEventBusOptions.ExchangeName,
+                AbpRabbitMqEventBusOptions.GetExchangeTypeOrDefault(),
+                durable: true
+            );
+        }
+        _exchangeCreated = true;
+    }
+
     private void SetEventMessageHeaders(IBasicProperties properties, Dictionary<string, object> headersArguments)
     {
         if (headersArguments == null)
@@ -306,7 +360,7 @@ public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDe
         var handlerFactoryList = new List<EventTypeWithEventHandlerFactories>();
 
         foreach (var handlerFactory in
-            HandlerFactories.Where(hf => ShouldTriggerEventForHandler(eventType, hf.Key)))
+                 HandlerFactories.Where(hf => ShouldTriggerEventForHandler(eventType, hf.Key)))
         {
             handlerFactoryList.Add(
                 new EventTypeWithEventHandlerFactories(handlerFactory.Key, handlerFactory.Value));

framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/RebusDistributedEventBus.cs

@@ -7,6 +7,7 @@ using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Options;
 using Rebus.Bus;
 using Rebus.Pipeline;
+using Rebus.Transport;
 using Volo.Abp.DependencyInjection;
 using Volo.Abp.EventBus.Distributed;
 using Volo.Abp.Guids;
@@ -149,6 +150,11 @@ public class RebusDistributedEventBus : DistributedEventBusBase, ISingletonDepen
     }
 
     protected async override Task PublishToEventBusAsync(Type eventType, object eventData)
+    {
+        await PublishAsync(eventType, eventData);
+    }
+
+    protected virtual async Task PublishAsync(Type eventType, object eventData)
     {
         if (AbpRebusEventBusOptions.Publish != null)
         {
@@ -218,7 +224,22 @@ public class RebusDistributedEventBus : DistributedEventBusBase, ISingletonDepen
         return PublishToEventBusAsync(eventType, eventData);
     }
 
-    public override async Task ProcessFromInboxAsync(
+    public async override Task PublishManyFromOutboxAsync(IEnumerable<OutgoingEventInfo> outgoingEvents, OutboxConfig outboxConfig)
+    {
+        var outgoingEventArray = outgoingEvents.ToArray();
+
+        using (var scope = new RebusTransactionScope())
+        {
+            foreach (var outgoingEvent in outgoingEventArray)
+            {
+                await PublishFromOutboxAsync(outgoingEvent, outboxConfig);
+            }
+            
+            await scope.CompleteAsync();
+        }
+    }
+
+    public async override Task ProcessFromInboxAsync(
         IncomingEventInfo incomingEvent,
         InboxConfig inboxConfig)
     {

framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/AbpEventBusBoxesOptions.cs

@@ -34,6 +34,11 @@ public class AbpEventBusBoxesOptions
     /// Default: 2 hours
     /// </summary>
     public TimeSpan WaitTimeToDeleteProcessedInboxEvents { get; set; }
+    
+    /// <summary>
+    /// Default: true
+    /// </summary>
+    public bool BatchPublishOutboxEvents { get; set; }
 
     public AbpEventBusBoxesOptions()
     {
@@ -43,5 +48,6 @@ public class AbpEventBusBoxesOptions
         PeriodTimeSpan = TimeSpan.FromSeconds(2);
         DistributedLockWaitDuration = TimeSpan.FromSeconds(15);
         WaitTimeToDeleteProcessedInboxEvents = TimeSpan.FromHours(2);
+        BatchPublishOutboxEvents = true;
     }
 }

framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/DistributedEventBusBase.cs

@@ -85,6 +85,11 @@ public abstract class DistributedEventBusBase : EventBusBase, IDistributedEventB
         OutboxConfig outboxConfig
     );
 
+    public abstract Task PublishManyFromOutboxAsync(
+        IEnumerable<OutgoingEventInfo> outgoingEvents,
+        OutboxConfig outboxConfig
+    );
+
     public abstract Task ProcessFromInboxAsync(
         IncomingEventInfo incomingEvent,
         InboxConfig inboxConfig);
@@ -101,7 +106,8 @@ public abstract class DistributedEventBusBase : EventBusBase, IDistributedEventB
         {
             if (outboxConfig.Selector == null || outboxConfig.Selector(eventType))
             {
-                var eventOutbox = (IEventOutbox)unitOfWork.ServiceProvider.GetRequiredService(outboxConfig.ImplementationType);
+                var eventOutbox =
+                    (IEventOutbox)unitOfWork.ServiceProvider.GetRequiredService(outboxConfig.ImplementationType);
                 var eventName = EventNameAttribute.GetNameOrDefault(eventType);
                 await eventOutbox.EnqueueAsync(
                     new OutgoingEventInfo(
@@ -135,7 +141,8 @@ public abstract class DistributedEventBusBase : EventBusBase, IDistributedEventB
             {
                 if (inboxConfig.EventSelector == null || inboxConfig.EventSelector(eventType))
                 {
-                    var eventInbox = (IEventInbox)scope.ServiceProvider.GetRequiredService(inboxConfig.ImplementationType);
+                    var eventInbox =
+                        (IEventInbox)scope.ServiceProvider.GetRequiredService(inboxConfig.ImplementationType);
 
                     if (!messageId.IsNullOrEmpty())
                     {

framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/IEventOutbox.cs

@@ -12,4 +12,6 @@ public interface IEventOutbox
     Task<List<OutgoingEventInfo>> GetWaitingEventsAsync(int maxCount, CancellationToken cancellationToken = default);
 
     Task DeleteAsync(Guid id);
+
+    Task DeleteManyAsync(IEnumerable<Guid> ids);
 }

framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/ISupportsEventBoxes.cs

@@ -1,3 +1,4 @@
+using System.Collections.Generic;
 using System.Threading.Tasks;
 
 namespace Volo.Abp.EventBus.Distributed;
@@ -9,6 +10,11 @@ public interface ISupportsEventBoxes
         OutboxConfig outboxConfig
     );
 
+    Task PublishManyFromOutboxAsync(
+        IEnumerable<OutgoingEventInfo> outgoingEvents,
+        OutboxConfig outboxConfig
+    );
+
     Task ProcessFromInboxAsync(
         IncomingEventInfo incomingEvent,
         InboxConfig inboxConfig

framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/OutboxSender.cs

@@ -1,4 +1,7 @@
 using System;
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.Extensions.DependencyInjection;
@@ -82,18 +85,14 @@ public class OutboxSender : IOutboxSender, ITransientDependency
                     }
 
                     Logger.LogInformation($"Found {waitingEvents.Count} events in the outbox.");
-
-                    foreach (var waitingEvent in waitingEvents)
+                    
+                    if (EventBusBoxesOptions.BatchPublishOutboxEvents)
                     {
-                        await DistributedEventBus
-                            .AsSupportsEventBoxes()
-                            .PublishFromOutboxAsync(
-                                waitingEvent,
-                                OutboxConfig
-                            );
-
-                        await Outbox.DeleteAsync(waitingEvent.Id);
-                        Logger.LogInformation($"Sent the event to the message broker with id = {waitingEvent.Id:N}");
+                        await PublishOutgoingMessagesInBatchAsync(waitingEvents);
+                    }
+                    else
+                    {
+                        await PublishOutgoingMessagesAsync(waitingEvents);
                     }
                 }
             }
@@ -108,4 +107,32 @@ public class OutboxSender : IOutboxSender, ITransientDependency
             }
         }
     }
+
+    protected virtual async Task PublishOutgoingMessagesAsync(List<OutgoingEventInfo> waitingEvents)
+    {
+        foreach (var waitingEvent in waitingEvents)
+        {
+            await DistributedEventBus
+                .AsSupportsEventBoxes()
+                .PublishFromOutboxAsync(
+                    waitingEvent,
+                    OutboxConfig
+                );
+
+            await Outbox.DeleteAsync(waitingEvent.Id);
+            
+            Logger.LogInformation($"Sent the event to the message broker with id = {waitingEvent.Id:N}");
+        }
+    }
+
+    protected virtual async Task PublishOutgoingMessagesInBatchAsync(List<OutgoingEventInfo> waitingEvents)
+    {
+        await DistributedEventBus
+            .AsSupportsEventBoxes()
+            .PublishManyFromOutboxAsync(waitingEvents, OutboxConfig);
+                    
+        await Outbox.DeleteManyAsync(waitingEvents.Select(x => x.Id).ToArray());
+        
+        Logger.LogInformation($"Sent {waitingEvents.Count} events to message broker");
+    }
 }

framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/ConsumerPool.cs

@@ -76,6 +76,7 @@ public class ConsumerPool : IConsumerPool, ISingletonDependency
 
             try
             {
+                consumer.Value.Unsubscribe();
                 consumer.Value.Close();
                 consumer.Value.Dispose();
             }

framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/KafkaMessageConsumer.cs

@@ -80,9 +80,11 @@ public class KafkaMessageConsumer : IKafkaMessageConsumer, ITransientDependency,
 
     protected virtual async Task Timer_Elapsed(AbpAsyncTimer timer)
     {
-        await CreateTopicAsync();
-        Consume();
-        Timer.Stop();
+        if (Consumer == null)
+        {
+            await CreateTopicAsync();
+            Consume();
+        }
     }
 
     protected virtual async Task CreateTopicAsync()
@@ -164,12 +166,21 @@ public class KafkaMessageConsumer : IKafkaMessageConsumer, ITransientDependency,
 
     public virtual void Dispose()
     {
+        Timer.Stop();
         if (Consumer == null)
         {
             return;
         }
 
-        Consumer.Close();
-        Consumer.Dispose();
+        try
+        {
+            Consumer.Unsubscribe();
+            Consumer.Close();
+            Consumer.Dispose();
+            Consumer = null;
+        }
+        catch (ObjectDisposedException)
+        {
+        }
     }
 }

framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/ProducerPool.cs

@@ -17,6 +17,8 @@ public class ProducerPool : IProducerPool, ISingletonDependency
     protected ConcurrentDictionary<string, Lazy<IProducer<string, byte[]>>> Producers { get; }
 
     protected TimeSpan TotalDisposeWaitDuration { get; set; } = TimeSpan.FromSeconds(10);
+    
+    protected TimeSpan DefaultTransactionsWaitDuration { get; set; } = TimeSpan.FromSeconds(30);
 
     public ILogger<ProducerPool> Logger { get; set; }
 
@@ -37,11 +39,18 @@ public class ProducerPool : IProducerPool, ISingletonDependency
         return Producers.GetOrAdd(
             connectionName, connection => new Lazy<IProducer<string, byte[]>>(() =>
             {
-                var config = Options.Connections.GetOrDefault(connection);
-
-                Options.ConfigureProducer?.Invoke(new ProducerConfig(config));
-
-                return new ProducerBuilder<string, byte[]>(config).Build();
+                var producerConfig = new ProducerConfig(Options.Connections.GetOrDefault(connection));
+                Options.ConfigureProducer?.Invoke(producerConfig);
+
+                if (producerConfig.TransactionalId.IsNullOrWhiteSpace())
+                {
+                    producerConfig.TransactionalId = Guid.NewGuid().ToString();
+                }
+                
+                var producer = new ProducerBuilder<string, byte[]>(producerConfig).Build();
+                producer.InitTransactions(DefaultTransactionsWaitDuration);
+                
+                return producer;
             })).Value;
     }
 
@@ -69,7 +78,7 @@ public class ProducerPool : IProducerPool, ISingletonDependency
         foreach (var producer in Producers.Values)
         {
             var poolItemDisposeStopwatch = Stopwatch.StartNew();
-
+        
             try
             {
                 producer.Value.Dispose();
@@ -77,19 +86,19 @@ public class ProducerPool : IProducerPool, ISingletonDependency
             catch
             {
             }
-
+        
             poolItemDisposeStopwatch.Stop();
-
+        
             remainingWaitDuration = remainingWaitDuration > poolItemDisposeStopwatch.Elapsed
                 ? remainingWaitDuration.Subtract(poolItemDisposeStopwatch.Elapsed)
                 : TimeSpan.Zero;
         }
-
+        
         poolDisposeStopwatch.Stop();
-
+        
         Logger.LogInformation(
             $"Disposed Kafka Producer Pool ({Producers.Count} producers in {poolDisposeStopwatch.Elapsed.TotalMilliseconds:0.00} ms).");
-
+        
         if (poolDisposeStopwatch.Elapsed.TotalSeconds > 5.0)
         {
             Logger.LogWarning(

framework/src/Volo.Abp.MongoDB/Volo/Abp/MongoDB/DistributedEvents/MongoDbContextEventOutbox.cs

@@ -68,4 +68,18 @@ public class MongoDbContextEventOutbox<TMongoDbContext> : IMongoDbContextEventOu
             await dbContext.OutgoingEvents.DeleteOneAsync(x => x.Id.Equals(id));
         }
     }
+
+    [UnitOfWork]
+    public async Task DeleteManyAsync(IEnumerable<Guid> ids)
+    {
+        var dbContext = (IHasEventOutbox)await MongoDbContextProvider.GetDbContextAsync();
+        if (dbContext.SessionHandle != null)
+        {
+            await dbContext.OutgoingEvents.DeleteManyAsync(dbContext.SessionHandle, x => ids.Contains(x.Id));
+        }
+        else
+        {
+            await dbContext.OutgoingEvents.DeleteManyAsync(x => ids.Contains(x.Id));
+        }
+    }
 }

framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/RabbitMqMessageConsumer.cs

@@ -176,14 +176,6 @@ public class RabbitMqMessageConsumer : IRabbitMqMessageConsumer, ITransientDepen
         }
         catch (Exception ex)
         {
-            if (ex is OperationInterruptedException operationInterruptedException &&
-                operationInterruptedException.ShutdownReason.ReplyCode == 406 &&
-                operationInterruptedException.Message.Contains("arg 'x-dead-letter-exchange'"))
-            {
-                Logger.LogException(ex, LogLevel.Warning);
-                await ExceptionNotifier.NotifyAsync(ex, logLevel: LogLevel.Warning);
-            }
-
             Logger.LogException(ex, LogLevel.Warning);
             await ExceptionNotifier.NotifyAsync(ex, logLevel: LogLevel.Warning);
         }