Kafka Consumer Idempotency for Sensor Streams

Scope: the consumer that writes readings to a database

This page is for engineers who own the service between a Kafka topic of sensor readings and a relational database: for example a topic bms.readings carrying AHU supply-air temperature, chiller kW and zone CO2 from building gateways every 10 to 60 seconds, consumed by a writer that inserts into PostgreSQL or MySQL. The question it answers is narrow: when Kafka redelivers, how do you make sure the database ends up with each reading once, and the consumer restarts from the right place?

It does not repeat the general material. For Kafka itself see Apache Kafka explained; for the end-to-end pipeline see designing an IoT data pipeline with Kafka; for finding and cleaning duplicates already in a table see handling duplicate telemetry events in SQL. The readings table design follows how to store IoT sensor data.

Three decisions are made here: what the idempotency key is, where the consumer offset is stored, and what goes inside one database transaction.

What Kafka guarantees, and where it stops

The Kafka design documentation names three delivery guarantees: at most once (messages may be lost, never redelivered), at least once (never lost, may be redelivered) and exactly once. It splits the problem into two halves: getting a message durably into the log, and consuming it. Different features cover each half.

The failure that matters for a database sink

The KafkaConsumer javadoc walks through the usual pattern: poll, buffer records, insert them into a database, then call commitSync(). If the process fails after the insert but before the commit, the consumer that takes over starts from the last committed offset and repeats the insert of the last batch. The javadoc calls this at-least-once delivery. Committing first and inserting second swaps the risk for data loss (at most once), which is rarely acceptable for metering or compliance data.

Auto-commit (enable.auto.commit=true, the client default) commits in the background on a timer, so it is not tied to your database commit at all; the next section sets out when that loses or repeats records. A database writer should set it to false and decide itself when a record counts as consumed.

The design documentation gives the way out: in many cases messages have a primary key, so the updates are idempotent and receiving the same message twice just overwrites a record with another copy of itself. Sensor readings have such a key.

How Kafka offset commits work: auto, manual and stored with the data

The KafkaConsumer javadoc distinguishes two positions for each partition. The position is the offset of the next record poll() will hand out; it advances automatically on every poll. The committed position is the last offset that has been stored securely; should the process fail and restart, that is the offset the consumer recovers to. A committed offset always names the next record to read, not the last one processed. Every duplicate or loss scenario below comes from the gap between these two positions and your database commit.

Commit after processing (at least once) or before (at most once)

The Kafka design documentation describes both orders. If the consumer reads the messages, saves its position, then processes them, a crash between the second and third steps means the next owner starts after messages that were never processed: at most once. If it reads, processes, then saves its position, a crash before the save means the next owner processes some messages again: at least once. Kafka gives at-least-once by default, and the documentation says at-most-once is something you choose by disabling producer retries and committing offsets before processing. For readings that feed energy reports, compliance records or maintenance decisions, losing data is normally worse than storing a duplicate that an upsert absorbs; that is a judgement for your use case, not a Kafka rule.

What is redelivered after a crash

Everything after the last committed offset. The javadoc's database example spells it out: if the process fails after the insert but before the commit, the process that takes over consumes from the last committed offset and repeats the insert of the last batch. With batch commits that is the whole batch, not only the record in flight, so a batch of 2,000 readings (an illustrative size) is written again in full. When the offset is stored in the database in the same transaction as the readings, the stored offset only moves when the readings are committed, so a restart that seeks to it re-reads exactly the records whose transaction did not commit.

Rebalances and max.poll.interval.ms

A consumer stays in its group only while it keeps calling poll(). If it does not call poll() within max.poll.interval.ms (default 300000 ms, five minutes), the configuration reference says it is considered failed and the group rebalances to give its partitions to another member. The javadoc adds that the consumer then proactively leaves the group, and that a later commitSync() fails with CommitFailedException: a safety mechanism which guarantees that only active members of the group are able to commit offsets. The new owner starts from the last committed offset, so the slow batch is processed again.

That safety mechanism protects offsets committed to Kafka, not your database. The ConsumerRebalanceListener javadoc describes the gap: if the session times out, the partitions may be reassigned before the consumer has a chance to revoke them gracefully, and when onPartitionsLost is called the partitions may already be owned by other members. Until the old process notices, it can still be in the middle of a database transaction for a partition that now belongs to someone else. That outdated owner (often called a zombie) can commit readings and a stored offset after the new owner has started. Idempotent upserts absorb most of that for readings, but a zombie can also overwrite a newer correction or move the stored offset in a way the new owner does not expect, so the database writes need a fencing check: see fencing an outdated consumer.

The javadoc lists the controls: raise max.poll.interval.ms (at the cost of slower rebalances), cap the work per poll with max.poll.records, or process in another thread and pause() the partitions while you do, committing only offsets that have been fully handled. For a database writer, a slow database (a long lock wait, a failover) is one likely way to miss the deadline, so measure batch write time against this setting. When partitions are assigned dynamically, use a ConsumerRebalanceListener: in onPartitionsRevoked commit (or abandon) the in-flight work, and in onPartitionsAssigned look up the stored offsets and seek() to them.

Producer retries and idempotence

On the producer side (the MQTT bridge or the gateway service that writes to Kafka), retries defaults to 2147483647 and retries continue until the send succeeds, fails with a non-transient error, or delivery.timeout.ms (default 120000 ms) expires. Without idempotence, a retry after a lost acknowledgement can write the same record twice. With enable.idempotence=true, the default when no conflicting settings are made, the producer ensures that exactly one copy of each message is written in the stream; it requires acks=all, retries above 0 and at most five in-flight requests per connection.

What it does not cover, from the KafkaProducer javadoc: re-sends made by your application (they cannot be de-duplicated), and messages from a new producer session after a restart. If a send still fails after its retries, the javadoc recommends shutting the producer down and checking the last produced message so it is not duplicated. Idempotence also says nothing about the consumer: a record written once can still be read twice. That is why the database needs its own idempotency key (next section but one).

Kafka transactions, read_committed, and why they stop at the database

With a transactional.id, a producer can write to several partitions and update the consumer's committed offsets in one atomic transaction. The design documentation says the consumer side must set isolation.level=read_committed and enable.auto.commit=false; the default isolation level, read_uncommitted, returns records from aborted transactions too. Together these give exactly-once processing when input and output are both Kafka topics, which is how Kafka Streams works.

A PostgreSQL or MySQL commit is not part of a Kafka transaction. The design documentation describes the classic answer, a two-phase commit between the stored consumer position and the consumer's output, and then the simpler one it recommends: store the offset in the same place as the output. So if the database commit and the Kafka offset commit are two separate commits, there is always a window in which one has happened and the other has not, whatever Kafka settings you use. Storing the offset in the database transaction removes that window; idempotent upserts make whatever replay remains harmless; a fencing check stops an outdated consumer from writing. Use all three, and do not describe the database result as exactly once without the first and the last. If upstream producers use transactions, still set read_committed on the writer, or it can store readings from transactions that were later aborted.

Choose the idempotency key

An idempotent write needs a key that is identical every time the same reading arrives, however it arrived. For telemetry there are three candidates:

Recommendation for building and IoT telemetry

This page uses (point_id, ts) as the primary key and the conflict target, which suits telemetry where every reading has a device or gateway timestamp. It is the key the queries already use, so it costs no extra index. Keep the source partition and offset as ordinary columns for lineage: when a value looks wrong, you can trace it back to the exact Kafka record. Normalise the timestamp before it becomes a key (UTC, a fixed precision such as milliseconds), or the same reading sent once as 09:00:00 and once as 09:00:00.000+01:00 may not match.

Store the offset in the same transaction as the readings

The KafkaConsumer javadoc describes storing offsets outside Kafka: if results go into a relational database, storing the offset there too lets you commit the results and the offset in a single transaction. Either both are stored, or neither is. The recipe is: set enable.auto.commit=false, save the offset that comes with each record, and on restart restore the position with seek(). When partitions are assigned dynamically, read the stored offsets in the rebalance listener's onPartitionsAssigned callback.

The data model is one extra table keyed by consumer group, topic and partition. It stores the next offset to read, which is the convention Kafka itself uses for committed offsets.

CREATE TABLE reading (
    point_id      integer NOT NULL,
    ts            timestamptz NOT NULL,
    value         double precision NOT NULL,
    quality       smallint NOT NULL DEFAULT 0,
    src_partition integer NOT NULL,      -- lineage only
    src_offset    bigint NOT NULL,       -- lineage only
    PRIMARY KEY (point_id, ts)           -- the idempotency key
);

CREATE TABLE consumer_offset (
    consumer_group  text    NOT NULL,
    topic           text    NOT NULL,
    kafka_partition integer NOT NULL,
    next_offset     bigint  NOT NULL,    -- offset of the next record to read
    updated_at      timestamptz NOT NULL DEFAULT now(),
    PRIMARY KEY (consumer_group, topic, kafka_partition)
);

BEGIN;

WITH batch (point_id, ts, value, quality, src_partition, src_offset) AS (
    VALUES (101, timestamptz '2026-10-27 09:00:00+00', 13.2, 0, 3, 1000),
           (102, timestamptz '2026-10-27 09:00:00+00', 612.0, 0, 3, 1001),
           (101, timestamptz '2026-10-27 09:01:00+00', 13.4, 0, 3, 1002)
), ins AS (
    INSERT INTO reading AS r (point_id, ts, value, quality, src_partition, src_offset)
    SELECT point_id, ts, value, quality, src_partition, src_offset FROM batch
    ON CONFLICT (point_id, ts) DO UPDATE
        SET value = EXCLUDED.value, quality = EXCLUDED.quality,
            src_partition = EXCLUDED.src_partition, src_offset = EXCLUDED.src_offset
        WHERE (r.value, r.quality) IS DISTINCT FROM (EXCLUDED.value, EXCLUDED.quality)
    RETURNING 1
)
SELECT count(*) AS rows_written FROM ins;

INSERT INTO consumer_offset AS c (consumer_group, topic, kafka_partition, next_offset)
VALUES ('bms-writer', 'bms.readings', 3, 1003)
ON CONFLICT (consumer_group, topic, kafka_partition) DO UPDATE
    SET next_offset = EXCLUDED.next_offset, updated_at = now()
    WHERE c.next_offset < EXCLUDED.next_offset;

COMMIT;

 rows_written
--------------
            3

Replay the same batch: nothing changes

Now simulate the crash case: the transaction committed, but the consumer died before it could move on, and after the restart Kafka hands it offsets 1000 to 1002 again. Running exactly the same transaction a second time (with the offset upsert wrapped in a RETURNING count) gave this, from the same run:

 rows_written
--------------
            0

 offset_rows_changed
---------------------
                   0

WITH upd AS (
    INSERT INTO consumer_offset AS c (consumer_group, topic, kafka_partition, next_offset)
    VALUES ('bms-writer', 'bms.readings', 3, 990)
    ON CONFLICT (consumer_group, topic, kafka_partition) DO UPDATE
        SET next_offset = EXCLUDED.next_offset, updated_at = now()
        WHERE c.next_offset < EXCLUDED.next_offset
    RETURNING 1
)
SELECT count(*) AS offset_rows_changed FROM upd;
--  offset_rows_changed
-- ---------------------
--                    0

SELECT kafka_partition, next_offset FROM consumer_offset
WHERE consumer_group = 'bms-writer' AND topic = 'bms.readings';
--  kafka_partition | next_offset
-- -----------------+-------------
--                3 |        1003

Why the statement is written this way

• ON CONFLICT ... DO UPDATE is atomic in PostgreSQL: the documentation guarantees an insert or an update outcome even under high concurrency, which is what should stop two consumers that briefly overlap during a rebalance from both inserting the same key. The runs on this page were single-session, so that overlap was not exercised here.
• The WHERE ... IS DISTINCT FROM guard skips rows whose value and quality have not changed. A replay then writes nothing, which avoids dead row versions and needless WAL; a genuine correction (same point and timestamp, new value) still updates. If your rule is "first value wins", use DO NOTHING instead.
• The data-modifying CTE is there only to count rows. PostgreSQL executes data-modifying statements in WITH exactly once and to completion, whether or not the outer query reads their output.
• The monotonic offset guard (WHERE c.next_offset < EXCLUDED.next_offset) stops the stored position going backwards if an old batch is committed late. It does not stop an outdated consumer from writing readings, and a zombie holding a later batch would pass it; that is what the owner epoch in the next section is for. How the guard behaves when two transactions update the same offset row at the same moment was not run here (single session only).

Duplicates inside one batch

Idempotency across batches is not the whole story. If one poll returns the same reading twice (a gateway resent it within seconds), a single INSERT ... ON CONFLICT DO UPDATE that contains both rows fails: PostgreSQL does not allow the command to affect the same row twice, and the documentation says rows proposed for insertion should not duplicate each other on the conflict key. The whole batch, and the offset update with it, rolls back.

INSERT INTO reading AS r (point_id, ts, value, quality, src_partition, src_offset)
VALUES (101, '2026-10-27 09:02:00+00', 13.5, 0, 3, 1003),
       (101, '2026-10-27 09:02:00+00', 13.5, 0, 3, 1004)
ON CONFLICT (point_id, ts) DO UPDATE SET value = EXCLUDED.value;

ERROR:  ON CONFLICT DO UPDATE command cannot affect row a second time
HINT:  Ensure that no rows proposed for insertion within the same command have duplicate constrained values.
SQLSTATE 21000

INSERT INTO reading AS r (point_id, ts, value, quality, src_partition, src_offset)
SELECT DISTINCT ON (point_id, ts) point_id, ts, value, quality, src_partition, src_offset
FROM (VALUES (101, timestamptz '2026-10-27 09:02:00+00', 13.5, 0, 3, 1003),
             (101, timestamptz '2026-10-27 09:02:00+00', 13.5, 0, 3, 1004))
     AS b (point_id, ts, value, quality, src_partition, src_offset)
ORDER BY point_id, ts, src_offset DESC
ON CONFLICT (point_id, ts) DO UPDATE SET value = EXCLUDED.value
RETURNING point_id, ts, src_offset;

--  point_id |           ts           | src_offset
-- ----------+------------------------+------------
--       101 | 2026-10-27 09:02:00+00 |       1004

Where the deeper duplicate handling lives

Do the in-batch de-duplication in the consumer (a map keyed by point and timestamp) or in SQL as above. ON CONFLICT DO NOTHING does not raise this error, because it never updates. Finding duplicates that are already stored, near-duplicates with slightly different timestamps, and cleaning them up is covered in handling duplicate telemetry events in SQL. A table without a unique key raises no error at all and simply stores both copies; with a unique key and a plain INSERT, the duplicate fails with SQLSTATE 23505.

The same pattern in MySQL

MySQL (InnoDB) supports the same design with INSERT ... ON DUPLICATE KEY UPDATE. Use GREATEST() to keep the stored offset monotonic. Note how MySQL reports the result: the affected-rows value is 1 for an inserted row, 2 for an updated row and 0 for an existing row set to its current values (unless the client connects with CLIENT_FOUND_ROWS).

CREATE TABLE reading (
    point_id   INT NOT NULL,
    ts         DATETIME(3) NOT NULL,      -- store UTC
    value      DOUBLE NOT NULL,
    src_offset BIGINT NOT NULL,
    PRIMARY KEY (point_id, ts)
) ENGINE=InnoDB;

CREATE TABLE consumer_offset (
    consumer_group  VARCHAR(100) NOT NULL,
    topic           VARCHAR(200) NOT NULL,
    kafka_partition INT NOT NULL,
    next_offset     BIGINT NOT NULL,
    PRIMARY KEY (consumer_group, topic, kafka_partition)
) ENGINE=InnoDB;

START TRANSACTION;
INSERT INTO reading (point_id, ts, value, src_offset) VALUES
    (101, '2026-10-27 09:00:00', 13.2, 1000),
    (102, '2026-10-27 09:00:00', 612.0, 1001),
    (101, '2026-10-27 09:01:00', 13.4, 1002) AS new
ON DUPLICATE KEY UPDATE value = new.value, src_offset = new.src_offset;
INSERT INTO consumer_offset VALUES ('bms-writer', 'bms.readings', 3, 1003) AS new
ON DUPLICATE KEY UPDATE next_offset = GREATEST(consumer_offset.next_offset, new.next_offset);
COMMIT;

first run:   INSERT INTO reading ...           affected rows: 3
             INSERT INTO consumer_offset ...   affected rows: 1
replay:      INSERT INTO reading ...           affected rows: 0
             INSERT INTO consumer_offset ...   affected rows: 0

SELECT COUNT(*) AS readings FROM reading;                    -- 3
SELECT kafka_partition, next_offset FROM consumer_offset;    -- 3 | 1003

Failure recovery: what happens in each case

The table combines the Kafka behaviour described above with the database design on this page. The "Kafka offsets only" column assumes the writer calls commitSync() after the database commit; the last column assumes offsets are stored in the same database transaction as the readings and restored with seek(). The retry limit and back-off are illustrative assumptions: choose them from how long your database failovers take, how often you see transient errors, and how long a partition may stall before lag alarms fire.

BEGIN;
INSERT INTO reading (point_id, ts, value, quality, src_partition, src_offset)
VALUES (101, timestamptz '2026-10-27 09:02:00+00', 13.5, 0, 3, 1003),
       (102, timestamptz '2026-10-27 09:01:00+00', NULL, 0, 3, 1004),
       (101, timestamptz '2026-10-27 09:03:00+00', 13.6, 0, 3, 1005);
-- ERROR:  null value in column "value" of relation "reading" violates not-null constraint
-- DETAIL:  Failing row contains (102, 2026-10-27 09:01:00+00, null, 0, 3, 1004).
-- SQLSTATE 23502
INSERT INTO consumer_offset AS c (consumer_group, topic, kafka_partition, next_offset)
VALUES ('bms-writer', 'bms.readings', 3, 1006)
ON CONFLICT (consumer_group, topic, kafka_partition) DO UPDATE
    SET next_offset = EXCLUDED.next_offset, updated_at = now()
    WHERE c.next_offset < EXCLUDED.next_offset;
-- ERROR:  current transaction is aborted, commands ignored until end of transaction block
-- SQLSTATE 25P02
ROLLBACK;

SELECT count(*) AS readings FROM reading;      -- 3
SELECT next_offset FROM consumer_offset
WHERE consumer_group = 'bms-writer' AND topic = 'bms.readings' AND kafka_partition = 3;   -- 1003

CREATE TABLE reading_reject (
    src_partition integer NOT NULL,
    src_offset    bigint  NOT NULL,
    raw_value     text,
    reason        text    NOT NULL,
    rejected_at   timestamptz NOT NULL DEFAULT now(),
    PRIMARY KEY (src_partition, src_offset)
);

BEGIN;
INSERT INTO reading (point_id, ts, value, quality, src_partition, src_offset)
VALUES (101, timestamptz '2026-10-27 09:02:00+00', 13.5, 0, 3, 1003),
       (101, timestamptz '2026-10-27 09:03:00+00', 13.6, 0, 3, 1005)
ON CONFLICT (point_id, ts) DO NOTHING;
INSERT INTO reading_reject (src_partition, src_offset, raw_value, reason)
VALUES (3, 1004, NULL, 'value missing after 3 attempts')
ON CONFLICT (src_partition, src_offset) DO NOTHING;
INSERT INTO consumer_offset AS c (consumer_group, topic, kafka_partition, next_offset)
VALUES ('bms-writer', 'bms.readings', 3, 1006)
ON CONFLICT (consumer_group, topic, kafka_partition) DO UPDATE
    SET next_offset = EXCLUDED.next_offset, updated_at = now()
    WHERE c.next_offset < EXCLUDED.next_offset;
COMMIT;

SELECT (SELECT count(*) FROM reading)        AS readings,
       (SELECT count(*) FROM reading_reject) AS rejects,
       (SELECT next_offset FROM consumer_offset WHERE kafka_partition = 3) AS next_offset;
--  readings | rejects | next_offset
-- ----------+---------+-------------
--         5 |       1 |        1006

Committing to Kafka as well: a design option for lag monitoring

When the database holds the position, the offsets in Kafka are not needed for recovery. The Kafka consumer-group tool reports each group's position and how far behind the end of the log it is, and the documentation we checked does not describe how it treats offsets kept outside Kafka. As a design option, not a documented Kafka recommendation, some writers also commit the same offsets to Kafka with commitAsync() after the database commit, so that standard lag tooling shows a meaningful figure, and on restart always seek to the database offsets. The alternative is to compute lag from the database offsets yourself. Either way, decide which copy is authoritative and document it; this page did not run a Kafka cluster to test either option.

Fencing an outdated consumer: an owner epoch per partition

Offsets in the database and idempotent upserts handle redelivery. They do not handle the outdated owner described under rebalances: a consumer that lost a partition, has not noticed yet, and commits a database transaction for it while the new owner is already working. Kafka's own fencing (the CommitFailedException on commitSync()) applies only to offsets committed to Kafka. The database needs its own fencing check, in the same transaction as the batch.

The safeguard. Store an owner epoch next to each partition's stored offset. When a consumer is assigned a partition, it claims it by writing its epoch, but only if the stored epoch is not higher. Every batch transaction then starts by updating the offset row on condition that the stored epoch is not higher than the consumer's own; if that update changes 0 rows, the consumer is outdated, writes nothing and rolls back. Because the readings are inserted in the same transaction and only if the offset update succeeded, the batch is accepted or rejected as a whole.

Where the epoch comes from. Two options:

• The consumer group generation. The Java client exposes groupMetadata(), documented as "Return the current group metadata associated with this consumer", and ConsumerGroupMetadata.generationId(), documented as "Returns the generation ID of the consumer group". Read it in onPartitionsAssigned and use it as the epoch. The javadoc we checked says no more than that: it does not state that the value increases every time a partition changes owner, and the client documents two rebalance protocols (GroupProtocol.CLASSIC and GroupProtocol.CONSUMER). Check that behaviour for your client version and protocol before relying on it.
• An epoch issued by the database. In onPartitionsAssigned, run UPDATE ... SET owner_epoch = owner_epoch + 1 ... RETURNING owner_epoch, next_offset and keep the returned epoch. Each claim gets a higher number than any earlier owner, and the same statement returns the offset to seek() to. This relies only on PostgreSQL behaviour, so it is the safer choice where the generation semantics are in doubt.

The rebalance listener is where the claim belongs: the ConsumerRebalanceListener javadoc recommends committing offsets to Kafka or a custom store in onPartitionsRevoked to prevent duplicate data, and says onPartitionsAssigned runs when a rebalance completes. It also notes that callbacks only serve as notification of an assignment change, so the database check, not the callback, is what actually rejects a late writer.

ALTER TABLE consumer_offset ADD COLUMN owner_epoch integer NOT NULL DEFAULT 0;
UPDATE consumer_offset SET owner_epoch = 7          -- the old owner's earlier claim
WHERE consumer_group = 'bms-writer' AND topic = 'bms.readings' AND kafka_partition = 3;

-- onPartitionsAssigned in the new owner: claim with epoch 8, read the offset to seek to
UPDATE consumer_offset
   SET owner_epoch = 8, updated_at = now()
 WHERE consumer_group = 'bms-writer' AND topic = 'bms.readings' AND kafka_partition = 3
   AND owner_epoch <= 8
RETURNING kafka_partition, next_offset, owner_epoch;

--  kafka_partition | next_offset | owner_epoch
-- -----------------+-------------+-------------
--                3 |        1006 |           8

WITH fence AS (
    UPDATE consumer_offset
       SET next_offset = 1009, updated_at = now()
     WHERE consumer_group = 'bms-writer' AND topic = 'bms.readings' AND kafka_partition = 3
       AND owner_epoch <= 7          -- fencing: stored epoch must not be newer than mine
       AND next_offset < 1009        -- monotonic offset guard
    RETURNING 1
), batch (point_id, ts, value, quality, src_partition, src_offset) AS (
    VALUES (101, timestamptz '2026-10-27 09:04:00+00', 13.7, 0, 3, 1006),
           (102, timestamptz '2026-10-27 09:02:00+00', 615.0, 0, 3, 1007),
           (101, timestamptz '2026-10-27 09:05:00+00', 13.8, 0, 3, 1008)
), ins AS (
    INSERT INTO reading AS r (point_id, ts, value, quality, src_partition, src_offset)
    SELECT point_id, ts, value, quality, src_partition, src_offset FROM batch
    WHERE EXISTS (SELECT 1 FROM fence)    -- readings only if the fence passed
    ON CONFLICT (point_id, ts) DO UPDATE
        SET value = EXCLUDED.value, quality = EXCLUDED.quality,
            src_partition = EXCLUDED.src_partition, src_offset = EXCLUDED.src_offset
        WHERE (r.value, r.quality) IS DISTINCT FROM (EXCLUDED.value, EXCLUDED.quality)
    RETURNING 1
)
SELECT (SELECT count(*) FROM fence) AS offset_rows_changed,
       (SELECT count(*) FROM ins)   AS rows_written;

--  offset_rows_changed | rows_written
-- ---------------------+--------------
--                    0 |            0

-- the same check as a separate first statement in an explicit transaction:
BEGIN;
UPDATE consumer_offset
   SET next_offset = 1009, updated_at = now()
 WHERE consumer_group = 'bms-writer' AND topic = 'bms.readings' AND kafka_partition = 3
   AND owner_epoch <= 7 AND next_offset < 1009
RETURNING next_offset;
-- (0 rows): the application stops here and rolls back without writing readings
ROLLBACK;

WITH fence AS (
    UPDATE consumer_offset
       SET next_offset = 1009, updated_at = now()
     WHERE consumer_group = 'bms-writer' AND topic = 'bms.readings' AND kafka_partition = 3
       AND owner_epoch <= 8          -- the current owner's epoch
       AND next_offset < 1009
    RETURNING 1
), batch (point_id, ts, value, quality, src_partition, src_offset) AS (
    VALUES (101, timestamptz '2026-10-27 09:04:00+00', 13.7, 0, 3, 1006),
           (102, timestamptz '2026-10-27 09:02:00+00', 615.0, 0, 3, 1007),
           (101, timestamptz '2026-10-27 09:05:00+00', 13.8, 0, 3, 1008)
), ins AS (
    INSERT INTO reading AS r (point_id, ts, value, quality, src_partition, src_offset)
    SELECT point_id, ts, value, quality, src_partition, src_offset FROM batch
    WHERE EXISTS (SELECT 1 FROM fence)
    ON CONFLICT (point_id, ts) DO UPDATE
        SET value = EXCLUDED.value, quality = EXCLUDED.quality,
            src_partition = EXCLUDED.src_partition, src_offset = EXCLUDED.src_offset
        WHERE (r.value, r.quality) IS DISTINCT FROM (EXCLUDED.value, EXCLUDED.quality)
    RETURNING 1
)
SELECT (SELECT count(*) FROM fence) AS offset_rows_changed,
       (SELECT count(*) FROM ins)   AS rows_written;

--  offset_rows_changed | rows_written
-- ---------------------+--------------
--                    1 |            3

SELECT kafka_partition, next_offset, owner_epoch FROM consumer_offset
WHERE consumer_group = 'bms-writer' AND topic = 'bms.readings';
--  kafka_partition | next_offset | owner_epoch
-- -----------------+-------------+-------------
--                3 |        1009 |           8

SELECT count(*) AS readings FROM reading;    -- 8

-- the database-issued epoch option: the next claim on partition 3
UPDATE consumer_offset
   SET owner_epoch = owner_epoch + 1, updated_at = now()
 WHERE consumer_group = 'bms-writer' AND topic = 'bms.readings' AND kafka_partition = 3
RETURNING next_offset, owner_epoch;
--  next_offset | owner_epoch
-- -------------+-------------
--         1009 |           9

What was and was not shown

The run was a single PostgreSQL session: the outdated and the current owner were simulated one after the other, so overlapping transactions from two consumers were not exercised here. The behaviour when they overlap rests on the PostgreSQL documentation for the default Read Committed level: a second transaction that tries to update the same offset row waits for the first to commit or roll back, and then the WHERE clause is re-evaluated against the updated version of the row. So if the new owner's claim commits first, the outdated owner's fence finds a higher epoch and changes 0 rows; if the outdated owner's batch commits first, the claim waits for it and then returns the offset after that batch, which is where the new owner seeks to. This only holds if the fence is the first statement of every batch transaction (so the row lock is held until commit) and the claim is committed before the new owner starts fetching. Verify it under your own concurrency before production.

Wording. With offsets committed in the same transaction as the readings and this fencing check, each Kafka record's effect on the database tables is applied once, as far as the documented behaviour goes. Without the fencing check, call the design at-least-once with idempotent writes, not exactly once. Side effects outside the transaction are never covered by either.

Operational workflow for the consumer

The order of operations is what makes the design safe. The loop below is pseudo-code (not run) using the Java client's method names.

config: enable.auto.commit=false, isolation.level=read_committed (if producers use transactions)

on partitions assigned (rebalance listener):
    for each partition p:
        claim: UPDATE consumer_offset SET owner_epoch = my_epoch      -- or owner_epoch + 1 (database-issued)
               WHERE group, topic, p AND owner_epoch <= my_epoch
               RETURNING next_offset, owner_epoch
        if a row came back: remember the epoch; consumer.seek(p, next_offset)
        if no row: a newer owner exists; do not process p
        (insert the row first if this partition has never been stored; otherwise auto.offset.reset applies)

loop:
    records = consumer.poll(timeout)
    batch   = parse, validate, normalise timestamps to UTC, de-duplicate on (point_id, ts)
    BEGIN
        fence first: UPDATE consumer_offset SET next_offset = last offset + 1
                     WHERE group, topic, p AND owner_epoch <= my_epoch AND next_offset < last offset + 1
        if 0 rows: ROLLBACK and stop processing p (this consumer is outdated)
        upsert batch into reading (ON CONFLICT (point_id, ts) ...)
    COMMIT
    on database error: ROLLBACK, retry with back-off up to a limit, then park the record in a reject table
    design option: consumer.commitAsync(same offsets)   -- lag monitoring only; restart always seeks to the database offsets

on partitions revoked: finish or abandon the in-flight transaction before returning

Operational notes

• Lag monitoring. If the database holds the real position, decide how lag is measured: from the database offsets, or by also committing the same offsets to Kafka after the database commit (a design option, described in the failure-recovery section). On restart, seek to the database offsets.
• Batch size. One transaction per poll amortises commit overhead; a few hundred to a few thousand readings per batch is an illustrative starting point, not a recommendation. Choose it from your write latency, max.poll.records and how much replay you accept after a crash, and keep the processing time inside max.poll.interval.ms, or the consumer leaves the group and its partitions move, which is itself a source of replays.
• Poison records. A record that always fails must not block the partition forever. Park it in a reject table inside the same transaction that advances the offset; a dead-letter topic, if you use one, is a separate Kafka write and not part of that transaction.
• Retention. Replays only work while Kafka still holds the records. Set topic retention longer than the longest database outage you plan to ride out.

Limitations and when not to use this pattern

Side effects outside the transaction. If the consumer also raises a BMS alarm, sends a notification or opens a work order in the CMMS, a replay repeats those actions. Give each one its own idempotency key (for example a unique constraint on (asset_id, alarm_type, alarm_start) in the work-order table, or an idempotency key the CMMS API accepts) or move them to a separate consumer that reads from the database.

Corrections and ordering. Last-write-wins with DO UPDATE assumes a later delivery is the better value. If a stale copy can arrive after a correction, compare a version or source timestamp in the WHERE clause instead of overwriting blindly.

Kafka-to-Kafka processing. When the output is another topic (enrichment, unit conversion, routing), use Kafka transactions or Kafka Streams instead; they are designed for that case.

Append-only stores without keys. Some analytical engines do not enforce unique keys at insert time. There, de-duplication moves to merge or query time and this pattern does not apply as written; check the engine's own documentation. See the time-series overview for the storage options and Kafka vs MQTT for where each transport fits.

Frequently asked questions

Sources

• Apache Kafka 4.4 documentation: Design (message delivery semantics, using transactions)
• Apache Kafka 4.4 documentation: Producer configs
• Apache Kafka 4.4 documentation: Consumer configs
• Apache Kafka 4.4 javadoc: KafkaConsumer
• Apache Kafka 4.4 javadoc: KafkaProducer
• PostgreSQL 18 documentation: INSERT
• PostgreSQL 18 documentation: WITH Queries
• MySQL 8.0 Reference Manual: INSERT ... ON DUPLICATE KEY UPDATE
• Apache Kafka 4.4 documentation: Kafka Connect user guide (error reporting and dead letter queue)
• Apache Kafka 4.4 documentation: Basic Kafka Operations (checking consumer position)
• Apache Kafka 4.4 javadoc: ConsumerGroupMetadata (generationId)
• Apache Kafka 4.4 javadoc: ConsumerRebalanceListener
• PostgreSQL 18 documentation: Transaction Isolation

Checked 27 October 2026.

How we research these articles: our editorial method. Research-based from official documentation; SQL examples are run on a local database where the page says so. CodeWithSQL earns nothing from the vendors mentioned.