From c80af1a6e58057a6c37c02fe19e97dc24ea8bcae Mon Sep 17 00:00:00 2001
From: Leonid Groshev <OtezVikentiy@gmail.com>
Date: Mon, 15 Jan 2024 15:00:29 +0300
Subject: [PATCH] added configuration for consumers max.poll.interval.ms

---
 src/Configuration/Type/MaxPollIntervalMs.php  | 65 +++++++++++++++++++
 src/DependencyInjection/Configuration.php     |  4 ++
 src/Resources/config/configuration_types.xml  |  4 ++
 .../Type/MaxPollIntervalMsTest.php            | 26 ++++++++
 4 files changed, 99 insertions(+)
 create mode 100644 src/Configuration/Type/MaxPollIntervalMs.php
 create mode 100644 tests/Unit/Configuration/Type/MaxPollIntervalMsTest.php

diff --git a/src/Configuration/Type/MaxPollIntervalMs.php b/src/Configuration/Type/MaxPollIntervalMs.php
new file mode 100644
index 0000000..811534c
--- /dev/null
+++ b/src/Configuration/Type/MaxPollIntervalMs.php
@@ -0,0 +1,65 @@
+<?php
+
+declare(strict_types=1);
+
+namespace StsGamingGroup\KafkaBundle\Configuration\Type;
+
+use StsGamingGroup\KafkaBundle\Configuration\Contract\CastValueInterface;
+use StsGamingGroup\KafkaBundle\Configuration\Contract\ConsumerConfigurationInterface;
+use StsGamingGroup\KafkaBundle\Configuration\Contract\KafkaConfigurationInterface;
+use StsGamingGroup\KafkaBundle\Configuration\Traits\SupportsConsumerTrait;
+use Symfony\Component\Console\Input\InputOption;
+
+class MaxPollIntervalMs implements ConsumerConfigurationInterface, KafkaConfigurationInterface, CastValueInterface
+{
+    use SupportsConsumerTrait;
+
+    public const NAME = 'max_poll_interval_ms';
+
+    public function getName(): string
+    {
+        return self::NAME;
+    }
+
+    public function getKafkaProperty(): string
+    {
+        return 'max.poll.interval.ms';
+    }
+
+    public function getMode(): int
+    {
+        return InputOption::VALUE_REQUIRED;
+    }
+
+    public function getDescription(): string
+    {
+        return sprintf(
+            <<<EOT
+            The maximum delay between invocations of poll() when using consumer group management. This places an upper
+            bound on the amount of time that the consumer can be idle before fetching more records. If poll() is not
+            called before expiration of this timeout, then the consumer is considered failed and the group will
+            rebalance in order to reassign the partitions to another member. For consumers using a non-null
+            group.instance.id which reach this timeout, partitions will not be immediately reassigned. Instead, the
+            consumer will stop sending heartbeats and partitions will be reassigned after expiration of
+            session.timeout.ms. This mirrors the behavior of a static consumer which has shutdown.
+            Defaults to %s ms. Set 0 to disable.
+            EOT,
+            $this->getDefaultValue()
+        );
+    }
+
+    public function isValueValid($value): bool
+    {
+        return is_numeric($value) && !str_contains((string)$value, '.') && $value >= 0;
+    }
+
+    public function getDefaultValue(): int
+    {
+        return 300000;
+    }
+
+    public function cast($validatedValue): int
+    {
+        return (int) $validatedValue;
+    }
+}
diff --git a/src/DependencyInjection/Configuration.php b/src/DependencyInjection/Configuration.php
index 5161818..3e7e08e 100644
--- a/src/DependencyInjection/Configuration.php
+++ b/src/DependencyInjection/Configuration.php
@@ -13,6 +13,7 @@
 use StsGamingGroup\KafkaBundle\Configuration\Type\EnableAutoOffsetStore;
 use StsGamingGroup\KafkaBundle\Configuration\Type\GroupId;
 use StsGamingGroup\KafkaBundle\Configuration\Type\LogLevel;
+use StsGamingGroup\KafkaBundle\Configuration\Type\MaxPollIntervalMs;
 use StsGamingGroup\KafkaBundle\Configuration\Type\MaxRetries;
 use StsGamingGroup\KafkaBundle\Configuration\Type\MaxRetryDelay;
 use StsGamingGroup\KafkaBundle\Configuration\Type\ProducerPartition;
@@ -143,6 +144,9 @@ private function addConsumerConfigurations(NodeBuilder $builder): void
             ->integerNode(StatisticsIntervalMs::NAME)
                 ->defaultValue((new StatisticsIntervalMs)->getDefaultValue())
             ->end()
+            ->integerNode(MaxPollIntervalMs::NAME)
+                ->defaultValue((new MaxPollIntervalMs)->getDefaultValue())
+            ->end()
             ->scalarNode(AutoOffsetReset::NAME)
                 ->defaultValue((new AutoOffsetReset)->getDefaultValue())
                 ->cannotBeEmpty()
diff --git a/src/Resources/config/configuration_types.xml b/src/Resources/config/configuration_types.xml
index b217f0e..f4021fd 100644
--- a/src/Resources/config/configuration_types.xml
+++ b/src/Resources/config/configuration_types.xml
@@ -29,6 +29,10 @@
                  id="sts_gaming_group_kafka.configuration.type.statistics_interval_ms">
             <tag name="sts_gaming_group_kafka.configuration.type"/>
         </service>
+        <service class="StsGamingGroup\KafkaBundle\Configuration\Type\MaxPollIntervalMs"
+                 id="sts_gaming_group_kafka.configuration.type.max_poll_interval_ms">
+            <tag name="sts_gaming_group_kafka.configuration.type"/>
+        </service>
         <service class="StsGamingGroup\KafkaBundle\Configuration\Type\Decoder" id="sts_gaming_group_kafka.configuration.type.decoder">
             <tag name="sts_gaming_group_kafka.configuration.type"/>
         </service>
diff --git a/tests/Unit/Configuration/Type/MaxPollIntervalMsTest.php b/tests/Unit/Configuration/Type/MaxPollIntervalMsTest.php
new file mode 100644
index 0000000..d4e2f10
--- /dev/null
+++ b/tests/Unit/Configuration/Type/MaxPollIntervalMsTest.php
@@ -0,0 +1,26 @@
+<?php
+
+declare(strict_types=1);
+
+namespace StsGamingGroup\KafkaBundle\Tests\Unit\Configuration\Type;
+
+use StsGamingGroup\KafkaBundle\Configuration\Contract\ConfigurationInterface;
+use StsGamingGroup\KafkaBundle\Configuration\Type\MaxPollIntervalMs;
+
+class MaxPollIntervalMsTest extends AbstractConfigurationTest
+{
+    protected function getConfiguration(): ConfigurationInterface
+    {
+        return new MaxPollIntervalMs();
+    }
+
+    protected function getValidValues(): array
+    {
+        return ['1', 2, 1000, '1000', 5000, 30000, 100000, 2_000_000];
+    }
+
+    protected function getInvalidValues(): array
+    {
+        return [-1, '-1', 1.51, '2.55', '', [], null, new \stdClass(), false, true];
+    }
+}