Package com.linkedin.venice.hadoop.task.datawriter

Class DualWriteVeniceWriter

java.lang.Object
com.linkedin.venice.writer.AbstractVeniceWriter<byte[],byte[],byte[]>
com.linkedin.venice.hadoop.task.datawriter.DualWriteVeniceWriter
All Implemented Interfaces:
Closeable, AutoCloseable
public class DualWriteVeniceWriter extends AbstractVeniceWriter<byte[],byte[],byte[]>
Wraps a Kafka-backed AbstractVeniceWriter and one or more ExternalStorageWriters (one per DUAL_WRITE target region) so each batch-push record is written to every external sink first and then produced to Kafka. The Kafka future is what the caller waits on for at-least-once semantics; external-sink durability is synchronous from the caller's point of view, so a failed external write throws before the Kafka produce starts and the Spark task is retried.

When the push targets multiple regions, the same record is fanned out to every regional writer before the Kafka produce. The guarantee is that no Kafka produce happens for a batch until every regional external write for that batch has succeeded by default: if any regional write fails (after its bounded retry) the exception propagates before the produce and fails the Spark task. When the optional fail-open mode is enabled, only the region that exhausted retries is disabled and reported; healthy external regions and Kafka continue. Note this is not an all-or-nothing write across the external sinks within a failed attempt — an earlier region in the fan-out may already hold the batch while a later one failed. Consistency is restored either by the whole-partition retry (fail-fast) or by flipping the failed region's version storageMode back to INTERNAL before EOP (fail-open mode).

Region fan-out is sequential. Each region's batchPut (including its bounded retries and backoff) completes before the next region's begins, on the single partition-writer task thread. This keeps the failure semantics and ordering simple, but the per-batch external-write latency is the sum across regions. TODO(future iteration): parallelize the per-region fan-out (e.g. a small per-task executor that issues the regional batchPuts concurrently and then joins, aggregating failures) when cross-region write latency dominates the push.

The writer buffers up to batchSize consecutive put records and flushes them as a single ExternalStorageWriter.batchPut(List) (per regional writer) before the corresponding Kafka produces fire. batchSize = 1 (the default) disables buffering: every record is forwarded immediately as a one-element batch, matching pre-batching semantics. flush() and close() drain any pending buffer before doing their own work, so the producer's record ordering is preserved.

update and delete are not supported — batch pushes from clean input never call either. Both throw UnsupportedOperationException so a stray invocation fails the Spark task loudly rather than silently leaving the external sink and Venice in divergent states.

Write-path timing. The wrapper measures how long this task spends in each of the two legs and reports the accrued time to the DataWriterTaskTracker:

  • external — per-region throttling wait, batchPut including its retries and retry backoff sleeps, external flush and external close.
  • venice — the kafkaWriter.put invocations plus kafkaWriter.flush() and kafkaWriter.close().
The two legs are measured over disjoint intervals on the single partition-writer task thread, so no wall-clock instant is counted twice. Timing uses System.nanoTime() (monotonic, immune to wall-clock adjustments) and is accrued in try/finally blocks so a failing batch still contributes the time it burned. Accrued time is published to the tracker as whole-millisecond deltas after every drain, flush and close, so work done while flushing/closing is always captured even if the task later fails.

Caveat: the two legs are not symmetric cost measurements. kafkaWriter.put(...) is an asynchronous enqueue — the actual produce work happens on the producer's background I/O thread and is mostly captured later, in kafkaWriter.flush(). Because that background thread keeps draining the producer's buffer while this task thread is blocked inside the external-write leg, some of the real Kafka produce cost is hidden from veniceWriteTimeMs whenever the external leg is the slower one. Treat externalStorageWriteTimeMs vs veniceWriteTimeMs as an indicator of which leg to look at first, not a precise cost split — it structurally under-reports the Venice leg's true cost, more so the slower external storage is.