diff --git a/CHANGELOG.md b/CHANGELOG.md index 43bda18..fdf713a 100644 --- a/CHANGELOG.md +++ b/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) diff --git a/docs/input-kafka.asciidoc b/docs/input-kafka.asciidoc index c94e52f..863f542 100644 --- a/docs/input-kafka.asciidoc +++ b/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 <> 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 <> 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 | <> |<>|No | <> |<>|No | <> |<>|No +| <> |<>, one of `["consumer_group", "share_group"]`|No | <> |<>|No | <> |<>|No | <> |<>|No @@ -175,6 +212,7 @@ See the https://kafka.apache.org/{kafka_client_doc}/documentation for more detai | <> |<>, one of `["PLAINTEXT", "SSL", "SASL_PLAINTEXT", "SASL_SSL"]`|No | <> |<>|No | <> |<>|No +| <> |<>, one of `["release", "reject"]`|No | <> |<>|No | <> |<>|No | <> |a valid filesystem path|No @@ -194,13 +232,18 @@ input plugins.   [id="plugins-{type}s-{plugin}-auto_commit_interval_ms"] -===== `auto_commit_interval_ms` +===== `auto_commit_interval_ms` * Value type is <> * 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 <> * 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 <> * Default value is `true` @@ -339,8 +427,14 @@ 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 <>. +===== + [id="plugins-{type}s-{plugin}-exclude_internal_topics"] -===== `exclude_internal_topics` +===== `exclude_internal_topics` * Value type is <> * There is no default value for this setting. @@ -348,6 +442,11 @@ consumer writes data fetched from the topic to the in-memory or persistent queue 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 <> * 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 <> * 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 -value. Use either the Schema Registry config option or the <> 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,7 +956,7 @@ 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 <> * Default value is `10000` milliseconds (10 seconds). @@ -833,6 +964,28 @@ The size of the TCP send buffer (SO_SNDBUF) to use when sending data 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` diff --git a/lib/logstash/inputs/kafka.rb b/lib/logstash/inputs/kafka.rb index 634ba22..d1ad460 100644 --- a/lib/logstash/inputs/kafka.rb +++ b/lib/logstash/inputs/kafka.rb @@ -45,8 +45,10 @@ # 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 +# 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. # # For more information see http://kafka.apache.org/documentation.html#theconsumer # @@ -100,8 +102,10 @@ class LogStash::Inputs::Kafka < LogStash::Inputs::Base # is to be able to track the source of requests beyond just ip/port by allowing # a logical application name to be included. config :client_id, :validate => :string, :default => "logstash" - # 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 + # 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. config :consumer_threads, :validate => :number, :default => 1 # If true, periodically commit to Kafka the offsets of messages already returned by the consumer. # This committed offset will be used when the process fails as the position from @@ -281,6 +285,33 @@ class LogStash::Inputs::Kafka < LogStash::Inputs::Base # otherwise auto-topic creation is not permitted. config :auto_create_topics, :validate => :boolean, :default => true + # Sets the consumer mode for this input. + # + # * `consumer_group` (default): Standard Kafka consumer group mode. Partitions are assigned + # exclusively to consumers; each partition is consumed by exactly one thread in the group. + # This mode uses `KafkaConsumer` internally. + # + # * `share_group`: Kafka 4.0+ Share Group mode (KIP-932). Provides queue semantics where + # records from any partition can be delivered to any consumer in the share group, and each + # record is delivered to exactly one consumer. Consumers must acknowledge each record. + # This mode uses `KafkaShareConsumer` internally and requires Kafka brokers 4.0 or later + # with `group.share.enable=true` on the broker. + # + # The following options are not supported in `share_group` mode and raise a configuration + # 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`, + # `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`. + # Successfully processed records are always acknowledged as ACCEPT. + # This setting controls what happens when an error occurs during record processing: + # + # * `release` (default): Re-queue the record for redelivery to any consumer in the share group. + # * `reject`: Permanently discard the record (it will not be redelivered). + config :share_group_acknowledgement_on_error, :validate => ["release", "reject"], :default => "release" + config :decorate_events, :validate => %w(none basic extended false true), :default => "none" attr_reader :metadata_mode @@ -294,15 +325,45 @@ def initialize(params = {}) super(params) end + SHARE_GROUP_INCOMPATIBLE_OPTIONS = %w[ + 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 + ].freeze + + CONSUMER_GROUP_ONLY_OPTIONS = %w[ + share_group_acknowledgement_on_error + ].freeze + public def register @runner_threads = [] @metadata_mode = extract_metadata_level(@decorate_events) reassign_dns_lookup @pattern ||= java.util.regex.Pattern.compile(@topics_pattern) unless @topics_pattern.nil? - check_schema_registry_parameters - set_group_protocol! + if @consumer_mode == "share_group" + validate_share_group_config! + ack_type_class = org.apache.kafka.clients.consumer.AcknowledgeType + @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 end METADATA_NONE = Set[].freeze @@ -329,20 +390,35 @@ def extract_metadata_level(decorate_events_setting) public def run(logstash_queue) - @runner_consumers = consumer_threads.times.map do |i| - thread_group_instance_id = consumer_threads > 1 && group_instance_id ? "#{group_instance_id}-#{i}" : group_instance_id - subscribe(create_consumer("#{client_id}-#{i}", thread_group_instance_id)) + if @consumer_mode == "share_group" + @runner_consumers = consumer_threads.times.map do |i| + subscribe(create_share_consumer("#{client_id}-#{i}")) + end + @runner_threads = @runner_consumers.map.with_index do |consumer, i| + share_group_thread_runner(logstash_queue, consumer, "kafka-input-worker-#{client_id}-#{i}") + end + else + @runner_consumers = consumer_threads.times.map do |i| + thread_group_instance_id = consumer_threads > 1 && group_instance_id ? "#{group_instance_id}-#{i}" : group_instance_id + subscribe(create_consumer("#{client_id}-#{i}", thread_group_instance_id)) + end + @runner_threads = @runner_consumers.map.with_index { |consumer, i| thread_runner(logstash_queue, consumer, + "kafka-input-worker-#{client_id}-#{i}") } end - @runner_threads = @runner_consumers.map.with_index { |consumer, i| thread_runner(logstash_queue, consumer, - "kafka-input-worker-#{client_id}-#{i}") } @runner_threads.each(&:start) @runner_threads.each(&:join) end # def run public def stop - # if we have consumers, wake them up to unblock our runner threads - @runner_consumers && @runner_consumers.each(&:wakeup) + # wake up consumers to unblock the poll loop in runner threads + @runner_consumers && @runner_consumers.each do |consumer| + begin + consumer.wakeup + rescue => e + logger.debug("Error waking up consumer on stop", :error => e.message) + end + end end public @@ -440,46 +516,63 @@ def maybe_commit_offset(consumer) end private + def build_common_props(client_id) + props = java.util.Properties.new + kafka = org.apache.kafka.clients.consumer.ConsumerConfig + + props.put(kafka::BOOTSTRAP_SERVERS_CONFIG, bootstrap_servers) + props.put(kafka::CLIENT_DNS_LOOKUP_CONFIG, client_dns_lookup) + props.put(kafka::CLIENT_ID_CONFIG, client_id) + props.put(kafka::CONNECTIONS_MAX_IDLE_MS_CONFIG, connections_max_idle_ms.to_s) unless connections_max_idle_ms.nil? + props.put(kafka::FETCH_MAX_BYTES_CONFIG, fetch_max_bytes.to_s) unless fetch_max_bytes.nil? + props.put(kafka::FETCH_MAX_WAIT_MS_CONFIG, fetch_max_wait_ms.to_s) unless fetch_max_wait_ms.nil? + props.put(kafka::FETCH_MIN_BYTES_CONFIG, fetch_min_bytes.to_s) unless fetch_min_bytes.nil? + props.put(kafka::GROUP_ID_CONFIG, group_id) + props.put(kafka::KEY_DESERIALIZER_CLASS_CONFIG, key_deserializer_class) + props.put(kafka::MAX_PARTITION_FETCH_BYTES_CONFIG, max_partition_fetch_bytes.to_s) unless max_partition_fetch_bytes.nil? + props.put(kafka::MAX_POLL_RECORDS_CONFIG, max_poll_records.to_s) unless max_poll_records.nil? + props.put(kafka::MAX_POLL_INTERVAL_MS_CONFIG, max_poll_interval_ms.to_s) unless max_poll_interval_ms.nil? + props.put(kafka::METADATA_MAX_AGE_CONFIG, metadata_max_age_ms.to_s) unless metadata_max_age_ms.nil? + props.put(kafka::RECEIVE_BUFFER_CONFIG, receive_buffer_bytes.to_s) unless receive_buffer_bytes.nil? + props.put(kafka::RECONNECT_BACKOFF_MS_CONFIG, reconnect_backoff_ms.to_s) unless reconnect_backoff_ms.nil? + props.put(kafka::RECONNECT_BACKOFF_MAX_MS_CONFIG, reconnect_backoff_max_ms.to_s) unless reconnect_backoff_max_ms.nil? + props.put(kafka::REQUEST_TIMEOUT_MS_CONFIG, request_timeout_ms.to_s) unless request_timeout_ms.nil? + 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? + if security_protocol == "SSL" + set_trustore_keystore_config(props) + elsif security_protocol == "SASL_PLAINTEXT" + set_sasl_config(props) + elsif security_protocol == "SASL_SSL" + set_trustore_keystore_config(props) + set_sasl_config(props) + end + + props + end + def create_consumer(client_id, group_instance_id) begin - props = java.util.Properties.new + props = build_common_props(client_id) kafka = org.apache.kafka.clients.consumer.ConsumerConfig 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::BOOTSTRAP_SERVERS_CONFIG, bootstrap_servers) - props.put(kafka::CHECK_CRCS_CONFIG, check_crcs.to_s) unless check_crcs.nil? - props.put(kafka::CLIENT_DNS_LOOKUP_CONFIG, client_dns_lookup) - props.put(kafka::CLIENT_ID_CONFIG, client_id) - props.put(kafka::CONNECTIONS_MAX_IDLE_MS_CONFIG, connections_max_idle_ms.to_s) unless connections_max_idle_ms.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::FETCH_MAX_BYTES_CONFIG, fetch_max_bytes.to_s) unless fetch_max_bytes.nil? - props.put(kafka::FETCH_MAX_WAIT_MS_CONFIG, fetch_max_wait_ms.to_s) unless fetch_max_wait_ms.nil? - props.put(kafka::FETCH_MIN_BYTES_CONFIG, fetch_min_bytes.to_s) unless fetch_min_bytes.nil? - props.put(kafka::GROUP_ID_CONFIG, group_id) props.put(kafka::GROUP_INSTANCE_ID_CONFIG, group_instance_id) unless group_instance_id.nil? props.put(kafka::GROUP_PROTOCOL_CONFIG, group_protocol) props.put(kafka::HEARTBEAT_INTERVAL_MS_CONFIG, heartbeat_interval_ms.to_s) unless heartbeat_interval_ms.nil? props.put(kafka::ISOLATION_LEVEL_CONFIG, isolation_level) - props.put(kafka::KEY_DESERIALIZER_CLASS_CONFIG, key_deserializer_class) - props.put(kafka::MAX_PARTITION_FETCH_BYTES_CONFIG, max_partition_fetch_bytes.to_s) unless max_partition_fetch_bytes.nil? - props.put(kafka::MAX_POLL_RECORDS_CONFIG, max_poll_records.to_s) unless max_poll_records.nil? - props.put(kafka::MAX_POLL_INTERVAL_MS_CONFIG, max_poll_interval_ms.to_s) unless max_poll_interval_ms.nil? - props.put(kafka::METADATA_MAX_AGE_CONFIG, metadata_max_age_ms.to_s) unless metadata_max_age_ms.nil? props.put(kafka::PARTITION_ASSIGNMENT_STRATEGY_CONFIG, partition_assignment_strategy_class) unless partition_assignment_strategy.nil? - props.put(kafka::RECEIVE_BUFFER_CONFIG, receive_buffer_bytes.to_s) unless receive_buffer_bytes.nil? - props.put(kafka::RECONNECT_BACKOFF_MS_CONFIG, reconnect_backoff_ms.to_s) unless reconnect_backoff_ms.nil? - props.put(kafka::RECONNECT_BACKOFF_MAX_MS_CONFIG, reconnect_backoff_max_ms.to_s) unless reconnect_backoff_max_ms.nil? - props.put(kafka::REQUEST_TIMEOUT_MS_CONFIG, request_timeout_ms.to_s) unless request_timeout_ms.nil? - 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::SESSION_TIMEOUT_MS_CONFIG, session_timeout_ms.to_s) unless session_timeout_ms.nil? - props.put(kafka::VALUE_DESERIALIZER_CLASS_CONFIG, value_deserializer_class) - props.put(kafka::CLIENT_RACK_CONFIG, client_rack) unless client_rack.nil? - props.put("security.protocol", security_protocol) unless security_protocol.nil? if schema_registry_url props.put(kafka::VALUE_DESERIALIZER_CLASS_CONFIG, Java::io.confluent.kafka.serializers.KafkaAvroDeserializer.java_class) serdes_config = Java::io.confluent.kafka.serializers.AbstractKafkaSchemaSerDeConfig @@ -495,14 +588,6 @@ def create_consumer(client_id, group_instance_id) props.put(serdes_config::BASIC_AUTH_CREDENTIALS_SOURCE, 'URL') end end - if security_protocol == "SSL" - set_trustore_keystore_config(props) - elsif security_protocol == "SASL_PLAINTEXT" - set_sasl_config(props) - elsif security_protocol == "SASL_SSL" - set_trustore_keystore_config(props) - set_sasl_config(props) - end if schema_registry_ssl_truststore_location props.put('schema.registry.ssl.truststore.location', schema_registry_ssl_truststore_location) props.put('schema.registry.ssl.truststore.password', schema_registry_ssl_truststore_password.value) @@ -565,4 +650,85 @@ def header_with_value(header) !header.nil? && !header.value.nil? && !header.key.nil? end + def validate_share_group_config! + incompatible = original_params.keys & SHARE_GROUP_INCOMPATIBLE_OPTIONS + unless incompatible.empty? + raise LogStash::ConfigurationError, + "The following options are not compatible with share_group consumer_mode and must be removed: #{incompatible.join(', ')}" + 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) + + # Share consumer defaults to implicit acknowledgement, which disallows explicit + # acknowledge() calls. The plugin acknowledges each record after processing, so + # explicit mode is required. + props.put("share.acknowledgement.mode", "explicit") + + org.apache.kafka.clients.consumer.KafkaShareConsumer.new(props) + rescue => e + logger.error("Unable to create Kafka share consumer from given configuration", + :kafka_error_message => e, + :cause => e.respond_to?(:getCause) ? e.getCause() : nil) + raise e + end + end + + def share_group_thread_runner(logstash_queue, consumer, name) + java.lang.Thread.new do + LogStash::Util::set_thread_name(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| + 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) + end + end + ensure + consumer.close + end + end + end + + def commit_share_acknowledgements(consumer) + begin + consumer.commitSync + rescue org.apache.kafka.common.errors.WakeupException => e + logger.debug("Wake up from share consumer commitSync", :kafka_error_message => e) + raise e unless stop? + rescue StandardError => e + logger.error("Unable to commit share group acknowledgements", + :kafka_error_message => e, + :cause => e.respond_to?(:getCause) ? e.getCause() : nil) + end + end + end #class LogStash::Inputs::Kafka diff --git a/spec/integration/inputs/kafka_spec.rb b/spec/integration/inputs/kafka_spec.rb index f9a28f7..dcb0e08 100644 --- a/spec/integration/inputs/kafka_spec.rb +++ b/spec/integration/inputs/kafka_spec.rb @@ -130,6 +130,64 @@ def send_message(record) end end + context "#share-group" do + let(:share_group_msg_count) { 20 } + let(:share_group_config) do + { + 'topics' => ['logstash_share_group_topic'], + 'group_id' => "sg-integration-#{rand(36**6).to_s(36)}", + 'consumer_mode' => 'share_group', + 'consumer_threads' => 2, + 'bootstrap_servers' => 'localhost:9092', + } + end + + it "raises ConfigurationError when auto_offset_reset is set with share_group" do + bad_config = share_group_config.merge('auto_offset_reset' => 'earliest') + kafka_input = LogStash::Inputs::Kafka.new(bad_config) + expect { kafka_input.register }.to raise_error(LogStash::ConfigurationError, /auto_offset_reset/) + end + + it "raises ConfigurationError when share_group_acknowledgement_on_error is set with consumer_group" do + bad_config = share_group_config.merge('consumer_mode' => 'consumer_group', 'share_group_acknowledgement_on_error' => 'reject') + kafka_input = LogStash::Inputs::Kafka.new(bad_config) + expect { kafka_input.register }.to raise_error(LogStash::ConfigurationError, /share_group_acknowledgement_on_error/) + end + + it "consumes messages produced while the consumer is running" do + # Share groups default to share.auto.offset.reset=latest, so messages must + # be produced AFTER the consumer joins the group to guarantee delivery. + queue = Queue.new + kafka_input = LogStash::Inputs::Kafka.new(share_group_config) + kafka_input.register + t = Thread.new { kafka_input.run(queue) } + begin + t.run + sleep 8 # share group partition assignment takes ~5s; wait for STABLE epoch + + share_group_msg_count.times do |i| + record = org.apache.kafka.clients.producer.ProducerRecord.new( + 'logstash_share_group_topic', "share-msg-#{i}") + send_message(record) + end + + wait(timeout_seconds).for { queue.length }.to eq(share_group_msg_count) + expect(queue.length).to eq(share_group_msg_count) + ensure + kafka_input.do_stop + t.kill + t.join(30) + end + end + + it "check_crcs is accepted in share_group mode without error" do + config = share_group_config.merge('check_crcs' => true) + kafka_input = LogStash::Inputs::Kafka.new(config) + expect { kafka_input.register }.not_to raise_error + kafka_input.do_stop rescue nil + end + end + context "#kafka-decorate" do it "should show the right topic and group name in decorated kafka section" do start = LogStash::Timestamp.now.time.to_i diff --git a/spec/unit/inputs/kafka_spec.rb b/spec/unit/inputs/kafka_spec.rb index 0c1da64..9803d95 100644 --- a/spec/unit/inputs/kafka_spec.rb +++ b/spec/unit/inputs/kafka_spec.rb @@ -84,6 +84,11 @@ expect { subject.register }.to_not raise_error end + it "raises ConfigurationError when share_group_acknowledgement_on_error is set in consumer_group mode" do + config['share_group_acknowledgement_on_error'] = 'reject' + expect { subject.register }.to raise_error(LogStash::ConfigurationError, /share_group_acknowledgement_on_error/) + end + context "when the deprecated `default` is specified" do let(:config) { common_config.merge('client_dns_lookup' => 'default') } @@ -495,4 +500,178 @@ expect( subject.enable_auto_commit ).to be true end end + + context 'share group mode' do + let(:share_consumer_double) { double('kafka-share-consumer') } + let(:config) { common_config.merge('consumer_mode' => 'share_group') } + + before do + allow(share_consumer_double).to receive(:subscribe) + allow(share_consumer_double).to receive(:wakeup) + allow(share_consumer_double).to receive(:close) + end + + describe '#register' do + it "registers without error" do + expect { subject.register }.not_to raise_error + end + + it "raises ConfigurationError when group_instance_id is set" do + config['group_instance_id'] = 'my-instance' + expect { subject.register }.to raise_error(LogStash::ConfigurationError, /group_instance_id/) + end + + it "raises ConfigurationError when partition_assignment_strategy is set" do + config['partition_assignment_strategy'] = 'sticky' + expect { subject.register }.to raise_error(LogStash::ConfigurationError, /partition_assignment_strategy/) + end + + it "raises ConfigurationError when schema_registry_url is set" do + config['schema_registry_url'] = 'http://localhost:8081' + expect { subject.register }.to raise_error(LogStash::ConfigurationError, /schema_registry_url/) + end + + %w[ + enable_auto_commit + auto_commit_interval_ms + heartbeat_interval_ms + session_timeout_ms + isolation_level + auto_offset_reset + exclude_internal_topics + group_protocol + ].each do |opt| + it "raises ConfigurationError when #{opt} is set in the pipeline config" do + config[opt] = case opt + when 'enable_auto_commit', 'exclude_internal_topics' then 'true' + when 'auto_commit_interval_ms', 'heartbeat_interval_ms', 'session_timeout_ms' then '5000' + when 'isolation_level' then 'read_committed' + when 'auto_offset_reset' then 'earliest' + when 'group_protocol' then 'consumer' + end + expect { subject.register }.to raise_error(LogStash::ConfigurationError, /#{opt}/) + end + end + + it "accepts check_crcs in share_group mode (supported by KafkaShareConsumer)" do + config['check_crcs'] = true + expect { subject.register }.not_to raise_error + end + + it "accepts auto_create_topics in share_group mode (supported by KafkaShareConsumer)" do + config['auto_create_topics'] = true + expect { subject.register }.not_to raise_error + end + end + + describe '#create_share_consumer' do + it "creates a KafkaShareConsumer with correct bootstrap_servers and group_id" do + expect(org.apache.kafka.clients.consumer.KafkaShareConsumer). + to receive(:new).with(hash_including( + 'bootstrap.servers' => 'localhost:9092', + 'group.id' => 'logstash' + )).and_return(share_consumer_double) + + expect(subject.send(:create_share_consumer, 'test-share-client-0')).to be share_consumer_double + end + + it "does not set enable.auto.commit, isolation.level or auto.offset.reset" do + expect(org.apache.kafka.clients.consumer.KafkaShareConsumer). + to receive(:new) do |props| + expect(props).not_to have_key('enable.auto.commit') + expect(props).not_to have_key('isolation.level') + expect(props).not_to have_key('auto.offset.reset') + share_consumer_double + end + + subject.send(:create_share_consumer, 'test-share-client-0') + end + + it "sets share.acknowledgement.mode to explicit" do + expect(org.apache.kafka.clients.consumer.KafkaShareConsumer). + to receive(:new).with(hash_including('share.acknowledgement.mode' => 'explicit')). + and_return(share_consumer_double) + + subject.send(:create_share_consumer, 'test-share-client-0') + end + + context 'with client_rack' do + let(:config) { super().merge('client_rack' => 'US-EAST-1') } + + it "passes client.rack to the share consumer" do + expect(org.apache.kafka.clients.consumer.KafkaShareConsumer). + to receive(:new).with(hash_including('client.rack' => 'US-EAST-1')). + and_return(share_consumer_double) + + subject.send(:create_share_consumer, 'test-share-client-0') + end + end + end + + describe '#running in share_group mode' do + let(:q) { Queue.new } + let(:payload) do + 5.times.map { org.apache.kafka.clients.consumer.ConsumerRecord.new("logstash", 0, 0, "key", "value") } + end + + before do + subject.register + expect(subject).to receive(:create_share_consumer).once.and_return(share_consumer_double) + polled = false + allow(share_consumer_double).to receive(:poll) do + if polled + [] + else + polled = true + payload + end + end + allow(share_consumer_double).to receive(:acknowledge) + allow(share_consumer_double).to receive(:commitSync) + end + + it "processes events and acknowledges each record" do + expect(share_consumer_double).to receive(:acknowledge).exactly(5).times + expect(share_consumer_double).to receive(:commitSync).at_least(:once) + + t = Thread.new { sleep(1); subject.do_stop } + subject.run(q) + t.join + + expect(q.size).to eq(5) + end + + it "acknowledges with ACCEPT on successful record processing" do + ack_accept = org.apache.kafka.clients.consumer.AcknowledgeType::ACCEPT + expect(share_consumer_double).to receive(:acknowledge).with(anything, ack_accept).exactly(5).times + allow(share_consumer_double).to receive(:commitSync) + + t = Thread.new { sleep(1); subject.do_stop } + subject.run(q) + t.join + end + end + + describe 'acknowledgement_on_error' do + before { subject.register } + + context 'with release strategy (default)' do + it "uses RELEASE ack type on error" do + expect(subject.instance_variable_get(:@share_group_ack_error_type)).to eq( + org.apache.kafka.clients.consumer.AcknowledgeType::RELEASE + ) + end + end + + context 'with reject strategy' do + let(:config) { super().merge('share_group_acknowledgement_on_error' => 'reject') } + + it "uses REJECT ack type on error" do + expect(subject.instance_variable_get(:@share_group_ack_error_type)).to eq( + org.apache.kafka.clients.consumer.AcknowledgeType::REJECT + ) + end + end + end + end end