fix: handle revoke with no partitions assigned (issue #214)

BEagle1984 · Nov 25, 2023 · 1623072 · 1623072
1 parent f2a3821
commit 1623072
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 3 deletions.
diff --git a/Directory.Build.props b/Directory.Build.props
@@ -1,6 +1,6 @@
 <Project>
   <PropertyGroup Label="Package information">
-    <BaseVersionSuffix>-beta.7</BaseVersionSuffix>
+    <BaseVersionSuffix>-beta.8</BaseVersionSuffix>
     <BaseVersion>4.4.0$(BaseVersionSuffix)</BaseVersion>
     <DatabasePackagesRevision>1</DatabasePackagesRevision>
     <DatabasePackagesVersionSuffix>$(BaseVersionSuffix)</DatabasePackagesVersionSuffix>

diff --git a/src/Silverback.Integration.Kafka/Messaging/Broker/Kafka/ConsumerChannelsManager.cs b/src/Silverback.Integration.Kafka/Messaging/Broker/Kafka/ConsumerChannelsManager.cs
@@ -43,7 +43,7 @@ public ConsumerChannelsManager(
             Func<ISequenceStore> sequenceStoreFactory,
             ISilverbackLogger logger)
             : base(
-                consumer.Endpoint.ProcessPartitionsIndependently ? partitions.Count : 1,
+                GetChannelsCount(partitions, consumer),
                 consumer.Endpoint.BackpressureLimit,
                 sequenceStoreFactory)
         {
@@ -76,7 +76,7 @@ public void StartReading()
         {
             if (_consumer.Endpoint.ProcessPartitionsIndependently)
                 _partitions.ForEach(StartReading);
-            else
+            else if (_partitions.Count > 0)
                 StartReading(0);
         }
 
@@ -161,6 +161,11 @@ protected override void Dispose(bool disposing)
             _readingSemaphoreSlim.Dispose();
         }
 
+        private static int GetChannelsCount(IReadOnlyList<TopicPartition> partitions, KafkaConsumer consumer) =>
+            consumer.Endpoint.ProcessPartitionsIndependently
+                ? partitions.Count
+                : Math.Min(partitions.Count, 1);
+
         [SuppressMessage("", "VSTHRD110", Justification = Justifications.FireAndForget)]
         private void StartReading(int channelIndex)
         {