diff --git a/CHANGELOG.md b/CHANGELOG.md index 33f2242..9cd9c6d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,6 @@ +## 12.2.0 + - Added `commit_after_pq_fsync` option to the input: commit Kafka offsets only after the Logstash persistent queue has fsynced the polled batch to disk [#272](https://github.com/logstash-plugins/logstash-integration-kafka/pull/272) + ## 12.1.5 - Upgrades `httpcore5` dependency to v5.4.2 [#270](https://github.com/logstash-plugins/logstash-integration-kafka/pull/270) diff --git a/docs/input-kafka.asciidoc b/docs/input-kafka.asciidoc index c94e52f..87651f3 100644 --- a/docs/input-kafka.asciidoc +++ b/docs/input-kafka.asciidoc @@ -122,6 +122,7 @@ See the https://kafka.apache.org/{kafka_client_doc}/documentation for more detai | <> |<>|No | <> |<>|No | <> |<>|No +| <> |<>|No | <> |<>|No | <> |<>|No | <> |<>|No @@ -277,6 +278,30 @@ The setting corresponds with Kafka's `broker.rack` configuration. NOTE: Available only for Kafka 2.4.0 and higher. See https://cwiki.apache.org/confluence/display/KAFKA/KIP-392%3A+Allow+consumers+to+fetch+from+closest+replica[KIP-392]. +[id="plugins-{type}s-{plugin}-commit_after_pq_fsync"] +===== `commit_after_pq_fsync` + + * Value type is <> + * Default value is `false` + +When set to `true`, Kafka offsets are committed to the broker only after the Logstash +persistent queue has fsynced the polled batch to disk. This guarantees at-least-once +delivery across Logstash crashes: an offset is never committed for events that could +still be lost with the process. + +Requires Logstash to run with a persistent queue (`queue.type: persisted`); pipeline +startup fails with a configuration error otherwise. Requires Logstash X.Y or later. +Forces `enable_auto_commit` to `false`, since the Kafka client's background committer +would bypass the fsync gate. + +Each poll batch incurs one fsync, which bounds consumer throughput by fsync latency; +increase `max_poll_records` to amortize the cost. + +NOTE: The fsync holds the pipeline's shared queue lock for its duration, briefly +blocking all other inputs in the same pipeline. To avoid unintended throughput +contention, run this input in a dedicated pipeline (via `pipelines.yml`) so that +the lock is not shared with unrelated inputs. + [id="plugins-{type}s-{plugin}-connections_max_idle_ms"] ===== `connections_max_idle_ms` diff --git a/lib/logstash/inputs/kafka.rb b/lib/logstash/inputs/kafka.rb index 634ba22..413222b 100644 --- a/lib/logstash/inputs/kafka.rb +++ b/lib/logstash/inputs/kafka.rb @@ -103,10 +103,14 @@ class LogStash::Inputs::Kafka < LogStash::Inputs::Base # 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 config :consumer_threads, :validate => :number, :default => 1 - # If true, periodically commit to Kafka the offsets of messages already returned by the consumer. + # 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 # which the consumption will begin. config :enable_auto_commit, :validate => :boolean, :default => true + # If true, Kafka offsets are committed only after the Logstash persistent queue + # has fsynced the polled batch to disk. Requires `queue.type: persisted` and + # forces `enable_auto_commit` to false. + config :commit_after_pq_fsync, :validate => :boolean, :default => false # 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. config :exclude_internal_topics, :validate => :string @@ -303,6 +307,7 @@ def register check_schema_registry_parameters set_group_protocol! + validate_pq_fsync_config! end METADATA_NONE = Set[].freeze @@ -329,14 +334,17 @@ def extract_metadata_level(decorate_events_setting) public def run(logstash_queue) + check_pq_fsync_support!(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)) end + @thread_errors = java.util.concurrent.CopyOnWriteArrayList.new @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) + raise @thread_errors[0] unless @thread_errors.empty? end # def run public @@ -364,9 +372,15 @@ def thread_runner(logstash_queue, consumer, name) records = do_poll(consumer) unless records.empty? records.each { |record| handle_record(record, codec_instance, logstash_queue) } + checkpoint_persistent_queue!(logstash_queue) if @commit_after_pq_fsync maybe_commit_offset(consumer) end end + rescue => e + # Capture unexpected failures (e.g. PQ checkpoint error) so run can re-raise + # them after all threads finish, allowing the inputworker to restart the input. + # Suppress during orderly shutdown — stop? means we were asked to exit. + @thread_errors << e unless stop? ensure consumer.close end @@ -423,6 +437,18 @@ def maybe_set_metadata(event, record) end end + # Blocks until the PQ has fsynced everything pushed so far. Re-raises on + # failure so the offset commit is skipped and this consumer thread stops: + # the events' durability is unknown, so the offsets must stay uncommitted + # for another consumer to re-poll after rebalance. + def checkpoint_persistent_queue!(logstash_queue) + logstash_queue.checkpoint! + rescue => e + logger.error("PQ checkpoint failed; Kafka offsets will not be committed, consumer stopping", + :error => e.message, :cause => e.respond_to?(:getCause) ? e.getCause : nil) + raise + end + def maybe_commit_offset(consumer) begin consumer.commitSync if @enable_auto_commit.eql?(false) @@ -524,6 +550,44 @@ def create_consumer(client_id, group_instance_id) end end + # commit_after_pq_fsync needs a durable queue and manual offset commits. + # Validated here because register failures abort pipeline startup, while + # exceptions from run are retried forever by the pipeline's inputworker. + def validate_pq_fsync_config! + return unless @commit_after_pq_fsync + + queue_type = pipeline_queue_type + unless queue_type == 'persisted' + raise LogStash::ConfigurationError, + "commit_after_pq_fsync requires Logstash to be configured with a persistent queue " \ + "(queue.type: persisted), detected queue.type: #{queue_type.inspect}" + end + + if @enable_auto_commit + logger.warn("commit_after_pq_fsync is enabled; forcing enable_auto_commit to false") + @enable_auto_commit = false + end + end + + # Defense in depth: on Logstash versions whose write client predates the + # checkpoint! API this turns a mid-stream NoMethodError into a clear error. + # NOTE: inputworker retries run-time failures every second, so this logs + # repeatedly by design — the message must stay self-explanatory. + def check_pq_fsync_support!(logstash_queue) + return unless @commit_after_pq_fsync + return if logstash_queue.respond_to?(:checkpoint!) + + raise LogStash::ConfigurationError, + "commit_after_pq_fsync requires a Logstash version whose queue write client " \ + "supports checkpoint! — upgrade Logstash to X.Y or later" + end + + def pipeline_queue_type + execution_context&.pipeline&.settings&.get('queue.type') + rescue StandardError + nil + end + # In order to use group_protocol => consumer, heartbeat_interval_ms, session_timeout_ms and partition_assignment_strategy need to be unset # If any of these are not using the default value of the plugin, we raise a configuration error def set_group_protocol! diff --git a/spec/unit/inputs/kafka_spec.rb b/spec/unit/inputs/kafka_spec.rb index 0c1da64..e5e0b63 100644 --- a/spec/unit/inputs/kafka_spec.rb +++ b/spec/unit/inputs/kafka_spec.rb @@ -95,6 +95,140 @@ end end + describe 'commit_after_pq_fsync' do + let(:config) { common_config.merge('commit_after_pq_fsync' => true) } + + context 'with a persisted queue' do + before { allow(subject).to receive(:pipeline_queue_type).and_return('persisted') } + + it 'registers successfully' do + expect { subject.register }.not_to raise_error + end + + it 'forces enable_auto_commit to false' do + subject.register + expect(subject.enable_auto_commit).to be false + end + + context 'when enable_auto_commit is explicitly true' do + let(:config) { super().merge('enable_auto_commit' => true) } + + it 'warns and forces it to false' do + expect(subject.logger).to receive(:warn).with(/forcing enable_auto_commit to false/) + subject.register + expect(subject.enable_auto_commit).to be false + end + end + + context 'when enable_auto_commit is explicitly false' do + let(:config) { super().merge('enable_auto_commit' => false) } + + it 'does not warn' do + expect(subject.logger).not_to receive(:warn).with(/enable_auto_commit/) + subject.register + end + end + end + + context 'with a memory queue' do + before { allow(subject).to receive(:pipeline_queue_type).and_return('memory') } + + it 'raises a configuration error' do + expect { subject.register }.to raise_error(LogStash::ConfigurationError, /queue\.type: persisted/) + end + end + + context 'when the queue type cannot be determined' do + before { allow(subject).to receive(:pipeline_queue_type).and_return(nil) } + + it 'raises a configuration error (fail-safe)' do + expect { subject.register }.to raise_error(LogStash::ConfigurationError, /queue\.type: persisted/) + end + end + + context 'at run time' do + before { allow(subject).to receive(:pipeline_queue_type).and_return('persisted') } + + it 'raises when the queue write client does not support checkpoint!' do + subject.register + # a plain Ruby Queue stands in for an old-Logstash write client: no checkpoint! + expect { subject.run(Queue.new) }.to raise_error(LogStash::ConfigurationError, /checkpoint!/) + end + end + + context 'when disabled (default)' do + let(:config) { common_config } + + it 'does not query the pipeline queue type' do + expect(subject).not_to receive(:pipeline_queue_type) + subject.register + end + end + + context 'when running' do + let(:config) { common_config.merge('commit_after_pq_fsync' => true, 'client_id' => 'test') } + let(:q) do + queue = Queue.new + def queue.checkpoint!; end # PQ write-client API stand-in + queue + end + + before do + allow(subject).to receive(:pipeline_queue_type).and_return('persisted') + expect(subject).to receive(:create_consumer).once.and_return(consumer_double) + allow(consumer_double).to receive(:wakeup) + allow(consumer_double).to receive(:close) + allow(consumer_double).to receive(:subscribe) + polled = false + allow(consumer_double).to receive(:poll) do + if polled + [] + else + polled = true + payload + end + end + subject.register + end + + def run_until_stopped + t = Thread.new do + sleep(1) + subject.do_stop + end + subject.run(q) + t.join + end + + it 'checkpoints the queue before committing offsets' do + expect(q).to receive(:checkpoint!).ordered + expect(consumer_double).to receive(:commitSync).ordered + run_until_stopped + end + + it 'processes events into the queue' do + allow(consumer_double).to receive(:commitSync) + run_until_stopped + expect(q.size).to eq(10) + end + + it 'does not commit offsets when checkpoint! raises, and surfaces the error' do + allow(q).to receive(:checkpoint!).and_raise(IOError.new('disk full')) + expect(consumer_double).not_to receive(:commitSync) + expect { run_until_stopped }.to raise_error(IOError, 'disk full') + end + + context 'when the option is disabled' do + let(:config) { common_config.merge('client_id' => 'test') } + + it 'never calls checkpoint!' do + expect(q).not_to receive(:checkpoint!) + run_until_stopped + end + end + end + end + describe '#running' do let(:q) { Queue.new } let(:config) { common_config.merge('client_id' => 'test') } diff --git a/version b/version index f34cd31..6853326 100644 --- a/version +++ b/version @@ -1 +1 @@ -12.1.5 +12.2.0