Fix event consumer.

backend/src/Squidex.Infrastructure/EventSourcing/Grains/BatchSubscriber.cs

@@ -22,6 +22,11 @@ namespace Squidex.Infrastructure.EventSourcing.Grains
         private readonly IEventSubscription eventSubscription;
         private readonly IDataflowBlock pipelineEnd;
 
+        public object Sender
+        {
+            get { return eventSubscription.Sender; }
+        }
+
         private sealed class Job
         {
             public StoredEvent? StoredEvent { get; set; }
@@ -91,11 +96,11 @@ namespace Squidex.Infrastructure.EventSourcing.Grains
 
                         if (exception != null)
                         {
-                            await grain.OnErrorAsync(exception);
+                            await grain.OnErrorAsync(Sender, exception);
                         }
                         else
                         {
-                            await grain.OnEventsAsync(GetEvents(jobsBySender), GetPosition(jobsBySender));
+                            await grain.OnEventsAsync(Sender, GetEvents(jobsBySender), GetPosition(jobsBySender));
                         }
                     }
                 }

backend/src/Squidex.Infrastructure/EventSourcing/Grains/EventConsumerGrain.cs

@@ -81,8 +81,13 @@ namespace Squidex.Infrastructure.EventSourcing.Grains
             return State.ToInfo(eventConsumer!.Name).AsImmutable();
         }
 
-        public Task OnEventsAsync(IReadOnlyList<Envelope<IEvent>> events, string position)
+        public Task OnEventsAsync(object sender, IReadOnlyList<Envelope<IEvent>> events, string position)
         {
+            if (!ReferenceEquals(sender, currentSubscriber?.Sender))
+            {
+                return Task.CompletedTask;
+            }
+
             return DoAndUpdateStateAsync(async () =>
             {
                 await DispatchAsync(events);
@@ -91,8 +96,13 @@ namespace Squidex.Infrastructure.EventSourcing.Grains
             });
         }
 
-        public Task OnErrorAsync(Exception exception)
+        public Task OnErrorAsync(object sender, Exception exception)
         {
+            if (!ReferenceEquals(sender, currentSubscriber?.Sender))
+            {
+                return Task.CompletedTask;
+            }
+
             return DoAndUpdateStateAsync(() =>
             {
                 Unsubscribe();

backend/tests/Squidex.Infrastructure.Tests/EventSourcing/MongoEventStoreFixture.cs

@@ -27,18 +27,21 @@ namespace Squidex.Infrastructure.EventSourcing
             mongoClient = new MongoClient(connectionString);
             mongoDatabase = mongoClient.GetDatabase($"EventStoreTest");
 
-            Dispose();
-
             BsonJsonConvention.Register(JsonSerializer.Create(JsonHelper.DefaultSettings()));
 
             EventStore = new MongoEventStore(mongoDatabase, notifier);
             EventStore.InitializeAsync().Wait();
         }
 
-        public void Dispose()
+        public void Cleanup()
         {
             mongoClient.DropDatabase("EventStoreTest");
         }
+
+        public void Dispose()
+        {
+            Cleanup();
+        }
     }
 
     public sealed class MongoEventStoreDirectFixture : MongoEventStoreFixture

backend/tests/Squidex.Infrastructure.Tests/EventSourcing/MongoParallelInsertTests.cs

@@ -212,6 +212,7 @@ namespace Squidex.Infrastructure.EventSourcing
         public MongoParallelInsertTests(MongoEventStoreReplicaSetFixture fixture)
         {
             _ = fixture;
+            _.Cleanup();
 
             var typeNameRegistry = new TypeNameRegistry().Map(typeof(MyEvent), "My");