JDBC Source Connector

Summary

The JDBC Source Connector connects to any database with a JDBC driver and continuously polls for new or updated rows. It's perfect for database replication, data warehousing, and building event-driven architectures from database changes.

WHAT IS THE JDBC SOURCE CONNECTOR?

The JDBC Source Connector uses standard JDBC (Java Database Connectivity) to query databases and stream results to Kafka. It supports any database with a JDBC driver, including MySQL, PostgreSQL, Oracle, SQL Server, and many others.

Let's visualize how the JDBC Source Connector works. It queries the database periodically and produces results as Kafka messages.

The connector queries the database at regular intervals defined by the poll interval. It uses various strategies to detect new or updated rows, which we'll explore shortly.

QUERY MODES

The JDBC Source Connector supports multiple query modes to detect changes in your database.

There are three primary modes for querying data.

Bulk Mode: Queries the entire table on each poll. This mode doesn't track offsets and will re-read all data every time. Use this only for small, static lookup tables.

Incrementing Mode: Uses an auto-incrementing column (like an ID) to track progress. The connector remembers the highest ID it has seen and queries for rows with IDs greater than that value. This mode works well for insert-only tables but won't detect updates.

Timestamp Mode: Uses a timestamp column (like created_at or updated_at) to detect new or modified rows. The connector tracks the latest timestamp and queries for rows with timestamps greater than or equal to that value. This mode detects both inserts and updates.

Timestamp + Incrementing Mode: Combines both approaches for maximum reliability. It uses timestamps to detect changes and IDs to handle multiple rows with the same timestamp. This is the recommended mode for most use cases.

INCREMENTING MODE EXAMPLE

Let's see how incrementing mode works in practice.

The connector remembers the highest ID value and only queries for rows with larger IDs. This is efficient and avoids re-reading the entire table on each poll.

TIMESTAMP MODE EXAMPLE

look at timestamp mode, which can detect both new and updated rows.

Timestamp mode captures updates to existing rows because the updated_at timestamp changes when a row is modified. This makes it ideal for change data capture scenarios where you need to track all changes.

CONNECTOR CONFIGURATION

The following shows a typical JDBC Source Connector configuration.

1{
2  "name": "jdbc-source-users",
3  "config": {
4    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
5    "tasks.max": "1",
6    "connection.url": "jdbc:mysql://localhost:3306/mydb",
7    "connection.user": "kafka_connect",
8    "connection.password": "password",
9    "table.whitelist": "users,orders",
10    "mode": "timestamp+incrementing",
11    "timestamp.column.name": "updated_at",
12    "incrementing.column.name": "id",
13    "topic.prefix": "mysql-",
14    "poll.interval.ms": "5000"
15  }
16}

connector.class: The JDBC Source Connector class from Confluent or another provider.

connection.url: JDBC connection string for your database.

connection.user and connection.password: Database credentials.

table.whitelist: Comma-separated list of tables to ingest. Alternatively, use table.blacklist to exclude specific tables.

mode: The query mode - incrementing, timestamp, timestamp+incrementing, or bulk.

timestamp.column.name: The timestamp column to use for timestamp mode.

incrementing.column.name: The auto-incrementing column to use for incrementing mode.

topic.prefix: Prefix for Kafka topic names. Each table gets its own topic, like mysql-users and mysql-orders.

poll.interval.ms: How often to poll the database for changes, in milliseconds.

SCHEMA EVOLUTION

The JDBC Source Connector automatically detects database schema changes and updates the Kafka message schema accordingly.

When you add a column to your database table, the JDBC Source Connector automatically includes it in subsequent messages. If you're using Avro and Schema Registry, the new schema version is registered automatically.

PERFORMANCE CONSIDERATIONS

When using the JDBC Source Connector at scale, there are several performance considerations.

Batching: The connector reads multiple rows per poll and produces them in batches to Kafka for better throughput.

Parallelism: You can increase tasks.max to parallelize ingestion across multiple tables. Each task handles a subset of tables.

Query Optimization: Ensure your incrementing and timestamp columns are indexed in the database to make queries efficient.

Poll Interval: Balance between freshness and database load. Shorter intervals mean lower latency but more database queries.

By using multiple tasks, you can ingest from multiple tables concurrently, improving overall throughput.

Let's recap what we learned about the JDBC Source Connector:

Key ideas

The JDBC Source Connector streams data from relational databases to Kafka using standard JDBC.

It supports multiple query modes: bulk, incrementing, timestamp, and timestamp+incrementing for different use cases.

Incrementing mode uses auto-increment IDs to track new rows. Timestamp mode detects both new and updated rows using timestamp columns.

Configuration is flexible, supporting table whitelisting, custom polling intervals, and parallel task execution.

Schema evolution is handled automatically when database schemas change.

For production use, optimize queries with proper indexes and balance poll intervals for freshness versus database load.