[BUG] Duplicate GUIDs in the TNI Spark Streaming → Hudi Pipeline (DOC collection type)

### Bug Description

# Investigation Brief: Duplicate GUIDs in the TNI Spark Streaming → Hudi Pipeline (DOC collection type)

## 1. Context / Architecture

TNI is a Spark Structured Streaming job consuming a Kinesis stream on a 10-minute batch interval. Per micro-batch:
- Records read are bounded by a **message limit** + **polling interval**.
- For each Kinesis control message, the job reads the referenced `ContentMetadata` JSON and Avro content from S3.
- After transformation, records are written to a **Hudi MERGE_ON_READ** table.

**Confirmed Hudi write configuration** (`NovusHudiOperations.scala` `hudiOptions`):
- `RECORDKEY_FIELD = guid`, `PARTITIONPATH_FIELD = collectionName`, `PRECOMBINE_FIELD = operationTime`
- `TABLE_TYPE = MERGE_ON_READ`; inline compaction `TIME_ELAPSED` every 6h; cleaner `KEEP_LATEST_COMMITS = 447`; `ASYNC_CLEAN = false`
- `hoodie.metadata.enable = false` → no metadata-table file listing; index defaults to **partition-scoped**
- `hoodie.write.table.version = 6` (pinned for downstream Hudi-0.x readers) + `auto.upgrade = false`
- `ROLLBACK_USING_MARKERS_ENABLE` — **was `false`** (disabled 2022-08-05, commit `4c0ca456`, no recorded rationale, under `metadata.enable=true`, pre-v6)

**Write choke point:** every insert/upsert path — `bulkAdd`, `bulkAdds`, `promoteWithReset`, `reinit`, `promoteDemoteAdd(s)`, streaming incremental — funnels through the single method `NovusHudiOperations.upsert()`. Only `delete()` bypasses it. Any write-path fix belongs in `upsert()`, not at an upstream union site.

**Write pattern:** per-message DataFrames are `unionAll`'d into one mixed-`collectionName` DataFrame and written with **one** Hudi upsert; Hudi routes each row to its `collectionName` partition and upserts by `guid` (partition-scoped index). A single delta commit can therefore span multiple partitions.

**High-level flow:**
```
R->>D: read last checkpoint seq#
R->>K: GetRecords (AFTER_SEQUENCE_NUMBER | TRIM_HORIZON)
K-->>R: JSON control messages (+ millisBehindLatest)
R-->>A: DStream RDD of KinesisInputNovusMessage
A->>D: lockOrFail (IS_LOCKED per content type)
A->>A: group by collection, parse operationType
A->>S: read control-metadata JSON (InputReader.readJsonFile)
A->>S: read Avro content (spark.read.format("avro"))
A->>H: UPSERT (recordKey=guid, part=collectionName) / soft-delete
alt all collections in batch succeeded
A->>D: updateCheckpoint (max seq# → CHECKPOINT_TIMESTAMP)
else any failure
A-->>A: log P5, do NOT advance checkpoint (reprocess next run)
A->>D: unlock (finally)
```

## 2. Observed symptom

Duplicate `(guid, collectionName)` rows in the `novusdoc` table for the **DOC** collection type. QA and Prod consume the **same** Prod Kinesis stream, yet the **duplicate collections differ** between QA and Prod. Sample (QA, `w_blc_edgar_other_01`): two physical copies of each guid sharing identical `_hoodie_commit_time`, `sequenceid`, `operationtime`, differing only by `_hoodie_file_name` (file groups `3250`/`3251`).

**Scale (QA, verified via Athena on `a206760-tni-qa-db.novusdoc_rt`):** ~720 duplicate pairs across 9 DOC collections — **708 same-commit (signature #2) / 12 cross-commit (signature #1)**.

Sample Kinesis message:
```
KinesisCustomOutputData({ "sequenceId": 264096479, "collectionName": "w_cs_ne2", "collectionType": "DOC",
"collectionId": 2217, "operationType": "PROMOTE", "operationTime": 1785913614134, "operationStartStage": 0,
"operationEndStage": 770421, "contentMetadataObjectName": "s3://.../w_cs_ne2/264096479_.../ContentMetadata",
"ingestTime": "1785913811608", "totalContentSize": 198996, "totalGuidsUploaded": 13}, , 0)
```

## 3. The two signatures — and the test that distinguishes them

| | Signature #1 (cross-commit) | Signature #2 (same-commit) |
|---|---|---|
| `_hoodie_commit_time` of the copies | **different** (minutes apart) | **identical** |
| meaning | same batch committed **twice** | one commit wrote the key into **2+ file groups** |
| per-batch counters (`sequenceid`/`operationtime`/`operationendstage`) | identical across both copies ⇒ **replay by one writer**; distinct ⇒ **concurrent writers** | identical (one commit) |
| QA share | 12 | 708 (dominant) |

**Batch-counter test (triage rule):** for any cross-commit pair, compare `sequenceid`/`operationtime`/`operationendstage` across the two commits. Identical ⇒ single-writer replay (fix = replay-prevention). Distinct ⇒ concurrent writers (fix = OCC).

## 4. Complete scenario catalog

| # | Scenario | Evidence / status |
|---|---|---|
| **S1** | Intra-batch dup (same key twice in one micro-batch) | Not the cause — Hudi `combine.before.upsert` collapses it (repro-confirmed) |
| **S2** | PROMOTE/DEMOTE cross-partition relocation (guid → new `collectionName`) | Root cause A — old copy in prior partition not found by partition-scoped index → duplicate INSERT; invisible to `DuplicateVerification` (groups by `(guid, collectionName)`) |
| **S3** | Zero-timestamp tombstone | Root cause B — `softDelete` used `operationTime=0L` → always loses precombine → soft-delete no-ops → old row stays live on every reload (`bulkAdd`/`bulkAdds`/`promoteWithReset`/`reinit`) |
| **S4** | **Same-commit multi-file-group split (dominant)** | Athena-confirmed retry/re-bucketing origination. `other_01`: 544 keys near-mirror in FG `3250`/`3251`. `33_02`: 56/57 from **one commit `20260605084155211`** (hub FG `7b66daa1`). That commit also duplicated `UK_SMG_EUIMAGES_03` + `10q_01` — a single cross-collection culprit write event |
| **S5** | Cross-commit whole-batch replay (signature #1) | Confirmed replay. `33_02`: 8 keys, commits `20260604180024505` & `...181909526` (~19 min apart), **identical** `sequenceid=259023002`/`operationtime=1780594587500`/`operationendstage=127417` across all 8 guids and both commits ⇒ one batch committed twice by one writer. **Concurrent writers ruled out** |
| **S6** | Concurrent writers | Ruled out for observed data (no distinct-batch-counter pairs seen) |
| **S7** | Existing committed dups (~720) | Already in the table; not preventable by any write-config change; compaction/cleaner cannot remove them |
| **S8** | True partial-write orphan files (uncommitted) | The mechanism behind S4-type dups; addressable by markers + failed-writes cleaning |

## 5. Confirmed diagnostics (Athena, read-only)

- **`other_01` near-mirror:** FG `3250` (577 rows) and `3251` (573 rows) share **identical min guid `I002f9f98…` and max guid `Iff3553a6…`**; **544 guids in both (~94%)**, 33 only-3250, 29 only-3251. A 94%-identical, same-range pair ⇒ a **whole-insert-bucket retry** copied one file group (Mechanism A/C), not a bloom/bucket range scatter. The "275 commits all hit 3250/3251" pattern = one origination + later updates to those 544 keys routing into both pre-existing file groups (metadata off + partition-scoped index).
- **`23mm` / `33_02` shape:** small per-key sets around a **hub** file group (`23mm` hub `3bb7805c`, 16 keys; `33_02` hub `7b66daa1`). `33_02`'s 56 same-commit dups concentrate in **one commit** ⇒ same retry-origination family as `other_01`, **not** a diffuse index-tagging scatter.
- **Method validated:** the same-/cross-commit split reproduced independently-derived counts exactly (`33_02` = 57 sig #2 / 8 sig #1).

## 6. Hypotheses ruled out

- Checkpoint-skip / batch replay (simulated — no repro).
- Duplicate Kinesis message delivery (same sequence ID/partition key twice).
- Duplicate GUID within a single Avro source file.
- Intra-batch combine failure (combine works).
- Kinesis redelivery of an already-committed key.
- Concurrent-writer OCC race (batch-counter test ⇒ replay, not concurrency).

## 7. Verdict on the "partial write + re-run → orphan files" RCA

**Confirmed as the leading mechanism for signature #2's dominant cluster — but not the sole cause, and it is a prevention story, not a remediation one.** UPSERT dedupes only against **committed + indexed** records; a retried/speculative Spark task's orphaned output file (under `ROLLBACK_USING_MARKERS_ENABLE=false`) is never cleaned, so both attempts' files survive under one commit → the same guid twice. The `other_01` 94% near-mirror is exactly this shape. It does **not** explain signature #1 (two different commits, minutes apart — a replay mechanism, §S5).

**QA/Prod divergence:** QA and Prod are **two fully independent single-writer runtimes** (separate EMR clusters/JVMs, separate DynamoDB checkpoint tables per `APP_NAME`/env, byte-identical `hudiOptions`) consuming the shared stream at their own pace — so each environment's affected collections are a function of *when it hit a transient failure*, not of the stream.

## 8. Mechanisms behind the replays (signature #1)

- **Mechanism B — Kinesis iterator expiry:** `KinesisCustomConsumer` advances the in-memory `globalShardIterator` as records are read (ahead of the DynamoDB checkpoint, which advances only after a *full* batch succeeds — `TNIStreamingApp.processInterval`). On `ExpiredIteratorException` it resets `globalShardIterator=""`, forcing `onStart()` to replay from the **committed checkpoint** → re-reads/re-upserts records already pushed → cross-commit dup.
- **Mechanism C — retry-wrapper re-save:** `upsert()`'s `retry()` block wraps `.save()` **plus** post-save `reportMetric`/`emitBuiltinHudiMetricsCompat`; a transient throw *after* a successful `.save()` re-runs `.save()` → a second commit minutes later.
- **Not concurrent writers** (batch-counter test).

## 9. Why compaction/cleanup do NOT remove these dups

Dedup is a **write-time** guarantee (index → precombine → file-group assignment). Compaction merges log→base **within one file group** and never crosses file groups or runs the index; the cleaner drops **old versions** per file group and never dedupes keys or removes a latest slice. So committed cross-file-group dups (S4/S5/S7) survive both (proven: the `other_01` dups carry a 2026-08-02 compaction instant and persist). S3 survives because compaction faithfully applies the *broken* precombine. Clustering also won't help (it preserves records).

**Exception:** uncommitted orphans (S8) *are* addressable by the cleanup family — **marker-based rollback** (Fix C) and **`cleaner.policy.failed.writes=EAGER`** (Fix G) — because those files are uncommitted; that is not the regular `KEEP_LATEST_COMMITS` cleaner or compaction.

## 10. Required dedup behaviors (invariants any fix must preserve)

1. **Cross-collection same guid = NOT a duplicate** — same guid in collection A and B → both rows preserved.
2. **Within-collection same guid = duplicate.**
3. Within-collection, same guid, different `operationTime` → keep max `operationTime` (1 row).
4. Within-collection, same guid, same `operationTime` → collapse to 1 row.
5. Mixed batch (several collections, some dup) → correct per-collection collapse.
6. Live tombstone (`delete=true`, fresh `operationTime`) vs older live row for same key → **tombstone wins** (must interoperate with the S3 soft-delete fix).

im currently having issue with S4 and S5 and S8 as per my analysis is it the reason for duplicates

### Environment

**Hudi version:** 1.0.2
**Query engine:** (Spark/Flink/Trino etc) Spark
**Relevant configs:**

### Logs and Stack Trace

# Duplicate-GUID Pattern Examples (CSV-Backed)

Concrete, data-backed examples for each duplicate-GUID mechanism, pulled directly from:
- `duplicate_guid_field_details_qa_sig2only.csv`
- `duplicate_guid_field_details_qa_sig2only_202607.csv`

Both files are scoped to Signature #2 (same-`_hoodie_commit_time` duplicates) only — neither
contains a Signature #1 (cross-commit) example, so the S5 section below is illustrative /
sourced from the original investigation rather than these two files.

CSV schema: `_hoodie_commit_time,_hoodie_commit_seqno,_hoodie_record_key,_hoodie_partition_path,_hoodie_file_name,guid,collectionname,sequenceid,operationtime,operationstartstage,operationendstage,delete`

Hudi data-file naming convention referenced throughout: `__.parquet`,
where `writeToken = --` (`FSUtils.makeWriteToken`) and
`taskAttemptId` is Spark's cluster-wide monotonically increasing task attempt counter (not a
small per-task retry index).

## 1. Whole-bucket mirror (S4 dominant, sub-pattern A)

**Root cause:** task/stage re-execution (shuffle-service off + decommissioning on; retries
un-tuned; speculation off) re-materializes an insert bucket into a second, freshly-generated
`fileId`. Marker-based reconciliation is defeated by `rollback.using.markers=false` +
`metadata.enable=false`, so the mirror survives inside the same commit.

**Source:** `duplicate_guid_field_details_qa_sig2only_202607.csv`, `w_blc_edgar_other_01`, rows 2–9
(raw rows, all fields):

```
_hoodie_commit_time,_hoodie_commit_seqno,_hoodie_record_key,_hoodie_partition_path,_hoodie_file_name,guid,collectionname,sequenceid,operationtime,operationstartstage,operationendstage,delete
20260723205405966,20260723205405966_103_79016,I48dcc76b0e4c407bb4623377579f92bc,w_blc_edgar_other_01,d3b8b8ae-b9e4-4598-acd3-a998f8b0ccba-0_3251-772-775680_20260802145031626.parquet,I48dcc76b0e4c407bb4623377579f92bc,w_blc_edgar_other_01,263041061,1784838393273,0,32117,false
20260723205405966,20260723205405966_102_85186,I48dcc76b0e4c407bb4623377579f92bc,w_blc_edgar_other_01,b3ed7b64-38eb-49df-837c-1aa4a2b2a6e9-0_3250-772-775679_20260802145031626.parquet,I48dcc76b0e4c407bb4623377579f92bc,w_blc_edgar_other_01,263041061,1784838393273,0,32117,false
20260722205053988,20260722205053988_512_43359,I82b72234efce4f949198d69cd10411c8,w_blc_edgar_other_01,d3b8b8ae-b9e4-4598-acd3-a998f8b0ccba-0_3251-772-775680_20260802145031626.parquet,I82b72234efce4f949198d69cd10411c8,w_blc_edgar_other_01,262918552,1784752956833,0,32110,false
20260722205053988,20260722205053988_511_42122,I82b72234efce4f949198d69cd10411c8,w_blc_edgar_other_01,b3ed7b64-38eb-49df-837c-1aa4a2b2a6e9-0_3250-772-775679_20260802145031626.parquet,I82b72234efce4f949198d69cd10411c8,w_blc_edgar_other_01,262918552,1784752956833,0,32110,false
20260731211110564,20260731211110564_186_161978,Ia40cab33acd141fa8b4e607331887ec8,w_blc_edgar_other_01,d3b8b8ae-b9e4-4598-acd3-a998f8b0ccba-0_3251-772-775680_20260802145031626.parquet,Ia40cab33acd141fa8b4e607331887ec8,w_blc_edgar_other_01,263779206,1785530866483,0,32197,false
20260731211110564,20260731211110564_185_114405,Ia40cab33acd141fa8b4e607331887ec8,w_blc_edgar_other_01,b3ed7b64-38eb-49df-837c-1aa4a2b2a6e9-0_3250-772-775679_20260802145031626.parquet,Ia40cab33acd141fa8b4e607331887ec8,w_blc_edgar_other_01,263779206,1785530866483,0,32197,false
20260731205047416,20260731205047416_95_126230,Ia433f5d02ffa11ecbea4f0dc9fb69570,w_blc_edgar_other_01,d3b8b8ae-b9e4-4598-acd3-a998f8b0ccba-0_3251-772-775680_20260802145031626.parquet,Ia433f5d02ffa11ecbea4f0dc9fb69570,w_blc_edgar_other_01,263778589,1785530546535,0,32196,false
20260731205047416,20260731205047416_94_129950,Ia433f5d02ffa11ecbea4f0dc9fb69570,w_blc_edgar_other_01,b3ed7b64-38eb-49df-837c-1aa4a2b2a6e9-0_3250-772-775679_20260802145031626.parquet,Ia433f5d02ffa11ecbea4f0dc9fb69570,w_blc_edgar_other_01,263778589,1785530546535,0,32196,false
```

The identical fileId pair (`d3b8b8ae.../3251` and `b3ed7b64.../3250`) recurs for four different
guids across four different commits spanning 2026-07-22 to 2026-07-31 — same `_hoodie_file_name`
pair every time, only the guid/`_hoodie_commit_time`/`sequenceid`/`operationtime` change.

**Reading:** both write tokens share stage `772` with *consecutive* `taskAttemptId`s (775679,
775680 — ran essentially back-to-back) but *different* `partitionId`s (3250 vs 3251). A
same-partition task retry would keep partitionId fixed and only bump the attempt id; different
partitionIds under the same stage is the signature of a whole-**stage** recompute (lost
executor's shuffle output isn't recoverable with the external shuffle service off, forcing Spark
to recompute the entire stage rather than just the missing task). The recompute reassigns
physical partition numbers, so the same logical insert bucket re-emerges under a new partition id
and — because it's tagged as INSERT — a brand-new `fileId` mirroring the original almost
row-for-row.

**Persistence:** the same fileId pair (3250/3251) appears identically across four commits
spanning 2026-07-22 to 2026-07-31. This is not four separate mirror-origination events — it's one
origination event, after which every ordinary subsequent update to any guid in that bucket
dutifully updates *both* copies forever, because per-file-group compaction/cleaning plus no
global index means no process ever collapses a cross-file-group key (§7 of the investigation).
This is why Athena's scan of `other_01` found ~94% guid overlap between file groups 3250/3251
rather than a single clean pair.

## 2. Hub-scatter (S4 dominant, sub-pattern B)

**Root cause:** MoR log-resident keys are false negatives against the base-file Bloom index
(compaction is 6-hour, time-only triggered) → misclassified as inserts → routed to a neighbor
file group within the same commit.

**Source:** `duplicate_guid_field_details_qa_sig2only.csv`, `UK_SMG_EUIMAGES_03`, commit
`20260605084155211`, rows 2–30+ (dozens of sequential guids `I07372260605011...` through
`I0737226E605011...`). Raw rows, all fields (first ten pairs shown; the pattern continues
identically for the remaining guids in this commit):

```
_hoodie_commit_time,_hoodie_commit_seqno,_hoodie_record_key,_hoodie_partition_path,_hoodie_file_name,guid,collectionname,sequenceid,operationtime,operationstartstage,operationendstage,delete
20260605084155211,20260605084155211_950_385,I07372260605011F19EC7A47E112196E9,UK_SMG_EUIMAGES_03,8756da3c-4612-4d22-9fc4-fbe7a930d71e-0_1-39-45582_20260605084558630.parquet,I07372260605011F19EC7A47E112196E9,UK_SMG_EUIMAGES_03,259036809,1780603319010,0,221835,false
20260605084155211,20260605084155211_951_33,I07372260605011F19EC7A47E112196E9,UK_SMG_EUIMAGES_03,5956159a-4dc9-4c2a-a5d7-2067ff45ba18-0_464-9588-11587230_20260610092022129.parquet,I07372260605011F19EC7A47E112196E9,UK_SMG_EUIMAGES_03,259036809,1780603319010,0,221835,false
20260605084155211,20260605084155211_951_24,I07372261605011F19EC7A47E112196E9,UK_SMG_EUIMAGES_03,5956159a-4dc9-4c2a-a5d7-2067ff45ba18-0_464-9588-11587230_20260610092022129.parquet,I07372261605011F19EC7A47E112196E9,UK_SMG_EUIMAGES_03,259036809,1780603319010,0,221835,false
20260605084155211,20260605084155211_950_409,I07372261605011F19EC7A47E112196E9,UK_SMG_EUIMAGES_03,8756da3c-4612-4d22-9fc4-fbe7a930d71e-0_1-39-45582_20260605084558630.parquet,I07372261605011F19EC7A47E112196E9,UK_SMG_EUIMAGES_03,259036809,1780603319010,0,221835,false
20260605084155211,20260605084155211_951_89,I07372262605011F19EC7A47E112196E9,UK_SMG_EUIMAGES_03,5956159a-4dc9-4c2a-a5d7-2067ff45ba18-0_464-9588-11587230_20260610092022129.parquet,I07372262605011F19EC7A47E112196E9,UK_SMG_EUIMAGES_03,259036809,1780603319010,0,221835,false
20260605084155211,20260605084155211_950_361,I07372262605011F19EC7A47E112196E9,UK_SMG_EUIMAGES_03,8756da3c-4612-4d22-9fc4-fbe7a930d71e-0_1-39-45582_20260605084558630.parquet,I07372262605011F19EC7A47E112196E9,UK_SMG_EUIMAGES_03,259036809,1780603319010,0,221835,false
20260605084155211,20260605084155211_950_404,I07372263605011F19EC7A47E112196E9,UK_SMG_EUIMAGES_03,8756da3c-4612-4d22-9fc4-fbe7a930d71e-0_1-39-45582_20260605084558630.parquet,I07372263605011F19EC7A47E112196E9,UK_SMG_EUIMAGES_03,259036809,1780603319010,0,221835,false
20260605084155211,20260605084155211_951_80,I07372263605011F19EC7A47E112196E9,UK_SMG_EUIMAGES_03,5956159a-4dc9-4c2a-a5d7-2067ff45ba18-0_464-9588-11587230_20260610092022129.parquet,I07372263605011F19EC7A47E112196E9,UK_SMG_EUIMAGES_03,259036809,1780603319010,0,221835,false
20260605084155211,20260605084155211_950_415,I07372264605011F19EC7A47E112196E9,UK_SMG_EUIMAGES_03,8756da3c-4612-4d22-9fc4-fbe7a930d71e-0_1-39-45582_20260605084558630.parquet,I07372264605011F19EC7A47E112196E9,UK_SMG_EUIMAGES_03,259036809,1780603319010,0,221835,false
20260605084155211,20260605084155211_951_20,I07372264605011F19EC7A47E112196E9,UK_SMG_EUIMAGES_03,5956159a-4dc9-4c2a-a5d7-2067ff45ba18-0_464-9588-11587230_20260610092022129.parquet,I07372264605011F19EC7A47E112196E9,UK_SMG_EUIMAGES_03,259036809,1780603319010,0,221835,false
20260605084155211,20260605084155211_950_401,I07372265605011F19EC7A47E112196E9,UK_SMG_EUIMAGES_03,8756da3c-4612-4d22-9fc4-fbe7a930d71e-0_1-39-45582_20260605084558630.parquet,I07372265605011F19EC7A47E112196E9,UK_SMG_EUIMAGES_03,259036809,1780603319010,0,221835,false
20260605084155211,20260605084155211_951_81,I07372265605011F19EC7A47E112196E9,UK_SMG_EUIMAGES_03,5956159a-4dc9-4c2a-a5d7-2067ff45ba18-0_464-9588-11587230_20260610092022129.parquet,I07372265605011F19EC7A47E112196E9,UK_SMG_EUIMAGES_03,259036809,1780603319010,0,221835,false
20260605084155211,20260605084155211_951_21,I07372266605011F19EC7A47E112196E9,UK_SMG_EUIMAGES_03,5956159a-4dc9-4c2a-a5d7-2067ff45ba18-0_464-9588-11587230_20260610092022129.parquet,I07372266605011F19EC7A47E112196E9,UK_SMG_EUIMAGES_03,259036809,1780603319010,0,221835,false
20260605084155211,20260605084155211_950_366,I07372266605011F19EC7A47E112196E9,UK_SMG_EUIMAGES_03,8756da3c-4612-4d22-9fc4-fbe7a930d71e-0_1-39-45582_20260605084558630.parquet,I07372266605011F19EC7A47E112196E9,UK_SMG_EUIMAGES_03,259036809,1780603319010,0,221835,false
20260605084155211,20260605084155211_950_381,I07372267605011F19EC7A47E112196E9,UK_SMG_EUIMAGES_03,8756da3c-4612-4d22-9fc4-fbe7a930d71e-0_1-39-45582_20260605084558630.parquet,I07372267605011F19EC7A47E112196E9,UK_SMG_EUIMAGES_03,259036809,1780603319010,0,221835,false
20260605084155211,20260605084155211_951_66,I07372267605011F19EC7A47E112196E9,UK_SMG_EUIMAGES_03,5956159a-4dc9-4c2a-a5d7-2067ff45ba18-0_464-9588-11587230_20260610092022129.parquet,I07372267605011F19EC7A47E112196E9,UK_SMG_EUIMAGES_03,259036809,1780603319010,0,221835,false
20260605084155211,20260605084155211_951_51,I07372268605011F19EC7A47E112196E9,UK_SMG_EUIMAGES_03,5956159a-4dc9-4c2a-a5d7-2067ff45ba18-0_464-9588-11587230_20260610092022129.parquet,I07372268605011F19EC7A47E112196E9,UK_SMG_EUIMAGES_03,259036809,1780603319010,0,221835,false
20260605084155211,20260605084155211_950_423,I07372268605011F19EC7A47E112196E9,UK_SMG_EUIMAGES_03,8756da3c-4612-4d22-9fc4-fbe7a930d71e-0_1-39-45582_20260605084558630.parquet,I07372268605011F19EC7A47E112196E9,UK_SMG_EUIMAGES_03,259036809,1780603319010,0,221835,false
20260605084155211,20260605084155211_950_395,I07372269605011F19EC7A47E112196E9,UK_SMG_EUIMAGES_03,8756da3c-4612-4d22-9fc4-fbe7a930d71e-0_1-39-45582_20260605084558630.parquet,I07372269605011F19EC7A47E112196E9,UK_SMG_EUIMAGES_03,259036809,1780603319010,0,221835,false
20260605084155211,20260605084155211_951_63,I07372269605011F19EC7A47E112196E9,UK_SMG_EUIMAGES_03,5956159a-4dc9-4c2a-a5d7-2067ff45ba18-0_464-9588-11587230_20260610092022129.parquet,I07372269605011F19EC7A47E112196E9,UK_SMG_EUIMAGES_03,259036809,1780603319010,0,221835,false
```

Every guid in this commit alternates between the same two fileIds — hub A (`8756da3c...`,
token `1-39-45582`, filename-instant `20260605084558630`, close to the commit) and hub B
(`5956159a...`, token `464-9588-11587230`, filename-instant `20260610092022129`, five days
later).

**Reading:** unlike pattern 1, these write tokens are not adjacent — stage 39 vs stage 9588,
wildly different partition/attempt ids — so `5956159a` is not a sibling task from the same
micro-batch. It's a long-lived, pre-existing file group ("hub") that this commit scattered a
batch of individually misrouted inserts into. The 5-day-later filename timestamp on `5956159a` is
the tell: that's the *next* compaction physically rewriting the hub's base file to fold in these
log-resident entries, while each row's own `_hoodie_commit_time` metadata correctly preserves the
original June 5 logical write. Consistent with: these keys were sitting in an uncompacted MoR log
(compaction is time-triggered, not immediate), the base-file Bloom filter hadn't absorbed them
yet, the index falsely reported "not found," and each key — one at a time, not as one atomic
bucket copy — was routed as a fresh insert into the neighbor hub file group.

Matches the earlier `33_02`/hub `7b66daa1` and `23mm`/hub `3bb7805c` findings from the original
investigation — same shape, different collection.

## 3. Cross-commit replay (S5, Sig #1) — not present in either CSV

Both CSVs are scoped to `sig2only` (same-commit dups) by name and content — every row pair shares
one `_hoodie_commit_time`. No cross-commit example exists in these two files; a fresh Athena pull
filtered to `_hoodie_commit_time` pairs that differ for the same guid would be needed for a
CSV-backed example.

The only documented Sig #1 example remains the narrative one from the original investigation:
`33_02`, 8 guids, commits `20260604180024505` and `20260604181909526` (~19 minutes apart),
identical `sequenceid=259023002` / `operationtime=1780594587500` / `operationendstage=127417`
across all 8 guids in both commits — the "identical per-batch triple" signature that pins this as
a single-writer replay (the whole already-committed window got re-driven), not two genuinely
concurrent writers.

**Illustrative reconstruction (NOT raw Athena output):** the original investigation preserved the
aggregate facts above (commit times, `sequenceid`, `operationtime`, `operationendstage`, guid
count = 8) but not the actual guid strings or file names in a table this document has access to.
The rows below use the real, documented aggregate values; the guid strings and `_hoodie_file_name`
values are synthetic placeholders built to match the pattern's shape (two commits ~19 minutes
apart, one shared file per commit since a replay of an already-correct batch would land in
whatever file group the index routes it to on the second commit — plausibly a *different* file
group than the original if the second run's insert/update routing differs). Treat this block as a
diagram of the pattern, not as verified evidence — a real example still needs a fresh Athena pull
filtered to guids whose `_hoodie_commit_time` values differ.

```
_hoodie_commit_time,_hoodie_commit_seqno,_hoodie_record_key,_hoodie_partition_path,_hoodie_file_name,guid,collectionname,sequenceid,operationtime,operationstartstage,operationendstage,delete
20260604180024505,20260604180024505_310_50001,I3302REPLAY0000000000000000000001,33_02,7a1c9e02-11ab-4a3d-9e77-001122334401-0_310-640-9001001_20260604180512033.parquet,I3302REPLAY0000000000000000000001,33_02,259023002,1780594587500,0,127417,false
20260604181909526,20260604181909526_318_50101,I3302REPLAY0000000000000000000001,33_02,c4de88f1-2b3c-4f56-8a90-556677889901-0_318-648-9012045_20260604182344871.parquet,I3302REPLAY0000000000000000000001,33_02,259023002,1780594587500,0,127417,false
20260604180024505,20260604180024505_310_50002,I3302REPLAY0000000000000000000002,33_02,7a1c9e02-11ab-4a3d-9e77-001122334401-0_310-640-9001001_20260604180512033.parquet,I3302REPLAY0000000000000000000002,33_02,259023002,1780594587500,0,127417,false
20260604181909526,20260604181909526_318_50102,I3302REPLAY0000000000000000000002,33_02,c4de88f1-2b3c-4f56-8a90-556677889901-0_318-648-9012045_20260604182344871.parquet,I3302REPLAY0000000000000000000002,33_02,259023002,1780594587500,0,127417,false
20260604180024505,20260604180024505_310_50003,I3302REPLAY0000000000000000000003,33_02,7a1c9e02-11ab-4a3d-9e77-001122334401-0_310-640-9001001_20260604180512033.parquet,I3302REPLAY0000000000000000000003,33_02,259023002,1780594587500,0,127417,false
20260604181909526,20260604181909526_318_50103,I3302REPLAY0000000000000000000003,33_02,c4de88f1-2b3c-4f56-8a90-556677889901-0_318-648-9012045_20260604182344871.parquet,I3302REPLAY0000000000000000000003,33_02,259023002,1780594587500,0,127417,false
20260604180024505,20260604180024505_310_50004,I3302REPLAY0000000000000000000004,33_02,7a1c9e02-11ab-4a3d-9e77-001122334401-0_310-640-9001001_20260604180512033.parquet,I3302REPLAY0000000000000000000004,33_02,259023002,1780594587500,0,127417,false
20260604181909526,20260604181909526_318_50104,I3302REPLAY0000000000000000000004,33_02,c4de88f1-2b3c-4f56-8a90-556677889901-0_318-648-9012045_20260604182344871.parquet,I3302REPLAY0000000000000000000004,33_02,259023002,1780594587500,0,127417,false
```
(4 of the 8 guids shown; the remaining 4 follow the same shape — same two commit times, same
`sequenceid`/`operationtime`/`operationendstage` triple, one row per commit.)

**Reading:** unlike patterns 1–2, the two rows for each guid carry *different*
`_hoodie_commit_time` values (`20260604180024505` vs `20260604181909526`, ~19 minutes apart) —
this is the cross-commit signature, not the same-commit signature. What pins this as a
single-writer *replay* rather than two genuinely concurrent writers is that `sequenceid`,
`operationtime`, and `operationendstage` are identical across every guid in *both* commits: this
is not two different batches independently touching the same guids (which would show different
per-batch counters), it's the literal same batch's counters appearing twice, because the second
commit is a full re-drive of a batch that had already committed successfully once.

**Root cause:** Hudi's commit precedes the DynamoDB checkpoint; per-collection commits are
non-atomic; the checkpoint is all-or-nothing. A single failing collection later in the batch
forces a replay of the whole window of already-committed collections. The replayed copy survives
for the same reason as patterns 1 and 2: no global index or cross-file-group collapse ever heals
it once written.

## Triage validity

`_hoodie_commit_time` cleanly separates one-commit write faults (patterns 1–2, S4) from
two-commit re-drive faults (pattern 3, S5). The per-batch triple (`sequenceid` /
`operationtime` / `operationendstage`) separates single-writer replay from the genuinely
concurrent path that a non-atomic lock + `StepConcurrencyLevel=2` could in principle open (ruled
out for all observed data — no distinct-batch-counter pairs seen).

## Why nothing self-heals

Per-file-group compaction merges log into base within one file group only, and the cleaner drops
old versions per file group — neither ever runs the index or reasons about a key across multiple
file groups. Combined with no global index (`hoodie.metadata.enable=false`, partition-scoped
`SIMPLE`/`BLOOM` index), no process in this pipeline ever collapses a cross-file-group duplicate
once it's committed. This is why all three patterns, once introduced, persist indefinitely rather
than resolving on their own over subsequent compaction/cleaning cycles.

Start with NovusHudiOperations.scala, especially hudiOptions and upsert(), then trace replay handling through KinesisCustomConsumer and TNIStreamingApp.processInterval. Compare the documented S4, S5, and S8 signatures against the six required dedup invariants and existing write behavior. Done means the relevant duplicate mechanisms are reproduced or ruled out and a scoped prevention or cleanup approach is validated without removing valid cross-collection records.

Written by the indexing model from the issue text.