refactored kafka

EspadaKomanda · Nov 21, 2024 · a93c227 · a93c227
1 parent 27a256f
commit a93c227
Show file tree

Hide file tree

Showing 14 changed files with 111 additions and 42 deletions.
diff --git a/ApiGatewayService/Kafka/KafkaRequestService.cs b/ApiGatewayService/Kafka/KafkaRequestService.cs
@@ -178,7 +178,8 @@ public async Task<bool> Produce(string topicName, Message<string, string> messag
                 bool IsTopicExists = IsTopicAvailable(topicName);
                 if (IsTopicExists && IsTopicPendingMessageBusExist( responseTopic))
                 {
-                    var deliveryResult = await _producer.ProduceAsync(topicName, message);
+                    var deliveryResult = await _producer.ProduceAsync(
+                    new TopicPartition(topicName, new Partition(0));
                     if (deliveryResult.Status == PersistenceStatus.Persisted)
                     {
                         _logger.LogInformation("Message delivery status: Persisted {Result}", deliveryResult.Value);
@@ -200,7 +201,8 @@ public async Task<bool> Produce(string topicName, Message<string, string> messag
                 bool IsTopicCreated = _kafkaTopicManager.CreateTopic(topicName, Convert.ToInt32(Environment.GetEnvironmentVariable("PARTITIONS_STANDART")), Convert.ToInt16(Environment.GetEnvironmentVariable("REPLICATION_FACTOR_STANDART")));
                 if (IsTopicCreated && IsTopicPendingMessageBusExist( responseTopic))
                 {
-                    var deliveryResult = await _producer.ProduceAsync(topicName, message);
+                    var deliveryResult = await _producer.ProduceAsync(
+                    new TopicPartition(topicName, new Partition(0));
                     if (deliveryResult.Status == PersistenceStatus.Persisted)
                     {
                         _logger.LogInformation("Message delivery status: Persisted {Result}", deliveryResult.Value);
@@ -235,7 +237,10 @@ private bool IsTopicPendingMessageBusExist(string responseTopic)
         private async Task Consume(IConsumer<string,string> localConsumer,string topicName)
         {   
             topicCount++;
-            localConsumer.Subscribe(topicName);
+             var partitions = new List<TopicPartition>();
+                partitions.Add(new TopicPartition(topicName, 0));
+
+            localConsumer.Assign(partitions);
             while (true)
             {
                 ConsumeResult<string, string> result = localConsumer.Consume();

diff --git a/ApiGatewayService/Kafka/KafkaService.cs b/ApiGatewayService/Kafka/KafkaService.cs
@@ -28,7 +28,10 @@ protected void ConfigureConsumer(string topicName)
             _consumer = new ConsumerBuilder<string, string>(config).Build();
             if(IsTopicAvailable(topicName))
             {
-                _consumer.Subscribe(topicName);
+                var partitions = new List<TopicPartition>();
+                partitions.Add(new TopicPartition(topicName, 0));
+
+                _consumer.Assign(partitions);
                 return;
             }
             throw new ConsumerTopicUnavailableException("Topic unavailable");
@@ -78,7 +81,8 @@ public async Task<bool> Produce( string topicName,Message<string, string> messag
             bool IsTopicExists = IsTopicAvailable(topicName);
             if (IsTopicExists)
             {
-                var deliveryResult = await _producer.ProduceAsync(topicName, message);
+                var deliveryResult = await _producer.ProduceAsync(
+                    new TopicPartition(topicName, new Partition(0));
                 if (deliveryResult.Status == PersistenceStatus.Persisted)
                 {
 
@@ -94,7 +98,8 @@ public async Task<bool> Produce( string topicName,Message<string, string> messag
             bool IsTopicCreated = _kafkaTopicManager.CreateTopic(topicName, Convert.ToInt32(Environment.GetEnvironmentVariable("PARTITIONS_STANDART")), Convert.ToInt16(Environment.GetEnvironmentVariable("REPLICATION_FACTOR_STANDART")));
             if (IsTopicCreated)
             {
-                var deliveryResult = await _producer.ProduceAsync(topicName, message);
+                var deliveryResult = await _producer.ProduceAsync(
+                    new TopicPartition(topicName, new Partition(0));
                 if (deliveryResult.Status == PersistenceStatus.Persisted)
                 {
                     _logger.LogInformation("Message delivery status: Persisted {Result}", deliveryResult.Value);

diff --git a/AuthService/Kafka/KafkaRequestService.cs b/AuthService/Kafka/KafkaRequestService.cs
@@ -178,7 +178,8 @@ public async Task<bool> Produce(string topicName, Message<string, string> messag
                 bool IsTopicExists = IsTopicAvailable(topicName);
                 if (IsTopicExists && IsTopicPendingMessageBusExist( responseTopic))
                 {
-                    var deliveryResult = await _producer.ProduceAsync(topicName, message);
+                    var deliveryResult = await _producer.ProduceAsync(
+                    new TopicPartition(topicName, new Partition(0));
                     if (deliveryResult.Status == PersistenceStatus.Persisted)
                     {
                         _logger.LogInformation("Message delivery status: Persisted {Result}", deliveryResult.Value);
@@ -200,7 +201,8 @@ public async Task<bool> Produce(string topicName, Message<string, string> messag
                 bool IsTopicCreated = _kafkaTopicManager.CreateTopic(topicName, Convert.ToInt32(Environment.GetEnvironmentVariable("PARTITIONS_STANDART")), Convert.ToInt16(Environment.GetEnvironmentVariable("REPLICATION_FACTOR_STANDART")));
                 if (IsTopicCreated && IsTopicPendingMessageBusExist( responseTopic))
                 {
-                    var deliveryResult = await _producer.ProduceAsync(topicName, message);
+                    var deliveryResult = await _producer.ProduceAsync(
+                    new TopicPartition(topicName, new Partition(0));
                     if (deliveryResult.Status == PersistenceStatus.Persisted)
                     {
                         _logger.LogInformation("Message delivery status: Persisted {Result}", deliveryResult.Value);
@@ -235,7 +237,10 @@ private bool IsTopicPendingMessageBusExist(string responseTopic)
         private async Task Consume(IConsumer<string,string> localConsumer,string topicName)
         {   
             topicCount++;
-            localConsumer.Subscribe(topicName);
+            var partitions = new List<TopicPartition>();
+                partitions.Add(new TopicPartition(topicName, 0));
+
+            localConsumer.Assign(partitions);
             while (true)
             {
                 ConsumeResult<string, string> result = localConsumer.Consume();

diff --git a/AuthService/Kafka/KafkaService.cs b/AuthService/Kafka/KafkaService.cs
@@ -27,7 +27,10 @@ protected void ConfigureConsumer(string topicName)
             _consumer = new ConsumerBuilder<string, string>(config).Build();
             if(IsTopicAvailable(topicName))
             {
-                _consumer.Subscribe(topicName);
+                var partitions = new List<TopicPartition>();
+                partitions.Add(new TopicPartition(topicName, 0));
+
+                _consumer.Assign(partitions);
                 return;
             }
             throw new ConsumerTopicUnavailableException("Topic unavailable");
@@ -77,7 +80,8 @@ public async Task<bool> Produce( string topicName,Message<string, string> messag
             bool IsTopicExists = IsTopicAvailable(topicName);
             if (IsTopicExists)
             {
-                var deliveryResult = await _producer.ProduceAsync(topicName, message);
+                var deliveryResult = await _producer.ProduceAsync(
+                    new TopicPartition(topicName, new Partition(0));
                 if (deliveryResult.Status == PersistenceStatus.Persisted)
                 {
 
@@ -93,7 +97,8 @@ public async Task<bool> Produce( string topicName,Message<string, string> messag
             bool IsTopicCreated = _kafkaTopicManager.CreateTopic(topicName, Convert.ToInt32(Environment.GetEnvironmentVariable("PARTITIONS_STANDART")), Convert.ToInt16(Environment.GetEnvironmentVariable("REPLICATION_FACTOR_STANDART")));
             if (IsTopicCreated)
             {
-                var deliveryResult = await _producer.ProduceAsync(topicName, message);
+                var deliveryResult = await _producer.ProduceAsync(
+                    new TopicPartition(topicName, new Partition(0));
                 if (deliveryResult.Status == PersistenceStatus.Persisted)
                 {
                     _logger.LogInformation("Message delivery status: Persisted {Result}", deliveryResult.Value);

diff --git a/EntertaimentService/Kafka/KafkaRequestService.cs b/EntertaimentService/Kafka/KafkaRequestService.cs
@@ -179,7 +179,8 @@ public async Task<bool> Produce(string topicName, Message<string, string> messag
                 bool IsTopicExists = IsTopicAvailable(topicName);
                 if (IsTopicExists && IsTopicPendingMessageBusExist( responseTopic))
                 {
-                    var deliveryResult = await _producer.ProduceAsync(topicName, message);
+                    var deliveryResult = await _producer.ProduceAsync(
+                    new TopicPartition(topicName, new Partition(0));;
                     if (deliveryResult.Status == PersistenceStatus.Persisted)
                     {
                         _logger.LogInformation("Message delivery status: Persisted {Result}", deliveryResult.Value);
@@ -201,7 +202,8 @@ public async Task<bool> Produce(string topicName, Message<string, string> messag
                 bool IsTopicCreated = _kafkaTopicManager.CreateTopic(topicName, Convert.ToInt32(Environment.GetEnvironmentVariable("PARTITIONS_STANDART")), Convert.ToInt16(Environment.GetEnvironmentVariable("REPLICATION_FACTOR_STANDART")));
                 if (IsTopicCreated && IsTopicPendingMessageBusExist( responseTopic))
                 {
-                    var deliveryResult = await _producer.ProduceAsync(topicName, message);
+                    var deliveryResult = await _producer.ProduceAsync(
+                    new TopicPartition(topicName, new Partition(0));;
                     if (deliveryResult.Status == PersistenceStatus.Persisted)
                     {
                         _logger.LogInformation("Message delivery status: Persisted {Result}", deliveryResult.Value);
@@ -236,7 +238,9 @@ private bool IsTopicPendingMessageBusExist(string responseTopic)
         private async Task Consume(IConsumer<string,string> localConsumer,string topicName)
         {   
             topicCount++;
-            localConsumer.Subscribe(topicName);
+            var partitions = new List<TopicPartition>();
+            partitions.Add(new TopicPartition(topicName, 0));
+            localConsumer.Assign(partitions);
             while (true)
             {
                 ConsumeResult<string, string> result = localConsumer.Consume();

diff --git a/EntertaimentService/Kafka/KafkaService.cs b/EntertaimentService/Kafka/KafkaService.cs
@@ -27,7 +27,10 @@ protected void ConfigureConsumer(string topicName)
             _consumer = new ConsumerBuilder<string, string>(config).Build();
             if(IsTopicAvailable(topicName))
             {
-                _consumer.Subscribe(topicName);
+                var partitions = new List<TopicPartition>();
+                partitions.Add(new TopicPartition(topicName, 0));
+
+                _consumer.Assign(partitions);
                 return;
             }
             throw new ConsumerTopicUnavailableException("Topic unavailable");
@@ -77,7 +80,8 @@ public async Task<bool> Produce( string topicName,Message<string, string> messag
             bool IsTopicExists = IsTopicAvailable(topicName);
             if (IsTopicExists)
             {
-                var deliveryResult = await _producer.ProduceAsync(topicName, message);
+                var deliveryResult = await _producer.ProduceAsync(
+                    new TopicPartition(topicName, new Partition(0));
                 if (deliveryResult.Status == PersistenceStatus.Persisted)
                 {
 
@@ -93,7 +97,8 @@ public async Task<bool> Produce( string topicName,Message<string, string> messag
             bool IsTopicCreated = _kafkaTopicManager.CreateTopic(topicName, Convert.ToInt32(Environment.GetEnvironmentVariable("PARTITIONS_STANDART")), Convert.ToInt16(Environment.GetEnvironmentVariable("REPLICATION_FACTOR_STANDART")));
             if (IsTopicCreated)
             {
-                var deliveryResult = await _producer.ProduceAsync(topicName, message);
+                var deliveryResult = await _producer.ProduceAsync(
+                    new TopicPartition(topicName, new Partition(0));
                 if (deliveryResult.Status == PersistenceStatus.Persisted)
                 {
                     _logger.LogInformation("Message delivery status: Persisted {Result}", deliveryResult.Value);

diff --git a/MailService/Kafka/KafkaRequestService.cs b/MailService/Kafka/KafkaRequestService.cs
@@ -178,7 +178,8 @@ public async Task<bool> Produce(string topicName, Message<string, string> messag
                 bool IsTopicExists = IsTopicAvailable(topicName);
                 if (IsTopicExists && IsTopicPendingMessageBusExist( responseTopic))
                 {
-                    var deliveryResult = await _producer.ProduceAsync(topicName, message);
+                    var deliveryResult = await _producer.ProduceAsync(
+                    new TopicPartition(topicName, new Partition(0));
                     if (deliveryResult.Status == PersistenceStatus.Persisted)
                     {
                         _logger.LogInformation("Message delivery status: Persisted {Result}", deliveryResult.Value);
@@ -200,7 +201,8 @@ public async Task<bool> Produce(string topicName, Message<string, string> messag
                 bool IsTopicCreated = _kafkaTopicManager.CreateTopic(topicName, Convert.ToInt32(Environment.GetEnvironmentVariable("PARTITIONS_STANDART")), Convert.ToInt16(Environment.GetEnvironmentVariable("REPLICATION_FACTOR_STANDART")));
                 if (IsTopicCreated && IsTopicPendingMessageBusExist( responseTopic))
                 {
-                    var deliveryResult = await _producer.ProduceAsync(topicName, message);
+                    var deliveryResult = await _producer.ProduceAsync(
+                    new TopicPartition(topicName, new Partition(0));
                     if (deliveryResult.Status == PersistenceStatus.Persisted)
                     {
                         _logger.LogInformation("Message delivery status: Persisted {Result}", deliveryResult.Value);
@@ -235,7 +237,10 @@ private bool IsTopicPendingMessageBusExist(string responseTopic)
         private async Task Consume(IConsumer<string,string> localConsumer,string topicName)
         {   
             topicCount++;
-            localConsumer.Subscribe(topicName);
+            var partitions = new List<TopicPartition>();
+                partitions.Add(new TopicPartition(topicName, 0));
+
+            localConsumer.Assign(partitions);
             while (true)
             {
                 ConsumeResult<string, string> result = localConsumer.Consume();

diff --git a/MailService/Kafka/KafkaService.cs b/MailService/Kafka/KafkaService.cs
@@ -27,7 +27,10 @@ protected void ConfigureConsumer(string topicName)
             _consumer = new ConsumerBuilder<string, string>(config).Build();
             if(IsTopicAvailable(topicName))
             {
-                _consumer.Subscribe(topicName);
+                var partitions = new List<TopicPartition>();
+                partitions.Add(new TopicPartition(topicName, 0));
+
+                _consumer.Assign(partitions);
                 return;
             }
             throw new ConsumerTopicUnavailableException("Topic unavailable");
@@ -77,7 +80,8 @@ public async Task<bool> Produce( string topicName,Message<string, string> messag
             bool IsTopicExists = IsTopicAvailable(topicName);
             if (IsTopicExists)
             {
-                var deliveryResult = await _producer.ProduceAsync(topicName, message);
+                var deliveryResult = await _producer.ProduceAsync(
+                    new TopicPartition(topicName, new Partition(0));
                 if (deliveryResult.Status == PersistenceStatus.Persisted)
                 {
 
@@ -93,7 +97,8 @@ public async Task<bool> Produce( string topicName,Message<string, string> messag
             bool IsTopicCreated = _kafkaTopicManager.CreateTopic(topicName, Convert.ToInt32(Environment.GetEnvironmentVariable("PARTITIONS_STANDART")), Convert.ToInt16(Environment.GetEnvironmentVariable("REPLICATION_FACTOR_STANDART")));
             if (IsTopicCreated)
             {
-                var deliveryResult = await _producer.ProduceAsync(topicName, message);
+                var deliveryResult = await _producer.ProduceAsync(
+                    new TopicPartition(topicName, new Partition(0));
                 if (deliveryResult.Status == PersistenceStatus.Persisted)
                 {
                     _logger.LogInformation("Message delivery status: Persisted {Result}", deliveryResult.Value);

diff --git a/PromoService/Kafka/KafkaRequestService.cs b/PromoService/Kafka/KafkaRequestService.cs
@@ -178,7 +178,8 @@ public async Task<bool> Produce(string topicName, Message<string, string> messag
                 bool IsTopicExists = IsTopicAvailable(topicName);
                 if (IsTopicExists && IsTopicPendingMessageBusExist( responseTopic))
                 {
-                    var deliveryResult = await _producer.ProduceAsync(topicName, message);
+                    var deliveryResult = await _producer.ProduceAsync(
+                    new TopicPartition(topicName, new Partition(0));
                     if (deliveryResult.Status == PersistenceStatus.Persisted)
                     {
                         _logger.LogInformation("Message delivery status: Persisted {Result}", deliveryResult.Value);
@@ -200,7 +201,8 @@ public async Task<bool> Produce(string topicName, Message<string, string> messag
                 bool IsTopicCreated = _kafkaTopicManager.CreateTopic(topicName, Convert.ToInt32(Environment.GetEnvironmentVariable("PARTITIONS_STANDART")), Convert.ToInt16(Environment.GetEnvironmentVariable("REPLICATION_FACTOR_STANDART")));
                 if (IsTopicCreated && IsTopicPendingMessageBusExist( responseTopic))
                 {
-                    var deliveryResult = await _producer.ProduceAsync(topicName, message);
+                    var deliveryResult = await _producer.ProduceAsync(
+                    new TopicPartition(topicName, new Partition(0));
                     if (deliveryResult.Status == PersistenceStatus.Persisted)
                     {
                         _logger.LogInformation("Message delivery status: Persisted {Result}", deliveryResult.Value);
@@ -235,7 +237,10 @@ private bool IsTopicPendingMessageBusExist(string responseTopic)
         private async Task Consume(IConsumer<string,string> localConsumer,string topicName)
         {   
             topicCount++;
-            localConsumer.Subscribe(topicName);
+             var partitions = new List<TopicPartition>();
+                partitions.Add(new TopicPartition(topicName, 0));
+
+            localConsumer.Assign(partitions);
             while (true)
             {
                 ConsumeResult<string, string> result = localConsumer.Consume();