Commit stream offsets in the same transaction as the data

How to do it

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 describes. Events that fail to parse go to a dead letter queue, as Set aside messages that keep failing says, and never stop the stream.

Examples

Python example

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

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]);
  });
}

How to 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.

Sources

• Apache Kafka design on delivery semantics

Related rules

• DEN-7 Make every load safe to rerun in Data engineering
• EVT-23 Set aside messages that keep failing in Integration and events
• EVT-11 Process each message once in Integration and events