[samza][producer] Isolate STREAM writes with shared workers - #3000
Open
kvargha wants to merge 3 commits into
Open
[samza][producer] Isolate STREAM writes with shared workers#3000kvargha wants to merge 3 commits into
kvargha wants to merge 3 commits into
Conversation
Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
Contributor
There was a problem hiding this comment.
Pull request overview
Introduces a shared, bounded partition-striped executor in venice-common and applies it to isolate STREAM-mode write submission off the caller thread (notably for VeniceSystemProducer), while preserving FIFO ordering per partition/stripe and keeping producer-specific semantics (futures, sticky failure handling, flush fences, shutdown behavior) in the producer layers.
Changes:
- Added
PartitionStripedExecutor(bounded, blocking admission; deterministic partition→stripe routing; per-stripe FIFO workers) and a dedicated unit test suite for its core guarantees. - Added partition-routing helpers (
getPartitionId) toAbstractVeniceWriterand implementations/delegation inVeniceWriterandBatchingVeniceWriterto enable stable upstream striping. - Added
VeniceSystemProducerasync STREAM write dispatcher/command plumbing plus extensive deterministic tests; refactoredPartitionedProducerExecutorto compose the shared striped executor.
Reviewed changes
Copilot reviewed 15 out of 15 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
| internal/venice-common/src/main/java/com/linkedin/venice/utils/concurrent/PartitionStripedExecutor.java | New shared bounded striped executor kernel used by producers. |
| internal/venice-common/src/test/java/com/linkedin/venice/utils/concurrent/PartitionStripedExecutorTest.java | Deterministic tests for striping, FIFO, backpressure, interrupt/shutdown semantics, shared await deadline. |
| internal/venice-common/src/main/java/com/linkedin/venice/writer/AbstractVeniceWriter.java | Adds default getPartitionId API (legacy routes to 0). |
| internal/venice-common/src/main/java/com/linkedin/venice/writer/VeniceWriter.java | Implements getPartitionId via key serialization + configured partitioner. |
| internal/venice-common/src/main/java/com/linkedin/venice/writer/BatchingVeniceWriter.java | Delegates getPartitionId to internal writer using serialized key bytes. |
| internal/venice-common/src/test/java/com/linkedin/venice/writer/VeniceWriterUnitTest.java | Adds routing determinism + delegation coverage for new getPartitionId. |
| internal/venice-common/src/main/java/com/linkedin/venice/ConfigKeys.java | Adds VSP STREAM worker/queue configs (count + per-stripe capacity). |
| integrations/venice-samza/src/main/java/com/linkedin/venice/samza/VeniceSystemProducerWriteCommand.java | New immutable write command + submission/durable future state machine + awaitSubmission helper. |
| integrations/venice-samza/src/main/java/com/linkedin/venice/samza/VeniceSystemProducerWriteDispatcher.java | New STREAM async dispatcher using partition-striped workers, flush fencing, sticky failure, drain semantics. |
| integrations/venice-samza/src/main/java/com/linkedin/venice/samza/VeniceSystemProducer.java | Wires async dispatch into STREAM send/put/delete/envelope send; validates configs; adjusts stop/flush behavior. |
| integrations/venice-samza/src/test/java/com/linkedin/venice/samza/VeniceSystemProducerWriteCommandTest.java | Tests command lifecycle (submission vs durable), interrupt behavior, error identity propagation. |
| integrations/venice-samza/src/test/java/com/linkedin/venice/samza/VeniceSystemProducerWriteDispatcherTest.java | Deterministic tests for dispatch contract, FIFO/isolation, sticky failures, flush fence, stop drain, deadlock regressions. |
| integrations/venice-samza/src/test/java/com/linkedin/venice/samza/VeniceSystemProducerTest.java | Updates/extends producer tests for async submission semantics, config validation, kill switch, batching path. |
| clients/venice-producer/src/main/java/com/linkedin/venice/producer/PartitionedProducerExecutor.java | Refactors Online Producer executor to compose PartitionStripedExecutor; adjusts awaitTermination interrupt-drain semantics. |
| clients/venice-producer/src/test/java/com/linkedin/venice/producer/PartitionedProducerExecutorTest.java | Adds regression test for interrupt-resistant draining behavior. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
Comment on lines
279
to
+283
| if (callbackExecutor != null) { | ||
| long remainingNanos = deadlineNanos - System.nanoTime(); | ||
| if (remainingNanos <= 0) { | ||
| return false; | ||
| try { | ||
| callbackTerminated = callbackExecutor.awaitTermination(Math.max(0, remainingNanos), TimeUnit.NANOSECONDS); | ||
| } catch (InterruptedException e) { |
Comment on lines
+793
to
+796
| if (workerCount == 0) { | ||
| // Kill switch: leave validatedWorkerCount at 0 so no dispatcher is created (every write runs inline). | ||
| return; | ||
| } |
Comment on lines
+85
to
+89
| /** | ||
| * Routes {@code command} to the stripe owning its Venice partition and returns its durable future after | ||
| * bounded admission. Never waits for the writer. A rejected admission (dispatcher stopped or kernel shutdown) | ||
| * fails the command's submission and records a sticky failure. | ||
| */ |
Comment on lines
+100
to
+106
| int partition = writer.getPartitionId(command.getKey()); | ||
| try { | ||
| kernel.submit(partition, () -> execute(command)); | ||
| } catch (RuntimeException e) { | ||
| recordSticky(e); | ||
| runDurableCompletion(command.finishSubmission(e)); | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Problem Statement
VeniceSystemProducerSTREAM writes currently invoke the coreVeniceWriteron the caller thread. LazySTART_OF_SEGMENTproduction can wait for a PubSub acknowledgement during Kafka leader movement. For Flink callers, this blocks the single-threaded mailbox from processing its checkpoint barrier; one delayed subtask then prevents the global checkpoint from completing.Online Producer already uses partition-based workers, but its executor implementation was producer-specific and could not be reused safely by
VeniceSystemProducer.Solution
Add a minimal
PartitionStripedExecutorinvenice-commonand use it from both producer APIs. The shared class owns only bounded blocking admission, deterministic partition-to-stripe routing, per-stripe FIFO workers, queue measurements, and primitive shutdown. Futures, callbacks, metrics, flush policy, writer ownership, and failure handling remain producer-specific.Online Producer retains its existing API, configuration, defaults, callback executor, thread and metric names, inline mode, rejection fallback, and close behavior. Its adapter now composes the shared executor instead of maintaining duplicate worker-array and striping logic.
For
VeniceSystemProducerSTREAM writes:send(Object, Object)path used by Flink returns after queue admission without waiting forVeniceWriter.put,delete, orupdateto return.put,delete, and Samza envelopesendstill wait through writer submission, preserving their existing compatibility contract.flush()remains a global pre-fence durability boundary.flush().0, BATCH, and STREAM_REPROCESSING retain the inline path.Defaults match Online Producer:
venice.system.producer.worker.count=4venice.system.producer.worker.queue.capacity=100000per worker stripeBatchingVeniceWriteronly gains partition-routing delegation. Its buffering, checker, flush, close, and ownership behavior are unchanged. Future Online ProducerVeniceWriterHookintegrations remain writer-construction concerns and require no shared-executor changes.The VSP-owned completion handoff is used only when a writer callback or failure completes synchronously on a stripe worker. It prevents user continuations from self-waiting on that worker. Normal Kafka callbacks continue completing directly. The handoff uses a fixed number of lazily started daemon threads; its ordinary queue is intentionally not awaited during stop because completions may execute arbitrary user continuations.
Code changes
Concurrency-Specific Checks
Both reviewer and PR author to verify
synchronized,RWLock) are used where needed.ConcurrentHashMap,CopyOnWriteArrayList).How was this PR tested?
Coverage includes:
VeniceWriterblocked onSTART_OF_SEGMENT, proving protected send returns while public compatibility wrappers still wait for submission.Erroridentity, and uninterruptible ownership after admission.VeniceWriterpartition routing andBatchingVeniceWriterdelegation.Validation results:
spotlessCheckandgit diff --checkpassed.Does this PR introduce any user-facing or breaking changes?
STREAM-mode protected
VeniceSystemProducer.send(Object, Object)now uses four partition-striped workers by default and returns after bounded admission. Writer-level failures after admission are reported through the returned future and stickyflush()failure. Publicput,delete, and Samza envelopesendretain their writer-submission wait. Setvenice.system.producer.worker.count=0to restore fully inline execution.🤖 Generated with GitHub Copilot CLI