Trendyol#29 prevents to initialize kafka consumers for empty topic pa…

…rtitions
ahmetfurkankavraz · May 20, 2024 · 2eecccf · 2eecccf
1 parent e60a4c0
commit 2eecccf
Showing 1 changed file with 3 additions and 4 deletions.
diff --git a/src/Services/Implementations/KafkaRetryJobService.cs b/src/Services/Implementations/KafkaRetryJobService.cs
@@ -72,9 +72,6 @@ private async Task MoveMessagesForTopic(
             Action<IConsumer<string,string>, ConsumeResult<string,string>> consumerCommitStrategy
         )
         {
-            var consumer = _kafkaService.GetKafkaConsumer();
-            var producer = _kafkaService.GetKafkaProducer();
-
             var messageConsumeLimitPerTopic = _configuration.MessageConsumeLimitPerTopic;
 
             foreach (var (topicPartition, lag) in topicPartitionsWithLag)
@@ -83,6 +80,8 @@ Action<IConsumer<string,string>, ConsumeResult<string,string>> consumerCommitStr
                 {
                     continue;
                 }
+                var consumer = _kafkaService.GetKafkaConsumer();
+                var producer = _kafkaService.GetKafkaProducer();
                 var errorTopic = topicPartition.Topic;
 
                 try
@@ -95,7 +94,7 @@ Action<IConsumer<string,string>, ConsumeResult<string,string>> consumerCommitStr
 
                     while (currentLag > 0 && messageConsumeLimitPerTopic > 0)
                     {
-                        var result = consumer.Consume(TimeSpan.FromSeconds(10));
+                        var result = consumer.Consume(TimeSpan.FromSeconds(3));
 
                         if (result is null)
                         {