feat kafka: support group.instance.id option

<section id="quibbler-autodescription">

#### Поддержка статических членов группы потребителей Kafka 📝

- 🔄 Добавлена возможность использовать `{pod_name}` в `group_instance_id` для динамической подстановки имени пода, что позволяет корректно настраивать статические члены группы потребителей.
- 🛠 Обновлена логика обработки `group_id` и `group_instance_id`, чтобы поддерживать подстановку имени пода из переменной окружения `YA_K8S_POD_NAME`.
- 🧪 Обновлены конфигурации и тесты для корректной работы с новой функциональностью, включая добавление фикстур для эмуляции переменных окружения в тестах.
- 📦 Расширена схема конфигурации Kafka, чтобы поддерживать новое поле `group_instance_id` и обновленное описание для `env_pod_name`.

<a href="https://nda.ya.ru/t/qa0kX64r7DqvtN"><font size="2">Autodescription by Yandex Code Assistant</font></a>
</section>
commit_hash:cb19e17260acf07bdd50e89c05195a24e25fab64

Author: jdillinger

kafka/include/userver/kafka/impl/configuration.hpp

@@ -52,6 +52,7 @@ struct ConsumerConfiguration final {
     std::string group_id;
     std::string auto_offset_reset{"smallest"};
     std::optional<std::string> env_pod_name{};
+    std::optional<std::string> group_instance_id{};
     std::chrono::milliseconds max_callback_duration{300000};
 
     RdKafkaOptions rd_kafka_options;

kafka/src/kafka/consumer_component.yaml

@@ -18,6 +18,15 @@ properties:
             with the client instance which made it,
             which can be helpful in debugging and troubleshooting scenarios.
         default: userver
+    group_instance_id:
+        type: string
+        description: |
+            Enable static group membership. Static group members are able
+            to leave and rejoin a group within the configured session.timeout.ms
+            without prompting a group rebalance. This should be used in combination
+            with a larger session.timeout.ms to avoid group rebalances caused by
+            transient unavailability (e.g. process restarts)
+        default: none
     topics:
         type: array
         description: list of topics consumer subscribes
@@ -26,13 +35,11 @@ properties:
             description: topic name
     max_batch_size:
         type: integer
-        description: maximum number of messages consumer waits for new messages before
-            calling a callback
+        description: maximum number of messages consumer waits for new messages before calling a callback
         default: 1
     poll_timeout:
         type: string
-        description: maximum amount of time consumer waits for new messages before
-            calling a callback
+        description: maximum amount of time consumer waits for new messages before calling a callback
         default: 1s
     message_key_log_format:
         type: string
@@ -78,10 +85,9 @@ properties:
     env_pod_name:
         type: string
         description: |
-            if defined and `group_id` value contains
-            `{pod_name}` substring, the substring
-            is replaced with the value of the environment
-            variable `env_pod_name`
+            if defined and `group_id` or `group_instance_id` value contains
+            `{pod_name}` substring, the substring is replaced with
+            the value of the environment variable `env_pod_name`
         default: none
     security_protocol:
         type: string

kafka/src/kafka/impl/configuration.cpp

@@ -84,7 +84,7 @@ std::string ResolveGroupId(const ConsumerConfiguration& configuration) {
     }
 
     const auto pos = group_id.find(kPodNameSubstr);
-    if (group_id.find(kPodNameSubstr) != std::string::npos) {
+    if (pos != std::string::npos) {
         const auto pod_name = ResolvePodName(*configuration.env_pod_name);
 
         return group_id.replace(pos, kPodNameSubstr.size(), pod_name);
@@ -93,6 +93,28 @@ std::string ResolveGroupId(const ConsumerConfiguration& configuration) {
     return group_id;
 }
 
+std::optional<std::string> ResolveGroupInstanceId(const ConsumerConfiguration& configuration) {
+    static constexpr std::string_view kPodNameSubstr{"{pod_name}"};
+
+    if (!configuration.group_instance_id.has_value()) {
+        return std::nullopt;
+    }
+    auto group_instance_id{configuration.group_instance_id.value()};
+
+    if (!configuration.env_pod_name.has_value()) {
+        return std::nullopt;
+    }
+
+    const auto pos = group_instance_id.find(kPodNameSubstr);
+    if (pos != std::string::npos) {
+        const auto pod_name = ResolvePodName(*configuration.env_pod_name);
+
+        return group_instance_id.replace(pos, kPodNameSubstr.size(), pod_name);
+    }
+
+    return group_instance_id;
+}
+
 }  // namespace
 
 CommonConfiguration Parse(const yaml_config::YamlConfig& config, formats::parse::To<CommonConfiguration>) {
@@ -169,6 +191,7 @@ ConsumerConfiguration Parse(const yaml_config::YamlConfig& config, formats::pars
     if (config.HasMember(kEnvPodNameField)) {
         consumer.env_pod_name = config[kEnvPodNameField].As<std::string>();
     }
+    consumer.group_instance_id = config["group_instance_id"].As<std::optional<std::string>>(std::nullopt);
 
     return consumer;
 }
@@ -327,10 +350,19 @@ void Configuration::SetRdKafka(const RdKafkaOptions& rd_kafka_options) {
 void Configuration::SetConsumer(const ConsumerConfiguration& configuration) {
     const auto group_id = ResolveGroupId(configuration);
     UINVARIANT(!group_id.empty(), "Consumer group_id must not be empty");
+    const auto group_instance_id = ResolveGroupInstanceId(configuration);
 
-    LOG_INFO("Consumer '{}' is going to join group '{}'", name_, group_id);
+    LOG_INFO(
+        "Consumer '{}' is going to join group '{}' as instance '{}'",
+        name_,
+        group_id,
+        group_instance_id.value_or("<chosen by broker>")
+    );
 
     SetOption("group.id", group_id);
+    if (group_instance_id.has_value()) {
+        SetOption("group.instance.id", group_instance_id.value());
+    }
     SetOption("enable.auto.commit", "false");
     SetOption("auto.offset.reset", configuration.auto_offset_reset);
     SetOption("max.poll.interval.ms", configuration.max_callback_duration);

kafka/src/kafka/producer_component.yaml

@@ -47,8 +47,7 @@ properties:
         default: 1000000
     message_send_max_retries:
         type: integer
-        description: maximum number of send request retries until `delivery_timeout`
-            reached
+        description: maximum number of send request retries until `delivery_timeout` reached
         default: 2147483647
     retry_backoff:
         type: string