Merge pull request #199 from bazen-teklehaymanot/fix-partition-consumer

superstreamlabs · Feb 5, 2024 · 14b9a47 · 14b9a47
2 parents be2b50b + d8c2857
commit 14b9a47
Showing 1 changed file with 5 additions and 0 deletions.
diff --git a/src/Memphis.Client/MemphisClient.Consumer.cs b/src/Memphis.Client/MemphisClient.Consumer.cs
@@ -77,6 +77,11 @@ public async Task<MemphisConsumer> CreateConsumer(MemphisConsumerOptions consume
  var consumer = new MemphisConsumer(this, consumerOptions, createConsumerResponse.PartitionsUpdate.PartitionsList);
  _consumerCache.AddOrUpdate(consumer.Key, consumer, (_, _) => consumer);
 
+ if(createConsumerResponse is { PartitionsUpdate: { } partitionsUpdate })
+ {
+ _stationPartitions.AddOrUpdate(consumerOptions.StationName, partitionsUpdate, (_, _) => partitionsUpdate);
+ }
+
  await ListenForSchemaUpdate(consumerOptions.StationName);
 
  return consumer;