Merge pull request #130 from leroy-merlin-br/feat/improve-config-to-a…

…ccept-max-pool-interval-ms feat: improve config to accept max poll interval ms
leroy-merlin-br · Sep 4, 2023 · 530f03b · 530f03b
2 parents f06105e + 3676365
commit 530f03b
Show file tree

Hide file tree

Showing 5 changed files with 19 additions and 0 deletions.
diff --git a/config/kafka.php b/config/kafka.php
@@ -102,6 +102,10 @@
                         // time we need to wait until receiving a message?
                         'timeout' => 20000,
 
+                        // A max interval for consumer to make poll calls. That means: how much
+                        // time we need to wait for poll calls until consider the consumer has inactive.
+                        'max_poll_interval_ms' => 300000,
+
                         // Once you've enabled this, the Kafka consumer will commit the
                         // offset of the last message received in response to its poll() call
                         'auto_commit' => true,

diff --git a/src/Connectors/Consumer/HighLevel.php b/src/Connectors/Consumer/HighLevel.php
@@ -14,9 +14,15 @@ class HighLevel implements ConnectorInterface
     public function getConsumer(bool $autoCommit, AbstractConfigManager $configManager): ConsumerInterface
     {
         $conf = $this->getConf($configManager);
+        $maxPollIntervalMs = (int) $configManager->get('max_poll_interval_ms');
 
         $conf->set('group.id', $configManager->get('consumer_group'));
         $conf->set('auto.offset.reset', $configManager->get('offset_reset'));
+        $conf->set(
+            'max.poll.interval.ms',
+            $maxPollIntervalMs ?: 300000
+        );
+
         if (!$autoCommit) {
             $conf->set('enable.auto.commit', 'false');
         }

diff --git a/src/Connectors/Consumer/LowLevel.php b/src/Connectors/Consumer/LowLevel.php
@@ -15,7 +15,14 @@ class LowLevel implements ConnectorInterface
     public function getConsumer(bool $autoCommit, AbstractConfigManager $configManager): ConsumerInterface
     {
         $conf = $this->getConf();
+        $maxPollIntervalMs = (int) $configManager->get('max_poll_interval_ms');
+
         $conf->set('group.id', $configManager->get('consumer_group'));
+        $conf->set(
+            'max.poll.interval.ms',
+            $maxPollIntervalMs ?: 300000
+        );
+
         if (!$autoCommit) {
             $conf->set('enable.auto.commit', 'false');
         }

diff --git a/tests/Unit/Connectors/Consumer/HighLevelTest.php b/tests/Unit/Connectors/Consumer/HighLevelTest.php
@@ -20,6 +20,7 @@ public function testItShouldMakeConnectorSetup(): void
             'topic_id' => 'some_topic',
             'offset_reset' => 'earliest',
             'timeout' => 1000,
+            'max_poll_interval_ms' => 900000,
         ]);
         $connector = new HighLevel();
 

diff --git a/tests/Unit/Connectors/Consumer/LowLevelTest.php b/tests/Unit/Connectors/Consumer/LowLevelTest.php
@@ -19,6 +19,7 @@ public function testItShouldMakeConnectorSetup(): void
             'consumer_group' => 'some-group',
             'topic' => 'some_topic',
             'offset_reset' => 'earliest',
+            'max_poll_interval_ms' => 900000,
             'offset' => 0,
             'partition' => 1,
         ]);