filebeat kafka input: add group_instance_id for static group membership (#51772) (#51886)

* filebeat kafka input: add group_instance_id for static group membership

The Kafka input now has a `group_instance_id` option wiring it to Consumer.Group.InstanceId (KIP-345): a member that rejoins within
`session_timeout` keeps its partitions instead of rebalancing.
Static membership needs Kafka `version` >= 2.3.0, so the input
fails fast with a clear error when it is set on the default 2.1.0.
Unset by default; existing configs are unaffected.

Assisted-By: Claude Code
(cherry picked from commit 6811ebc)

Co-authored-by: Anderson Queiroz <anderson.queiroz@elastic.co>

Author: mergify[bot]

changelog/fragments/1783416326-filebeat-kafka-group-instance-id.yaml

@@ -0,0 +1,36 @@
+# Kind can be one of:
+# - breaking-change: a change to previously-documented behavior
+# - deprecation: functionality that is being removed in a later release
+# - bug-fix: fixes a problem in a previous version
+# - enhancement: extends functionality but does not break or fix existing behavior
+# - feature: new functionality
+# - known-issue: problems that we are aware of in a given version
+# - security: impacts on the security of a product or a user’s deployment.
+# - upgrade: important information for someone upgrading from a prior version
+# - other: does not fit into any of the other categories
+kind: enhancement
+
+# Change summary; a 80ish characters long description of the change.
+summary: Add `group_instance_id` (KIP-345 static group membership) to the Filebeat Kafka input.
+
+# Long description; in case the summary is not enough to describe the change
+# this field accommodate a description without length limits.
+description: >
+  The Filebeat Kafka input now exposes `group_instance_id`, wired to Sarama's
+  `Consumer.Group.InstanceId`. Setting a stable, unique id per consumer instance
+  enables Kafka static group membership (KIP-345): a member that restarts and
+  rejoins within `session_timeout` is recognized as the same member, avoiding the
+  rebalance storms that dynamic membership causes during rolling restarts of
+  multi-replica deployments. Requires `version` >= 2.3.0. Unset by default, so
+  existing behavior is unchanged.
+
+# Affected component; usually one of "elastic-agent", "fleet-server", "filebeat", "metricbeat", "auditbeat", "all", etc.
+component: "filebeat"
+
+# PR URL; optional; the PR number that added the changeset.
+# If not present is automatically filled by the tooling finding the PR where this changelog fragment has been added.
+# pr: https://github.com/elastic/beats/pull/0000
+
+# Issue URL; optional; the GitHub issue related to this changeset (either closes or is part of).
+# If not present is automatically filled by the tooling with the issue linked to the PR number.
+issue: https://github.com/elastic/beats/issues/51768

docs/reference/filebeat/filebeat-input-kafka.md

@@ -128,6 +128,25 @@ stack: ga 9.3.8+, ga 9.4.4+, ga 9.5
 How often the consumer sends heartbeats to the broker. This must be lower than `session_timeout`, and is typically set to no more than a third of that value. Default is 3s.
 
 
+### `group_instance_id` [_group_instance_id]
+```{applies_to}
+stack: ga 9.3+
+```
+
+A stable identifier that enables Kafka static group membership ([KIP-345](https://cwiki.apache.org/confluence/display/KAFKA/KIP-345%3A+Introduce+static+membership+protocol+to+reduce+consumer+rebalances)). When set, a consumer that restarts and rejoins the group within [`session_timeout`](#_session_timeout) is recognized as the same member and keeps its partition assignment, avoiding the rebalances that a rolling restart of multiple instances would otherwise trigger. Requires `version` to be at least `2.3.0`.
+
+Each consumer instance sharing a `group_id` must use a **unique** `group_instance_id`. Two live members with the same value cause the broker to fence one of them. Set `group_instance_id` to a value that is unique to each consumer instance and stable across restarts. The option is unset by default, in which case the consumer uses dynamic membership.
+
+```yaml
+- type: kafka
+  hosts: ["kafka-broker:9092"]
+  topics: ["my-topic"]
+  group_id: "filebeat"
+  version: "2.3.0"
+  group_instance_id: "<unique-instance-id>"
+```
+
+
 ### `max_wait_time` [_max_wait_time]
 
 How long to wait for the minimum number of input bytes while reading. Default is 250ms.

filebeat/_meta/config/filebeat.inputs.reference.yml.tmpl

@@ -703,6 +703,10 @@ filebeat.inputs:
   # How often the consumer sends heartbeats. Must be lower than session_timeout.
   #heartbeat_interval: 3s
 
+  # Static group membership id (KIP-345). Set a unique, stable value per instance
+  # so a restart does not trigger a rebalance. Requires version >= 2.3.0.
+  #group_instance_id: "<unique-instance-id>"
+
   # How long to wait for the minimum number of input bytes while reading.
   #max_wait_time: 250ms
 

filebeat/filebeat.reference.yml

@@ -1122,6 +1122,10 @@ filebeat.inputs:
   # How often the consumer sends heartbeats. Must be lower than session_timeout.
   #heartbeat_interval: 3s
 
+  # Static group membership id (KIP-345). Set a unique, stable value per instance
+  # so a restart does not trigger a rebalance. Requires version >= 2.3.0.
+  #group_instance_id: "<unique-instance-id>"
+
   # How long to wait for the minimum number of input bytes while reading.
   #max_wait_time: 250ms
 

filebeat/input/kafka/config.go

@@ -38,6 +38,7 @@ type kafkaInputConfig struct {
 	Hosts                    []string          `config:"hosts" validate:"required"`
 	Topics                   []string          `config:"topics" validate:"required"`
 	GroupID                  string            `config:"group_id" validate:"required"`
+	GroupInstanceID          string            `config:"group_instance_id"`
 	ClientID                 string            `config:"client_id"`
 	Version                  kafka.Version     `config:"version"`
 	InitialOffset            initialOffset     `config:"initial_offset"`
@@ -180,6 +181,13 @@ func newSaramaConfig(config kafkaInputConfig, logger *logp.Logger) (*sarama.Conf
 	k.Consumer.Group.Session.Timeout = config.SessionTimeout
 	k.Consumer.Group.Heartbeat.Interval = config.HeartbeatInterval
 
+	if config.GroupInstanceID != "" {
+		if !version.IsAtLeast(sarama.V2_3_0_0) {
+			return nil, fmt.Errorf("group_instance_id requires 'version' >= 2.3.0 for static group membership (KIP-345); configured version is %q", config.Version)
+		}
+		k.Consumer.Group.InstanceId = config.GroupInstanceID
+	}
+
 	k.Net.DialTimeout = config.Timeout
 	k.Net.ReadTimeout = config.Timeout
 	k.Net.WriteTimeout = config.Timeout

filebeat/input/kafka/config_test.go

@@ -18,6 +18,7 @@
 package kafka
 
 import (
+	"strings"
 	"testing"
 	"time"
 
@@ -39,6 +40,8 @@ func TestNewSaramaConfigDefaults(t *testing.T) {
 	assert.Equal(t, 30*time.Second, saramaConfig.Net.DialTimeout)
 	assert.Equal(t, 30*time.Second, saramaConfig.Net.ReadTimeout)
 	assert.Equal(t, 30*time.Second, saramaConfig.Net.WriteTimeout)
+	assert.Empty(t, saramaConfig.Consumer.Group.InstanceId,
+		"group_instance_id must be unset by default so consumers keep dynamic membership")
 }
 
 // TestNewSaramaConfigTimeoutOverrides verifies that the session_timeout,
@@ -62,3 +65,57 @@ func TestNewSaramaConfigTimeoutOverrides(t *testing.T) {
 	assert.Equal(t, 60*time.Second, saramaConfig.Net.WriteTimeout)
 	assert.Equal(t, 15*time.Second, saramaConfig.Net.KeepAlive)
 }
+
+// TestNewSaramaConfigGroupInstanceID verifies that group_instance_id is
+// propagated to sarama's Consumer.Group.InstanceId, enabling Kafka static
+// group membership (KIP-345), when a compatible protocol version is set.
+func TestNewSaramaConfigGroupInstanceID(t *testing.T) {
+	config := defaultConfig()
+	config.Version = "2.3.0"
+	config.GroupInstanceID = "filebeat-pod-1"
+
+	saramaConfig, err := newSaramaConfig(config, logp.NewNopLogger())
+	require.NoError(t, err)
+
+	assert.Equal(t, "filebeat-pod-1", saramaConfig.Consumer.Group.InstanceId,
+		"group_instance_id must be propagated to sarama's Consumer.Group.InstanceId")
+}
+
+// TestNewSaramaConfigGroupInstanceIDRequiresVersion verifies that setting
+// group_instance_id with a protocol version below 2.3.0 (including the 2.1.0
+// default) fails early with a clear, Filebeat-oriented error rather than
+// sarama's opaque "need Version >= 2.3" message.
+func TestNewSaramaConfigGroupInstanceIDRequiresVersion(t *testing.T) {
+	config := defaultConfig() // Version defaults to 2.1.0
+	config.GroupInstanceID = "filebeat-pod-1"
+
+	_, err := newSaramaConfig(config, logp.NewNopLogger())
+	require.Error(t, err, "group_instance_id below version 2.3.0 must be rejected")
+	assert.ErrorContains(t, err, "group_instance_id requires 'version' >= 2.3.0",
+		"error must carry the stable, searchable message naming the option and required version")
+}
+
+// TestNewSaramaConfigGroupInstanceIDInvalid verifies that malformed
+// group_instance_id values are rejected by sarama's own validation (length,
+// reserved names, and the allowed character set) even when the version is
+// compatible.
+func TestNewSaramaConfigGroupInstanceIDInvalid(t *testing.T) {
+	tests := map[string]string{
+		"dot":            ".",
+		"dot-dot":        "..",
+		"illegal char":   "has space",
+		"too long (250)": strings.Repeat("a", 250),
+	}
+
+	for name, id := range tests {
+		t.Run(name, func(t *testing.T) {
+			config := defaultConfig()
+			config.Version = "2.3.0"
+			config.GroupInstanceID = id
+
+			_, err := newSaramaConfig(config, logp.NewNopLogger())
+			assert.Error(t, err,
+				"invalid group_instance_id %q must be rejected", id)
+		})
+	}
+}

x-pack/filebeat/filebeat.reference.yml

@@ -2822,6 +2822,10 @@ filebeat.inputs:
   # How often the consumer sends heartbeats. Must be lower than session_timeout.
   #heartbeat_interval: 3s
 
+  # Static group membership id (KIP-345). Set a unique, stable value per instance
+  # so a restart does not trigger a rebalance. Requires version >= 2.3.0.
+  #group_instance_id: "<unique-instance-id>"
+
   # How long to wait for the minimum number of input bytes while reading.
   #max_wait_time: 250ms