Add share group consumer

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 12.2.0
+  - Add share group consumer support to Kafka input via `consumer_mode => "share_group"` (Kafka 4.0+ KIP-932) [#252](https://github.com/logstash-plugins/logstash-integration-kafka/pull/252)
+
 ## 12.1.3
   - Upgrades `httpcore5` dependency to v5.3.6 [#262](https://github.com/logstash-plugins/logstash-integration-kafka/pull/262)
 

docs/input-kafka.asciidoc

@@ -53,13 +53,49 @@ The Logstash Kafka consumer handles group management and uses the default offset
 strategy using Kafka topics.
 
 Logstash instances by default form a single logical group to subscribe to Kafka topics
-Each Logstash Kafka consumer can run multiple threads to increase read throughput. Alternatively, 
+Each Logstash Kafka consumer can run multiple threads to increase read throughput. Alternatively,
 you could run multiple Logstash instances with the same `group_id` to spread the load across
 physical machines. Messages in a topic will be distributed to all Logstash instances with
 the same `group_id`.
 
-Ideally you should have as many threads as the number of partitions for a perfect balance --
-more threads than partitions means that some threads will be idle
+The number of consumer threads this input should spawn for fetching events and pushing them into the pipeline's queue.
+
+When configured with `consumer_mode => consumer_group`, each partition in the topic can only be assigned to a single consumer at a time, so if the _total_ number of consumer threads across all consumers in the group exceeds the number of partitions, some of those threads will be idle.
+
+This constraint doesn't apply when configured with `consumer_mode => share_group` which is not partition-oriented.
+
+[id="plugins-{type}s-{plugin}-share-groups"]
+==== Kafka Share Groups (queue semantics)
+
+Setting <<plugins-{type}s-{plugin}-consumer_mode>> to `share_group` enables Kafka 4.0+ Share
+Groups (KIP-932). Share groups provide *queue semantics*: any consumer in the group can read
+from any partition, and each record is delivered to exactly one consumer. Unlike consumer
+groups, the number of active consumers is no longer bounded by the partition count.
+
+Requirements:
+
+* Kafka brokers 4.0 or later with `group.share.enable=true`
+* The `group_id` setting identifies the share group name
+
+Share group consumers must acknowledge each record.
+After an event is accepted by the pipeline's queue, it is acknowledged as `ACCEPT` to ensure that no other consumer in the share group receives it.
+When an event is either _not_ accepted by the pipeline's queue or an error occurs before it is offered to the queue, the <<plugins-{type}s-{plugin}-share_group_acknowledgement_on_error>> option controls how the event is acknowledged (`release` for redelivery or `reject` to permanently discard).
+
+Example configuration:
+
+[source,ruby]
+--
+input {
+  kafka {
+    bootstrap_servers => "localhost:9092"
+    topics => ["my-topic"]
+    group_id => "my-share-group"
+    consumer_mode => "share_group"
+    consumer_threads => 4
+    share_group_acknowledgement_on_error => "release"
+  }
+}
+--
 
 For more information see https://kafka.apache.org/{kafka_client_doc}/documentation.html#theconsumer
 
@@ -123,6 +159,7 @@ See the https://kafka.apache.org/{kafka_client_doc}/documentation for more detai
 | <<plugins-{type}s-{plugin}-client_id>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-client_rack>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-connections_max_idle_ms>> |<<number,number>>|No
+| <<plugins-{type}s-{plugin}-consumer_mode>> |<<string,string>>, one of `["consumer_group", "share_group"]`|No
 | <<plugins-{type}s-{plugin}-consumer_threads>> |<<number,number>>|No
 | <<plugins-{type}s-{plugin}-decorate_events>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-enable_auto_commit>> |<<boolean,boolean>>|No
@@ -175,6 +212,7 @@ See the https://kafka.apache.org/{kafka_client_doc}/documentation for more detai
 | <<plugins-{type}s-{plugin}-security_protocol>> |<<string,string>>, one of `["PLAINTEXT", "SSL", "SASL_PLAINTEXT", "SASL_SSL"]`|No
 | <<plugins-{type}s-{plugin}-send_buffer_bytes>> |<<number,number>>|No
 | <<plugins-{type}s-{plugin}-session_timeout_ms>> |<<number,number>>|No
+| <<plugins-{type}s-{plugin}-share_group_acknowledgement_on_error>> |<<string,string>>, one of `["release", "reject"]`|No
 | <<plugins-{type}s-{plugin}-ssl_endpoint_identification_algorithm>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-ssl_key_password>> |<<password,password>>|No
 | <<plugins-{type}s-{plugin}-ssl_keystore_location>> |a valid filesystem path|No
@@ -194,13 +232,18 @@ input plugins.
 &nbsp;
 
 [id="plugins-{type}s-{plugin}-auto_commit_interval_ms"]
-===== `auto_commit_interval_ms` 
+===== `auto_commit_interval_ms`
 
   * Value type is <<number,number>>
   * Default value is `5000`.
 
 The frequency in milliseconds that the consumer offsets are committed to Kafka.
 
+[NOTE]
+=====
+This option is supported only with `consumer_mode => consumer_group`.
+=====
+
 [id="plugins-{type}s-{plugin}-auto_offset_reset"]
 ===== `auto_offset_reset` 
 
@@ -215,6 +258,12 @@ What to do when there is no initial offset in Kafka or if an offset is out of ra
 * none: throw exception to the consumer if no previous offset is found for the consumer's group
 * anything else: throw exception to the consumer.
 
+[NOTE]
+=====
+This option is supported only with `consumer_mode => consumer_group`.
+For `share_group` mode, the equivalent is the broker/group-level `share.auto.offset.reset` setting.
+=====
+
 [id="plugins-{type}s-{plugin}-bootstrap_servers"]
 ===== `bootstrap_servers` 
 
@@ -285,14 +334,53 @@ https://cwiki.apache.org/confluence/display/KAFKA/KIP-392%3A+Allow+consumers+to+
 
 Close idle connections after the number of milliseconds specified by this config.
 
+[id="plugins-{type}s-{plugin}-consumer_mode"]
+===== `consumer_mode`
+
+* Value can be either of: `consumer_group`, `share_group`
+* Default value is `consumer_group`
+
+Sets the consumer mode for this input.
+
+`consumer_group` (default) uses the standard Kafka consumer group protocol (`KafkaConsumer`).
+Partitions are assigned exclusively to consumers — each partition is consumed by exactly one
+thread in the group at a time.
+
+`share_group` uses the Kafka 4.0+ Share Group protocol (`KafkaShareConsumer`, KIP-932).
+Share groups provide *queue semantics*: records from any partition can be fetched by any
+consumer in the group, and each record is delivered to exactly one consumer. There is no
+partition assignment, so any number of consumers can scale independently without being
+limited by the partition count.
+
+To use share groups, your Kafka brokers must be version 4.0 or later and must have
+`group.share.enable=true` set in the broker configuration.
+
+[NOTE]
+====
+Setting any of the following options in the pipeline config raises a configuration error in
+`share_group` mode, since they do not apply to the Share Group protocol: `group_protocol`,
+`group_instance_id`, `partition_assignment_strategy`, `heartbeat_interval_ms`,
+`session_timeout_ms`, `isolation_level`, `enable_auto_commit`, `auto_commit_interval_ms`,
+`auto_offset_reset`, `exclude_internal_topics`, `schema_registry_url`.
+
+Some of these have broker/group-level equivalents for share groups that can be configured
+directly on the broker (for example, `auto.offset.reset` → `share.auto.offset.reset`,
+`heartbeat.interval.ms` → `share.heartbeat.interval.ms`,
+`session.timeout.ms` → `share.session.timeout.ms`,
+`isolation.level` → `share.isolation.level`).
+====
+
 [id="plugins-{type}s-{plugin}-consumer_threads"]
-===== `consumer_threads` 
+===== `consumer_threads`
 
   * Value type is <<number,number>>
   * Default value is `1`
 
-Ideally you should have as many threads as the number of partitions for a perfect
-balance — more threads than partitions means that some threads will be idle
+The number of consumer threads this input should spawn for fetching events and pushing them into the pipeline's queue.
+
+When configured with `consumer_mode => consumer_group`, each partition in the topic can only be assigned to a single consumer at a time, so if the _total_ number of consumer threads across all consumers in the group exceeds the number of partitions, some of those threads will be idle.
+
+This constraint doesn't apply when configured with `consumer_mode => share_group` which is not partition-oriented.
 
 [id="plugins-{type}s-{plugin}-decorate_events"]
 ===== `decorate_events` 
@@ -327,7 +415,7 @@ A topic will be auto-created only if this configuration is set to `true` and a
 otherwise auto-topic creation is not permitted. 
 
 [id="plugins-{type}s-{plugin}-enable_auto_commit"]
-===== `enable_auto_commit` 
+===== `enable_auto_commit`
 
   * Value type is <<boolean,boolean>>
   * Default value is `true`
@@ -339,15 +427,26 @@ If true, periodically commit to Kafka the offsets of messages already returned b
 the consumer. If value is `false` however, the offset is committed every time the
 consumer writes data fetched from the topic to the in-memory or persistent queue.
 
+[NOTE]
+=====
+This option is supported only with `consumer_mode => consumer_group`.
+Share groups manage acknowledgement per record rather than via committed offsets — see <<plugins-{type}s-{plugin}-share_group_acknowledgement_on_error>>.
+=====
+
 [id="plugins-{type}s-{plugin}-exclude_internal_topics"]
-===== `exclude_internal_topics` 
+===== `exclude_internal_topics`
 
   * Value type is <<string,string>>
   * There is no default value for this setting.
 
 Whether records from internal topics (such as offsets) should be exposed to the consumer.
 If set to true the only way to receive records from an internal topic is subscribing to it.
 
+[NOTE]
+=====
+This option is supported only with `consumer_mode => consumer_group`.
+=====
+
 [id="plugins-{type}s-{plugin}-fetch_max_bytes"]
 ===== `fetch_max_bytes` 
 
@@ -414,6 +513,11 @@ You can set this value to use information such as a hostname, an IP, or anything
 NOTE: In cases when multiple threads are configured and `consumer_threads` is greater than one, a suffix is appended to
 the `group_instance_id` to avoid collisions.
 
+[NOTE]
+=====
+This option is supported only with `consumer_mode => consumer_group`.
+=====
+
 [id="plugins-{type}s-{plugin}-group_protocol"]
 ===== `group_protocol`
 
@@ -428,21 +532,31 @@ Specifies the consumer group rebalance protocol used by the Kafka client.
 When using `consumer`, the following settings **cannot be configured**:
 `partition_assignment_strategy`, `heartbeat_interval_ms`, and `session_timeout_ms`.
 
+[NOTE]
+=====
+This option is supported only with `consumer_mode => consumer_group`.
+=====
 
 [id="plugins-{type}s-{plugin}-heartbeat_interval_ms"]
-===== `heartbeat_interval_ms` 
+===== `heartbeat_interval_ms`
 
   * Value type is <<number,number>>
   * Default value is `3000` milliseconds (3 seconds).
 
-The expected time between heartbeats to the consumer coordinator. Heartbeats are used to ensure 
+The expected time between heartbeats to the consumer coordinator. Heartbeats are used to ensure
 that the consumer's session stays active and to facilitate rebalancing when new
 consumers join or leave the group. The value must be set lower than
 `session.timeout.ms`, but typically should be set no higher than 1/3 of that value.
 It can be adjusted even lower to control the expected time for normal rebalances.
 
+[NOTE]
+=====
+This option is supported only with `consumer_mode => consumer_group`.
+For `share_group` mode, the equivalent is the broker/group-level `share.heartbeat.interval.ms` setting.
+=====
+
 [id="plugins-{type}s-{plugin}-isolation_level"]
-===== `isolation_level` 
+===== `isolation_level`
 
   * Value type is <<string,string>>
   * Default value is `"read_uncommitted"`
@@ -452,6 +566,12 @@ transactional messages which have been committed. If set to `read_uncommitted` (
 return all messages, even transactional messages which have been aborted. Non-transactional messages will be returned
 unconditionally in either mode.
 
+[NOTE]
+=====
+This option is supported only with `consumer_mode => consumer_group`.
+For `share_group` mode, the equivalent is the broker/group-level `share.isolation.level` setting.
+=====
+
 [id="plugins-{type}s-{plugin}-jaas_path"]
 ===== `jaas_path` 
 
@@ -548,6 +668,12 @@ partition ownership amongst consumer instances, supported options are:
 These map to Kafka's corresponding https://kafka.apache.org/{kafka_client_doc}/javadoc/org/apache/kafka/clients/consumer/ConsumerPartitionAssignor.html[`ConsumerPartitionAssignor`]
 implementations.
 
+[NOTE]
+=====
+This option is supported only with `consumer_mode => consumer_group`.
+Share groups do not assign partitions exclusively to consumers, so partition assignment strategies do not apply.
+=====
+
 [id="plugins-{type}s-{plugin}-poll_timeout_ms"]
 ===== `poll_timeout_ms` 
 
@@ -796,6 +922,11 @@ The schemas must follow a naming convention with the pattern <topic name>-value.
 Use either the Schema Registry config option or the
 <<plugins-{type}s-{plugin}-value_deserializer_class>> config option, but not both.
 
+[NOTE]
+=====
+Schema Registry options (including `schema_registry_key`, `schema_registry_secret`, `schema_registry_proxy`, `schema_registry_validation`, and `schema_registry_ssl_*`) are supported only with `consumer_mode => consumer_group`.
+=====
+
 [id="plugins-{type}s-{plugin}-schema_registry_validation"]
 ===== `schema_registry_validation`
 
@@ -825,14 +956,36 @@ Security protocol to use, which can be either of PLAINTEXT,SSL,SASL_PLAINTEXT,SA
 The size of the TCP send buffer (SO_SNDBUF) to use when sending data
 
 [id="plugins-{type}s-{plugin}-session_timeout_ms"]
-===== `session_timeout_ms` 
+===== `session_timeout_ms`
 
   * Value type is <<number,number>>
   * Default value is `10000` milliseconds (10 seconds).
 
 The timeout after which, if the `poll_timeout_ms` is not invoked, the consumer is marked dead
 and a rebalance operation is triggered for the group identified by `group_id`
 
+[NOTE]
+=====
+This option is supported only with `consumer_mode => consumer_group`.
+For `share_group` mode, the equivalent is the broker/group-level `share.session.timeout.ms` setting.
+=====
+
+[id="plugins-{type}s-{plugin}-share_group_acknowledgement_on_error"]
+===== `share_group_acknowledgement_on_error`
+
+* Value can be either of: `release`, `reject`
+* Default value is `release`
+
+Controls how a record is acknowledged when the event is either not accepted by the pipeline's queue or an error occurs before it is offered.
+
+* `release` (default): The record is released back to the share group for redelivery to any available consumer. This strategy favors consistency over availability, which can allow a single malformed event to be retried indefinitely.
+* `reject`: The record is permanently discarded and will not be redelivered. This strategy favors availability over consistency and risks data loss.
+
+[NOTE]
+=====
+This option is supported only with `consumer_mode => share_group`.
+=====
+
 [id="plugins-{type}s-{plugin}-ssl_endpoint_identification_algorithm"]
 ===== `ssl_endpoint_identification_algorithm`