Expose persistent? and checkpoint! on queue write clients for durable input acknowledgement by LolloneS · Pull Request #19362 · elastic/logstash

LolloneS · 2026-07-31T14:04:23Z

Release notes

Added persistent? and checkpoint! methods to the queue write client used by input plugins, allowing inputs to fsync the persistent queue before acknowledging upstream systems.

What does this PR do?

Adds a small durability API to the queue write-client hierarchy so input plugins can guarantee events are persisted to disk before acknowledging them upstream (e.g. committing Kafka offsets):

Queue#ensurePersisted() — new no-arg method that takes the queue lock and fsyncs the head page up to the last written sequence number (delegates to the existing Page#ensurePersistedUpto; a no-op when the last head checkpoint already covers all writes). Tail pages need no handling: page rotation already checkpoints and fsyncs.
JRubyAckedQueueExt#ensure_persisted — Ruby-visible bridge, wrapping IOException in a Ruby IOError (mirrors rubyWrite).
JRubyAbstractQueueWriteClientExt — exposes persistent? and checkpoint! via @JRubyMethod, delegating to new abstract hooks isPersistent() / doCheckpoint(ThreadContext) (same pattern as rubyPush→doPush).
JrubyAckedWriteClientExt — persistent? → true; checkpoint! → fsync via ensure_persisted.
JrubyMemoryWriteClientExt — persistent? → false; checkpoint! raises NotImplementedError (raising rather than silently no-oping means a caller that skipped its own PQ guard fails loudly instead of committing offsets without durability).
JRubyWrappedWriteClientExt (what input plugins actually receive in run) — delegates both methods to the wrapped client, with no metrics wrapping (a checkpoint is not an event push).

The change is purely additive: 0 deletions, no existing method's behavior changes, and nothing in core calls the new API — it is only reachable by plugins that opt in.

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

Today an input plugin cannot know when the persistent queue has fsynced the events it pushed: queue << event returns once the event lands in the head-page buffer. For sources with commit/ack semantics (Kafka being the driving case), this means offsets can be committed for events that a crash would lose — silently breaking the at-least-once guarantee users expect from queue.type: persisted.

With this API, the Kafka input (companion PR: logstash-plugins/logstash-integration-kafka — adds an opt-in commit_after_pq_fsync option) can call checkpoint! after pushing a poll batch and only then commit offsets, closing the window. Other acknowledging inputs can adopt the same pattern.

Checklist

My code follows the style guidelines of this project
I have commented my code, particularly in hard-to-understand areas
I have made corresponding changes to the documentation
I have made corresponding change to the default configuration files (and/or docker env variables)
I have added tests that prove my fix is effective or that my feature works

Author's Checklist

New Java tests: QueueTest (fsync-up-to-last-write + no-op/idempotency), JrubyAckedWriteClientExtTest (end-to-end fsync via checkpoint.head), JrubyMemoryWriteClientExtTest, JRubyWrappedWriteClientExtTest (delegation)

How to test this PR locally

./gradlew :logstash-core:javaTests --tests "org.logstash.ackedqueue.QueueTest" --tests "org.logstash.ext.*WriteClient*"

Full :logstash-core:javaTests passes.

🤖 Generated with Claude Code

github-actions · 2026-07-31T14:04:48Z

🤖 GitHub comments

Just comment with:

run docs-build : Re-trigger the docs validation. (use unformatted text in the comment!)
run exhaustive tests : Run the exhaustive tests Buildkite pipeline.

mergify · 2026-07-31T14:36:15Z

This pull request does not have a backport label. Could you fix it @LolloneS? 🙏
To fixup this pull request, you need to add the backport labels for the needed
branches, such as:

backport-8./d is the label to automatically backport to the 8./d branch. /d is the digit.
If no backport is necessary, please add the backport-skip label

…st closed queue Page.ensurePersistedUpto used a strict > comparison against an exclusive upper bound, causing checkpoint! to no-op when exactly one event was written since the last checkpoint — silently breaking the at-least-once guarantee for single-record Kafka poll batches. Change the comparison to >=. Queue.ensurePersisted dereferenced headPage with no closed-queue check; close() nulls headPage under the same lock after its own final fsync, so a Kafka consumer thread racing shutdown would NPE. Return early when isClosed() — close() has already fsynced everything by that point so no data is lost. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

github-actions · 2026-08-06T13:41:27Z

TL;DR

The :lab_coat: x-pack integration job failed in the Monitoring TLS hot-reload integration flow while waiting for the log line rebuilt elasticsearch client ... on certificate change. The PR changes are in acked queue write/checkpoint code paths, not X-Pack monitoring, so this currently looks more like a flaky/infrastructure-style CI failure than a confirmed regression from this diff.

Remediation

Re-run the failed :lab_coat: x-pack integration step first.
If it reproduces, capture a full (non-truncated) log and inspect x-pack/qa/integration/monitoring/monitoring_tls_hot_reload_spec.rb:63-67 (wait_for_log_line) and :95 (expected rebuild log), plus x-pack/lib/helpers/elasticsearch_client_holder.rb:106 (debug log emission: rebuilt elasticsearch client ... on certificate change).
I checked for an existing flaky-test issue with this signature and found no match; if it repeats, file one and link it here.

Investigation details

Root Cause

The run terminates while polling for a specific debug log pattern in the test Monitoring TLS hot-reload (the config.reload.automatic: true context), indicating the certificate-change rebuild signal was not observed in time. The visible log content does not include a Java/Ruby exception or compile/runtime crash tied to the PR's modified files.

Relevant code paths:

x-pack/qa/integration/monitoring/monitoring_tls_hot_reload_spec.rb:95
- wait_for_log_line(/rebuilt elasticsearch client.*on certificate change/)
x-pack/qa/integration/monitoring/monitoring_tls_hot_reload_spec.rb:63-67
- raises Log pattern not yet visible while polling
x-pack/lib/helpers/elasticsearch_client_holder.rb:106
- logger.debug("rebuilt elasticsearch client ... on certificate change")

PR diff context inspected:

Changes are under logstash-core/src/main/java/org/logstash/ackedqueue/* and logstash-core/src/main/java/org/logstash/ext/* (queue persistence/checkpoint APIs/tests), with no modifications in x-pack/qa/integration/monitoring/* or x-pack/lib/helpers/elasticsearch_client_holder.rb.

Evidence

Build: https://buildkite.com/elastic/logstash-pull-request-pipeline/builds/5150
Job/step: :lab_coat: x-pack integration
Key log excerpt:
- Failed (Log pattern not yet visible: /rebuilt elasticsearch client.*on certificate change/)
- repeated backoff sleeps (0.02, 0.04, 0.08, ... 1.28)

Verification

Not run (read-only CI log and source analysis).

Follow-up

If retry fails with the same signature, treat this as a flaky candidate in Monitoring TLS hot-reload and track under a new flaky-test issue.

What is this? | From workflow: PR Buildkite Detective

Give us feedback! React with 🚀 if perfect, 👍 if helpful, 👎 if not.

infra-vault-gh-plugin-prod · 2026-08-06T15:00:35Z

💛 Build succeeded, but was flaky

Buildkite Build
Commit: 87e4f58

Failed CI Steps

History

💔 Build #5150 failed 6fbef91
💔 Build #5135 failed b142b11

LolloneS added 4 commits July 31, 2026 14:53

Add Queue#ensurePersisted to fsync all writes so far

27b6e23

Expose persistent? and checkpoint! on queue write clients

3ad9817

Pin memory write client persistent?/checkpoint! behavior with tests

28256c9

Delegate persistent? and checkpoint! through wrapped write client

b142b11

LolloneS mentioned this pull request Jul 31, 2026

Add commit_after_pq_fsync: commit Kafka offsets only after the persistent queue fsyncs logstash-plugins/logstash-integration-kafka#272

Draft

2 tasks

Merge branch 'main' into pq-write-client-checkpoint-api

87e4f58

Navigation Menu

Uh oh!

Expose persistent? and checkpoint! on queue write clients for durable input acknowledgement - #19362

Expose persistent? and checkpoint! on queue write clients for durable input acknowledgement#19362
LolloneS wants to merge 6 commits into
elastic:mainfrom
LolloneS:pq-write-client-checkpoint-api

LolloneS commented Jul 31, 2026

Uh oh!

github-actions Bot commented Jul 31, 2026

Uh oh!

mergify Bot commented Jul 31, 2026

Uh oh!

github-actions Bot commented Aug 6, 2026

Root Cause

Evidence

Verification

Follow-up

Uh oh!

infra-vault-gh-plugin-prod Bot commented Aug 6, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

Uh oh!

Conversation

LolloneS commented Jul 31, 2026

Release notes

What does this PR do?

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

Checklist

Author's Checklist

How to test this PR locally

Uh oh!

github-actions Bot commented Jul 31, 2026

🤖 GitHub comments

Uh oh!

mergify Bot commented Jul 31, 2026

Uh oh!

github-actions Bot commented Aug 6, 2026

TL;DR

Remediation

Root Cause

Evidence

Verification

Follow-up

Uh oh!

infra-vault-gh-plugin-prod Bot commented Aug 6, 2026

💛 Build succeeded, but was flaky

Failed CI Steps

History

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant