In a Dataflow pipeline, what does PubsubIO's withIdAttribute change about deduplication?

## Two different duplicate problems "Duplicates from Pub/Sub" covers two distinct situations, and the whole value of this question is telling them apart. **Redelivery of the same message.** Pub/Sub may deliver a message more than once — for instance when an acknowledgement is delayed. Each delivery carries the *same* service-assigned message ID. **Republication of the same logical event.** An upstream service publishes an event, does not get a confirmation, retries, and publishes again. Now there are two genuinely different messages in the topic with two different message IDs that happen to describe the same real-world event. When a Beam pipeline reads Pub/Sub on the Dataflow runner, the runner handles the first case for you: it deduplicates on the Pub/Sub message ID, which is why Dataflow can offer exactly-once *processing* of records within the pipeline. It cannot handle the second case, because from the runner's point of view those are two unrelated messages. ## What withIdAttribute does `withIdAttribute` changes the dedup key from "whatever the service assigned" to "an attribute the publisher set": ```java PCollection<PubsubMessage> events = p.apply( PubsubIO.readMessages() .fromSubscription("projects/p/subscriptions/events") .withIdAttribute("event_id")); ``` In the Python SDK the same idea is `id_label`: ```python events = p | beam.io.ReadFromPubSub( subscription='projects/p/subscriptions/events', id_label='event_id', with_attributes=True) ``` Now the source treats two messages carrying the same `event_id` attribute as the same element and emits only one of them. If the publisher stamps a stable idempotency key — an order ID, an event UUID generated once before the first publish attempt — then publisher retries stop reaching your pipeline as duplicates. There is a companion knob worth knowing in the same breath: `withTimestampAttribute` (Java) / `timestamp_attribute` (Python), which tells the source to take each element's event time from a message attribute rather than from the Pub/Sub publish time. It is a different concern — event time versus identity — but it is configured the same way and is the other reason to set attributes on the publisher side. ## The three conditions this depends on 1. **The publisher must set the attribute, on every message.** A message without the attribute has nothing to dedupe on. This is a contract between the publishing service and the pipeline, and it is the part that breaks in practice: one producer in a fleet forgets, and duplicates start appearing. 2. **The value must be stable across retries.** Generating a fresh UUID inside the retry loop defeats the whole mechanism — the key must be computed once, before the first publish attempt, and reused by every retry of that same logical event. 3. **Deduplication is bounded in time, not permanent.** The source keeps a record of recently-seen IDs; it does not maintain a lifetime registry of every key it has ever observed. A copy republished long after the original will pass through. Treat the feature as protection against near-term republishing, not as a general-purpose idempotency store. ## What it does not give you This is source-side deduplication. It says nothing about what happens at the *other* end of the pipeline. Dataflow's exactly-once story is scoped to processing within a running job: if you cancel and restart, if you drain and relaunch, or if your sink is not idempotent, records can still be written twice downstream. The durable pattern is unchanged by anything on the source side: give each record a business key and make the sink an upsert or a keyed merge, so re-writing a record is a no-op. Equally, if what you need is deduplication over a long, explicitly-controlled horizon, or deduplication on a key computed from the message *body* rather than an attribute, that belongs in the pipeline as a deliberate stateful step (Beam has a `Deduplicate` transform with a configurable duration) rather than as a source option. ## How this shows up in an interview Usually as a scenario: "our downstream table has duplicate rows and we read from Pub/Sub — why?" The strong answer separates the layers out loud. Redelivery of one message is already handled by the runner. Duplicate *publishes* are not, and `withIdAttribute` with a publisher-supplied stable key is the source-side fix, subject to the publisher honouring the contract and to the bounded dedup horizon. And whatever you do at the source, the sink still has to be idempotent, because job restarts and replays exist.