Offset Management #

When we design fault-tolerant distributed data stream processing systems, one of the biggest challenges is maintaining read position synchronization. As discussed in the previous chapter, storing commit offsets in Kafka’s internal storage (__consumer_offsets) — whether using auto-commit or manual commit — is never atomic with our business database transaction writes. This is because writing to a Kafka broker and writing to an RDBMS database (like PostgreSQL or MySQL) are two separate network I/O operations not wrapped in one global transaction. If the application crashes between those two writes, we can’t avoid data duplication risk. To achieve true Exactly-Once processing guarantees without the overhead of internal Kafka transactions, advanced microservices architectures apply the External Offset Storage strategy, managing offsets independently using the consumer.seek() API.

The Non-Atomic Distributed System Problem #

By default, the common consumer workflow that writes data to a database is as follows:

  1. The consumer calls .poll() and obtains a message with Offset 500.
  2. The consumer processes the data and writes the result to the business database table orders (success).
  3. The consumer sends a manual commit commitSync(Offset 500) to the Kafka broker (fails because the network dropped).
  4. Problem: The business data is already stored in the database, but the Kafka coordinator considers the last offset still 499. When the consumer recovers, it reprocesses message 500, triggering data duplication in the database.

We can’t wrap the SQL data storage operation in PostgreSQL and the Kafka broker commit operation in the same ACID transaction block. They’re separate storage systems (non-atomic distributed writes).

Solution: Storing Offsets in an External Database (RDBMS) #

The most elegant way to solve this non-atomic problem is to move the commit offset storage location from Kafka’s internal __consumer_offsets topic into a special table in our own business RDBMS database.

The Local Database Transaction (ACID) Concept #

By storing offsets in the same business database as our business data, we can wrap the business data storage operation and the Kafka offset number update into one same local database transaction (BEGIN TRANSACTION … COMMIT):

BEGIN TRANSACTION;

-- 1. Write business transaction data
INSERT INTO orders (order_id, customer_id, amount) VALUES ('TX-1001', 'CUST-99', 500000.0);

-- 2. Update the Kafka offset for the related partition
UPDATE kafka_offsets 
SET last_offset = 501 
WHERE topic = 'orders-topic' AND partition_id = 2;

COMMIT;

Why Is This Approach 100% Safe? #

Inside a structured RDBMS, transactions are guaranteed to comply with ACID principles (Atomicity, Consistency, Isolation, Durability).

  • If the database connection drops midway, or one of the two SQL commands above fails, the database performs a full rollback for both command lines. There will never be a condition where business data is stored but the offset doesn’t advance, or vice versa.
  • The Kafka broker side never receives offset commits at all (we disable auto-commit and never call commitSync or commitAsync on our consumer client). The Kafka coordinator considers our group offset still at the old number, but we ignore that broker-side offset status and use our local database offset status as the absolute truth.

Transactional Workflow to a MySQL Database #

The following flow chart visualizes how the data read cycle, ACID database transaction execution combining business data writes with Kafka offsets, and rollback handling if a crash occurs:

flowchart TD
    Start["1. Application Calls poll()"] --> Cons["Consumer Receives Message Batch (Offset 500-505)"]
    Cons --> DB_Tx["2. Open Local Database Transaction (DB.beginTransaction)"]
    
    subgraph Database["One Atomic MySQL Database Transaction"]
        direction TB
        DB_Write["3. Write Business Data (Insert into orders)"] --> DB_Offset["4. Write Kafka Offset (Update kafka_offsets set offset = 506 where partition = 0)"]
    end
    
    DB_Tx --> DB_Write
    DB_Offset --> CommitCheck{"Did all database writes succeed?"}
    
    CommitCheck -- "Yes" --> DB_Commit["5. Commit Transaction (DB.commit)"]
    DB_Commit --> ProcessNext["6. Ready to poll the next data batch"]
    
    CommitCheck -- "No (Crash / Error)" --> DB_Rollback["5. Rollback Transaction (DB.rollback)"]
    DB_Rollback --> SeekBack["6. Re-fetch Last Offset in Database & Call consumer.seek(partition, 500)"]
    SeekBack --> ProcessNext
    
    style Start stroke:#e5e7eb
    style DB_Tx stroke:#f57c00,stroke-width:2px
    style Database stroke:#e5e7eb
    style DB_Write stroke:#0288d1,stroke-width:2px
    style DB_Offset stroke:#0288d1,stroke-width:2px
    style DB_Commit stroke:#2e7d32,stroke-width:2px
    style DB_Rollback stroke:#c62828,stroke-width:2px

The Dynamic Offset Seeking Mechanism #

To apply this strategy, our consumer client must be able to jump past the broker’s default read offset position and dynamically direct the read pointer to the offset recorded in our database. This task is done using the consumer.seek() API.

How to Use consumer.seek() #

The seek method lets us manually move a partition’s read pointer:

// Directing the consumer to read Partition 2 starting from Offset 501
consumer.seek(new TopicPartition("orders-topic", 2), 501);

Integration with ConsumerRebalanceListener #

Because partition ownership can move between consumers during rebalance processes, we can’t call seek only at initial application startup. We must register a ConsumerRebalanceListener object when calling the .subscribe() function.

This listener has two main methods intercepted when a rebalance happens:

  • onPartitionsRevoked: Called before our partitions are revoked. Here we must ensure all tasks on those partitions are finished writing and committed to the database.
  • onPartitionsAssigned: Called after new partitions are allocated to our consumer. This is where we must run SQL queries against the business database to find the last successfully committed offset for each of those new partitions, then call consumer.seek() for each partition before the first .poll() begins.

Java Implementation: Unified Business & Offset Transactions #

Here’s a Java code comparison between the unsafe standard offset writing (anti-pattern) versus safe transactional database writing using JDBC and a custom ConsumerRebalanceListener:

// ANTI-PATTERN: Writing to the database and committing offsets to Kafka separately
// Very prone to data duplication if the internet connection drops before the offset commit is successfully sent
public class UnsafeDatabaseWriter {
    public void consume(KafkaConsumer<String, String> consumer, Connection dbConnection) {
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                try {
                    // 1. Write business data to the database
                    saveOrderToDb(dbConnection, record.value());
                    
                    // 2. Commit to the Kafka broker separately (Non-Atomic)
                    consumer.commitSync(); // ✗ Crash risk here triggers data duplication
                } catch (Exception e) {
                    System.err.println("Failed to write data: " + e.getMessage());
                }
            }
        }
    }
    private void saveOrderToDb(Connection conn, String data) {}
}

// CORRECT: Using a Local ACID Database Transaction to unify Business Data and Kafka Offsets
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.time.Duration;
import java.util.Collection;
import java.util.Collections;
import java.util.Properties;

public class TransactionalOffsetConsumer {
    private final Connection dbConnection;
    private final KafkaConsumer<String, String> consumer;
    private final String topicName = "orders-topic";

    public TransactionalOffsetConsumer(Properties kafkaProps, Connection dbConnection) {
        this.dbConnection = dbConnection;
        // ✓ CORRECT: Disable Kafka auto-commit absolutely
        kafkaProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        this.consumer = new KafkaConsumer<>(kafkaProps);
    }

    public void start() {
        // ✓ CORRECT: Register a custom Rebalance Listener when subscribing to the topic
        consumer.subscribe(Collections.singletonList(topicName), new DatabaseOffsetRebalanceListener());

        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (TopicPartition partition : records.partitions()) {
                    for (ConsumerRecord<String, String> record : records.records(partition)) {
                        // ✓ Run the write inside a single database transaction
                        processRecordWithinDatabaseTransaction(record);
                    }
                }
            }
        } finally {
            consumer.close();
        }
    }

    private void processRecordWithinDatabaseTransaction(ConsumerRecord<String, String> record) {
        try {
            // ✓ CORRECT: Disable JDBC auto-commit to start a local database transaction
            dbConnection.setAutoCommit(false);

            // 1. Write business data
            String insertOrderSql = "INSERT INTO orders (order_id, val) VALUES (?, ?)";
            try (PreparedStatement orderStmt = dbConnection.prepareStatement(insertOrderSql)) {
                orderStmt.setString(1, record.key());
                orderStmt.setString(2, record.value());
                orderStmt.executeUpdate();
            }

            // 2. Write the Kafka offset to a special table in the same database
            String updateOffsetSql = "INSERT INTO kafka_offsets (topic, partition_id, last_offset) " +
                                     "VALUES (?, ?, ?) " +
                                     "ON CONFLICT (topic, partition_id) " +
                                     "DO UPDATE SET last_offset = EXCLUDED.last_offset";
            try (PreparedStatement offsetStmt = dbConnection.prepareStatement(updateOffsetSql)) {
                offsetStmt.setString(1, record.topic());
                offsetStmt.setInt(2, record.partition());
                // Store the current record offset + 1 (pointing to the next read position)
                offsetStmt.setLong(3, record.offset() + 1);
                offsetStmt.executeUpdate();
            }

            // ✓ CORRECT: Commit the entire local database transaction (Atomic)
            dbConnection.commit();
        } catch (SQLException e) {
            try {
                // Cancel all operations if one of the SQL commands fails
                dbConnection.rollback();
            } catch (SQLException ex) {
                System.err.println("Failed to rollback: " + ex.getMessage());
            }
            System.err.println("Transaction failed, rolling back: " + e.getMessage());
        }
    }

    // Custom Listener to direct the consumer read position during Rebalance
    private class DatabaseOffsetRebalanceListener implements ConsumerRebalanceListener {
        @Override
        public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
            // Before partitions are revoked, make sure all data has been committed to the database
            try {
                dbConnection.commit();
            } catch (SQLException e) {
                System.err.println("Failed to commit database during rebalance: " + e.getMessage());
            }
        }

        @Override
        public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
            // After new partitions are allocated, find the last offset from the database
            for (TopicPartition partition : partitions) {
                long lastCommittedOffset = getOffsetFromDatabase(partition);
                
                if (lastCommittedOffset != -1) {
                    // ✓ CORRECT: Force the consumer to move its read pointer to the database offset
                    consumer.seek(partition, lastCommittedOffset);
                } else {
                    // If the data doesn't exist in the database yet, use the default seek to beginning / end
                    consumer.seekToBeginning(Collections.singletonList(partition));
                }
            }
        }

        private long getOffsetFromDatabase(TopicPartition partition) {
            String sql = "SELECT last_offset FROM kafka_offsets WHERE topic = ? AND partition_id = ?";
            try (PreparedStatement stmt = dbConnection.prepareStatement(sql)) {
                stmt.setString(1, partition.topic());
                stmt.setInt(2, partition.partition());
                try (ResultSet rs = stmt.executeQuery()) {
                    if (rs.next()) {
                        return rs.getLong("last_offset");
                    }
                }
            } catch (SQLException e) {
                System.err.println("Failed to read offset from the database: " + e.getMessage());
            }
            return -1; // Fallback if the data is empty
        }
    }
}

Weaknesses and Complexity of the External Pattern #

Although the External Offset Storage strategy provides an absolute transactional Exactly-Once guarantee, this pattern isn’t always the top choice because it brings several technical consequences:

1. Increased Database Load #

Storing offsets in an RDBMS means adding a write query (insert/update) for every batch or every message record. At high throughput (e.g., tens of thousands of messages per second), this can trigger write bottlenecks on our RDBMS database.

2. Loss of Built-in Lag Monitoring (Consumer Lag Metric) #

Kafka’s standard built-in monitoring tools (like the kafka-consumer-groups.sh command or Prometheus exporter metrics) measure consumer lag by comparing the last offset on the broker with the commit offset in the __consumer_offsets topic.

  • Because we disable auto-commit and never send commits to the broker, the committed offset value on the broker stays at the initial number or zero.
  • The built-in lag indicator detects very high lag (false alarms), even though our consumer application is actually running normally reading offsets from the database.
  • Solution: We must create custom lag metric visualizations by making monitoring scripts that periodically compare the Kafka topic high watermark offset with the last offset stored in our kafka_offsets database table.

Summary #

  • Non-Atomic Distributed Writes: Writing offsets to the Kafka broker and writing business transactions to a local RDBMS database can’t be done atomically by default.
  • External Offset Storage: An architecture solution moving commit offset storage from the internal __consumer_offsets topic to our RDBMS database table.
  • ACID Transaction: Unifying business data modification operations and Kafka offset number updates into one local database transaction.
  • seek() API: The Kafka consumer client function to jump past broker offsets and dynamically direct the read pointer to a specific offset.
  • Rebalance Listener: The mandatory interceptor component for dynamically seeking offsets from the database right after a partition rebalance process completes.
  • Monitoring Challenge: Using the external pattern causes Kafka’s built-in lag indicator to be inaccurate, requiring custom lag metric dashboards.

← Previous: Auto vs Manual Commit Next: Consumer Group →

About | Author | Content Scope | Editorial Policy | Privacy Policy | Disclaimer | Contact