Source Offsets and Restart Behavior

Summary

Source connector offsets are checkpoints that track where the connector has read up to in the external system. When a connector restarts, it uses these offsets to resume from exactly where it left off, preventing data loss or duplication.

WHY OFFSETS MATTER

Offsets are crucial for reliability and fault tolerance in Kafka Connect.

Let's see what would happen without proper offset management.

Without offsets, a connector restart creates an impossible choice: re-process all data and create duplicates, skip ahead and lose data, or manually intervene every time. Offsets solve this by automatically tracking progress.

HOW SOURCE OFFSETS WORK

Source connectors define their own offset format based on what they're reading. The offset structure varies by connector type because different systems have different ways of tracking position.

Each connector uses offset keys and values that make sense for its source system. FileStream uses filename and byte position. JDBC uses table names and timestamp or ID values. Debezium uses database-specific log positions.

OFFSET STORAGE LOCATIONS

Where are offsets stored? It depends on whether you're running in standalone or distributed mode.

In standalone mode, offsets are stored in a local file specified in the worker configuration. This file is simple but not fault-tolerant - if the file is lost, offsets are lost.

In distributed mode, offsets are stored in an internal Kafka topic, typically named "connect-offsets". This provides fault tolerance, replication, and enables multiple workers to coordinate.

OFFSET COMMIT FLOW

Let's trace how offsets are committed during normal connector operation.

The connector reads a batch of data, produces messages to Kafka, and only commits the offset after receiving acknowledgments from Kafka. This ensures that offsets always represent data that has been successfully written to Kafka.

RESTART BEHAVIOR

When a source connector restarts, it reads its last committed offset and resumes from that position.

If an offset exists, the connector resumes from that exact position. If no offset is found (first run or offset was lost), the connector either starts from the beginning or uses a configured starting position.

EXAMPLE: JDBC SOURCE OFFSETS

The following shows a concrete example with JDBC Source Connector offsets in timestamp+incrementing mode.

Offset Key

1{
2  "table": "users"
3}

Offset Value

1{
2  "timestamp": "2024-01-15 10:30:00.000",
3  "incrementing": 1543
4}

The offset key identifies which table this offset belongs to. The offset value contains both the last timestamp and the last incrementing ID value seen. On restart, the connector will query for rows where timestamp >= '2024-01-15 10:30:00.000' AND id > 1543.

EXAMPLE: DEBEZIUM OFFSETS

Debezium offsets are more complex because they track exact positions in database transaction logs.

Debezium MySQL Offset Key

1{
2  "server": "mysql-server-1"
3}

Debezium MySQL Offset Value

1{
2  "file": "mysql-bin.000042",
3  "pos": 154897,
4  "row": 1,
5  "server_id": 223344,
6  "event": 2
7}

The offset contains the binlog filename, byte position within that file, row number, and other metadata. This allows Debezium to resume reading from the exact transaction in the binlog.

OFFSET COMMIT FREQUENCY

How often should offsets be committed? This is controlled by the offset.flush.interval.ms configuration in the worker properties.

Frequent offset commits mean less data to re-process if the connector restarts, but create more storage overhead. Infrequent commits improve performance but mean more potential duplicates on restart. The default is typically 60 seconds, which balances both concerns.

HANDLING OFFSET CORRUPTION OR LOSS

Sometimes offsets can become corrupted or lost. There are strategies to handle this.

Your recovery strategy depends on your requirements:

Reset to beginning: Use when data volume is manageable and duplicates can be handled or deduplicated downstream.

Reset to timestamp: For JDBC connectors, you can configure a starting timestamp to skip old data.

Manual offset edit: In distributed mode, you can use tools to manually edit the offset topic. This requires understanding the offset format.

Start fresh: Accept that historical data is lost and start from current time. Only use when historical data isn't critical.

BEST PRACTICES FOR OFFSET MANAGEMENT

Monitor Offset Lag: Track how far behind your connector is from the source system. Large or growing lag indicates problems.

Backup Offset Storage: In standalone mode, backup the offset file regularly. In distributed mode, ensure the offset topic has sufficient replication.

Test Restart Behavior: Regularly test that connectors restart correctly and don't create duplicates or lose data.

Configure Appropriate Flush Intervals: Balance between performance and potential duplicate data on restart.

Use Distributed Mode for Production: The fault-tolerant offset storage is worth the additional complexity.

Idempotent Downstream Processing: Design downstream consumers to handle duplicates gracefully, as exactly-once isn't always guaranteed.

Let's recap what we learned about source connector offsets and restart behavior:

Key ideas

Source connectors use offsets to track their position in external systems, enabling fault-tolerant restarts.

Offset format varies by connector type - JDBC uses timestamps/IDs, Debezium uses log positions, FileStream uses file positions.

In standalone mode, offsets are stored in local files. In distributed mode, they're stored in internal Kafka topics.

Offsets are committed after messages are successfully produced to Kafka, ensuring data isn't lost.

On restart, connectors read their last offset and resume from that exact position.

Offset commit frequency balances performance against potential duplicate data on restart.

Proper offset management is critical for reliable, production-ready data pipelines.

This completes Chapter 5 on Source Connectors! In the next chapter, we'll explore Sink Connectors and how data flows from Kafka into external systems.