Building a Change Data Capture (CDC) System Using PostgreSQL Triggers and Java

1. Introduction

Modern applications often need to keep data synchronized across multiple databases or systems. For example, an organization might maintain its primary operational database while also maintaining a separate database for reporting, analytics, search, or downstream application processing.

One common approach is to periodically copy data from the source database to the destination database. However, repeatedly copying entire tables can be inefficient, especially when only a small percentage of records have changed.

Change Data Capture (CDC) provides a more efficient approach by identifying and capturing changes made to the source data. Instead of transferring entire tables repeatedly, a CDC system captures individual data changes, such as INSERT, UPDATE, and DELETE operations, and propagates them to another system.

In this tutorial, we will build a simple CDC pipeline using PostgreSQL and Java.

We will use PostgreSQL database triggers to capture changes made to a source table. These changes will be recorded in a dedicated CDC events table. A Java-based CDC connector will periodically read these events using JDBC and apply the corresponding changes to a destination PostgreSQL database.

The connector will also implement batch processing, pagination, checkpoint management, transaction handling, retry logic, and database connection recovery.

What We Will Build

Our implementation consists of three main components:

1. Source PostgreSQL Database: Contains an employees table. Database triggers capture INSERT, UPDATE, and DELETE operations and record the changes in a cdc_events table.
2. Java CDC Connector: Periodically polls the CDC events table, retrieves pending events, processes them in batches, and applies the changes to the destination database.
3. Destination PostgreSQL Database: Maintains a synchronized copy of the employees table and a checkpoint table that tracks the last successfully processed event.

By the end of this tutorial, we will have a working CDC pipeline that can capture database changes and propagate them to another PostgreSQL database.

We will also test the implementation by inserting, updating, and deleting employee records, and examine how the connector handles batch processing, restarts, and database connection failures.

Note: This tutorial implements trigger-based CDC. It does not read PostgreSQL's Write-Ahead Log (WAL). We will discuss the differences between trigger-based CDC and WAL-based CDC later in the tutorial.

Further Reading: Before exploring the implementation, you may find it helpful to review the Change Data Capture (CDC) Pattern, which explains the fundamental concepts, different CDC approaches, architecture, and key design considerations.

2. Understanding the Architecture

Before looking at the SQL scripts and Java implementation, let's understand the overall architecture of our CDC pipeline.

The system uses two separate PostgreSQL databases:

• Source database (cdc_source): Stores the original employee records and captures changes using database triggers.
• Destination database (cdc_destination): Stores the synchronized employee records and maintains the CDC processing checkpoint.

A Java application, called the CDC connector, runs independently of both databases. It connects to the source and destination databases using JDBC and transfers captured changes from the source to the destination.

Figure 1: Detailed architecture of the PostgreSQL trigger-based CDC pipeline.

2.1 Source PostgreSQL Database

The source database contains two important database objects:

1. employees table

This is the primary table where application data is stored. For this demonstration, we use an employee table containing fields such as employee ID, employee name, department, and salary.

Applications perform normal INSERT, UPDATE, and DELETE operations against this table.

2. cdc_events table

This table stores the changes captured by the database triggers.

Each event contains information such as:

• event_id: A unique event identifier generated using a PostgreSQL sequence.
• operation: The type of database operation (INSERT, UPDATE, or DELETE).
• record_id: The employee ID associated with the change.
• changed_data: The employee data captured as JSONB.
• created_at: The timestamp when the event was recorded.

A PostgreSQL trigger function executes after an INSERT, UPDATE, or DELETE operation on the employees table. It records the corresponding change in cdc_events.

An important property of this design is that the source data modification and the corresponding CDC event insertion occur within the same database transaction. If the source transaction rolls back, the CDC event is rolled back as well.

This prevents the event table from recording a committed change that never actually occurred in the source database.

2.2 Java CDC Connector

The Java CDC connector is responsible for moving captured changes from the source database to the destination database.

It uses JDBC to communicate with both PostgreSQL databases.

The connector performs the following operations:

1. Polling the source database

The connector periodically queries the cdc_events table for events whose IDs are greater than the last processed event ID stored in the checkpoint table.

Events are retrieved in ascending event ID order.

2. Pagination and batch processing

The connector retrieves a limited number of events in each query. This is controlled by the fetch-size configuration.

The retrieved events are then divided into smaller processing batches according to the configured batch size.

For example, if the fetch size is 4 and the batch size is 3, the connector retrieves up to four events at a time and processes them in batches of three and one.

3. Applying changes to the destination

For each event, the connector examines the operation type and performs the corresponding action against the destination employees table:

• INSERT: Inserts a new employee record.
• UPDATE: Updates an existing employee record.
• DELETE: Deletes the employee record.

4. Checkpoint management

After successfully applying a batch, the connector updates the checkpoint in the destination database.

The checkpoint records the last successfully processed event ID. When the connector restarts, it reads this value and resumes processing from the next event.

5. Error handling and recovery

The connector includes transaction rollback, retry handling with exponential backoff, and database connection recovery.

If a retryable transaction error occurs, the connector rolls back the current transaction and retries the batch. If a database connection failure occurs, the connector attempts to reconnect and resume from the stored checkpoint.

2.3 Destination PostgreSQL Database

The destination database maintains a synchronized copy of the source employees table.

It contains two important tables:

1. employees table

This table stores employee records replicated from the source database.

The Java connector applies INSERT, UPDATE, and DELETE operations to this table based on the events retrieved from the source database.

2. cdc_checkpoint table

This table maintains the connector's processing position.

It stores the connector name, the last successfully processed event ID, and the timestamp of the checkpoint update.

The destination data changes and checkpoint update are committed within the same destination database transaction. This ensures that the checkpoint does not advance independently of the corresponding data changes.

If a destination transaction fails, both the data changes and checkpoint update are rolled back.

2.4 End-to-End CDC Processing Flow

Let's walk through the complete processing flow.

1. An application inserts, updates, or deletes an employee record in the source employees table.
2. The PostgreSQL trigger executes and records the change in the source cdc_events table.
3. The Java CDC connector periodically polls the source database for events that have not yet been processed.
4. The connector retrieves events using pagination and divides them into processing batches.
5. For each event, the connector applies the corresponding INSERT, UPDATE, or DELETE operation to the destination employees table.
6. After processing a batch successfully, the connector updates the checkpoint with the last event ID in that batch.
7. The destination data changes and checkpoint update are committed together.
8. If a retryable transaction error occurs, the connector rolls back and retries the batch. If a connection failure occurs, it reconnects and resumes using the checkpoint.
9. The polling cycle continues, allowing subsequent source database changes to be synchronized.

2.5 Important Transaction Boundaries

There are two important transaction boundaries in this architecture.

Source transaction

The source employee modification and the insertion of its corresponding CDC event are part of the same source database transaction.

This ensures that the CDC event is committed only when the source data change is committed.

Destination transaction

The connector applies all employee changes in a batch and updates the checkpoint within one destination database transaction.

If any operation in the batch fails, the transaction is rolled back. This prevents the checkpoint from advancing when the corresponding batch has not been successfully committed.

These transaction boundaries are important for maintaining consistency between the source event stream and the destination data.

2.6 Architecture Summary

The overall architecture can be summarized as follows:

This design provides a practical foundation for understanding how a trigger-based CDC pipeline works.

One limitation to keep in mind is that a sequence-generated event ID does not necessarily represent transaction commit order when multiple source transactions run concurrently. We will discuss this ordering consideration and other limitations later in the tutorial.

3. Setting Up the PostgreSQL Databases

Before implementing the CDC logic, we need to set up the source and destination PostgreSQL databases.

We will use two separate databases:

• cdc_source – Stores the original employee records and captures database changes.
• cdc_destination – Stores the synchronized employee records and CDC checkpoint information.

3.1 Prerequisites

Make sure the following software is installed before proceeding.

Verify your installations

Check the Java version:

java -version 

Check Maven:

mvn -version 

Check PostgreSQL:

psql --version 

You should have PostgreSQL running and be able to connect using the PostgreSQL command-line client (psql).

Download the source code

The complete project, including the SQL scripts and Java implementation, is available in the GitHub repository:

PostgreSQL Trigger-Based CDC Demo – GitHub

Clone the repository:

git clone -b dev https://github.com/knowledgebase21st/Software-Engineering.git 

Navigate to the project directory:

cd cdc-demo 

The project contains the SQL scripts required to create the databases and tables, along with the Java CDC connector implementation.

In the next section, we will create the source PostgreSQL database and its employee table.

3.2 Creating the Source Database

The first step is to create the source PostgreSQL database named cdc_source.

This database will store the original employee records and the CDC events generated by database triggers.

Our project includes the following SQL script:

sql/01_create_source_database.sql 

SQL Script: 01_create_source_database.sql

-- ============================================================ -- File: 01_create_source_database.sql -- Purpose: Create the source database for CDC demonstration -- ============================================================ -- Create the source database if it does not already exist. SELECT 'CREATE DATABASE cdc_source' WHERE NOT EXISTS ( SELECT FROM pg_database WHERE datname = 'cdc_source' )\gexec -- Verify database creation. SELECT datname FROM pg_database WHERE datname = 'cdc_source'; 

Understanding the SQL script

Let's look at the important parts.

1. Checking whether the database exists

SELECT 'CREATE DATABASE cdc_source' WHERE NOT EXISTS ( SELECT FROM pg_database WHERE datname = 'cdc_source' ) 

PostgreSQL maintains a system catalog named pg_database, which contains information about databases in the PostgreSQL server.

The query checks whether a database named cdc_source already exists.

If it does not exist, the query returns the text:

CREATE DATABASE cdc_source 

If the database already exists, the query returns no rows.

2. Executing the CREATE DATABASE command

\gexec 

\gexec is a PostgreSQL psql meta-command. It executes the SQL statements returned by the preceding query.

In this case, it executes CREATE DATABASE cdc_source only when the database does not already exist.

This makes the script safe to run multiple times without attempting to create an existing database.

3. Verifying database creation

SELECT datname FROM pg_database WHERE datname = 'cdc_source'; 

This query checks the PostgreSQL system catalog again and displays the database name if it exists.

Executing the script

Connect to the default PostgreSQL database (postgres) using psql:

psql -U postgres -d postgres 

You can execute the SQL script from the terminal using:

psql -U postgres -d postgres -f sql/01_create_source_database.sql 

Make sure you run this command from the project root directory.

Expected output

If the database does not already exist, the script creates it and displays its name during verification.

 datname ------------ cdc_source (1 row) 

If the database already exists, the creation step is skipped and the verification query still displays cdc_source.

The source database is now ready. In the next section, we will create the employees table and the CDC events table.

Technical Note: The \gexec command is a psql meta-command, not a standard SQL statement. Therefore, this script should be executed using the PostgreSQL command-line client (psql). It will not work directly in pgAdmin's Query Tool or other SQL editors that do not support psql meta-commands.

3.3 Creating the Source Tables

After creating the cdc_source database, the next step is to create the tables required for the CDC pipeline.

We will create two tables:

1. employees – Stores the original employee records. This is the source table where INSERT, UPDATE, and DELETE operations will occur.
2. cdc_events – Stores the change events captured by the PostgreSQL trigger. The Java CDC connector will periodically read these events and apply the corresponding changes to the destination database.

Step 1: Create the Tables

The following SQL script creates both tables.

File: sql/02_create_source_tables.sql

-- ============================================================ -- File: 02_create_source_tables.sql -- Purpose: Create source employee and CDC event tables -- Database: cdc_source -- ============================================================ -- Create the employees table. CREATE TABLE IF NOT EXISTS employees ( id SERIAL PRIMARY KEY, name VARCHAR(100) NOT NULL, department VARCHAR(100), salary NUMERIC(10,2) ); -- Create the CDC events table. CREATE TABLE IF NOT EXISTS cdc_events ( event_id BIGSERIAL PRIMARY KEY, operation VARCHAR(10) NOT NULL, record_id INTEGER NOT NULL, changed_data JSONB, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); -- Verify table creation. SELECT table_name FROM information_schema.tables WHERE table_schema = 'public' AND table_name IN ('employees', 'cdc_events') ORDER BY table_name; 

Step 2: Understanding the Employees Table

The employees table stores the original employee data.

The id column uniquely identifies each employee record. It will also be used by the CDC connector to identify which destination record needs to be inserted, updated, or deleted.

Step 3: Understanding the CDC Events Table

The cdc_events table acts as the change log for our CDC pipeline. Whenever an employee record is inserted, updated, or deleted, the PostgreSQL trigger will capture the operation and store an event in this table.

A few important details:

• BIGSERIAL: Automatically generates a unique event ID using a sequence. The Java connector uses this ID when reading and tracking CDC events.
• JSONB: PostgreSQL's binary JSON data type. It allows the captured employee data to be stored in a structured format.
• operation: Identifies the type of database change so the connector knows which operation to apply to the destination table.
• record_id: Identifies the employee record associated with the event.

The trigger function that populates this table will be implemented in Section 4.

Step 4: Execute the SQL Script

Connect to the cdc_source database using psql:

psql -U postgres -d cdc_source 

Alternatively, execute the script directly from the project root directory:

psql -U postgres -d cdc_source -f sql/02_create_source_tables.sql 

Step 5: Verify Table Creation

The final query in the script checks the PostgreSQL information_schema.tables catalog to confirm that both tables exist in the public schema.

You should see output similar to the following:

 table_name ------------- cdc_events employees (2 rows) 

Both tables are now ready. In the next section, we will create the destination database and prepare it to receive the synchronized employee records.

3.4 Creating the Destination Database

Now that the source database and its tables are ready, the next step is to create the destination database.

The destination database, cdc_destination, will store the employee records synchronized from the source database by our Java CDC connector.

The connector reads change events from cdc_source and applies the corresponding INSERT, UPDATE, and DELETE operations to the destination database.

Step 1: Create the Destination Database

The following SQL script creates the cdc_destination database if it does not already exist.

File: sql/05_create_destination_database.sql

-- ============================================================ -- File: 05_create_destination_database.sql -- Purpose: Create the destination database for CDC -- ============================================================ -- Create the destination database if it does not already exist. SELECT 'CREATE DATABASE cdc_destination' WHERE NOT EXISTS ( SELECT FROM pg_database WHERE datname = 'cdc_destination' )\gexec -- Verify database creation. SELECT datname FROM pg_database WHERE datname = 'cdc_destination'; 

Step 2: Understanding the SQL Script

The script uses the PostgreSQL system catalog pg_database to check whether a database named cdc_destination already exists.

Let's look at the important parts:

• Database existence check: The SELECT statement checks pg_database for an existing database named cdc_destination.
• Conditional creation: If the database does not exist, the query returns the SQL statement CREATE DATABASE cdc_destination.
• \gexec: This psql meta-command executes the SQL statement returned by the query. If the database already exists, no statement is returned, so no creation is attempted.
• Verification: The final query checks pg_database again and displays the destination database name.

This approach makes the script safe to execute more than once without attempting to recreate an existing database.

Step 3: Execute the SQL Script

Connect to the default PostgreSQL database (postgres) using psql:

psql -U postgres -d postgres 

Alternatively, execute the script directly from the project root directory:

psql -U postgres -d postgres -f sql/05_create_destination_database.sql 

Step 4: Verify Database Creation

After successful execution, the verification query should display output similar to the following:

 datname ----------------- cdc_destination (1 row) 

The cdc_destination database is now ready.

Technical Note: Just like the source database creation script, this script uses \gexec, which is a psql meta-command rather than standard SQL. Execute it using the PostgreSQL command-line client (psql). It will not work directly in pgAdmin's Query Tool or other SQL editors that do not support psql meta-commands.

In the next section, we will create the destination tables required to store synchronized employee records and maintain CDC checkpoint information.

3.5 Creating the Destination Tables

Now that the cdc_destination database has been created, we need to create the tables required by the CDC pipeline.

We will create two tables:

1. employees – Stores employee records synchronized from the source database.
2. cdc_checkpoint – Maintains the last processed CDC event ID, allowing the Java connector to resume processing from its previous position after a restart.

Step 1: Create the Destination Employees Table

The destination employees table has the same columns as the source employees table. However, there is one important difference: the id column is defined as INTEGER PRIMARY KEY instead of SERIAL PRIMARY KEY.

The destination receives employee IDs from the source database, so it does not need to generate new IDs.

File: sql/06_create_destination_tables.sql

-- ============================================================ -- File: 06_create_destination_tables.sql -- Purpose: Create destination employee table -- Database: cdc_destination -- ============================================================ -- Create the destination employees table. CREATE TABLE IF NOT EXISTS employees ( id INTEGER PRIMARY KEY, name VARCHAR(100) NOT NULL, department VARCHAR(100), salary NUMERIC(10,2) ); -- Verify table creation. SELECT table_name FROM information_schema.tables WHERE table_schema = 'public' AND table_name = 'employees'; 

Step 2: Understanding the Destination Employees Table

The destination table stores the employee records replicated from the source database.

Why is id defined as INTEGER instead of SERIAL?

In the source database, PostgreSQL automatically generates employee IDs using SERIAL.

However, the destination database must preserve the original employee IDs. The Java connector reads the employee ID from the CDC event and uses that same ID when inserting or updating the destination record.

Therefore, the destination table uses INTEGER PRIMARY KEY rather than generating its own IDs.

Step 3: Create the CDC Checkpoint Table

The cdc_checkpoint table maintains the processing position of the Java CDC connector.

The connector uses this table to determine which CDC events have already been processed. After successfully applying a batch of changes to the destination database, it updates the checkpoint with the last processed event ID.

File: sql/07_create_checkpoint_table.sql

-- ============================================================ -- File: 07_create_checkpoint_table.sql -- Purpose: Create CDC connector checkpoint table -- Database: cdc_destination -- ============================================================ CREATE TABLE IF NOT EXISTS cdc_checkpoint ( connector_name VARCHAR(100) PRIMARY KEY, last_processed_event_id BIGINT NOT NULL DEFAULT 0, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); -- Initialize the checkpoint for our CDC connector. INSERT INTO cdc_checkpoint ( connector_name, last_processed_event_id ) VALUES ( 'employee-cdc-connector', 0 ) ON CONFLICT (connector_name) DO NOTHING; -- Verify checkpoint initialization. SELECT * FROM cdc_checkpoint WHERE connector_name = 'employee-cdc-connector'; 

Step 4: Understanding the CDC Checkpoint Table

Let's understand the purpose of each column.

A few important details:

• connector_name: Identifies the connector whose processing position is being tracked. In our implementation, the connector name is employee-cdc-connector.
• last_processed_event_id: Initially set to 0. As the connector processes CDC events, this value is updated to reflect the last processed event ID.
• updated_at: Defaults to the current timestamp when the checkpoint row is inserted.

The initialization statement uses:

ON CONFLICT (connector_name) DO NOTHING; 

This prevents duplicate checkpoint entries if the initialization script is executed more than once. If a checkpoint row already exists for the connector, its existing value is preserved.

The Java connector will use this checkpoint to resume processing after a restart.

Step 5: Execute the SQL Scripts

Connect to the cdc_destination database using psql:

psql -U postgres -d cdc_destination 

Alternatively, execute both scripts directly from the project root directory:

psql -U postgres -d cdc_destination -f sql/06_create_destination_tables.sql 

psql -U postgres -d cdc_destination -f sql/07_create_checkpoint_table.sql 

Execute the scripts in this order: first create the destination employees table, then create and initialize the checkpoint table.

Step 6: Verify Table Creation and Checkpoint Initialization

The first script verifies that the destination employees table exists.

Expected output:

 table_name ------------- employees (1 row) 

The second script verifies that the checkpoint entry has been initialized.

Expected output will be similar to:

 connector_name | last_processed_event_id | updated_at ------------------------+-------------------------+---------------------------- employee-cdc-connector | 0 | 2026-11-06 ... (1 row) 

The timestamp will reflect the time when the checkpoint row was created.

Both destination tables are now ready.

Summary

At this point, we have created the source and destination databases and all the tables required for our CDC pipeline.

In the next section, we will implement the PostgreSQL trigger-based CDC mechanism that captures employee INSERT, UPDATE, and DELETE operations and records them in the cdc_events table.

4. Implementing CDC Using PostgreSQL Triggers

In the previous section, we created the source and destination databases along with the tables required for our CDC pipeline.

Now we will implement the change capture mechanism using PostgreSQL triggers.

PostgreSQL triggers allow us to execute a function automatically when specified database operations occur. In our implementation, a trigger on the source employees table captures INSERT, UPDATE, and DELETE operations and records the corresponding changes in the cdc_events table.

We will implement this using two SQL scripts:

1. 03_create_cdc_function.sql – Defines the trigger function that captures employee changes.
2. 04_create_cdc_trigger.sql – Creates the trigger that invokes the function whenever an employee record is modified.

4.1 Creating the CDC Trigger Function

The first step is to create a PostgreSQL trigger function named capture_employee_changes().

This function determines which database operation occurred and records the corresponding employee data in the cdc_events table.

File: sql/03_create_cdc_function.sql

-- ============================================================ -- File: 03_create_cdc_function.sql -- Purpose: Create the CDC trigger function -- Database: cdc_source -- ============================================================ CREATE OR REPLACE FUNCTION capture_employee_changes() RETURNS TRIGGER AS $$ BEGIN IF (TG_OP = 'INSERT') THEN INSERT INTO cdc_events ( operation, record_id, changed_data ) VALUES ( 'INSERT', NEW.id, to_jsonb(NEW) ); RETURN NEW; ELSIF (TG_OP = 'UPDATE') THEN INSERT INTO cdc_events ( operation, record_id, changed_data ) VALUES ( 'UPDATE', NEW.id, to_jsonb(NEW) ); RETURN NEW; ELSIF (TG_OP = 'DELETE') THEN INSERT INTO cdc_events ( operation, record_id, changed_data ) VALUES ( 'DELETE', OLD.id, to_jsonb(OLD) ); RETURN OLD; END IF; Source：https://dev.to/sanjayghosh/building-a-change-data-capture-cdc-system-using-postgresql-triggers-and-java-ggi
My favorites

Related posts

Designing an AI incident agent for 3 a.m.: making memory visible

Picture the person who uses an incident tool. It's the middle of the night. An alert just woke them up. Customers are af

0dev.to UX

From Best Buy To Meta Lab: The Current XR Retail Experience

Von der Brille zur VR-Arena: Wie ein Quest-Spiel mein Amblyopie-Training veränderte – ein technischer und persönlicher R

0dev.to UX

Stamp an Expiry on Every AI Approval Before the Agent Acts

At 9:40 on a Monday, a support agent approved a refund. The evidence card looked clean and complete. The order state pri

0dev.to UX

Same AI, two answers: what happened when we gave our on-call agent a memory

Picture the person who uses an incident tool. It's the middle of the night. An alert just woke them up. Customers are af

0dev.to UX

Same AI, two answers: what happened when we gave our on-call agent a memory.

Picture the person who uses an incident tool. It's the middle of the night. An alert just woke them up. Customers are af

0dev.to UX

Same AI, two answers: what happened when we gave our on-call agent a memory..

When people talk about making AI better, they usually talk about bigger models. More parameters, longer context, a newer

0dev.to UX