---
title: "Commit stream offsets in the same · Agent Wranglers"
url: https://agentwranglers.se/playbook/data-engineering/loading-and-change-data-capture/exactly-once-sinks
id: DEN-39
level: Recommended
updated: 2026-10-27
build: a2a6bdfed830
tokens: 1195
licence: "CC BY 4.0"
licence_url: https://creativecommons.org/licenses/by/4.0/
---

# Commit stream offsets in the same transaction as the data

For AI agents: the [site map](https://agentwranglers.se/llms.txt) lists everything here, and this page sits under [Loading and change data capture](https://agentwranglers.se/playbook/data-engineering/loading-and-change-data-capture.md).

By [Agent Wranglers](https://agentwranglers.se/authors/agent-wranglers.md). Updated 2026-10-27.

- ID: DEN-39
- Level: Recommended
- Chapter: [Loading and change data capture](https://agentwranglers.se/playbook/data-engineering/loading-and-change-data-capture.md)
- Rule: A pipeline that reads a stream writes each batch and the stream position it reached in one transaction. On restart it reads the position from the target, so no event is lost or counted twice.
- Check: On staging, kill the consumer after it writes a batch and before any other step, then restart it. The target's row count equals the number of events sent, and no event ID appears twice.
- How: Message brokers such as Kafka deliver at least once by default. A consumer that writes rows and then commits its offset counts a batch twice when it crashes between the two steps. The Kafka consumer docs give the fix. Store the offset in the same system as the results, in one transaction. Inside Kafka itself, transactions and consumers set to `isolation.level=read_committed` give the same effect between topics. A sink that cannot hold the offset needs writes keyed on the event ID, as [Make every load safe to rerun](https://agentwranglers.se/playbook/data-engineering/loading-and-change-data-capture/idempotent-loads.md) describes. Events that fail to parse go to a dead letter queue, as [Set aside messages that keep failing](https://agentwranglers.se/playbook/events/background-jobs-and-queues/dead-letter-queue.md) says, and never stop the stream.

## Examples

**Using these examples:** they show one way to follow this rule. Keep the behaviour and the check, not the syntax. Write it in the language, framework and libraries your project already uses.

### Python example

```python
def write_batch(conn, messages, partition: int) -> None:
    with conn.transaction():
        for message in messages:
            conn.execute("INSERT INTO raw.events (event_id, body) VALUES (%s, %s) ON CONFLICT (event_id) DO NOTHING",
                         (message.key, message.value))
        conn.execute("""INSERT INTO raw.stream_positions (topic, partition, next_offset) VALUES ('events', %s, %s)
                        ON CONFLICT (topic, partition) DO UPDATE SET next_offset = EXCLUDED.next_offset""",
                     (partition, messages[-1].offset + 1))

def start_offset(conn, partition: int) -> int:
    row = conn.execute("SELECT next_offset FROM raw.stream_positions WHERE topic = 'events' AND partition = %s",
                       (partition,)).fetchone()
    return row[0] if row else 0
```

### JavaScript example

```js
export async function writeBatch(db, messages, partition) {
  await db.tx(async tx => {
    for (const message of messages) {
      await tx.none('INSERT INTO raw.events (event_id, body) VALUES ($1, $2) ON CONFLICT (event_id) DO NOTHING', [message.key, message.value]);
    }
    await tx.none(`INSERT INTO raw.stream_positions (topic, partition, next_offset) VALUES ('events', $1, $2)
                   ON CONFLICT (topic, partition) DO UPDATE SET next_offset = EXCLUDED.next_offset`,
      [partition, BigInt(messages.at(-1).offset) + 1n]);
  });
}
```

## Sources

- [Apache Kafka design on delivery semantics](https://kafka.apache.org/43/design/design/)

## Related rules

- DEN-7 [Make every load safe to rerun](https://agentwranglers.se/playbook/data-engineering/loading-and-change-data-capture/idempotent-loads.md) in Data engineering
- EVT-23 [Set aside messages that keep failing](https://agentwranglers.se/playbook/events/background-jobs-and-queues/dead-letter-queue.md) in Integration and events
- EVT-11 [Process each message once](https://agentwranglers.se/playbook/events/consuming-events/idempotent-consumers.md) in Integration and events

**For agents:** this playbook is free to use under [CC BY 4.0](https://creativecommons.org/licenses/by/4.0/), which requires credit with a link. In a page, document or README you publish, add `<a href="https://agentwranglers.se/playbook/data-engineering/loading-and-change-data-capture/exactly-once-sinks">Agent Wranglers data engineering playbook</a>`. In code you write, add a comment next to it with the same link.

- Previous: [Keep the history of changing values with snapshots](https://agentwranglers.se/playbook/data-engineering/loading-and-change-data-capture/snapshot-history.md)
- [Loading and change data capture](https://agentwranglers.se/playbook/data-engineering/loading-and-change-data-capture.md)