docs/rfc/0007-synchronous-elasticsearch-writes.md
Jaeger's Elasticsearch/OpenSearch (ES/OS) trace writer enqueues spans into an asynchronous client-side bulk buffer and returns success to its caller before the data is durable in the backend. This silently violates the tracestore.Writer contract — WriteTraces returns nil even when the eventual bulk flush fails — and, on the Kafka ingester path, causes Jaeger to commit Kafka offsets for data that was never persisted, i.e. silent trace loss on backend outage or overload (#8476).
This gap is unchanged by RFC 0006 M6 (#8944), which migrated the span write path off olivere/elastic's BulkProcessor onto an owned bounded BulkWriter over esutil.BulkIndexer. M6 fixed the unbounded-memory bug (#2192) with a hard byte cap, but the new BulkWriter.Add is still fire-and-forget — errors surface only in OnFailure callbacks — so WriteTraces still returns nil at enqueue time. M6 swapped the async buffer; it did not make writes synchronous.
This RFC establishes the facts that shape the fix — a single _bulk HTTP request to ES/OS is already synchronous and durable; the asynchrony is entirely a client-side artifact — and shows that the ClickHouse-style server-side batched-insert model the reporter asked about does not exist in ES/OS. It then proposes an opt-in synchronous write mode: issue one synchronous, size-bounded _bulk request per batch, checking item-level results and returning a real error, so the writer contract and Kafka at-least-once are restored. Crucially, batching must then move to a blocking, coalescing batcher that holds each caller until its batch is durable — which the OTel exporterhelper already provides (queue.wait_for_result + queue.batch; the blocking and error-fan-out are confirmed in its source and tests, §4.2), while the fire-and-forget pipeline batch processor must be removed because it would re-break the guarantee. At-least-once additionally requires the Kafka receiver's message_marking.after: true (its default marks offsets before the write). Both modes ship; the operator chooses, and the required end-to-end configuration is documented.
This is a design exploration, not a committed decision. It builds directly on the owned esclient.BulkWriter M6 delivered, adding a complementary synchronous write primitive over the same transport.
The v2 writer interface is explicit (internal/storage/v2/api/tracestore/writer.go):
// WriteTraces ... if any of the spans fail to be written an error is returned.
// Compatible with OTLP Exporter API.
WriteTraces(ctx context.Context, td ptrace.Traces) error
The ES implementation does not honor it. TraceWriter.WriteTraces converts the batch and, per span, calls the fire-and-forget SpanWriter.WriteSpan, then unconditionally returns nil (internal/storage/v2/elasticsearch/tracestore/writer.go):
func (t *TraceWriter) WriteTraces(_ context.Context, td ptrace.Traces) error {
dbSpans := ToDBModel(td)
for i := range dbSpans {
span := &dbSpans[i]
t.spanWriter.WriteSpan(model.EpochMicrosecondsAsTime(span.StartTime), span)
}
return nil
}
WriteSpan ultimately calls s.bulkWriter.Add(esclient.BulkItem{…}) (core/writer.go). BulkWriter.Add returns nothing (esclient/interfaces.go — Add(item BulkItem)); it encodes the document and hands it to the background esutil.BulkIndexer (esclient/bulk.go). The buffer flushes later on its own triggers, and per-item failures surface only in the indexer's OnFailure callback, which logs and updates the bulk_index metrics but cannot influence the already-returned WriteTraces. (Before M6 this was olivere's BulkProcessor with an After callback — same fire-and-forget shape, different library.)
Cassandra and ClickHouse v2 writers honor the same contract synchronously (they errors.Join per-span failures / return Append/Send errors). ES is the outlier.
Jaeger v2 is an OpenTelemetry Collector assembly. The ingester pipeline is:
kafka receiver ──▶ batch processor ──▶ jaeger_storage_exporter ──▶ WriteTraces
(config-kafka-ingester.yaml). The storage exporter wraps WriteTraces in exporterhelper with no sending queue and retry disabled by default (storageexporter/factory.go, config.go), so ConsumeTraces → pushTraces → WriteTraces is a straight synchronous call — the only asynchronous hop is the esutil.BulkIndexer buffer inside the writer.
There are actually two independent commit-before-durable gaps, and both must be closed for at-least-once. Verified against the receiver's source (the contrib kafkareceiver v0.155.0 is franz-go-based, consumer_franz.go):
message_marking.after: false, the consume loop calls client.MarkCommitRecords(msg) before handleMessage (i.e. before ConsumeTraces) — the offset is queued for commit regardless of the downstream result. Only message_marking.after: true moves the mark after a successful ConsumeTraces, and only then does a returned error rewind/pause the partition instead of advancing it (on_error: false, the default). So at-least-once requires message_marking.after: true no matter how the writer behaves.nil before the data is durable. Even with after: true, WriteTraces returns at enqueue time, so the receiver marks a batch that is not yet in ES. The sequence:
WriteTraces enqueues them into esutil.BulkIndexer and returns nil.after: true).The offset is already gone. This converts any transient backend problem into permanent, invisible trace loss, and defeats the entire point of buffering traces in Kafka. The same gap removes backpressure: the pipeline cannot slow down or apply retry because it never learns the write failed. This RFC fixes gap (2); the required message_marking.after: true for gap (1) is an operator configuration this RFC documents (§4.5) rather than a code change.
This is not a new capability so much as a return to the original design, made possible again by the v2 pipeline. The ES writer started synchronous, exactly like Cassandra:
spanstore.Writer interface was WriteSpan(span *model.Span) error — a single span, no batch method (introduced in PR #47, db6587b9, 2017-02-15; today internal/storage/v1/api/spanstore/interface.go with a ctx). Any batching therefore had to be hidden below the interface.f3717836, 2017-06-19) added plugin/storage/es/spanstore/writer.go, whose WriteSpan issued blocking .Do(s.ctx) calls per document and returned the write error synchronously — the same shape Cassandra still has.e52ecffb, 2018-01-23, Pavol Loffay) swapped the per-span .Do(s.ctx) for .Add() into a shared background elastic.BulkProcessor, and introduced the .bulk.* knobs (later the BulkProcessing struct, PR #6090). The exact diff:
- _, err := s.client.Index().Index(indexName).Type(spanType).BodyJson(&elasticSpan).Do(s.ctx)
+ s.client.Index().Index(indexName).Type(spanType).BodyJson(&elasticSpan).Add()
writeSpan/writeService lost their error returns in that commit — because with a one-span interface, issuing an HTTP round-trip per span is far too slow at collector scale, and the only way to batch under WriteSpan(one span) was a buffer with the outcome reported asynchronously. The async model was the correct answer to the v1 constraint.The v2 tracestore.Writer interface removes that constraint — WriteTraces(ctx, ptrace.Traces) already hands the writer a whole batch. Synchronous batched writes are now expressible without one-request-per-span; the historical reason for going async no longer binds. (M6, PR #8944, later swapped the olivere processor for an owned esutil.BulkIndexer, but preserved the async, fire-and-forget shape — §2.4.)
client.Bulk().Do(ctx) plus resp.Errors checking. The right direction, closed pending a design decision.atomic.Pointer and return it on the next WriteTraces. A minimal patch that closes the contract gap in the loosest sense but attributes the error to the wrong batch and still commits the failing batch's offset.This RFC exists to pick the direction those PRs were waiting on.
Three facts, each load-bearing for the design and each easy to get wrong.
_bulk request is synchronous and durableA single _bulk HTTP request is not asynchronous on the server. Under the default index.translog.durability: request, ES/OS fsync and commit the translog on the primary and every allocated replica before responding, so a 200 means "all acknowledged writes have been committed to disk" (ES translog settings). OpenSearch is identical. The async durability mode only changes when the fsync happens (every sync_interval, default 5s) — it does not make the request itself asynchronous, and is a separate, orthogonal knob.
Implication: durability is a property Jaeger already gets for free from a single bulk round-trip. The async behavior Jaeger exhibits is purely client-side — an artifact of the client-side bulk buffer (esutil.BulkIndexer today), not of ES.
Search visibility is a separate axis governed by refresh (index.refresh_interval, default 1s; the refresh bulk parameter defaults to false). A normal bulk 200 means the docs are durable, not necessarily searchable yet (refresh parameter).
This distinction does not affect this RFC: the writer contract and Kafka at-least-once only require durability (no acknowledged-but-lost data). We deliberately do not propose refresh=wait_for/true on writes — forcing refresh would wreck indexing throughput for no correctness benefit. Near-real-time search remains the existing ~1s behavior.
The reporter asked whether ES has a ClickHouse-style server-side batched insert — where the server coalesces many independent client requests into one flush and can block the client until that flush completes.
ClickHouse has this. async_insert=1 writes incoming INSERTs into a server-side in-memory buffer flushed on thresholds (async_insert_max_data_size ≈100 MiB, async_insert_busy_timeout_ms ≈200 ms, async_insert_max_query_number =450), "invisible to clients … merging insert traffic from multiple sources." With wait_for_async_insert=1 (default) the client's INSERT blocks until the buffered batch is flushed to disk, yielding a synchronous durability ack on top of server-side batching (ClickHouse async inserts). This is exactly the model the reporter described.
ES/OS have no equivalent. There is no server-side mode that buffers across separate client requests with an optional wait-for-flush ack. The only batching primitives are:
_bulk API — batches many docs, but within one client request; no cross-request coalescing;esutil.BulkIndexer — what Jaeger now uses — olivere BulkProcessor, Logstash/Beats) — batching lives in the client;translog.durability: async — an fsync-timing knob, still per-request, not coalescing.So the ClickHouse option is off the table for ES/OS. The equivalent has to be built where ES puts it — at the client / pipeline — which is precisely what this RFC does: let the pipeline (Kafka + batch processor) form the batch, and make the client's per-batch write synchronous. (The one place ES could "block until a batch fills" is the client-side buffer's FlushInterval — but that blocks nothing and acks nothing, which is the bug.)
Post-M6, the esutil.BulkIndexer is configured from the same bulk_processing block (config.go), mapped to the indexer's FlushBytes / FlushInterval / Workers (esclient/bulk.go); Jaeger defaults: max_bytes 5 MB, flush_interval 200 ms, workers 1. FlushBytes is now a hard byte ceiling — the buffer flushes before exceeding it — which is what closed the unbounded-memory / 413 Request Entity Too Large bug #2192 in M6. (max_actions no longer maps to anything: esutil has no doc-count trigger. This is a minor config-surface wrinkle M6 introduced, orthogonal to this RFC.) These knobs shape client-side flushing; none of them make a write synchronous or its failure observable to the caller.
Goal: make it possible for WriteTraces to return a truthful error for the batch it was given, so that (on Kafka) the offset commits only after that batch is durable — without collapsing throughput. Note "possible", not "always on": async writing is a legitimate mode for the unbatched direct-ingest path (§3.1), so this is about adding a synchronous option and choosing sensible defaults, not deleting the async one.
The options:
WriteSpan enqueues into the async esutil.BulkIndexer and WriteTraces returns nil immediately; failures land in OnFailure callbacks that only log and count. This is today's behavior. It is the bug for the Kafka-buffered topology (silent loss, §1.2), but — importantly — it is also the right default for direct, unbatched ingest (§3.1), which is why it is a real option and not merely a strawman baseline.atomic.Pointer and return it from the next WriteTraces. A minimal patch that surfaces some error eventually, without changing the async model.WriteTraces, then inspect results. Spec-conformant in principle, but it flushes the whole shared buffer per call._bulk per batch: assemble the batch's documents, issue a blocking round-trip, check item-level results, and return a real error (§4). Batching moves to a blocking pipeline batcher (§4.2). This is the model the reporter proposed.Criteria:
WriteTraces return real write errors?Options as columns, criteria as rows. 🟢 good · 🟡 partial/caveated · 🔴 poor.
| Criterion | A. Async status quo (post-M6) | B. Sticky-error (#8651) | C. Per-call buffer drain | D. Sync batch write (rec.) |
|---|---|---|---|---|
| Contract (real errors) | 🔴 always nil | 🟡 delayed, best-effort | 🟢 | 🟢 |
| Correct offset / at-least-once | 🔴 | 🔴¹ | 🟢 | 🟢 |
| Attribution | 🔴 | 🔴¹ | 🟢 | 🟢 |
| Throughput | 🟢 | 🟢 | 🔴² | 🟡³ |
| Backpressure | 🔴 | 🔴 | 🟡 | 🟢 |
| Complexity | 🟢 exists | 🟢 small | 🔴² | 🟡 moderate |
Legend / footnotes:
esutil.BulkIndexer exposes no per-call flush — only Close() (a full shutdown-flush). Approximating "drain now" means closing and recreating the indexer per batch, which is absurd. Option C is effectively unavailable on the current library, not merely slow.The matrix scores against the Kafka at-least-once goal. Read that way: B is a stop-gap that never withholds the failing batch's offset; C is spec-conformant but destroys batching and is anyway unimplementable on esutil (footnote 2); D is the only option that restores the contract and at-least-once with correct attribution, and is the model the reporter proposed. A scores red across the correctness rows — but only because those rows encode a goal that does not apply to every topology (§3.1).
So the recommendation is not simply "pick D" — it is:
_bulk requests. That virtue is not exclusive to async: D + a blocking batcher (§4.2) — or even D behind the plain batch processor, if one is willing to trade the durability guarantee for raw batching — reproduces A's request shape while keeping (or knowingly dropping) the error signal. Choosing D therefore does not cost throughput wherever batching exists.es.write.synchronous) that flips the default over the standard Jaeger alpha→beta→stable lifecycle. This gives operators time to add the required pipeline shape (§4.5) before sync becomes default.Async writing is not merely the old behavior to be tolerated until removed — it is the correct mode for one real topology: direct, unbatched ingest. When spans arrive over TCP straight into the collector (OTLP/Jaeger receivers → storage, no Kafka), there is no natural batch boundary and no offset to protect. Forcing a synchronous _bulk per inbound request would mean either tiny, inefficient bulks or a blocking batcher that adds flush_timeout latency to every request for a guarantee (at-least-once against a durable log) that this topology does not even have — the client, not a Kafka offset, owns retry. There, fire-and-forget client-side batching is exactly right.
Synchronous writes earn their cost specifically where there is a durable buffer to be honest with — the Kafka ingester — and where losing an ack means losing data that Kafka still believes was delivered. This is why the two modes coexist: async for unbatched direct ingest, sync for Kafka-buffered ingest. The feature gate governs the default; whether async is ever fully removed (vs. retained as a non-default option for the direct-ingest path) is left open (§7).
Replace the per-span enqueue with a per-batch synchronous write:
WriteTraces converts the whole ptrace.Traces batch to []dbmodel.Span (as today)._bulk body containing every span document and any service/operation documents the batch requires (§4.3), split into size-bounded sub-requests (§4.4).esclient (§5) — one _bulk round-trip that returns the parsed response — inspects resp.Errors and each item's status, and returns an error if the transport failed or any item failed.nil does the exporter return success → the Kafka offset commits.This requires the core writer to expose a batch, error-returning entry point (the #8476 discussion sketched exactly this):
// core.Writer
WriteSpans(ctx context.Context, spans []dbmodel.Span) error // replaces fire-and-forget WriteSpan
TraceWriter.WriteTraces calls it once per batch and returns its error verbatim. The SpanWriter's existing responsibilities (tag elevation, index/rotation target selection, @timestamp for data streams, service dedup cache) are unchanged — only the sink changes from "enqueue into shared buffer" to "append into this batch's bulk request."
Synchronous writes move the batching problem, they do not remove it. Batch size now determines throughput, so batches must still be formed somewhere — but that "somewhere" must block the origin caller until its spans are durably written, or the whole guarantee is lost again. This is the central design constraint, and it has one sharp consequence:
The OTel batch processor must be removed for sync mode. The pipeline batch processor is fire-and-forget: ConsumeTraces buffers the spans and returns nil immediately, and the batch is flushed to the exporter later on a separate goroutine. Placed in front of a synchronous writer, it re-introduces the exact bug this RFC fixes — the Kafka offset commits when the processor buffers, not when ES persists. So sync mode with the batch processor is strictly worse than useless. It has to go.
That leaves the question the reporter raised: if the pipeline no longer batches, and upstream messages are small (SDKs or a default-config collector emitting small ptrace.Traces), then one synchronous bulk per message is too small and throughput suffers. Sizing bulks well and keeping writes synchronous requires a blocking, coalescing batcher — one that merges spans from many concurrent ConsumeTraces calls into one large bulk and unblocks each caller with that bulk's durable result. This is precisely the ClickHouse wait_for_async_insert=1 model (§2.3), realized in the collector rather than the server.
This mechanism already exists in the OTel exporterhelper the storage exporter is built on, and its behavior is confirmed by reading the source (v0.155.0), not just the config. exporterhelper's QueueBatchConfig (which the storage exporter already wires via WithQueue, storageexporter/factory.go) exposes wait_for_result and a nested batch block. Tracing the code:
wait_for_result: true). memoryQueue.Offer creates a blockingDone{ch chan error}, enqueues, then blocks on <-done.ch; the consumer signals it with bd.ch <- err after export. So the calling goroutine returns the export result, not an enqueue ack. (internal/queue/memory_queue.go.)partitionBatcher accumulates each caller's done into a multiDone []queue.Done; on flush it runs done.OnDone(consumeFunc(ctx, req)), and multiDone.OnDone loops calling every merged caller's done with the same single error (internal/queuebatch/partition_batcher.go). Framework test TestPartitionBatcher_MergeError asserts exactly this fan-out; TestQueueBatch_BatchBlocking asserts concurrent Sends block until the batch flushes. Both wait_for_result and batch may be set together (the only rejected combination is wait_for_result with a persistent disk queue — irrelevant here, since sync mode wants the durable ack, not a spooling queue).So the required semantics are not merely plausible — they are what the code does: N partition-consumer ConsumeTraces calls merge into one bulk, each blocks until that bulk is durably written, and all of them receive its error. This resolves the concern flagged in the prior draft. One real consequence remains (§4.6): the merge (MoveAndAppendTo) fuses all callers' spans, so the result is strictly all-or-nothing — there is no per-caller partial result.
Three ways to form the batch:
queue.wait_for_result: true + queue.batch on the storage exporter, pipeline batch processor removed. Config-only, no new code. Works for every source (Kafka and direct OTLP) because it coalesces at the write boundary regardless of upstream message size, including across concurrent Kafka partitions.kgo.FetchTopicPartition.Records) into one ConsumeTraces call, then mark all N offsets together. This is appealing: one caller ⇒ one bulk ⇒ one all-or-nothing result ⇒ trivial, monotonic per-partition offset handling — no cross-partition coalescing, no multiDone fan-out, no wait_for_result machinery. However, the current contrib receiver does not do this: verified in consumer_franz.go, the loop is for _, msg := range msgs { handleMessage(pc, msg) } — strictly one record per ConsumeTraces, with no message-count/batch config knob. Realizing it means an upstream contrib feature (accumulate per-partition records before the consumer call) or a thin Jaeger-owned receiver wrapper. Worth pursuing upstream, but it is code, not config.ConsumeTraces calls, flushes one bulk on size/time, and blocks each caller until the flush completes. Same semantics as (1) but owned by us; only worth building if (1) proves unfit.Comparison (🟢 good · 🟡 partial/caveated · 🔴 poor):
| Criterion | 1. Exporter batcher | 2. Receiver batching | 3. Custom processor |
|---|---|---|---|
| Available today (no new code) | 🟢 config-only | 🔴 needs upstream/custom receiver | 🔴 new component |
| Coalesces across partitions | 🟢 | 🔴 per-partition only¹ | 🟢 |
| Offset/commit simplicity | 🟡 many blocked callers | 🟢 one caller, mark N together | 🟡 |
| Per-caller failure attribution | 🔴 all-or-nothing² | 🟡 whole-partition batch² | 🟡 could preserve, if built for it |
| Works for direct-OTLP (non-Kafka) | 🟢 | 🔴 Kafka-only | 🟢 |
Recommendation: ship with (1) — it exists, it is config-only, and it covers both Kafka and direct-OTLP. Pursue (2) upstream in parallel as the cleaner long-term shape for the Kafka ingester (simplest offsets). Keep (3) as a fallback only.
The redundant client-side buffer (esutil.BulkIndexer) is removed in sync mode. Batching happens once, at the blocking batcher, where it is observable and tunable — not hidden in a fire-and-forget client buffer that discards the durability signal.
WriteSpan today also writes a service:operation dedup document per new (service, operation) pair, gated by an in-memory TTL cache (core/service_operation.go). These documents already carry a deterministic _id — the FNV-64a hashCode(service) of {serviceName, operationName} — so repeated writes are idempotent upserts on the ES side; retries never duplicate them.
Two adjustments for sync mode:
writeCache(cacheKey, …) immediately after Add() — i.e. it marks the service "written" at enqueue time. In sync mode the cache must be updated only after the bulk succeeds; otherwise a failed-then-retried batch would find the key cached, skip the service document on retry, and — although the ES-side hash _id makes the document itself recoverable if any later span re-emits it — risk leaving a gap until then. Moving the cache write past the durable ack closes that window. (Same "cache after durable write" discipline as RFC 0004 §3.9.)http.max_content_length (default 100 MB) and 413. The primary control is the batcher's max_size (§4.2/§4.5), but the sync writer keeps a safety-net cap at max_bytes: it splits an oversized batch into sub-requests, issues them in sequence, and joins their errors. This is the same byte discipline M6's async FlushBytes already enforces (#2192 is fixed for the async path) — the sync path must carry it too rather than inherit it, since it does not go through esutil's buffer.exporterhelper retry). Item-level 429/503 (backpressure) therefore retry naturally. We do not silently swallow partial failures._id (§4.7). Today Jaeger sets no _id for spans (server-generated), so any retry — Kafka re-delivery or exporterhelper retry — creates duplicate span docs. Under async fire-and-forget this was a rare, tolerated edge case; under synchronous at-least-once writing, retries are the normal error-recovery path, so duplication moves from an edge case to a first-order concern (at high ingest a retry storm multiplies stored volume). The fix is the pattern the service documents already use (§4.3) and that Cassandra has always used — its traces primary key is (trace_id, span_id, span_hash), where span_hash is a content hash, so re-inserting an identical span is an idempotent upsert while genuinely-distinct spans that share (trace_id, span_id) stay separate. Give each ES/OS span a deterministic content-hash _id and a re-sent identical span upserts instead of duplicating. This makes whole-batch retry idempotent (which is why §4.6 needs no per-chunk retry bookkeeping) and is therefore promoted from a future RFC 0004 cross-reference to an in-scope milestone here — at-least-once is not production-safe without it. Two design notes carried into §4.7: (a) traceID+spanID+startTime (RFC 0004's sketch) is insufficient — the shared-span model lets a client and server span share (traceID, spanID), so a content hash is required, as Cassandra does; (b) the _id↔op_type interaction matters — op_type: index upserts cleanly, but data streams force op_type: create, under which a retried _id returns 409 that the writer must treat as idempotent success, not a failure.Two independent settings, on two components. First, the storage backend's write mode:
elasticsearch:
write_mode: async # async (default) | sync
bulk_processing: # shapes async esutil.BulkIndexer flushing; in sync mode only `max_bytes` (safety-net chunk cap)
max_bytes: 5000000
flush_interval: 200ms
workers: 1
async (default): today's esutil.BulkIndexer behavior (post-M6), unchanged — fully backward compatible.sync: §4.1. max_bytes is reused as the safety-net chunk ceiling (§4.4); flush_interval and workers are inert (each write is one blocking round-trip).The default value of write_mode (when unset) is governed by a feature gate — e.g. es.write.synchronous — following Jaeger's standard lifecycle: alpha (default async, opt into sync) → beta (default sync, opt out to async) → stable. An explicit write_mode in config always wins over the gate, so operators can pin either mode regardless of lifecycle stage (§3, point 3).
Second — and this is the part operators must get right — the pipeline must batch with a blocking batcher and must not use the fire-and-forget batch processor (§4.2). The recommended shape for the Kafka ingester:
service:
pipelines:
traces:
receivers: [kafka]
processors: [] # NO batch processor in sync mode (it is fire-and-forget)
exporters: [jaeger_storage_exporter]
receivers:
kafka:
message_marking:
after: true # REQUIRED: mark offset only after ConsumeTraces succeeds
on_error: false # (default) failed write → rewind/pause partition, do not advance
exporters:
jaeger_storage_exporter:
trace_storage: some_storage
queue:
wait_for_result: true # block each caller until its batch is durable
# note: wait_for_result is incompatible with a persistent (storage:) queue
batch:
sizer: bytes
min_size: 3000000 # coalesce small messages up to ~3 MB
max_size: 5000000 # match bulk_processing.max_bytes so the sync writer's safety-net split (§4.4) rarely triggers
flush_timeout: 200ms # bound added latency for low-traffic streams
Three things must line up, and all are the operator's choice — Jaeger ships both modes and does not pick for them:
elasticsearch.write_mode: sync (the durable writer),message_marking.after: true (§1.2 — otherwise the offset commits before the write regardless), andqueue.wait_for_result + queue.batch) with the fire-and-forget batch processor removed (§4.2).Any one of the three missing silently degrades the guarantee: write_mode: sync with a batch processor still present, or with message_marking.after: false, loses data exactly as before. This coupling must be documented prominently (§8). A startup warning when write_mode: sync coexists with a pipeline batch processor is worth considering, though the exporter cannot see the full pipeline graph (§7 Q2).
A related but separate misconfiguration — queue.batch.max_size exceeding bulk_processing.max_bytes (§4.4) — does not have that visibility problem: both values belong to the exporter itself. jaeger_storage_exporter's start() already resolves trace_storage to its ES factory via jaegerstorage.GetTraceStoreFactory, and the exporter's own Config already carries queue.batch.max_size, so it can read the resolved factory's bulk_processing.max_bytes and compare the two before serving traffic. Because this is a deterministic comparison of two configured integers, not a runtime heuristic that needs operational experience to trust, it should hard-fail startup when write_mode: sync and queue.batch.max_size > bulk_processing.max_bytes, rather than only warn.
Concurrency and offsets are handled by the receiver + franz-go — Jaeger does not hand-roll them. This is worth stating because Jaeger v1's Kafka consumer did track offsets manually and it was error-prone. In the contrib franz-go receiver (consumer_franz.go, verified):
fetch.EachPartition(... go func(pc, msgs))); partitions are processed concurrently, and within a partition records are processed serially (for _, msg := range msgs).MarkCommitRecords + CommitMarkedOffsets/autocommit); the integrator never computes offsets. On a failed write with after: true, the partition is rewound to the failed record (SetOffsets) or paused, so nothing past the failure commits.So "consume several chunks in parallel and commit their offsets independently" is exactly what happens: parallelism = number of partitions, each committing its own contiguous, monotonic range. Those concurrent per-partition ConsumeTraces calls are precisely the callers the exporter batcher (§4.2 option 1) coalesces and then unblocks together.
Fine-grained acking — is there room to use ES per-item bulk results? The reporter asked whether, since the ES _bulk response reports per-document status, the ack could be finer than all-or-nothing (commit the messages whose docs all succeeded, retry only the ones that failed). The answer, from the code, is not with the built-in batcher, and not cheaply:
MergeSplit fuses all callers' spans via ResourceSpans().MoveAndAppendTo(...) before the export. By the time the bulk runs, the mapping "this _bulk item ↔ this Kafka message/offset" is gone. consumeFunc returns a single error; multiDone gives every caller that same error. ES per-item results have nowhere to be routed back to.resp.items[i].status → offset. That is real complexity for limited payoff: ES partial-bulk failures cluster into two regimes — whole-target unavailable (every item fails; all-or-nothing retry is already correct) and poison documents (a few items rejected for mapping/validation reasons that will fail identically on every retry, so per-message retry just loops). Neither regime is meaningfully helped by finer acking; the poison case argues instead for a dead-letter path, which is orthogonal.Recommendation: keep offset attribution all-or-nothing per batch — but this is not the same as ignoring per-item results. It is correct for at-least-once because a retry of already-durable spans is a no-op once they carry a deterministic _id (§4.7), not a duplicate. What all-or-nothing does not by itself solve is the poison document: a doc ES/OS rejects deterministically (mapping conflict, malformed field). Failing the batch forever on a poison doc means the offset never advances and the partition is re-delivered in an infinite loop — head-of-line blocking. At production ingest (e.g. 100 MB/s) there is no "operator intervenes": a stalled partition is an outage, and back-pressure propagates until the pipeline falls over. So poison handling is not a deferrable "future item" — a conformant synchronous writer must dispose of poison documents autonomously so the partition always makes progress. That mechanism is specified in §4.8; the per-item _bulk results this writer already parses are used there to classify terminal vs. transient failures (not to attribute offsets).
_idSynchronous at-least-once makes retries a normal, frequent event, so the write must be idempotent: a re-sent span must not create a second document. Service documents already achieve this with a deterministic hash _id (§4.3); Cassandra has always achieved it by putting a content hash (span_hash) in the span primary key. Spans in ES/OS get the same treatment:
_id = a content hash of the span. Compute a stable hash over the span's canonical bytes (as Cassandra's dbmodel.Span.Hash does — gob-encode the span with the hash field zeroed; the ES writer would hash its own dbmodel.Span). A re-sent identical span produces the same _id and does not duplicate.traceID+spanID+startTime. RFC 0004 sketched that key, but it is unsafe: in the shared-span (Zipkin-style) model a client and server span legitimately share (traceID, spanID) and can share startTime; keying on that would collapse two distinct spans into one. A content hash distinguishes them, exactly as Cassandra's span_hash does — which is why we match Cassandra rather than the sketch.op_type interaction (load-bearing). With op_type: index (legacy indices) a duplicate _id overwrites — a clean idempotent upsert, no special handling. With op_type: create (required by data streams, RFC 0004) a duplicate _id is rejected 409 version_conflict; the synchronous writer must recognize a 409 on its own deterministic _id as idempotent success (the span is already durable) and not surface it as a batch error. Without that, every legitimate retry under data streams would look like a failure and stall the offset._id disables ES's auto-ID fast path (auto-IDs skip the version lookup), a modest per-doc indexing cost. This is the deliberate trade — bounded overhead for exactly-stored-once under retry — and it is worth it precisely in the topology where retries happen (Kafka at-least-once). For the async/direct-ingest path it can remain opt-in.This is delivered as its own milestone and its own PR (M4 below): it is self-contained, independently valuable (it also removes duplicate-on-retry for the async path), changes observable wire behavior (dedup, snapshot-visible), and carries the data-stream op_type/409 subtlety — all reasons to review it in isolation rather than fold it into the sync-wiring milestone. It should land before write_mode: sync becomes a default, so that at-least-once is idempotent the moment it is on.
A poison document fails deterministically on every attempt (mapping conflict, field-type clash, oversized field). Under synchronous at-least-once with message_marking.after: true, failing the batch holds the offset — correct for a transient failure, fatal for a poison one: the partition re-delivers the same poison forever (§4.6). The system must resolve this without a human, because at scale a stalled partition is an outage, not a ticket. The design:
_bulk result. The writer already parses per-item status (M2). Split failures into transient — 429 (backpressure), 503, connection/timeout, whole-target-unavailable — and terminal — 400, mapping/parse/validation rejections, and any status that will not change on replay.exporterhelper/Kafka re-delivery re-sends; idempotent _id (§4.7) means the already-durable items are no-ops, so only the genuinely-missing items are (re)written. This is the normal back-pressure path.*-dead-letter index/data stream, or a dead-letter Kafka topic) with the rejection reason attached. Once the poison items are safely in the dead-letter sink, the batch is complete — the good items are durable, the poison items are preserved out-of-band — so WriteTraces returns nil and the offset advances. The partition never blocks; no data is silently dropped; the poison is inspectable later.dead_lettered counter (by reason) and a log per poison doc so humans are notified asynchronously — but the pipeline has already recovered on its own. Optionally cap dead-letter volume with an alarm to catch a systemic mapping regression (a flood of terminal failures) versus the occasional bad document.This preserves the two invariants together: no silent loss (poison goes to the dead-letter sink, not /dev/null) and no head-of-line blocking (the live partition always progresses). It depends on §4.7 (so batch retry for the transient case doesn't duplicate) and is delivered as its own milestone (M7 below), since a durable, autonomous dead-letter path is a substantial, independently-testable component.
This work is small and lands cleanly on top of the client foundation RFC 0006 is building:
esclient.BulkWriter and the bounded bulk indexer over esutil.BulkIndexer exist and carry all span/service writes; olivere is gone from the write path (it remains only for the reader/control-plane, being migrated by later 0006 milestones). M6 delivered the async half; this RFC adds the synchronous half over the same esclient transport. The sync path is a new, distinct primitive — a blocking bulk on esclient (e.g. Bulk(ctx, items) (BulkResult, error)), reusing 0006's owned request/response types and the shared transport pool — not a call into the existing esutil.BulkIndexer and not a method on the async BulkWriter. The two are peers selected by write_mode: SpanWriter holds either the async BulkWriter (today) or a synchronous bulk client, chosen at construction from config. Why the sync path does not reuse esutil.BulkIndexer — which already builds _bulk NDJSON and, since go-elasticsearch v9.4.2, even offers a blocking Flush — is explained in §5.1.index→create) and adds @timestamp. The sync writer inherits both transparently — it writes whatever documents SpanWriter produces (which already emits esclient.BulkItems with the right OpType). The service-doc cache ordering (§4.3) reuses the same "cache after durable write" discipline data streams' trace-timestamp index needs.There is no conflict; there is shared surface. Treat "synchronous write mode" as a peer of M6's async BulkWriter on the same esclient, with the contract/Kafka fix (this RFC) as the motivating requirement for the synchronous primitive.
esutil.BulkIndexer for the synchronous path?esutil.BulkIndexer (the helper M6's async path is built on) already does two things a synchronous writer needs: it builds the _bulk NDJSON from structured items, and — since go-elasticsearch v9.4.2 — it exposes a blocking Flush(ctx) that drains the buffer and waits for the flush to complete. So the natural question is why the sync path does not simply Add the items and Flush. The obstacle is its result model, not its ability to flush synchronously:
Flush returns nil even when documents are rejected. Its return value reports only transport-level health — a transport error, a non-2xx overall response, or a body-parse failure. A _bulk request in which some or all items are individually rejected (a mapping rejection, a version conflict) is still a successful HTTP 200 at the transport level, so Flush returns nil; each item's real outcome is delivered only to the OnSuccess/OnFailure closure passed on Add. A conformant WriteTraces must instead return "error iff any item was rejected" (§1.1, §4.8) — a per-batch durability verdict that Flush's return value structurally cannot carry, because it never inspects the per-item statuses on the caller's behalf. This is not about exposing a structured per-item error list: the synchronous writer's return is also a single error. It is about that single error meaning "a document in this batch was not stored," whereas Flush's single error means only "the HTTP call failed" — a distinction the caller cannot recover from Flush alone.BulkIndexer is a single worker-pooled object; Flush "drains all currently queued items," and "items added concurrently with Flush may or may not be included." Under the blocking batcher (§4.2) many partition-consumer goroutines call the writer at once, so caller A's Flush would flush caller B's queued items and fire B's callbacks — there is no per-call boundary. Recovering one requires either (a) a fresh BulkIndexer per Bulk call — spinning up worker goroutines and a channel for a single HTTP POST — or (b) serializing all callers behind a lock and reconciling each call's failures out of the callback stream. Both add machinery and concurrency hazard that a single request does not warrant.A synchronous _bulk is fundamentally one request → one response → inspect the items. Re-implementing esutil's small NDJSON encoder (~15 lines) and issuing that one blocking POST /_bulk over the shared transport — then parsing the response into owned types — is simpler, has no worker pool, no callbacks, and no shared buffer, and yields exactly the per-batch verdict the writer contract needs. That is what esclient.SyncBulkWriter (M2) does, and why it is a peer of the async BulkIndexer rather than a wrapper around it.
write_mode: async remains the default in the alpha stage of the es.write.synchronous feature gate (§4.5); existing deployments see no behavior change.write_mode: sync together with message_marking.after: true and a blocking pipeline batcher, removing the fire-and-forget batch processor (§4.2/§4.5). Over the gate's alpha→beta→stable lifecycle the default moves to sync, giving operators a release window to adopt the pipeline shape before it becomes default.write_mode: async explicitly (overrides the gate). No data migration.Q1 — Exporter batcher (now) vs. receiver-level batching (later)?
The blocking-and-fan-out semantics of queue.wait_for_result + queue.batch are confirmed in the exporterhelper v0.155.0 source and its tests (§4.2), so option 1 is safe to ship as config-only. The genuinely open item is whether to also invest in receiver-level per-partition batching (§4.2 option 2) — simpler offsets, but needs an upstream contrib change or a Jaeger receiver wrapper. Recommendation: ship option 1; open an upstream issue for option 2 and adopt it if accepted. Pin the exporterhelper version, since the batcher internals are internal/ and could change.
Q2 — Startup validation of the sync + batch-processor misconfiguration.
write_mode: sync with a fire-and-forget batch processor still in the pipeline silently defeats the guarantee (§4.5). Can the exporter detect this at startup and warn/fail? The exporter does not see the full pipeline graph, so detection may not be feasible from inside the component; the fallback is prominent documentation. Recommendation: investigate a collector-level check; ship docs regardless.
Q3 — Interface doc for WriteTraces async allowance.
The tracestore.Writer doc comment should state explicitly whether an implementation may be asynchronous. Today the comment implies synchronous semantics that only Cassandra/ClickHouse honor. Recommendation: document that a conformant WriteTraces returns write errors synchronously, and that async mode is a deliberate, documented deviation trading the guarantee for throughput on unbatched ingest (§3.1) — not an accident.
Q4 — Deterministic _id for span dedup? → Resolved: in scope (§4.7).
Originally deferred to RFC 0004. Reconsidered: synchronous at-least-once makes retries routine, so idempotent writes are a prerequisite for production safety, not a nicety. Decision: adopt a deterministic content-hash _id for spans (matching Cassandra's span_hash, not traceID+spanID+startTime — see §4.7 for why the shared-span model requires a content hash), delivered as its own milestone/PR ahead of the default flip to sync. Remaining sub-question: the exact hash input (reuse Cassandra's gob-of-zeroed-span, or a cheaper canonical form) — settle during M4. The op_type: create 409-as-idempotent-success handling (§4.7) is the one non-obvious implementation point.
Q6 — Dead-letter sink: separate index or Kafka topic? (§4.8)
Autonomous poison-pill disposal (§4.8) needs a durable sink. A *-dead-letter index/data stream on the same cluster is simplest (no new dependency, reuses the ES client) but shares the cluster's fate; a dead-letter Kafka topic is more decoupled but reintroduces a Kafka dependency on the write side. Recommendation: default to a dead-letter index (available everywhere the writer already is), leave a Kafka-topic sink as a configurable alternative for Kafka-centric deployments; decide during M7.
Q5 — Is async ever removed, or retained for direct ingest?
The feature gate flips the default to sync, but async remains the better fit for unbatched direct ingest (§3.1). Two end-states: (a) at "stable", remove async entirely and require the blocking batcher even for direct ingest (accepting its flush_timeout latency); or (b) keep async as a permanent non-default option for the collector→storage path, only deprecating it for the Kafka ingester. Recommendation: decide at the beta→stable boundary once operational data exists; lean toward (b), since async is genuinely correct for that topology.
The work is decomposed into independently shippable milestones, each leaving the tree green with no user-visible behavior change until the mode is explicitly opted into. Milestone status is tracked here (✅ Done, with the delivering PR) as work lands.
Every milestone that introduces functional code must carry end-to-end coverage for that code in the same milestone — not defer it to a later validation step. The internal/storage/integration suite runs against real ES 7–9 / OS 1–3 containers in CI, so a new esclient primitive is exercised there through the owned client, and a wired write-path change is validated by running that suite in the new mode. A milestone whose only tests are httptest mocks does not count as covered.
core.Writer.WriteSpan(startTime, *span) with a batch, error-returning WriteSpans(ctx, []dbmodel.Span) error (§4.1) and route TraceWriter.WriteTraces through it; the async SpanWriter implements it by enqueuing each span into the existing esutil.BulkIndexer and returning nil (unchanged behavior — an async enqueue cannot fail synchronously). Clarify the tracestore.Writer doc comment on synchronous-error expectations and the documented async deviation (Q3). Exit: WriteTraces returns whatever WriteSpans returns; async behavior and wire format unchanged; the batch entry point that later milestones swap the sink under exists and is unit-tested.esclient (peer of M6's async BulkWriter, over the same transport — §5): one _bulk round-trip per size-bounded (max_bytes) chunk, parsing the response and returning transport + item-level errors from owned response types. Exit: byte-cap + item-error propagation proven by unit tests, and an ES/OS integration test (in the real ES 7–9 / OS 1–3 matrix) that round-trips documents to prove durability and forces a real item-level rejection to prove the error propagates — the primitive is exercised against a live backend the milestone it lands, not only via httptest.write_mode config + feature gate. Add write_mode: async|sync (explicit wins) and the es.write.synchronous feature gate governing the unset default (alpha: async — §4.5, §6); reuse max_bytes as chunk cap. Exit: config parse/validate + defaults + gate tests._id (idempotent writes) — §4.7. Give each span a deterministic content-hash _id (matching Cassandra's span_hash, not traceID+spanID+startTime), so a re-sent identical span upserts rather than duplicating. Handle the op_type: create (data-stream) case where a duplicate _id returns 409 by treating a 409 on our own _id as idempotent success. Its own PR — self-contained, independently valuable (removes duplicate-on-retry for async too), and wire-observable. Lands before the default flips to sync. Exit: writing the same span twice yields exactly one document, proven by an ES/OS integration test on both op_type: index and op_type: create; request snapshots updated for the new _id.TraceWriter.WriteTraces → WriteSpans → synchronous bulk; append service/operation docs into the same request; update the service cache only after success (§4.3). Exit: the full ES/OS trace-storage integration suite passes with write_mode: sync (parametrized alongside the async run) across the ES 7–9 / OS 1–3 matrix, so the wired write/read path is validated end-to-end in the new mode; plus a fault-injection integration test asserting a failing bulk makes WriteTraces return an error (and, on the ingester, the Kafka offset is not committed).queue.wait_for_result: true + queue.batch on the storage exporter, the pipeline batch processor removed, and the Kafka receiver's message_marking.after: true. The exporterhelper fan-out/blocking is already proven by its own tests (§4.2), so this step is mostly an example config + a Jaeger-side integration test. The one piece of new code: jaeger_storage_exporter.start() hard-fails when write_mode: sync and queue.batch.max_size > bulk_processing.max_bytes (§4.5). Exit: an ingester config that nacks every message in a failed batch and advances no offset; a mismatched max_size/max_bytes config fails startup._bulk failures into transient (retry the batch) vs. terminal (poison); route terminal items to a durable dead-letter sink (default: a *-dead-letter index) with the reason attached, then advance the offset. Emit a dead_lettered counter + log. This is what makes write_mode: sync production-safe: no head-of-line blocking and no silent loss. Required before the default flips to sync (beta). Exit: an integration test injecting a permanently-rejected document (e.g. a mapping conflict) shows the poison doc landing in the dead-letter sink, the good docs durable, the batch returning success, and the offset advancing — the partition never stalls; a transient failure still holds the offset._id + blocking batcher + poison dead-lettering + message_marking.after: true: kill/reject ES mid-stream, assert no offset advance and full recovery on ES return with no duplicates (idempotent _id) and no stall (poison dead-lettered). Exit: at-least-once demonstrated end-to-end, autonomously.write_mode and the co-required settings (message_marking.after: true, blocking batcher, no batch processor — §4.5), the deterministic-_id/idempotency behavior (§4.7), the dead-letter sink and its config (§4.8), upstream-sizing guidance, the durability-vs-searchability note (§2.2), and the acking rationale (§4.6). Exit: configuration guide updated.ConsumeTraces (§4.2 option 2). Independent of M1–M9; adopt if accepted.spanstore.Writer interface (WriteSpan, one span) (db6587b9, 2017-02-15) · PR #200 — original synchronous ES writer (f3717836, 2017-06-19) · PR #656 — "Use elasticsearch bulk API" (the sync→async switch) (e52ecffb, 2018-01-23) · PR #6090 — BulkProcessing config struct · PR #8944 — M6, owned esutil.BulkIndexerinternal/storage/v2/cassandra/tracestore/writer.go) — synchronous per-span, errors.Joinrefresh parameterasync_insert, wait_for_async_insert)exporterhelper — queue/batch settings (wait_for_result, batch) — blocking + fan-out verified in internal/queue/memory_queue.go and internal/queuebatch/partition_batcher.go (v0.155.0)kafkareceiver — message_marking, franz-go consumer — mark-before/after and per-partition concurrency verified in consumer_franz.go (v0.155.0)esutil.BulkIndexer