address PR review comments

- Move check_crcs and auto_create_topics to build_common_props (both supported by KafkaShareConsumer)
- Add CONSUMER_GROUP_ONLY_OPTIONS + validate_consumer_group_config! to raise on share_group_acknowledgement_on_error with consumer_group mode
- Release unprocessed records on shutdown in share_group_thread_runner instead of blocking on queue
- Replace ternary with if/else for @share_group_ack_error_type assignment
- Update consumer_threads docs with clearer partition/thread guidance per mode
- Update share group acknowledgement docs to clarify pipeline queue visibility boundary
- Update share_group_acknowledgement_on_error docs to frame around pipeline semantics
- Add note about broker/group-level equivalents for incompatible share_group options

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: orangecola

docs/input-kafka.asciidoc

@@ -58,10 +58,11 @@ you could run multiple Logstash instances with the same `group_id` to spread the
 physical machines. Messages in a topic will be distributed to all Logstash instances with
 the same `group_id`.
 
-In `consumer_group` mode, you should ideally have as many threads as the number of partitions
-for a perfect balance -- more threads than partitions means that some threads will be idle.
-This does not apply in `share_group` mode, where any thread can fetch from any partition, so
-the number of threads is not bounded by the partition count.
+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)
@@ -76,10 +77,9 @@ 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. Successful records are always acknowledged
-as `ACCEPT`. The <<plugins-{type}s-{plugin}-share_group_acknowledgement_on_error>> option
-controls acknowledgement for records that fail processing (`release` for redelivery or `reject`
-to permanently discard).
+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:
 
@@ -350,7 +350,13 @@ Setting any of the following options in the pipeline config raises a configurati
 `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`, `check_crcs`, `exclude_internal_topics`, `schema_registry_url`.
+`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"]
@@ -910,16 +916,10 @@ and a rebalance operation is triggered for the group identified by `group_id`
 * Default value is `release`
 * Only applicable when <<plugins-{type}s-{plugin}-consumer_mode>> is `share_group`
 
-Controls how a record is acknowledged when an error occurs while processing it.
-
-`release` (default): The record is released back to the share group for redelivery to any
-available consumer. Use this when processing errors are transient and the record should
-be retried.
-
-`reject`: The record is permanently discarded and will not be redelivered. Use this when
-the record is malformed and retrying would never succeed.
+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.
 
-Successfully processed records are always acknowledged as ACCEPT regardless of this setting.
+* `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.
 
 [id="plugins-{type}s-{plugin}-ssl_endpoint_identification_algorithm"]
 ===== `ssl_endpoint_identification_algorithm`

lib/logstash/inputs/kafka.rb

@@ -301,7 +301,7 @@ class LogStash::Inputs::Kafka < LogStash::Inputs::Base
   # error if set in the pipeline config: `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`,
-  # `check_crcs`, `exclude_internal_topics`, `schema_registry_url`.
+  # `exclude_internal_topics`, `schema_registry_url`.
   config :consumer_mode, :validate => ["consumer_group", "share_group"], :default => "consumer_group"
 
   # Controls how records are acknowledged when `consumer_mode` is set to `share_group`.
@@ -335,11 +335,14 @@ def initialize(params = {})
     enable_auto_commit
     auto_commit_interval_ms
     auto_offset_reset
-    check_crcs
     exclude_internal_topics
     schema_registry_url
   ].freeze
 
+  CONSUMER_GROUP_ONLY_OPTIONS = %w[
+    share_group_acknowledgement_on_error
+  ].freeze
+
   public
   def register
     @runner_threads = []
@@ -350,9 +353,14 @@ def register
     if @consumer_mode == "share_group"
       validate_share_group_config!
       ack_type_class = org.apache.kafka.clients.consumer.AcknowledgeType
-      @share_group_ack_error_type = @share_group_acknowledgement_on_error == "reject" ?
-        ack_type_class::REJECT : ack_type_class::RELEASE
+      @share_group_ack_error_type =
+        if @share_group_acknowledgement_on_error == "reject"
+          ack_type_class::REJECT
+        else
+          ack_type_class::RELEASE
+        end
     else
+      validate_consumer_group_config!
       check_schema_registry_parameters
       set_group_protocol!
     end
@@ -532,6 +540,8 @@ def build_common_props(client_id)
     props.put(kafka::RETRY_BACKOFF_MS_CONFIG, retry_backoff_ms.to_s) unless retry_backoff_ms.nil?
     props.put(kafka::SEND_BUFFER_CONFIG, send_buffer_bytes.to_s) unless send_buffer_bytes.nil?
     props.put(kafka::VALUE_DESERIALIZER_CLASS_CONFIG, value_deserializer_class)
+    props.put(kafka::ALLOW_AUTO_CREATE_TOPICS_CONFIG, auto_create_topics) unless auto_create_topics.nil?
+    props.put(kafka::CHECK_CRCS_CONFIG, check_crcs.to_s) unless check_crcs.nil?
     props.put(kafka::CLIENT_RACK_CONFIG, client_rack) unless client_rack.nil?
 
     props.put("security.protocol", security_protocol) unless security_protocol.nil?
@@ -554,8 +564,6 @@ def create_consumer(client_id, group_instance_id)
 
       props.put(kafka::AUTO_COMMIT_INTERVAL_MS_CONFIG, auto_commit_interval_ms.to_s) unless auto_commit_interval_ms.nil?
       props.put(kafka::AUTO_OFFSET_RESET_CONFIG, auto_offset_reset) unless auto_offset_reset.nil?
-      props.put(kafka::ALLOW_AUTO_CREATE_TOPICS_CONFIG, auto_create_topics) unless auto_create_topics.nil?
-      props.put(kafka::CHECK_CRCS_CONFIG, check_crcs.to_s) unless check_crcs.nil?
       props.put(kafka::ENABLE_AUTO_COMMIT_CONFIG, enable_auto_commit.to_s)
       props.put(kafka::EXCLUDE_INTERNAL_TOPICS_CONFIG, exclude_internal_topics) unless exclude_internal_topics.nil?
       props.put(kafka::GROUP_INSTANCE_ID_CONFIG, group_instance_id) unless group_instance_id.nil?
@@ -650,6 +658,14 @@ def validate_share_group_config!
     end
   end
 
+  def validate_consumer_group_config!
+    incompatible = original_params.keys & CONSUMER_GROUP_ONLY_OPTIONS
+    unless incompatible.empty?
+      raise LogStash::ConfigurationError,
+        "The following options require consumer_mode => 'share_group' and cannot be used with consumer_group mode: #{incompatible.join(', ')}"
+    end
+  end
+
   def create_share_consumer(client_id)
     begin
       props = build_common_props(client_id)
@@ -674,18 +690,23 @@ def share_group_thread_runner(logstash_queue, consumer, name)
       begin
         codec_instance = @codec.clone
         ack_accept = org.apache.kafka.clients.consumer.AcknowledgeType::ACCEPT
+        ack_release = org.apache.kafka.clients.consumer.AcknowledgeType::RELEASE
         until stop?
           records = do_poll(consumer)
           unless records.empty?
             records.each do |record|
-              begin
-                handle_record(record, codec_instance, logstash_queue)
-                consumer.acknowledge(record, ack_accept)
-              rescue => e
-                logger.error("Error processing record in share group mode, acknowledging with #{@share_group_acknowledgement_on_error}",
-                             :kafka_error_message => e,
-                             :cause => e.respond_to?(:getCause) ? e.getCause() : nil)
-                consumer.acknowledge(record, @share_group_ack_error_type)
+              if stop?
+                consumer.acknowledge(record, ack_release)
+              else
+                begin
+                  handle_record(record, codec_instance, logstash_queue)
+                  consumer.acknowledge(record, ack_accept)
+                rescue => e
+                  logger.error("Error processing record in share group mode, acknowledging with #{@share_group_acknowledgement_on_error}",
+                               :kafka_error_message => e,
+                               :cause => e.respond_to?(:getCause) ? e.getCause() : nil)
+                  consumer.acknowledge(record, @share_group_ack_error_type)
+                end
               end
             end
             commit_share_acknowledgements(consumer)