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

[LolloneS]

What does this PR do?

Adds an opt-in commit_after_pq_fsync boolean option (default false) to the Kafka input. When enabled, each poll batch is fsynced to Logstash's persistent queue before the corresponding offsets are committed to the broker:

• New config option commit_after_pq_fsync, documented in docs/input-kafka.asciidoc.
• Register-time validation: requires queue.type: persisted (read via the execution context's pipeline settings) and raises LogStash::ConfigurationError otherwise — validation happens in register because register failures abort pipeline startup, while exceptions from run are retried forever by the pipeline's inputworker. Also forces enable_auto_commit to false (with a warning if it was explicitly true), since the Kafka client's background committer would bypass the fsync gate.
• Run-time compatibility guard: raises a clear ConfigurationError when the queue write client predates the checkpoint! API (older Logstash), instead of a mid-stream NoMethodError.
• The gate itself: in thread_runner, after pushing a batch and before maybe_commit_offset, the plugin calls logstash_queue.checkpoint!. On checkpoint failure the error is logged and re-raised, so offsets are never committed for a batch whose durability is unknown; the consumer closes and its partitions rebalance, and the uncommitted offsets are re-polled.

Depends on the core API added in elastic/logstash#19362 (persistent? / checkpoint! on the queue write client). Version placeholders (X.Y in docs and the guard's error message, #PR in the changelog) will be filled once that PR lands in a release.

Why is it important/What is the impact to the user?

With a persistent queue, offsets are currently committed as soon as events reach the PQ's in-memory page buffer — before any fsync — in both commit modes (manual commitSync and the client's background auto-commit). A crash between the offset commit and the next PQ checkpoint loses events the broker already considers consumed, breaking at-least-once delivery. This option closes that window at poll-batch granularity, at the cost of one fsync per batch (increase max_poll_records to amortize).

Note on failure behavior: if checkpoint! fails (e.g. disk full) with consumer_threads => 1, the input stops consuming while the pipeline stays up, logging PQ checkpoint failed; Kafka offsets will not be committed, consumer stopping — operators may want to alert on that line. This mirrors the plugin's existing fatal-error paths (e.g. FencedInstanceIdException).

Checklist

• I have added tests that prove my fix is effective or that my feature works (7 new unit examples: validation matrix, forced auto-commit, ordered checkpoint!→commitSync, checkpoint-failure blocks commit, disabled-flag pin; full suite 91 examples, 0 failures)
• I have made corresponding changes to the documentation (docs/input-kafka.asciidoc, CHANGELOG, version bump to 12.2.0)

🤖 Generated with Claude Code