Add commit_after_pq_fsync: commit Kafka offsets only after the persistent queue fsyncs

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)
 

docs/input-kafka.asciidoc

@@ -122,6 +122,7 @@ See the https://kafka.apache.org/{kafka_client_doc}/documentation for more detai
 | <<plugins-{type}s-{plugin}-client_dns_lookup>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-client_id>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-client_rack>> |<<string,string>>|No
+| <<plugins-{type}s-{plugin}-commit_after_pq_fsync>> |<<boolean,boolean>>|No
 | <<plugins-{type}s-{plugin}-connections_max_idle_ms>> |<<number,number>>|No
 | <<plugins-{type}s-{plugin}-consumer_threads>> |<<number,number>>|No
 | <<plugins-{type}s-{plugin}-decorate_events>> |<<string,string>>|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 <<boolean,boolean>>
+  * 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` 
 

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!

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') }

version

@@ -1 +1 @@
-12.1.5
+12.2.0