Duplicate row insertion on retry in Pulsar IO JDBC sink

Question

The goal is to achieve exactly‑once write semantics when a Pulsar IO JDBC sink retries a message after a transient failure.
Transactional producers bypass broker‑side deduplication, so relying on deduplicationEnabled=true does not prevent duplicate inserts on retry. The sink receives the same message again and, if the JDBC operation is a plain INSERT, each retry creates another row unless the sink implements its own deduplication or upsert logic.
Because Pulsar provides no built‑in idempotent consumer framework, the application must track processed message IDs or make the sink operation idempotent, but the appropriate pattern and configuration remain unclear.
What configuration or sink design ensures idempotent writes during retries? Is enabling broker‑side deduplication sufficient when using transactional producers? Should the JDBC sink use upsert or an external processed‑ID store to avoid duplicates?

Tasadduq Burney · Answer

Idempotent write semantics for Pulsar IO JDBC sink
Assumption: the Pulsar IO JDBC sink currently uses a plain INSERT statement with no built‑in deduplication; idempotency must be introduced at the SQL or application level. When a message is retried after a transient failure, a plain INSERT creates a new row each time, producing duplicates. The sink does not rely on broker‑side deduplication because transactional producers bypass the broker’s deduplication logic.
Confirmed facts from available research:

Broker‑side deduplicationEnabled has no effect when transactional producers are used; the Pulsar broker does not deduplicate based on message ID after commit.
  Using an UPSERT (INSERT ... ON CONFLICT ... DO UPDATE) with a unique constraint on the message‑ID column prevents duplicate rows by updating the existing row on retry.
  Maintaining an external processed‑ID store (e.g., a lookup table keyed by Pulsar message ID) allows the sink to skip re‑processing a message that has already been written.

Likely explanation: a plain INSERT has no idempotency guard, so every retry inserts another row.
Steps needed for this case:

Add a UNIQUE or PRIMARY KEY on the message‑ID column in the JDBC target table.
  Configure the JDBC sink to use UPSERT syntax compatible with the database (e.g., INSERT ... ON CONFLICT DO UPDATE for PostgreSQL, INSERT ... SET ... ON DUPLICATE KEY UPDATE for MySQL).
  Map the Pulsar message ID to the unique column so the sink can locate the existing row on retry.
  Alternatively, configure the sink to check an external processed‑ID table before writing; if the ID exists, skip the insert.
  Verify that the transactional producer commits before the sink processes the next redelivery; use Pulsar admin API tracing if available.

Diagnostic needed: Does the target JDBC table have a UNIQUE or PRIMARY KEY on the message-ID column? This determines whether upsert or an external ID store is the viable path.

Duplicate row insertion on retry in Pulsar IO JDBC sink

1 answer

Idempotent write semantics for Pulsar IO JDBC sink

Your perspective belongs here.

0 question comments