perf(produce): add async commit pipeline to reduce FileCommitWaitTime by Mwea · Pull Request #525 · aiven/inkless

Mwea · 2026-03-03T13:56:53Z

Summary

Replace blocking upload wait with CompletableFuture callbacks. The commit thread no longer blocks on uploadFuture.get() - instead, commits chain via thenCombine() to preserve submission order while allowing parallel uploads.

Key changes:

FileCommitter now supports async (callback-based) and sync (blocking) modes, controlled by produce.async.commit.pipeline.enabled config
Async mode chains commits via CompletableFuture to preserve order without blocking the single commit thread
Add error isolation: failed commits don't block subsequent commits (using handle() to convert failures for ordering purposes only)
Add proper shutdown with awaitTermination() in close()
Add markReadyToCommit() so FileCommitWaitTime measures only the executor queue wait, not the cumulative chain wait

Tests:

Add CacheStoreJobTest for async callback behavior
Add tests for error isolation in both async and sync modes
Add tests for awaitTermination shutdown behavior
Update FileCommitterTest to use real executors with Awaitility

Benchmark results

Metric	Before (sync)	After (async)	Improvement
FileCommitWaitTime p99	227ms	~8ms	96.5%

Test plan

Unit tests pass for FileCommitterTest
Unit tests pass for CacheStoreJobTest
Integration test passes for WriterIntegrationTest
Manual testing with feature flag enabled in dev-env environment

Copilot

Pull request overview

This PR introduces an asynchronous (callback-based) commit pipeline for diskless produce to avoid blocking the single commit thread on upload completion, improving FileCommitWaitTime while preserving commit submission order.

Changes:

Add async commit pipeline to FileCommitter with ordered commit chaining, plus improved shutdown behavior (awaitTermination).
Refactor commit/cache-store jobs to be driven by upload completion callbacks instead of blocking on Future.get().
Add feature flag inkless.produce.async.commit.pipeline.enabled with documentation, and update/extend unit + integration tests.

Reviewed changes

Copilot reviewed 14 out of 14 changed files in this pull request and generated 5 comments.

Show a summary per file

File	Description
storage/inkless/src/main/java/io/aiven/inkless/produce/FileCommitter.java	Implements async commit chaining (ordered commits), callback-based cache store, and graceful shutdown.
storage/inkless/src/main/java/io/aiven/inkless/produce/FileCommitJob.java	Refactors commit job to run via upload-completion callback + adds `markReadyToCommit()` for metrics.
storage/inkless/src/main/java/io/aiven/inkless/produce/CacheStoreJob.java	Refactors cache store to run via upload-completion callback (no blocking on upload future).
storage/inkless/src/main/java/io/aiven/inkless/produce/UploadCompletionHandler.java	New handler interface for `CompletableFuture.handleAsync()` upload completion processing.
storage/inkless/src/main/java/io/aiven/inkless/produce/UploadCompletionConsumer.java	New consumer interface for `CompletableFuture.whenCompleteAsync()` upload completion processing.
storage/inkless/src/main/java/io/aiven/inkless/produce/Writer.java	Threads async pipeline flag through to `FileCommitter`.
storage/inkless/src/main/java/io/aiven/inkless/produce/AppendHandler.java	Wires config flag into `Writer` construction.
storage/inkless/src/main/java/io/aiven/inkless/config/InklessConfig.java	Adds `produce.async.commit.pipeline.enabled` config definition + accessor.
docs/inkless/configs.rst	Documents new async commit pipeline config option.
storage/inkless/src/test/java/io/aiven/inkless/produce/FileCommitterTest.java	Updates tests for async behavior and adds new async ordering/error-isolation/shutdown tests.
storage/inkless/src/test/java/io/aiven/inkless/produce/FileCommitJobTest.java	Updates commit job tests to use callback API and adds `markReadyToCommit()` metric test.
storage/inkless/src/test/java/io/aiven/inkless/produce/CacheStoreJobTest.java	New tests validating callback-driven cache store behavior.
storage/inkless/src/test/java/io/aiven/inkless/produce/WriterIntegrationTest.java	Enables async pipeline in integration test to validate end-to-end behavior.
storage/inkless/src/test/java/io/aiven/inkless/produce/WriterPropertyTest.java	Updates property test wiring for new `FileCommitter` ctor signature and shutdown behavior.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

You can also share your feedback on Copilot code review. Take the survey.

storage/inkless/src/main/java/io/aiven/inkless/produce/FileCommitter.java

storage/inkless/src/main/java/io/aiven/inkless/produce/FileCommitJob.java

storage/inkless/src/test/java/io/aiven/inkless/produce/FileCommitterTest.java

storage/inkless/src/main/java/io/aiven/inkless/produce/CacheStoreJob.java

Copilot

Pull request overview

Copilot reviewed 14 out of 14 changed files in this pull request and generated 5 comments.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

You can also share your feedback on Copilot code review. Take the survey.

storage/inkless/src/test/java/io/aiven/inkless/produce/FileCommitterTest.java

storage/inkless/src/main/java/io/aiven/inkless/produce/FileCommitter.java

Copilot

Copilot encountered an error and was unable to review this pull request. You can try again by re-requesting a review.

Copilot

Pull request overview

Copilot reviewed 14 out of 14 changed files in this pull request and generated 3 comments.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

You can also share your feedback on Copilot code review. Take the survey.

storage/inkless/src/test/java/io/aiven/inkless/produce/FileCommitterTest.java

jeqo · 2026-03-06T18:00:05Z

storage/inkless/src/main/java/io/aiven/inkless/produce/FileCommitter.java

+    // Tracks the previous commit to ensure commits happen in submission order, not upload completion order.
+    // Each new commit waits for both its upload AND the previous commit to complete before starting.
+    // Also used during shutdown to ensure all pending work completes before executors are shut down.
+    private CompletableFuture<?> previousCommitFuture = CompletableFuture.completedFuture(null);
+    // Tracks the last cache store future for shutdown coordination.
+    private CompletableFuture<?> lastCacheStoreFuture = CompletableFuture.completedFuture(null);


Wonder if instead of having these as separate futures, couldn't these be pipelined? This could simplify things a bit

Done. I've chained the cache store future with the commit future using thenCombine:

previousCommitFuture = commitFuture.thenCombine(cacheStoreFuture, (commitResult, cacheResult) -> commitResult);

This eliminates the separate lastCacheStoreFuture field and simplifies close() to wait on just the single combined future instead of CompletableFuture.allOf(pendingCommit, pendingCacheStore).

storage/inkless/src/main/java/io/aiven/inkless/config/InklessConfig.java

jeqo · 2026-03-10T11:01:26Z

storage/inkless/src/main/java/io/aiven/inkless/produce/UploadCompletionConsumer.java

Non of the interfaces introduced are actually used by type or reused. Both seem unnecessary.

jeqo

Changes look good but I think this is mixing a bunch of refactoring and feature work. We can clean this up moving the refactoring to a prep PR that can be quickly merged and do the feature review separately.

jeqo · 2026-03-11T19:22:53Z

storage/inkless/src/main/java/io/aiven/inkless/produce/FileCommitter.java

+        // Cache store runs after upload completes (in parallel with commit).
+        // It's tracked in the chain to provide backpressure if cache store is slow.
+        final CompletableFuture<?> cacheStoreFuture = uploadFuture
+                .whenCompleteAsync(cacheStoreJob::accept, executorServiceCacheStore);


We can use thenAcceptAsync instead and only call cache job on success.
Also there's no need to keep a pointer to the cacheStoreFuture AFAIC. We can keep it as fire and forget as before.

jeqo · 2026-03-11T19:24:04Z

storage/inkless/src/main/java/io/aiven/inkless/produce/CacheStoreJob.java

Can we consider moving this CacheStoreJob refactoring to a separate PR? It's changes are not crucial to this change.
We can make it implement Consumer<ObjectKey> and rename storeToCache to accept.

jeqo · 2026-03-11T19:29:48Z

storage/inkless/src/main/java/io/aiven/inkless/produce/FileCommitter.java

+                    // not the time waiting for upload completion or previous commits in the chain.
+                    commitJob.markReadyToCommit();
+                })
+                .handleAsync(commitJob, executorServiceCommit)


Similarly, using thenAplyAsync can simplify the commitJob.

…itTime Replace blocking commit pattern with pipelined CompletableFuture chain: - Upload starts immediately when file is closed - Commits are chained via previousCommitFuture to preserve submission order - Each commit waits for its upload AND previous commit before starting - Cache store is fire-and-forget (thenAcceptAsync on upload success) Key changes: - FileCommitJob now implements Function<ObjectKey, List<CommitBatchResponse>> instead of BiFunction, only handling the success path - CacheStoreJob now implements Consumer<ObjectKey>, invoked only on successful uploads via thenAcceptAsync - Upload failures are detected via CompletionException unwrapping in handleCommitResult() - markReadyToCommit() resets wait time measurement to exclude upload and previous commit wait times This reduces FileCommitWaitTime by removing thread blocking while waiting for uploads to complete. The commit thread now only runs the actual commit operation. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Mwea force-pushed the tchary/async-commit-pipeline branch 2 times, most recently from db78c51 to e1a8ebf Compare March 3, 2026 14:23

Mwea added a commit that referenced this pull request Mar 3, 2026

Merge PR #525: Async commit pipeline

078d259

Mwea marked this pull request as ready for review March 5, 2026 15:21

Mwea requested review from Copilot and jeqo and removed request for jeqo March 5, 2026 15:21

Copilot started reviewing on behalf of Mwea March 5, 2026 15:22 View session

Copilot AI reviewed Mar 5, 2026

View reviewed changes

Mwea requested a review from Copilot March 5, 2026 15:57

Copilot started reviewing on behalf of Mwea March 5, 2026 15:58 View session

Copilot AI reviewed Mar 5, 2026

View reviewed changes

Mwea requested a review from Copilot March 5, 2026 16:21

Copilot started reviewing on behalf of Mwea March 5, 2026 16:22 View session

Copilot AI reviewed Mar 5, 2026

View reviewed changes

Mwea requested a review from Copilot March 5, 2026 21:21

Copilot started reviewing on behalf of Mwea March 5, 2026 21:21 View session

Copilot AI reviewed Mar 5, 2026

View reviewed changes

jeqo reviewed Mar 6, 2026

View reviewed changes

Mwea force-pushed the tchary/async-commit-pipeline branch from e2c8d2c to 0e7afc7 Compare March 10, 2026 10:59

jeqo reviewed Mar 10, 2026

View reviewed changes

Mwea force-pushed the tchary/async-commit-pipeline branch 2 times, most recently from b731289 to 01f0578 Compare March 10, 2026 11:21

jeqo reviewed Mar 11, 2026

View reviewed changes

Mwea force-pushed the tchary/async-commit-pipeline branch 2 times, most recently from ffa7d1b to cb3dd06 Compare March 13, 2026 08:05

Mwea force-pushed the tchary/async-commit-pipeline branch from cb3dd06 to 7bbcc95 Compare March 18, 2026 17:36

Conversation

Mwea commented Mar 3, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Benchmark results

Test plan

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Uh oh!

Uh oh!

Uh oh!

jeqo Mar 6, 2026

Choose a reason for hiding this comment

Uh oh!

Mwea Mar 9, 2026

Choose a reason for hiding this comment

Uh oh!

Uh oh!

jeqo Mar 10, 2026

Choose a reason for hiding this comment

Uh oh!

jeqo left a comment

Choose a reason for hiding this comment

Uh oh!

jeqo Mar 11, 2026

Choose a reason for hiding this comment

Uh oh!

jeqo Mar 11, 2026

Choose a reason for hiding this comment

Uh oh!

jeqo Mar 11, 2026

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Mwea commented Mar 3, 2026 •

edited

Loading