feat(es): Add optional poison-pill drop for synchronous writes

First slice of RFC 0007 M5 (poison-pill handling). The synchronous bulk writer
now classifies each rejected _bulk item as transient (429 / 5xx / malformed
response — retry the batch) or terminal (a 4xx poison pill the backend will
reject identically on replay), and a new config selects what to do with terminal
rejections:

  elasticsearch:
    write_mode: sync
    poison_pill_handling: fail | drop   # default: fail

- fail (default, unchanged): any rejection fails the batch, so it is retried
  forever — safe but head-of-line-blocks a Kafka partition until the doc is fixed.
- drop: terminal rejections are logged, counted, and discarded so the batch
  completes and the offset advances; transient failures still fail the batch and
  retry. This is the "some operators are fine losing a rare malformed span"
  option and needs no extra pipeline.

The remaining M5 disposition, dead_letter (re-emit poison onto a separate OTel
pipeline via a traces->traces connector), is not implemented here; the RFC §4.8
and the M5 entry record it as the next slice, including the open component-shape
question to settle by prototype.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Signed-off-by: Yuri Shkuro <github@ysh.us>

Author: yurishkuro

docs/rfc/0007-synchronous-elasticsearch-writes.md

@@ -375,7 +375,12 @@ A **poison document** fails deterministically on every attempt (mapping conflict
 
 1. **Classify each item's `_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.
 2. **Transient → real error → retry the whole batch.** The offset is held; `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.
-3. **Terminal → dead-letter, then advance.** Re-emit each terminal item, with its rejection reason attached, onto a **separate OTel dead-letter pipeline** — not a sink Jaeger implements itself. The synchronous storage write is offered as a traces→traces **connector** (Jaeger already ships `forwardconnector`/`spanmetricsconnector`, so this is an established primitive) that writes the batch to storage and taps only the rejected spans to its connector output. The operator wires that dead-letter pipeline to any standard exporter — a Kafka exporter for a Kafka DLQ, `file`, `debug`/logging, or OTLP to another collector — so the sink is entirely their choice and there is **no dead-letter sink code in Jaeger**. Once the poison items are accepted by the dead-letter pipeline, the batch is **complete** — good items durable, poison preserved out-of-band — so the write returns success and the offset advances. The partition never blocks; no data is silently dropped; the poison is inspectable in whatever sink the operator chose.
+3. **Terminal → dispose, then advance.** The disposition is operator-selectable (`poison_pill_handling`), because tolerance for losing a malformed span differs:
+   - `fail` (default) — surface the rejection as a batch error, holding the offset. Safe but head-of-line-blocks the partition until the document is fixed; appropriate only where no span may ever be dropped and a human watches.
+   - `drop` — discard the poison document (logged and counted), completing the batch so the offset advances. The simplest way to guarantee the partition never stalls, for operators fine with losing a rare malformed span. No pipeline needed.
+   - `dead_letter` — re-emit each terminal item, with its reason attached, onto a **separate OTel dead-letter pipeline** rather than a sink Jaeger implements. The synchronous storage write is offered as a traces→traces **connector** (Jaeger already ships `forwardconnector`/`spanmetricsconnector`, so this is an established primitive) that writes the batch and taps only the rejected spans to its connector output; the operator wires that pipeline to any standard exporter — a Kafka exporter for a Kafka DLQ, `file`, `debug`/logging, or OTLP — so the sink is their choice with **no dead-letter sink code in Jaeger**.
+
+   Under `drop` and `dead_letter` the batch is **complete** once the poison is disposed — good items durable, poison discarded or preserved out-of-band — so the write returns success and the offset advances. The partition never blocks.
 4. **If the dead-letter pipeline rejects the items** (its exporter returns an error) → treat that as a transient failure, return an error, and hold the offset (don't advance on unconfirmed dead-lettering). The chosen dead-letter sink should be at least as available as the primary; a Kafka topic or a local `file` exporter satisfies this.
 5. **Observability, not intervention.** Emit a `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.
 
@@ -447,7 +452,7 @@ The work is decomposed into independently shippable milestones, each leaving the
 - **M2 — Synchronous bulk primitive. ✅ Done ([#8992](https://github.com/jaegertracing/jaeger/pull/8992)).** Add a new blocking bulk writer on `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.
 - **M3 — Deterministic span `_id` (idempotent writes) — §4.7. ✅ Done ([#9094](https://github.com/jaegertracing/jaeger/pull/9094)).** 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 the async path shipping today), and wire-observable. Lands **before** the sync wiring it makes safe (M4). *Exit:* writing the same span twice yields exactly one document, proven by an ES/OS integration test on `op_type: index` (the wired path); the `op_type: create` 409-as-idempotent-success path is covered by an `esclient` bulk unit test plus the live-409 assertion in the M2 integration test, with a data-stream end-to-end test to follow once data-stream rotation is wired; request snapshots updated for the new `_id`.
 - **M4 — Wire sync path + `write_mode` config. ✅ Done (refactor [#9097](https://github.com/jaegertracing/jaeger/pull/9097), then [#9093](https://github.com/jaegertracing/jaeger/pull/9093)).** First a no-behavior-change refactor (#9097) collapses the bulk-write API to a single `esclient.BatchWriter` (`WriteBatch(ctx, items) error`) that both the async indexer and the synchronous writer implement, so `WriteSpans` assembles the batch's documents once and writes them the same way in every mode — the write mode is not the writer's concern. Then #9093 introduces the `write_mode: async|sync` config (default `async`; explicit wins; unrecognized rejected by whole-config validation) **together with** the factory selecting the sync or async `BatchWriter` from it — not as a standalone knob (see the guiding constraint above). The service cache is marked only after a successful write (§4.3); `max_bytes` is the sync safety-net chunk cap. *Exit:* config parse/validate/default tests and writer unit tests (one batch write, error propagation, cache-after-durable); 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; plus a fault-injection test asserting a failing `_bulk` makes `WriteTraces` return an error. (The Kafka offset-not-committed assertion belongs with the ingester validation, M6.)
-- **M5 — Autonomous poison-pill handling (dead-letter) — §4.8.** Classify per-item `_bulk` failures into transient (retry the batch) vs. terminal (poison). Re-emit terminal items, with the rejection reason attached, onto a **separate OTel dead-letter pipeline** rather than a sink Jaeger implements: the sync storage write is offered as a traces→traces **connector** that taps rejected spans to its output, so the dead-letter sink is any standard exporter (a Kafka exporter, `file`, `debug`/logging, OTLP) — **no custom sink code in Jaeger**. Emit a `dead_lettered` counter + log, then advance the offset. 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`. Carries an open component-shape question (exporter-with-tap vs. connector, and its interaction with the blocking batcher's sending queue — §4.8) to settle by prototype. *Exit:* an integration test injecting a permanently-rejected document (e.g. a mapping conflict) shows the poison span emitted to the dead-letter pipeline, the good docs durable, the batch returning success, and the offset advancing — the partition never stalls; a transient failure still holds the offset.
+- **M5 — Poison-pill handling — §4.8.** Classify per-item `_bulk` failures into transient (retry the batch) vs. terminal (poison), and let the operator choose the terminal disposition: `fail` (default — retry forever, never drop), `drop` (discard the poison doc so the offset advances), or `dead_letter` (re-emit it onto a separate OTel pipeline). **Delivered so far:** the classification and the `poison_pill_handling: fail|drop` config for the synchronous writer (this PR) — `drop` is the "some operators are fine losing a rare malformed span" option and needs no pipeline. **Remaining:** `dead_letter`, which re-emits terminal items onto a **dead-letter pipeline** via a traces→traces **connector** so the sink is any standard exporter (Kafka, `file`, `debug`/logging, OTLP) with **no custom sink code in Jaeger**; it carries an open component-shape question (exporter-with-tap vs. connector, and its interaction with the blocking batcher's sending queue — §4.8) to settle by prototype. Poison handling is what makes `write_mode: sync` production-safe: no head-of-line blocking and no silent loss. Required before the default flips to `sync`. *Exit:* an integration test injecting a permanently-rejected document (e.g. a mapping conflict) shows, under `drop`, the poison discarded and the batch completing so the offset advances (partition never stalls) and, under `dead_letter`, the poison span emitted to the dead-letter pipeline; a transient failure still holds the offset.
 - **M6 — Ingester at-least-once: blocking batcher + end-to-end validation.** Provide the ingester pipeline shape that couples the Kafka offset to durability — `queue.wait_for_result: true` + `queue.batch` on the storage exporter, the pipeline `batch` processor removed, the Kafka receiver's `message_marking.after: true` — as an example config exercised in CI (the `exporterhelper` fan-out/blocking itself is already proven by its own tests, §4.2). Then validate at-least-once end-to-end on the Kafka path with the sync writer + deterministic `_id` + blocking batcher + poison dead-lettering (M5): 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; the example config nacks every message in a failed batch and advances no offset. (A standalone byte-sizing warning — local exporter hardening, not this milestone — shipped in [#9098](https://github.com/jaegertracing/jaeger/pull/9098).)
 - **M7 — Docs.** Document `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 pipeline and how to attach a sink (§4.8), upstream-sizing guidance, the durability-vs-searchability note (§2.2), and the acking rationale (§4.6). *Exit:* configuration guide updated.
 - **M8 — (Optional, upstream/custom) Throughput-decoupled consumer.** The per-partition receiver caps batch size at partition count (§4.6.1). Provide a consumer whose write parallelism is **independent of partition count** — a worker pool feeding the blocking batcher, plus a contiguous-offset serializer for safe out-of-order completion (the Jaeger v1 design, §4.6.1) — as a custom Jaeger Kafka receiver or an upstream contrib change. Subsumes the narrower per-partition coalescing idea (§4.2 option 2). Independent of M1–M7; adopt if/when throughput demands it.

internal/storage/elasticsearch/config/config.go

@@ -43,6 +43,23 @@ const (
 	WriteModeSync WriteMode = "sync"
 )
 
+// PoisonHandling selects what a synchronous writer does with a document the backend
+// rejects *terminally* — a "poison pill" (mapping conflict, malformed field, other
+// 4xx) that will fail identically on every retry. It has no effect in async mode.
+type PoisonHandling string
+
+const (
+	// PoisonFail fails the whole batch on any terminal rejection, so the write is
+	// retried indefinitely. This is the default: it never drops data, at the cost of
+	// head-of-line blocking on the Kafka ingest path until the document is fixed.
+	PoisonFail PoisonHandling = "fail"
+	// PoisonDrop discards a terminally-rejected document — logging and counting it —
+	// so the batch completes and the Kafka offset advances. Transient failures
+	// (429 / 5xx / transport) still fail the batch and are retried. Use this when
+	// dropping a rare malformed span is preferable to stalling a partition.
+	PoisonDrop PoisonHandling = "drop"
+)
+
 // IndexOptions describes the index format and rollover frequency
 type IndexOptions struct {
 	// Priority contains the priority of index template (ESv8 only).
@@ -175,6 +192,11 @@ type Configuration struct {
 	//     respecting the tracestore.Writer contract. See RFC 0007 for background.
 	// When empty, it defaults to "async". See config.EffectiveWriteMode().
 	WriteMode WriteMode `mapstructure:"write_mode"`
+	// PoisonPillHandling selects what the synchronous writer does with a document the
+	// backend rejects terminally: "fail" (default — retry forever, never drop) or
+	// "drop" (discard the poison doc so the offset advances). No effect in async mode.
+	// When empty it defaults to "fail". See config.EffectivePoisonHandling().
+	PoisonPillHandling PoisonHandling `mapstructure:"poison_pill_handling"`
 	// Version contains the backend version number (e.g. 7, 8, 9 for Elasticsearch,
 	// 101, 102, 103 for OpenSearch). If 0, it will be auto-detected from the server.
 	Version uint `mapstructure:"version"`
@@ -435,6 +457,9 @@ func (c *Configuration) ApplyDefaults(source *Configuration) {
 	if c.WriteMode == "" {
 		c.WriteMode = source.WriteMode
 	}
+	if c.PoisonPillHandling == "" {
+		c.PoisonPillHandling = source.PoisonPillHandling
+	}
 	if !c.Tags.AllAsFields {
 		c.Tags.AllAsFields = source.Tags.AllAsFields
 	}
@@ -607,6 +632,10 @@ func (c *Configuration) Validate() error {
 		return err
 	}
 
+	if err := validatePoisonHandling(c.PoisonPillHandling); err != nil {
+		return err
+	}
+
 	return validateLogLevel(c.LogLevel)
 }
 
@@ -619,6 +648,26 @@ func (c *Configuration) EffectiveWriteMode() WriteMode {
 	return WriteModeAsync
 }
 
+// EffectivePoisonHandling resolves the poison-pill policy Jaeger should use: the
+// explicit PoisonPillHandling from config, or PoisonFail when it is unset.
+func (c *Configuration) EffectivePoisonHandling() PoisonHandling {
+	if c.PoisonPillHandling != "" {
+		return c.PoisonPillHandling
+	}
+	return PoisonFail
+}
+
+// validatePoisonHandling rejects an unrecognized poison_pill_handling. An empty
+// value is allowed and resolves to the default (PoisonFail).
+func validatePoisonHandling(mode PoisonHandling) error {
+	switch mode {
+	case "", PoisonFail, PoisonDrop:
+		return nil
+	default:
+		return fmt.Errorf("unrecognized poison_pill_handling %q: valid values are %q and %q", mode, PoisonFail, PoisonDrop)
+	}
+}
+
 // validateWriteMode rejects an unrecognized write_mode. An empty value is allowed
 // and resolves to the default (WriteModeAsync). It mirrors validateLogLevel:
 // write_mode carries no govalidator struct tag, so the whole-config Validate must

internal/storage/elasticsearch/config/config_test.go

@@ -381,6 +381,15 @@ func TestValidate(t *testing.T) {
 			config:        &Configuration{Servers: []string{"localhost:8000/dummyserver"}, WriteMode: "eventually"},
 			expectedError: `unrecognized write_mode "eventually"`,
 		},
+		{
+			name:   "poison_pill_handling drop accepted",
+			config: &Configuration{Servers: []string{"localhost:8000/dummyserver"}, PoisonPillHandling: PoisonDrop},
+		},
+		{
+			name:          "unrecognized poison_pill_handling rejected",
+			config:        &Configuration{Servers: []string{"localhost:8000/dummyserver"}, PoisonPillHandling: "quarantine"},
+			expectedError: `unrecognized poison_pill_handling "quarantine"`,
+		},
 		{
 			name: "sniffing.use_https set is rejected",
 			config: &Configuration{
@@ -637,6 +646,23 @@ func TestEffectiveWriteMode(t *testing.T) {
 	assert.Equal(t, WriteModeSync, (&Configuration{WriteMode: WriteModeSync}).EffectiveWriteMode())
 }
 
+func TestEffectivePoisonHandling(t *testing.T) {
+	assert.Equal(t, PoisonFail, (&Configuration{}).EffectivePoisonHandling(), "unset defaults to fail")
+	assert.Equal(t, PoisonFail, (&Configuration{PoisonPillHandling: PoisonFail}).EffectivePoisonHandling())
+	assert.Equal(t, PoisonDrop, (&Configuration{PoisonPillHandling: PoisonDrop}).EffectivePoisonHandling())
+}
+
+func TestApplyDefaultsPoisonHandling(t *testing.T) {
+	source := &Configuration{PoisonPillHandling: PoisonDrop}
+	target := &Configuration{}
+	target.ApplyDefaults(source)
+	assert.Equal(t, PoisonDrop, target.PoisonPillHandling, "unset target inherits the source policy")
+
+	explicit := &Configuration{PoisonPillHandling: PoisonFail}
+	explicit.ApplyDefaults(source)
+	assert.Equal(t, PoisonFail, explicit.PoisonPillHandling, "explicit target policy is preserved")
+}
+
 func TestApplyForIndexPrefix(t *testing.T) {
 	tests := []struct {
 		testName     string