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:

Component	Responsibility
Source employees	Stores original employee records
PostgreSQL trigger	Captures INSERT, UPDATE, and DELETE operations
Source cdc_events	Stores captured database changes
Java CDC connector	Polls, processes, and transfers events
Destination employees	Stores synchronized employee records
Destination cdc_checkpoint	Tracks successfully processed events

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.

Software	Requirement	Purpose
Java	JDK 17 or later	Runs the Java CDC connector
Maven	Maven 3.x	Builds the Java application and manages dependencies
PostgreSQL	PostgreSQL 16.x or compatible	Provides the source and destination databases
Git	Optional	Clones the project repository

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.

Column	Description
id	Primary key. Uses SERIAL to generate employee IDs automatically.
name	Employee name. Cannot be NULL.
department	Employee's department.
salary	Employee salary, stored with two decimal places.

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.

Column	Description
event_id	Primary key. Uses BIGSERIAL to generate a unique event ID.
operation	Identifies the operation: INSERT, UPDATE, or DELETE.
record_id	Contains the ID of the employee record affected by the operation.
changed_data	Stores the captured record data in PostgreSQL's JSONB format.
created_at	Records when the event was created. Defaults to the current timestamp.

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.

Column	Description
id	Primary key. Stores the employee ID received from the source database.
name	Employee name. Cannot be NULL.
department	Employee's department.
salary	Employee salary, stored with two decimal places.

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.

Column	Description
connector_name	Primary key identifying the CDC connector.
last_processed_event_id	Stores the ID of the last successfully processed CDC event.
updated_at	Records when the checkpoint row was initially created.

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.

Database	Table	Purpose
cdc_source	employees	Stores original employee records.
cdc_source	cdc_events	Stores captured database change events.
cdc_destination	employees	Stores synchronized employee records.
cdc_destination	cdc_checkpoint	Tracks the connector's processing position.

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;

    RETURN NULL;

END;
$$ LANGUAGE plpgsql;

-- Verify function creation.
SELECT routine_name
FROM information_schema.routines
WHERE routine_schema = 'public'
  AND routine_name = 'capture_employee_changes';

Understanding the Trigger Function

Let's understand the important parts of this function.

1. Function Declaration

CREATE OR REPLACE FUNCTION capture_employee_changes()
RETURNS TRIGGER AS $$

• CREATE OR REPLACE FUNCTION creates the function or replaces its existing definition.
• capture_employee_changes() is the function name.
• RETURNS TRIGGER indicates that this function is designed to be invoked by a PostgreSQL trigger.
• $$ marks the beginning and end of the function body.

The function is written using PostgreSQL's procedural language, PL/pgSQL.

2. Identifying the Database Operation

The function uses the PostgreSQL trigger variable TG_OP to determine which operation caused the trigger to execute.

It checks for three possible operations:

• INSERT
• UPDATE
• DELETE

Based on the operation, the function captures the appropriate employee record and inserts a corresponding event into cdc_events.

3. Handling INSERT Operations

IF (TG_OP = 'INSERT') THEN

    INSERT INTO cdc_events (
        operation,
        record_id,
        changed_data
    )
    VALUES (
        'INSERT',
        NEW.id,
        to_jsonb(NEW)
    );

    RETURN NEW;

When a new employee record is inserted:

• TG_OP identifies the operation as INSERT.
• NEW represents the newly inserted employee record.
• NEW.id retrieves the employee ID.
• to_jsonb(NEW) converts the complete employee record into PostgreSQL's JSONB format.
• A new event is inserted into cdc_events with the operation type INSERT.

4. Handling UPDATE Operations

ELSIF (TG_OP = 'UPDATE') THEN

    INSERT INTO cdc_events (
        operation,
        record_id,
        changed_data
    )
    VALUES (
        'UPDATE',
        NEW.id,
        to_jsonb(NEW)
    );

    RETURN NEW;

When an existing employee record is updated:

• TG_OP identifies the operation as UPDATE.
• NEW represents the updated employee record.
• NEW.id identifies the employee.
• to_jsonb(NEW) captures the updated employee data.
• A new event is inserted into cdc_events with the operation type UPDATE.

Our implementation captures the complete updated record rather than only the columns that changed.

5. Handling DELETE Operations

ELSIF (TG_OP = 'DELETE') THEN

    INSERT INTO cdc_events (
        operation,
        record_id,
        changed_data
    )
    VALUES (
        'DELETE',
        OLD.id,
        to_jsonb(OLD)
    );

    RETURN OLD;

When an employee record is deleted:

• TG_OP identifies the operation as DELETE.
• OLD represents the employee record before deletion.
• OLD.id retrieves the employee ID.
• to_jsonb(OLD) converts the deleted record into JSONB format.
• A new event is inserted into cdc_events with the operation type DELETE.

Unlike INSERT and UPDATE, DELETE uses OLD because the record no longer exists in the source table after the deletion.

6. Understanding NEW and OLD

PostgreSQL provides two special variables that are particularly important when working with row-level triggers.

Variable	Description
NEW	Represents the new row being inserted or the updated row after an UPDATE operation.
OLD	Represents the existing row before an UPDATE or DELETE operation.

In our implementation:

• INSERT uses NEW.
• UPDATE uses NEW.
• DELETE uses OLD.

7. Returning from the Function

The function returns NEW for INSERT and UPDATE operations and OLD for DELETE operations.

It also contains a final RETURN NULL as a fallback if none of the three operation conditions matches.

Because this is an AFTER trigger, PostgreSQL does not use the returned row to modify the source table. The return value is effectively ignored for this trigger type.

Execute the SQL Script

Connect to the source database:

psql -U postgres -d cdc_source

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

psql -U postgres -d cdc_source -f sql/03_create_cdc_function.sql

The final query in the script verifies that the function was created successfully.

Expected output:

       routine_name
---------------------------
 capture_employee_changes
(1 row)

The CDC trigger function is now ready.

4.2 Creating the CDC Trigger

Creating the trigger function alone does not activate change capture. We must also create a trigger on the source employees table that invokes the function whenever an employee record is inserted, updated, or deleted.

File: sql/04_create_cdc_trigger.sql

-- ============================================================
-- File: 04_create_cdc_trigger.sql
-- Purpose: Create the CDC trigger on employees table
-- Database: cdc_source
-- ============================================================

-- Remove the trigger if it already exists.
DROP TRIGGER IF EXISTS employee_changes_trigger
ON employees;

-- Create the CDC trigger.
CREATE TRIGGER employee_changes_trigger
AFTER INSERT OR UPDATE OR DELETE
ON employees
FOR EACH ROW
EXECUTE FUNCTION capture_employee_changes();

-- Verify trigger creation.
SELECT trigger_name, event_manipulation
FROM information_schema.triggers
WHERE event_object_table = 'employees'
ORDER BY event_manipulation;

Understanding the Trigger Definition

Let's examine the important statements.

1. Removing an Existing Trigger

DROP TRIGGER IF EXISTS employee_changes_trigger
ON employees;

This statement removes the existing trigger if it is already present on the employees table.

Using IF EXISTS prevents an error if the trigger does not exist.

This makes the script easier to execute repeatedly.

2. Creating the Trigger

CREATE TRIGGER employee_changes_trigger
AFTER INSERT OR UPDATE OR DELETE
ON employees
FOR EACH ROW
EXECUTE FUNCTION capture_employee_changes();

Let's break down this statement:

Component	Description
employee_changes_trigger	Name of the trigger.
AFTER	Executes the trigger after the specified database operation.
INSERT OR UPDATE OR DELETE	Specifies the operations that activate the trigger.
ON employees	Attaches the trigger to the source employees table.
FOR EACH ROW	Executes the trigger separately for every affected row.
EXECUTE FUNCTION	Invokes the capture_employee_changes() function.

Because this is a row-level trigger, the function executes once for each affected row.

For example, if an UPDATE statement modifies five employee records, the trigger function executes five times, generating five CDC events.

3. Verifying Trigger Creation

The final query checks PostgreSQL's information_schema.triggers view to verify that the trigger exists and identify the operations associated with it.

Execute the SQL Script

Run the following command from the project root directory:

psql -U postgres -d cdc_source -f sql/04_create_cdc_trigger.sql

The verification query should return the trigger name and its associated operations:

       trigger_name       | event_manipulation
--------------------------+-------------------
 employee_changes_trigger | DELETE
 employee_changes_trigger | INSERT
 employee_changes_trigger | UPDATE
(3 rows)

The trigger is now installed on the source employees table.

4.3 Understanding the Transactional Behavior

An important characteristic of our implementation is that the employee change and the corresponding CDC event are written within the same source database transaction.

For example, when an application inserts a new employee:

1. PostgreSQL inserts the employee record into employees.
2. The AFTER INSERT trigger executes.
3. The trigger function inserts the corresponding event into cdc_events.
4. Both operations participate in the same transaction.

If the source transaction is committed, both the employee record and its CDC event are committed. If the transaction is rolled back, both are rolled back.

This ensures that the CDC event is not committed independently of the corresponding source database change.

The Java CDC connector will read these events from cdc_events and apply them to the destination database. We will implement that connector in a later section.

Summary

In this section, we implemented PostgreSQL trigger-based change capture.

• Created the capture_employee_changes() trigger function.
• Implemented handling for INSERT, UPDATE, and DELETE operations.
• Created the employee_changes_trigger on the source employees table.
• Configured the trigger to execute once for every affected row.
• Ensured that source changes and CDC event records participate in the same transaction.

Our PostgreSQL source database can now capture employee changes automatically. In the next section, we will begin implementing the Java CDC connector that reads these events and synchronizes them with the destination database.

5. Implementing the Java CDC Connector

In the previous section, we implemented PostgreSQL triggers to capture INSERT, UPDATE, and DELETE operations on the source employees table. These changes are recorded in the cdc_events table.

Now we will implement the Java CDC connector that reads these events and applies the corresponding changes to the destination database.

The connector uses JDBC to communicate with both PostgreSQL databases and Jackson to process the JSONB data stored in the CDC events.

5.1 Understanding the Connector Structure and Configuration

Before implementing the CDC processing logic, let's understand the structure of the Java connector and how its configuration is managed.

The main Java class is named CDCConnector and belongs to the following package:

package com.clockwise.cdc;

The connector uses several Java standard libraries and the Jackson library.

Required Imports

The following imports are used in the implementation:

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;

import java.util.ArrayList;
import java.util.List;
import java.math.BigDecimal;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Types;

import java.io.FileInputStream;
import java.io.IOException;
import java.util.Properties;

Here is the purpose of the important classes:

Class	Purpose
Connection	Represents a connection to a PostgreSQL database.
DriverManager	Establishes database connections using JDBC.
PreparedStatement	Executes parameterized SQL statements.
ResultSet	Reads query results from the database.
SQLException	Handles database-related exceptions.
JsonNode	Represents JSON data retrieved from CDC events.
ObjectMapper	Parses JSON strings into Jackson JSON objects.
BigDecimal	Handles employee salary values with decimal precision.
Properties	Loads configuration settings from a properties file.
ArrayList and List	Store and manage fetched CDC events.

Loading Configuration from application.properties

Rather than hardcoding database connection details and processing parameters throughout the Java code, the connector loads its configuration from an external file named application.properties.

The following configuration is used by our project:

File: application.properties

source.db.url=jdbc:postgresql://localhost:5432/cdc_source
destination.db.url=jdbc:postgresql://localhost:5432/cdc_destination
db.username=postgres

cdc.polling.interval.ms=2000
cdc.batch.size=3
cdc.fetch.size=4
cdc.max.retries=3
cdc.retry.delay.ms=1000
cdc.reconnect.delay.ms=5000

Let's understand these properties:

Property	Description
source.db.url	JDBC URL for connecting to the source database.
destination.db.url	JDBC URL for connecting to the destination database.
db.username	Database username used for both connections.
cdc.polling.interval.ms	Time interval between polling cycles, in milliseconds.
cdc.batch.size	Number of CDC events processed in a destination transaction.
cdc.fetch.size	Maximum number of CDC events retrieved in one source query.
cdc.max.retries	Maximum number of retries after the initial transaction attempt.
cdc.retry.delay.ms	Initial delay between transaction retry attempts.
cdc.reconnect.delay.ms	Delay before attempting to reconnect after a database connection failure.

The values shown above are the configuration used in our demonstration. We will examine batch processing, pagination, retries, and reconnection in more detail in the subsequent sections.

Reading the Configuration File

The connector loads the properties using the loadConfiguration() method.

private static Properties loadConfiguration() {

    Properties properties = new Properties();

    try (FileInputStream input =
                 new FileInputStream("application.properties")) {

        properties.load(input);

        System.out.println(
                "Configuration loaded from application.properties"
        );

    } catch (IOException e) {

        throw new IllegalStateException(
                "Unable to load application.properties", e
        );
    }

    return properties;
}

The method performs the following steps:

1. Creates a Properties object.
2. Opens application.properties using FileInputStream.
3. Loads the configuration values using properties.load().
4. Returns the populated Properties object.

If the configuration file cannot be opened or read, the method throws an IllegalStateException, preventing the connector from starting with missing configuration.

The file is opened using a relative path, so the connector expects application.properties to be available in its working directory.

Defining Configuration Constants

The connector loads the configuration once when the class is initialized and stores the values in static constants.

For example:

private static final Properties CONFIG = loadConfiguration();

private static final String SOURCE_URL =
        CONFIG.getProperty("source.db.url");

private static final String DESTINATION_URL =
        CONFIG.getProperty("destination.db.url");

private static final String DB_USER =
        CONFIG.getProperty("db.username", "postgres");

The getProperty() method retrieves the configured values. For the database username, the connector uses postgres as the default if the property is not specified.

The database password is handled separately:

private static final String DB_PASSWORD =
        System.getenv("DB_PASSWORD");

Instead of storing the database password in application.properties, the connector reads it from the DB_PASSWORD environment variable.

This avoids placing database credentials directly in the configuration file or source code.

CDC Processing Parameters

The connector also defines several parameters that control its processing behavior:

private static final int BATCH_SIZE =
        Integer.parseInt(
                CONFIG.getProperty("cdc.batch.size", "1")
        );

private static final int CDC_FETCH_SIZE =
        Integer.parseInt(
                CONFIG.getProperty("cdc.fetch.size", "100")
        );

private static final long POLLING_INTERVAL_MS =
        Long.parseLong(
                CONFIG.getProperty("cdc.polling.interval.ms", "2000")
        );

private static final int MAX_RETRIES =
        Integer.parseInt(
                CONFIG.getProperty("cdc.max.retries", "3")
        );

private static final long INITIAL_RETRY_DELAY_MS =
        Long.parseLong(
                CONFIG.getProperty("cdc.retry.delay.ms", "1000")
        );

private static final long RECONNECT_DELAY_MS =
        Long.parseLong(
                CONFIG.getProperty("cdc.reconnect.delay.ms", "5000")
        );

Notice that some properties have default values. These defaults are used when the corresponding property is missing from application.properties.

For example, if cdc.batch.size is not configured, the connector uses a default batch size of 1.

The connector also defines its name and a Jackson ObjectMapper:

private static final String CONNECTOR_NAME =
        "employee-cdc-connector";

private static final ObjectMapper OBJECT_MAPPER =
        new ObjectMapper();

private static volatile boolean running = true;

• CONNECTOR_NAME identifies this CDC connector when reading and updating its checkpoint.
• OBJECT_MAPPER is used to convert JSON event data into Java JSON objects.
• running controls the connector's execution lifecycle. It is declared volatile so changes made by another thread, such as the shutdown hook, are visible to the polling thread.

Summary

In this subsection, we examined the basic structure and configuration of the Java CDC connector.

• Identified the Java libraries used for database connectivity and JSON processing.
• Loaded database connection details and CDC parameters from application.properties.
• Read the database password from the DB_PASSWORD environment variable.
• Defined processing parameters such as batch size, fetch size, polling interval, and retry delays.
• Initialized the connector name, Jackson ObjectMapper, and execution flag.

With the configuration in place, we are ready to examine how the connector starts, validates its settings, establishes database connections, and begins processing CDC events.

5.2 Starting the CDC Connector

Now that we understand the connector configuration, let's examine how the Java application starts and manages its execution.

The main() method is the entry point of the CDC connector. It performs the initial configuration checks, registers a shutdown hook, establishes database connections, and starts the continuous CDC polling process.

Understanding the main() Method

The following is the relevant implementation from CDCConnector.java:

public static void main(String[] args) {

    if (DB_PASSWORD == null) {
        throw new IllegalStateException(
                "Please set the DB_PASSWORD environment variable."
        );
    }

    if (BATCH_SIZE < 1 || CDC_FETCH_SIZE < 1
            || POLLING_INTERVAL_MS < 1
            || MAX_RETRIES < 1 || INITIAL_RETRY_DELAY_MS < 1
            || RECONNECT_DELAY_MS < 1) {
        throw new IllegalArgumentException(
                "CDC configuration values must be positive.");
    }

    Runtime.getRuntime().addShutdownHook(new Thread(() -> {
        running = false;
        System.out.println("\nShutdown requested...");
    }));

    System.out.println("Starting CDC Connector...");
    System.out.println("CDC batch size: " + BATCH_SIZE);

    while (running) {

        try (
                Connection sourceConnection =
                        DriverManager.getConnection(
                                SOURCE_URL, DB_USER, DB_PASSWORD);

                Connection destinationConnection =
                        DriverManager.getConnection(
                                DESTINATION_URL, DB_USER, DB_PASSWORD)
        ) {

            System.out.println(
                    "Connected to source database: "
                            + sourceConnection.getCatalog()
            );

            System.out.println(
                    "Connected to destination database: "
                            + destinationConnection.getCatalog()
            );

            // This method reads the latest checkpoint on every
            // new connection session.
            runContinuousPolling(
                    sourceConnection,
                    destinationConnection
            );

        } catch (SQLException e) {

            if (!running) {
                break;
            }

            if (!isConnectionFailure(e)) {
                System.err.println(
                        "Non-connection database error. Stopping connector:"
                );
                e.printStackTrace();
                break;
            }

            System.err.println(
                    "Database connection lost. Reconnecting in "
                            + RECONNECT_DELAY_MS + " ms..."
            );

            try {
                Thread.sleep(RECONNECT_DELAY_MS);

            } catch (InterruptedException interrupted) {
                Thread.currentThread().interrupt();
                running = false;
            }

        } catch (Exception e) {

            System.err.println("CDC Connector failed:");
            e.printStackTrace();
            break;
        }
    }

    System.out.println("CDC Connector stopped.");
}

Let's examine the important parts of this method.

1. Validating the Database Password

if (DB_PASSWORD == null) {
    throw new IllegalStateException(
            "Please set the DB_PASSWORD environment variable."
    );
}

Before attempting to connect to either database, the connector checks whether the DB_PASSWORD environment variable is available.

If the password is missing, the application throws an IllegalStateException and stops execution.

This prevents the connector from attempting database connections without the required credentials.

2. Validating Configuration Parameters

if (BATCH_SIZE < 1 || CDC_FETCH_SIZE < 1
        || POLLING_INTERVAL_MS < 1
        || MAX_RETRIES < 1 || INITIAL_RETRY_DELAY_MS < 1
        || RECONNECT_DELAY_MS < 1) {
    throw new IllegalArgumentException(
            "CDC configuration values must be positive.");
}

The connector validates the numerical configuration parameters before starting.

All these values must be positive:

• Batch size
• Fetch size
• Polling interval
• Maximum retries
• Initial retry delay
• Reconnection delay

If any value is zero or negative, the connector throws an IllegalArgumentException.

This prevents invalid configuration values from causing unexpected behavior during CDC processing.

3. Registering a Shutdown Hook

Runtime.getRuntime().addShutdownHook(new Thread(() -> {
    running = false;
    System.out.println("\nShutdown requested...");
}));

The connector registers a JVM shutdown hook to handle graceful termination.

A shutdown hook is a thread that the JVM starts when it begins shutting down.

For example, when the user presses Ctrl+C in the terminal, the shutdown hook sets:

running = false;

Because the running variable is declared volatile, the polling thread can observe this change.

The connector then exits its execution loop and terminates.

4. Starting the Connector

System.out.println("Starting CDC Connector...");
System.out.println("CDC batch size: " + BATCH_SIZE);

These statements display startup information in the console.

For example:

Starting CDC Connector...
CDC batch size: 3

This provides basic visibility into the connector's startup and configured batch size.

5. Managing the Connector Lifecycle

while (running) {
    // Database connections and CDC polling
}

The connector runs inside a continuous loop as long as running remains true.

Inside this loop, it establishes connections to the source and destination databases and invokes runContinuousPolling().

The connections are created inside a Java try-with-resources statement:

try (
    Connection sourceConnection =
            DriverManager.getConnection(
                    SOURCE_URL, DB_USER, DB_PASSWORD);

    Connection destinationConnection =
            DriverManager.getConnection(
                    DESTINATION_URL, DB_USER, DB_PASSWORD)
) {
    // CDC processing
}

Try-with-resources automatically closes both database connections when execution leaves the block, including when an exception occurs.

We will examine database connection establishment in more detail in Section 5.3.

6. Handling Database Connection Failures

The main() method catches database exceptions:

catch (SQLException e) {

    if (!running) {
        break;
    }

    if (!isConnectionFailure(e)) {
        System.err.println(
                "Non-connection database error. Stopping connector:"
        );
        e.printStackTrace();
        break;
    }

    System.err.println(
            "Database connection lost. Reconnecting in "
                    + RECONNECT_DELAY_MS + " ms..."
    );

    try {
        Thread.sleep(RECONNECT_DELAY_MS);

    } catch (InterruptedException interrupted) {
        Thread.currentThread().interrupt();
        running = false;
    }
}

The connector distinguishes between connection failures and other database errors.

• Connection failure: The connector waits for the configured reconnection delay and then attempts to establish new database connections.
• Other database error: The connector logs the exception and stops execution rather than repeatedly retrying an error that may require intervention.

The method isConnectionFailure() determines whether a database exception represents a connection-related problem. We will examine its implementation in Section 8, Error Handling and Recovery.

7. Handling Other Exceptions

catch (Exception e) {

    System.err.println("CDC Connector failed:");
    e.printStackTrace();
    break;
}

Any other exception that is not handled by the preceding SQLException catch block is logged, and the connector stops.

This prevents the application from continuing in an unexpected state after an unhandled error.

8. Connector Shutdown

After the execution loop terminates, the following statement is executed:

System.out.println("CDC Connector stopped.");

This provides a final console message indicating that the connector has stopped.

Summary

In this subsection, we examined how the Java CDC connector starts and manages its execution lifecycle.

• Validates the database password and numerical configuration parameters.
• Registers a JVM shutdown hook for graceful termination.
• Starts the connector inside a continuous execution loop.
• Establishes connections to both PostgreSQL databases.
• Invokes the continuous polling method.
• Handles connection failures by waiting before attempting reconnection.
• Stops execution when it encounters non-connection database errors or other unhandled exceptions.

In the next subsection, we will examine how the connector establishes JDBC connections to the source and destination PostgreSQL databases.

5.3 Establishing Database Connections

The CDC connector needs to communicate with two PostgreSQL databases:

• Source database (cdc_source): Used to read CDC events from the cdc_events table.
• Destination database (cdc_destination): Used to apply employee changes and maintain the CDC checkpoint.

Java Database Connectivity (JDBC) provides the mechanism for establishing and managing these database connections.

Establishing JDBC Connections

Our implementation uses the JDBC DriverManager.getConnection() method to establish connections to both databases.

The relevant code from the main() method is:

try (
    Connection sourceConnection =
            DriverManager.getConnection(
                    SOURCE_URL, DB_USER, DB_PASSWORD);

    Connection destinationConnection =
            DriverManager.getConnection(
                    DESTINATION_URL, DB_USER, DB_PASSWORD)
) {

    System.out.println(
            "Connected to source database: "
                    + sourceConnection.getCatalog()
    );

    System.out.println(
            "Connected to destination database: "
                    + destinationConnection.getCatalog()
    );

    runContinuousPolling(
            sourceConnection,
            destinationConnection
    );
}

Let's understand the important parts.

1. Connecting to the Source Database

Connection sourceConnection =
        DriverManager.getConnection(
                SOURCE_URL, DB_USER, DB_PASSWORD);

This statement establishes a JDBC connection to the source PostgreSQL database.

The connection parameters are:

• SOURCE_URL – JDBC URL of the cdc_source database.
• DB_USER – PostgreSQL username.
• DB_PASSWORD – Database password obtained from the environment variable.

Once the connection is established, the connector can execute SQL queries against the source database to retrieve CDC events.

2. Connecting to the Destination Database

Connection destinationConnection =
        DriverManager.getConnection(
                DESTINATION_URL, DB_USER, DB_PASSWORD);

Similarly, this statement establishes a separate JDBC connection to the destination PostgreSQL database.

The connector uses this connection to:

• Insert new employee records.
• Update existing employee records.
• Delete employee records.
• Read and update the CDC checkpoint.

Keeping separate connections allows the connector to independently interact with both databases.

3. Verifying Database Connections

After establishing both connections, the connector displays the connected database names:

System.out.println(
        "Connected to source database: "
                + sourceConnection.getCatalog()
);

System.out.println(
        "Connected to destination database: "
                + destinationConnection.getCatalog()
);

The JDBC Connection.getCatalog() method returns the catalog associated with the connection. For PostgreSQL, this corresponds to the connected database name.

For our implementation, the expected output is:

Connected to source database: cdc_source
Connected to destination database: cdc_destination

This provides a simple way to confirm which databases the connector has connected to.

4. Managing Connection Resources

Notice that both connections are created inside a Java try-with-resources statement.

try (
    Connection sourceConnection = ...;
    Connection destinationConnection = ...
) {
    // CDC processing
}

Try-with-resources automatically closes resources when execution leaves the try block.

This is important because database connections consume resources on both the Java application and PostgreSQL server.

If an exception occurs during processing, the connections are still closed automatically when the try-with-resources block exits.

If establishing the second connection fails after the first connection has been successfully opened, Java also closes the first connection.

5. Passing Connections to the Polling Method

After both connections are established, the connector invokes:

runContinuousPolling(
        sourceConnection,
        destinationConnection
);

Both JDBC connections are passed to the polling method.

The source connection is used to retrieve CDC events, while the destination connection is used to apply the corresponding changes and manage checkpoints.

We will examine the continuous polling mechanism in the next subsection.

Summary

In this subsection, we examined how the Java CDC connector establishes database connections.

• Uses JDBC DriverManager.getConnection() to connect to PostgreSQL.
• Establishes separate connections to the source and destination databases.
• Uses the configured database URLs and credentials.
• Verifies the connected database names using getCatalog().
• Uses try-with-resources to ensure database connections are closed when the connection session ends.
• Passes both connections to the continuous polling method.

The connector is now connected to both databases and ready to begin processing CDC events.

5.4 Implementing Continuous CDC Polling

Once the connector establishes connections to both PostgreSQL databases, it starts continuously polling the source database for new CDC events.

The runContinuousPolling() method is responsible for managing this polling process.

It performs the following operations:

1. Reads the last processed event ID from the destination checkpoint table.
2. Starts the continuous polling loop.
3. Invokes the event-processing method to retrieve and process available CDC events.
4. Waits for the configured polling interval before checking again.
5. Stops polling when a shutdown is requested or the thread is interrupted.

Understanding the runContinuousPolling() Method

The implementation is shown below:

private static void runContinuousPolling(
        Connection sourceConnection,
        Connection destinationConnection) throws Exception {

    long lastProcessedEventId =
            getLastProcessedEventId(destinationConnection);

    System.out.println(
            "\nStarting continuous CDC polling..."
    );

    System.out.println(
            "Last processed event ID: " + lastProcessedEventId
    );

    System.out.println(
            "Polling interval: " + POLLING_INTERVAL_MS + " ms");

    System.out.println("Press Ctrl+C to stop the connector.\n");

    while (running && !Thread.currentThread().isInterrupted()) {

        lastProcessedEventId = processAvailableEvents(
                sourceConnection,
                destinationConnection,
                lastProcessedEventId
        );

        try {
            Thread.sleep(POLLING_INTERVAL_MS);

        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            running = false;
        }
    }

    System.out.println("CDC Connector stopped.");
}

Let's examine the important parts.

1. Reading the Last Processed Event ID

long lastProcessedEventId =
        getLastProcessedEventId(destinationConnection);

Before starting the polling loop, the connector reads the last successfully processed event ID from the cdc_checkpoint table in the destination database.

This allows the connector to resume processing from its previously recorded position when a new connection session starts.

For a new connector installation, the initial checkpoint value is 0, as initialized in Section 3.5.

We'll examine checkpoint management in detail in Section 7.

2. Displaying Polling Information

System.out.println(
        "\nStarting continuous CDC polling..."
);

System.out.println(
        "Last processed event ID: " + lastProcessedEventId
);

System.out.println(
        "Polling interval: " + POLLING_INTERVAL_MS + " ms");

These statements display useful information when the connector starts polling.

For example:

Starting continuous CDC polling...
Last processed event ID: 0
Polling interval: 2000 ms
Press Ctrl+C to stop the connector.

This helps monitor the connector's startup and current processing position.

3. Implementing the Continuous Polling Loop

while (running && !Thread.currentThread().isInterrupted()) {

    lastProcessedEventId = processAvailableEvents(
            sourceConnection,
            destinationConnection,
            lastProcessedEventId
    );

    // Wait before the next polling cycle.
}

The connector continues polling as long as both conditions are satisfied:

• running is true.
• The current thread has not been interrupted.

During each iteration, the connector invokes processAvailableEvents().

This method retrieves available CDC events from the source database, processes them, and returns the updated last processed event ID.

The returned value is used in the next polling iteration.

We'll examine the event retrieval and processing logic in Section 5.5.

4. Controlling the Polling Interval

Thread.sleep(POLLING_INTERVAL_MS);

After processing the currently available events, the connector waits for the configured polling interval before starting another polling cycle.

In our demonstration:

cdc.polling.interval.ms=2000

This means the connector waits for two seconds between polling cycles.

The polling interval controls how frequently the connector checks for newly available events when the current batch of available events has been processed.

5. Handling Thread Interruption

catch (InterruptedException e) {
    Thread.currentThread().interrupt();
    running = false;
}

If the polling thread is interrupted while sleeping, the connector handles the interruption by:

1. Restoring the thread's interrupted status using Thread.currentThread().interrupt().
2. Setting running to false to stop the polling loop.

This allows the connector to terminate its polling process cleanly.

Summary

In this subsection, we implemented the continuous polling mechanism.

• Reads the last processed event ID when a new connection session starts.
• Continuously invokes processAvailableEvents() to process available CDC events.
• Uses the configured polling interval to control polling frequency.
• Updates the local processing position using the value returned by the event-processing method.
• Supports graceful termination through the running flag and thread interruption.

The continuous polling mechanism is now in place. In the next subsection, we will examine how the connector retrieves CDC events from the source database and converts their JSON data into Java objects.

5.5 Reading and Processing CDC Events

Now that the continuous polling mechanism is in place, let's examine how the connector retrieves CDC events from the source database.

The processAvailableEvents() method is responsible for:

1. Querying the cdc_events table for unprocessed events.
2. Retrieving the event ID, operation type, employee record ID, and captured employee data.
3. Converting the JSON event data into Java objects using Jackson.
4. Organizing the retrieved events for processing.
5. Passing the events to the batch-processing method.
6. Returning the updated processing position.

Understanding the processAvailableEvents() Method

The method accepts three parameters:

• sourceConnection – JDBC connection to the source database.
• destinationConnection – JDBC connection to the destination database.
• lastProcessedEventId – The last event ID processed by the connector.

The implementation is shown below:

private static long processAvailableEvents(
        Connection sourceConnection,
        Connection destinationConnection,
        long lastProcessedEventId) throws Exception {

    String sql = """
            SELECT event_id, operation, record_id, changed_data
            FROM cdc_events
            WHERE event_id > ?
            ORDER BY event_id
            LIMIT ?
            """;

    while (true) {

        List<CDCEvent> events = new ArrayList<>();

        try (PreparedStatement statement =
                     sourceConnection.prepareStatement(sql)) {

            statement.setLong(1, lastProcessedEventId);
            statement.setInt(2, CDC_FETCH_SIZE);

            try (ResultSet resultSet = statement.executeQuery()) {

                while (resultSet.next()) {

                    long eventId = resultSet.getLong("event_id");
                    String operation = resultSet.getString("operation");
                    int recordId = resultSet.getInt("record_id");

                    String changedData =
                            resultSet.getString("changed_data");

                    JsonNode employeeData =
                            OBJECT_MAPPER.readTree(changedData);

                    events.add(new CDCEvent(
                            eventId,
                            operation,
                            recordId,
                            employeeData
                    ));
                }
            }
        }

        // No more events are currently available.
        if (events.isEmpty()) {
            break;
        }

        // Process the fetched events in configured batch sizes.
        for (int i = 0; i < events.size(); i += BATCH_SIZE) {

            int endIndex = Math.min(
                    i + BATCH_SIZE,
                    events.size()
            );

            List<CDCEvent> batch =
                    events.subList(i, endIndex);

            processBatchWithRetry(
                    destinationConnection,
                    batch
            );

            lastProcessedEventId =
                    batch.get(batch.size() - 1).eventId();
        }

        // Continue fetching the next page.
    }

    return lastProcessedEventId;
}

Let's examine the important parts.

1. Querying the CDC Events Table

The connector uses the following SQL query:

SELECT event_id, operation, record_id, changed_data
FROM cdc_events
WHERE event_id > ?
ORDER BY event_id
LIMIT ?;

This query retrieves CDC events that have not yet been processed according to the connector's current processing position.

The query uses two parameters:

• event_id > ? – Retrieves events with IDs greater than the last processed event ID.
• LIMIT ? – Restricts the number of events retrieved in a single query.

The results are ordered by event_id in ascending order.

This allows the connector to process the retrieved events in event ID order.

2. Executing the Query Using PreparedStatement

try (PreparedStatement statement =
             sourceConnection.prepareStatement(sql)) {

    statement.setLong(1, lastProcessedEventId);
    statement.setInt(2, CDC_FETCH_SIZE);

    try (ResultSet resultSet = statement.executeQuery()) {
        // Process results
    }
}

The connector uses a JDBC PreparedStatement to execute the query.

The two parameters are supplied using:

• setLong() for the last processed event ID.
• setInt() for the maximum number of events to retrieve.

Using a prepared statement separates SQL structure from parameter values and avoids constructing the query through string concatenation.

The query results are returned as a JDBC ResultSet.

3. Extracting CDC Event Information

The connector iterates through the result set:

while (resultSet.next()) {

    long eventId = resultSet.getLong("event_id");
    String operation = resultSet.getString("operation");
    int recordId = resultSet.getInt("record_id");

    String changedData =
            resultSet.getString("changed_data");
}

For each CDC event, it retrieves four values:

Column	Java variable	Description
event_id	eventId	Unique event ID.
operation	operation	INSERT, UPDATE, or DELETE.
record_id	recordId	ID of the affected employee.
changed_data	changedData	Employee data captured in JSONB format.

These values contain the information required to process the corresponding employee change.

4. Converting JSON Data into Java Objects

The changed_data column contains employee information in PostgreSQL's JSONB format.

The connector retrieves this value as a string and converts it into a Jackson JsonNode:

String changedData =
        resultSet.getString("changed_data");

JsonNode employeeData =
        OBJECT_MAPPER.readTree(changedData);

The ObjectMapper.readTree() method parses the JSON string and creates a tree representation of the JSON data.

For example, a captured employee record might look like this:

{
  "id": 101,
  "name": "John Smith",
  "department": "Engineering",
  "salary": 95000.00
}

After parsing, the connector can access individual fields using methods such as:

employeeData.get("id").asInt();
employeeData.get("name").asText();
employeeData.get("salary").decimalValue();

This makes it convenient to extract employee information when applying changes to the destination database.

5. Creating CDCEvent Objects

After extracting the event information and parsing the JSON data, the connector creates a CDCEvent object:

events.add(new CDCEvent(
        eventId,
        operation,
        recordId,
        employeeData
));

The CDCEvent record is defined as follows:

private record CDCEvent(
        long eventId,
        String operation,
        int recordId,
        JsonNode employeeData) {
}

Java records provide a concise way to represent immutable data objects.

Each CDCEvent contains:

• Event ID.
• Operation type.
• Employee record ID.
• Parsed employee data.

The connector stores the retrieved events in an ArrayList for subsequent processing.

6. Checking Whether Events Are Available

After executing the query, the connector checks whether any events were retrieved:

if (events.isEmpty()) {
    break;
}

If no unprocessed events are currently available, the method exits its fetching loop and returns the current processing position.

The continuous polling method will wait for the configured polling interval before checking the source database again.

If events are available, the connector passes them to the batch-processing logic.

7. Passing Events for Processing

The connector divides the retrieved events into batches and invokes:

processBatchWithRetry(
        destinationConnection,
        batch
);

After a batch has been successfully processed, the connector updates its local processing position:

lastProcessedEventId =
        batch.get(batch.size() - 1).eventId();

This value is then used when retrieving the next page of events.

The detailed batch-processing and pagination logic will be discussed in Section 6.

Summary

In this subsection, we examined how the Java connector retrieves and prepares CDC events for processing.

• Queries the source cdc_events table using the last processed event ID.
• Retrieves the event ID, operation, employee ID, and captured employee data.
• Uses JDBC PreparedStatement and ResultSet to execute and read the query.
• Converts JSONB event data into Jackson JsonNode objects.
• Creates CDCEvent records to represent individual changes.
• Passes the retrieved events to the batch-processing method.
• Updates the local processing position after each successfully processed batch.

The connector can now retrieve CDC events and prepare them for application to the destination database.

In the next subsection, we will examine how the connector applies INSERT, UPDATE, and DELETE operations to the destination employees table.

5.6 Applying Changes to the Destination Database

After retrieving and parsing CDC events from the source database, the connector must apply the corresponding changes to the destination employees table.

The applyChange() method is responsible for this operation.

It examines the CDC operation type and executes the appropriate SQL statement against the destination database.

Understanding the applyChange() Method

The implementation is shown below:

private static void applyChange(
        Connection connection,
        String operation,
        JsonNode employeeData) throws SQLException {

    int employeeId =
            employeeData.get("id").asInt();

    switch (operation) {

        case "INSERT" -> {

            String sql = """
                    INSERT INTO employees
                    (id, name, department, salary)
                    VALUES (?, ?, ?, ?)
                    """;

            try (PreparedStatement statement =
                         connection.prepareStatement(sql)) {

                statement.setInt(1, employeeId);
                statement.setString(
                        2, employeeData.get("name").asText());

                setDepartment(statement, 3, employeeData);
                setSalary(statement, 4, employeeData);

                statement.executeUpdate();
            }
        }

        case "UPDATE" -> {

            String sql = """
                    UPDATE employees
                    SET name = ?, department = ?, salary = ?
                    WHERE id = ?
                    """;

            try (PreparedStatement statement =
                         connection.prepareStatement(sql)) {

                statement.setString(
                        1, employeeData.get("name").asText());

                setDepartment(statement, 2, employeeData);
                setSalary(statement, 3, employeeData);

                statement.setInt(4, employeeId);

                int rowsUpdated = statement.executeUpdate();

                if (rowsUpdated != 1) {
                    throw new SQLException(
                            "Expected to update employee "
                                    + employeeId
                    );
                }
            }
        }

        case "DELETE" -> {

            String sql =
                    "DELETE FROM employees WHERE id = ?";

            try (PreparedStatement statement =
                         connection.prepareStatement(sql)) {

                statement.setInt(1, employeeId);

                int rowsDeleted = statement.executeUpdate();

                if (rowsDeleted != 1) {
                    throw new SQLException(
                            "Expected to delete employee "
                                    + employeeId
                    );
                }
            }
        }

        default -> throw new SQLException(
                "Unsupported CDC operation: " + operation
        );
    }
}

Let's examine how each operation is handled.

1. Extracting the Employee ID

int employeeId =
        employeeData.get("id").asInt();

The connector extracts the employee ID from the parsed JSON data.

This ID identifies the employee record that must be inserted, updated, or deleted in the destination database.

The destination uses the same employee ID as the source database to maintain consistency between the two databases.

2. Handling INSERT Operations

case "INSERT" -> {

    String sql = """
            INSERT INTO employees
            (id, name, department, salary)
            VALUES (?, ?, ?, ?)
            """;

    try (PreparedStatement statement =
                 connection.prepareStatement(sql)) {

        statement.setInt(1, employeeId);
        statement.setString(
                2, employeeData.get("name").asText());

        setDepartment(statement, 3, employeeData);
        setSalary(statement, 4, employeeData);

        statement.executeUpdate();
    }
}

When the operation is INSERT, the connector creates a new employee record in the destination database.

The prepared statement uses four parameters:

Parameter	Value
1	Employee ID
2	Employee name
3	Department
4	Salary

The employee ID and name are extracted directly from the JSON data.

The setDepartment() and setSalary() helper methods handle the remaining fields, including NULL values.

Finally, executeUpdate() executes the INSERT statement.

3. Handling UPDATE Operations

case "UPDATE" -> {

    String sql = """
            UPDATE employees
            SET name = ?, department = ?, salary = ?
            WHERE id = ?
            """;

    try (PreparedStatement statement =
                 connection.prepareStatement(sql)) {

        statement.setString(
                1, employeeData.get("name").asText());

        setDepartment(statement, 2, employeeData);
        setSalary(statement, 3, employeeData);

        statement.setInt(4, employeeId);

        int rowsUpdated = statement.executeUpdate();

        if (rowsUpdated != 1) {
            throw new SQLException(
                    "Expected to update employee "
                            + employeeId
            );
        }
    }
}

When the operation is UPDATE, the connector modifies the corresponding employee record in the destination database.

The SQL statement updates the employee's name, department, and salary using the employee ID as the condition.

After executing the statement, the connector checks the number of affected rows.

if (rowsUpdated != 1) {
    throw new SQLException(
            "Expected to update employee " + employeeId
    );
}

The implementation expects exactly one employee record to be updated.

If no matching employee exists, or an unexpected number of rows is affected, the connector throws an SQLException.

This prevents the connector from silently treating an unsuccessful update as successful.

4. Handling DELETE Operations

case "DELETE" -> {

    String sql =
            "DELETE FROM employees WHERE id = ?";

    try (PreparedStatement statement =
                 connection.prepareStatement(sql)) {

        statement.setInt(1, employeeId);

        int rowsDeleted = statement.executeUpdate();

        if (rowsDeleted != 1) {
            throw new SQLException(
                    "Expected to delete employee "
                            + employeeId
            );
        }
    }
}

When the operation is DELETE, the connector removes the corresponding employee record from the destination database.

The employee ID identifies the record to delete.

Similar to the UPDATE operation, the connector verifies that exactly one row was deleted. Otherwise, it throws an SQLException.

5. Handling Unsupported Operations

default -> throw new SQLException(
        "Unsupported CDC operation: " + operation
);

The default branch handles any operation type that is not supported by the connector.

Currently, the implementation supports:

• INSERT
• UPDATE
• DELETE

If an unexpected operation is received, the connector throws an exception instead of silently ignoring the event.

Summary

In this subsection, we examined how the Java CDC connector applies changes to the destination database.

• Extracts the employee ID from the parsed JSON event data.
• Uses a Java switch expression to identify the CDC operation.
• Executes INSERT statements for new employee records.
• Executes UPDATE statements for modified employee records.
• Executes DELETE statements for removed employee records.
• Uses JDBC PreparedStatement for parameterized SQL execution.
• Verifies that UPDATE and DELETE operations affect exactly one row.
• Rejects unsupported CDC operations.

The connector can now translate captured CDC events into corresponding database operations.

In the next subsection, we will examine how the connector handles nullable employee fields using the setDepartment() and setSalary() helper methods.

5.7 Handling Nullable Employee Fields

In the previous subsection, we examined how the connector applies INSERT, UPDATE, and DELETE operations to the destination database.

You may have noticed that the applyChange() method delegates the handling of department and salary values to two helper methods:

• setDepartment()
• setSalary()

These methods handle nullable employee fields and ensure that SQL NULL values are correctly passed to PostgreSQL.

1. Handling the Department Field

The implementation of setDepartment() is shown below:

private static void setDepartment(
        PreparedStatement statement,
        int parameterIndex,
        JsonNode employeeData) throws SQLException {

    JsonNode department =
            employeeData.get("department");

    if (department == null || department.isNull()) {
        statement.setNull(parameterIndex, Types.VARCHAR);
    } else {
        statement.setString(
                parameterIndex, department.asText());
    }
}

Let's understand how this method works.

1. Retrieves the department field from the JSON employee data.
2. Checks whether the field is missing or contains a JSON NULL value.
3. If the value is NULL, calls PreparedStatement.setNull() with Types.VARCHAR.
4. Otherwise, extracts the department value as a string and passes it to the prepared statement using setString().

For example, consider the following employee data:

{
  "id": 101,
  "name": "John Smith",
  "department": null,
  "salary": 95000.00
}

In this case, the method detects that the department value is NULL and sets the corresponding SQL parameter to NULL.

2. Handling the Salary Field

The implementation of setSalary() is shown below:

private static void setSalary(
        PreparedStatement statement,
        int parameterIndex,
        JsonNode employeeData) throws SQLException {

    JsonNode salary = employeeData.get("salary");

    if (salary == null || salary.isNull()) {
        statement.setNull(parameterIndex, Types.NUMERIC);
    } else {
        BigDecimal value = salary.decimalValue();
        statement.setBigDecimal(parameterIndex, value);
    }
}

The logic is similar to the department handling, but there is an important difference in the Java data type.

1. Retrieves the salary field from the JSON employee data.
2. Checks whether the field is missing or contains a JSON NULL value.
3. If the value is NULL, sets the SQL parameter to NULL using Types.NUMERIC.
4. Otherwise, converts the JSON numeric value into a Java BigDecimal.
5. Uses PreparedStatement.setBigDecimal() to pass the salary to PostgreSQL.

Why use BigDecimal for salary?

The PostgreSQL salary column is defined as:

salary NUMERIC(10,2)

Java's BigDecimal is suitable for handling decimal values because it provides decimal precision without the binary floating-point representation issues associated with float and double.

This is particularly useful when processing financial or monetary values.

3. Why Handle NULL Values Explicitly?

In our source database, the department and salary columns allow NULL values.

Therefore, the CDC connector must preserve those NULL values when synchronizing employee records.

JDBC provides the PreparedStatement.setNull() method to explicitly bind SQL NULL values to prepared statement parameters.

The implementation uses the appropriate SQL types:

Column	JDBC SQL type	Java value handling
department	Types.VARCHAR	String
salary	Types.NUMERIC	BigDecimal

This ensures that nullable employee fields are correctly represented in the destination database.

Summary

In this subsection, we examined how the connector handles nullable employee fields.

• Implemented setDepartment() to handle nullable VARCHAR values.
• Implemented setSalary() to handle nullable NUMERIC values.
• Used PreparedStatement.setNull() to explicitly bind SQL NULL values.
• Used BigDecimal to preserve decimal precision when processing salary values.

With these helper methods, the Java CDC connector can correctly handle employee records containing NULL department or salary values.

6. Batch Processing and Pagination

In the previous section, we implemented the Java CDC connector to retrieve events from the source database and apply the corresponding changes to the destination database.

However, processing a large number of CDC events individually can result in unnecessary database queries and transaction overhead.

To improve processing efficiency, our implementation uses two techniques:

1. Pagination: Retrieves CDC events from the source database in manageable groups.
2. Batch Processing: Applies multiple CDC events to the destination database within a single transaction.

These two mechanisms are controlled by separate configuration parameters: cdc.fetch.size and cdc.batch.size.

6.1 Understanding Fetch Size and Batch Size

Our connector uses the following configuration:

cdc.fetch.size=4
cdc.batch.size=3

Let's understand the difference.

Fetch Size

The fetch size determines the maximum number of CDC events retrieved from the source database in a single SQL query.

In our implementation, this value is used with the SQL LIMIT clause:

SELECT event_id, operation, record_id, changed_data
FROM cdc_events
WHERE event_id > ?
ORDER BY event_id
LIMIT ?;

For example, with cdc.fetch.size=4, each query retrieves at most four CDC events.

Batch Size

The batch size determines how many retrieved events are applied to the destination database within one transaction.

With cdc.batch.size=3, the connector processes up to three events in each destination transaction.

The difference can be summarized as follows:

Parameter	Purpose	Example
cdc.fetch.size	Maximum events retrieved per source query	4
cdc.batch.size	Maximum events processed per destination transaction	3

Keeping these parameters separate allows us to control source-side retrieval and destination-side transaction sizes independently.

6.2 Implementing Pagination

The processAvailableEvents() method implements pagination by repeatedly querying the source cdc_events table.

The relevant SQL query is:

SELECT event_id, operation, record_id, changed_data
FROM cdc_events
WHERE event_id > ?
ORDER BY event_id
LIMIT ?;

The connector supplies the query parameters using:

statement.setLong(1, lastProcessedEventId);
statement.setInt(2, CDC_FETCH_SIZE);

The query retrieves events whose IDs are greater than the last successfully processed event ID, in ascending order.

The LIMIT clause restricts each query to the configured fetch size.

After processing the retrieved events, the connector executes the query again to retrieve the next page.

The relevant loop is:

while (true) {

    List<CDCEvent> events = new ArrayList<>();

    // Retrieve the next page of CDC events.
    // ...

    if (events.isEmpty()) {
        break;
    }

    // Process the retrieved events.
    // ...
}

The loop continues until the query returns no additional events.

This allows the connector to process more events than can be retrieved in a single query.

6.3 Implementing Batch Processing

After retrieving a page of CDC events, the connector divides the events into smaller batches using the configured BATCH_SIZE.

The relevant implementation is:

for (int i = 0; i < events.size(); i += BATCH_SIZE) {

    int endIndex = Math.min(
            i + BATCH_SIZE,
            events.size()
    );

    List<CDCEvent> batch =
            events.subList(i, endIndex);

    processBatchWithRetry(
            destinationConnection,
            batch
    );

    lastProcessedEventId =
            batch.get(batch.size() - 1).eventId();
}

Let's understand how this works.

1. Iterating Through the Retrieved Events

for (int i = 0; i < events.size(); i += BATCH_SIZE)

The loop processes the retrieved events in groups determined by BATCH_SIZE.

For a batch size of 3, the starting index advances by three positions during each iteration.

2. Calculating the Batch Boundary

int endIndex = Math.min(
        i + BATCH_SIZE,
        events.size()
);

This calculates the ending index of the current batch.

The Math.min() method ensures that the ending index never exceeds the number of available events.

This is important when the final batch contains fewer events than the configured batch size.

3. Creating the Batch

List<CDCEvent> batch =
        events.subList(i, endIndex);

The subList() method creates a view of the retrieved events representing the current batch.

The connector then passes this batch to processBatchWithRetry().

4. Processing the Batch

processBatchWithRetry(
        destinationConnection,
        batch
);

This method applies the employee changes to the destination database and manages the destination transaction.

We will examine its transaction handling and retry mechanism in Section 8.

5. Updating the Local Processing Position

lastProcessedEventId =
        batch.get(batch.size() - 1).eventId();

After the batch-processing method returns successfully, the connector updates its local processing position to the event ID of the last event in that batch.

This value is used when retrieving subsequent events.

6.4 Understanding Pagination and Batch Processing Together

Let's consider an example using the configuration from our demonstration:

cdc.fetch.size=4
cdc.batch.size=3

Assume that eight CDC events are available, with event IDs from 1 through 8.

The connector processes them as follows:

First page

The source query retrieves four events:

Events: 1, 2, 3, 4

With a batch size of 3, these events are divided into:

Batch	Events	Events per transaction
Batch 1	1, 2, 3	3
Batch 2	4	1

After both batches are successfully processed, the local processing position becomes event ID 4.

Second page

The connector executes the source query again using:

WHERE event_id > 4

The next four events are retrieved:

Events: 5, 6, 7, 8

They are divided into:

Batch	Events	Events per transaction
Batch 3	5, 6, 7	3
Batch 4	8	1

After processing these batches, the local processing position becomes event ID 8.

Next query

The connector executes the query again using:

WHERE event_id > 8

If no additional events are available, the query returns an empty result. The pagination loop exits, and the connector returns to its continuous polling cycle.

The complete processing flow is illustrated below:

Figure 2: Example of pagination and batch processing using fetch size 4 and batch size 3.

6.5 Benefits of Batch Processing and Pagination

Using separate fetch and batch sizes provides several benefits:

• Controlled memory usage: The connector retrieves a limited number of events per source query instead of loading all available events at once.
• Reduced transaction overhead: Multiple employee changes are applied within a single destination transaction.
• Flexible configuration: Fetch size and batch size can be adjusted independently based on workload and database capacity.
• Continuous processing: Pagination allows the connector to process more events than the configured fetch size.

Summary

In this section, we examined how the CDC connector uses pagination and batch processing.

• Uses cdc.fetch.size to control the maximum number of events retrieved per source query.
• Uses cdc.batch.size to control the number of events applied per destination transaction.
• Implements pagination using a repeated query with LIMIT.
• Divides retrieved events into smaller batches using subList().
• Processes each batch through processBatchWithRetry().
• Continues fetching subsequent pages until no more events are available.

These mechanisms allow the connector to process multiple CDC events efficiently while controlling query sizes and destination transaction boundaries.

In the next section, we will examine Checkpoint Management, including how the connector records its processing position and resumes from that position after a restart.

7. Checkpoint Management

A CDC connector needs a reliable way to track which events have already been processed. Without this mechanism, the connector would have no way to determine where to resume after a restart.

In this project, we use a checkpoint table in the destination database to store the ID of the last successfully processed CDC event.

The Java connector reads this checkpoint when it starts a polling session and updates it after successfully applying each batch of events.

7.1 Understanding the Checkpoint Table

We created the cdc_checkpoint table in the destination database in Section 3.5.

The table has the following structure:

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
);

We also initialized a checkpoint entry for our connector:

INSERT INTO cdc_checkpoint (
    connector_name,
    last_processed_event_id
)
VALUES ('employee-cdc-connector', 0)
ON CONFLICT (connector_name) DO NOTHING;

The columns serve the following purposes:

• connector_name: Identifies the CDC connector. It is the primary key, allowing multiple connectors to maintain separate checkpoints.
• last_processed_event_id: Stores the ID of the last successfully processed event.
• updated_at: Records when the checkpoint was last updated. The Java connector explicitly refreshes this timestamp when advancing the checkpoint.

Initially, last_processed_event_id is set to 0, indicating that the connector has not yet processed any CDC events.

7.2 Reading the Last Processed Event ID

When a new database connection session is established, the connector retrieves the last processed event ID from the destination database.

The runContinuousPolling() method calls getLastProcessedEventId() before entering its polling loop.

Here is the implementation:

private static long getLastProcessedEventId(
        Connection connection) throws SQLException {

    String sql = """
            SELECT last_processed_event_id
            FROM cdc_checkpoint
            WHERE connector_name = ?
            """;

    try (PreparedStatement statement =
                 connection.prepareStatement(sql)) {

        statement.setString(1, CONNECTOR_NAME);

        try (ResultSet resultSet = statement.executeQuery()) {

            if (resultSet.next()) {
                return resultSet.getLong(
                        "last_processed_event_id");
            }

            throw new SQLException(
                    "CDC checkpoint not found: " + CONNECTOR_NAME
            );
        }
    }
}

The method performs the following operations:

1. Executes a query against the cdc_checkpoint table.
2. Uses the connector name to locate its checkpoint entry.
3. Retrieves and returns last_processed_event_id.
4. Throws an exception if the checkpoint entry does not exist.

The returned event ID becomes the starting position for the next CDC polling cycle.

7.3 Updating the Checkpoint

After a batch of CDC events has been successfully applied to the destination database, the connector updates the checkpoint.

The updateCheckpoint() method performs this operation:

private static void updateCheckpoint(
        Connection connection,
        long eventId) throws SQLException {

    String sql = """
            UPDATE cdc_checkpoint
            SET last_processed_event_id = ?,
                updated_at = CURRENT_TIMESTAMP
            WHERE connector_name = ?
            """;

    try (PreparedStatement statement =
                 connection.prepareStatement(sql)) {

        statement.setLong(1, eventId);
        statement.setString(2, CONNECTOR_NAME);

        int rowsUpdated = statement.executeUpdate();

        if (rowsUpdated != 1) {
            throw new SQLException(
                    "Failed to update CDC checkpoint."
            );
        }
    }
}

The method updates two columns:

• last_processed_event_id is set to the ID of the last event in the successfully processed batch.
• updated_at is refreshed with the current timestamp.

The method also verifies that exactly one checkpoint row was updated. If no matching row exists, it throws an exception rather than silently continuing.

7.4 Updating the Checkpoint Within the Destination Transaction

An important design consideration is ensuring that the destination data changes and checkpoint update are committed together.

In our implementation, the processBatchWithRetry() method handles both operations within the same destination database transaction.

The relevant part of the method is:

destinationConnection.setAutoCommit(false);

try {
    for (CDCEvent event : events) {

        applyChange(
                destinationConnection,
                event.operation(),
                event.employeeData()
        );
    }

    long lastEventId =
            events.get(events.size() - 1).eventId();

    updateCheckpoint(destinationConnection, lastEventId);

    destinationConnection.commit();

} catch (SQLException | RuntimeException e) {

    destinationConnection.rollback();
    throw e;
}

The sequence is important:

1. Disable auto-commit to begin an explicit transaction.
2. Apply all CDC events in the batch to the destination employees table.
3. Retrieve the event ID of the last event in the batch.
4. Update the checkpoint with that event ID.
5. Commit the transaction.

If an error occurs before the commit, the transaction is rolled back.

This means that the destination employee changes and the checkpoint update are treated as a single transaction.

For example:

• If all employee changes and the checkpoint update succeed, the transaction commits.
• If an employee change fails, the transaction is rolled back, including the checkpoint update.
• If the checkpoint update fails, the employee changes in that transaction are also rolled back.

This prevents the checkpoint from advancing when the corresponding destination changes have not been successfully committed.

7.5 Resuming Processing After a Restart

One of the main benefits of checkpoint management is that the connector does not need to start processing from the beginning every time it restarts.

Consider the following example:

Suppose the connector has processed events up to event ID 25.

The checkpoint table contains:

connector_name	last_processed_event_id
employee-cdc-connector	25

Now suppose the connector stops unexpectedly.

When it starts again, it establishes connections to the source and destination databases. The runContinuousPolling() method retrieves the checkpoint from the destination database.

The connector then queries the source CDC event table using:

SELECT event_id, operation, record_id, changed_data
FROM cdc_events
WHERE event_id > ?
ORDER BY event_id
LIMIT ?;

The parameter for event_id is set to 25.

Therefore, the connector looks for events with IDs greater than 25 and resumes processing from that position.

As subsequent batches are successfully committed, the checkpoint advances accordingly.

7.6 Benefits of Checkpoint Management

Checkpoint management provides several important benefits:

• Restart recovery: The connector can resume from its last recorded processing position after a restart.
• Reduced redundant processing: Previously checkpointed events are not normally fetched again during the next polling session.
• Consistent destination state: Destination changes and checkpoint updates are committed within the same transaction.
• Progress tracking: The checkpoint table provides a simple way to monitor the connector's processing progress.

In this project, checkpoint management works together with batch processing and transaction handling to provide a basic recovery mechanism for the CDC pipeline.

8. Error Handling and Recovery

A CDC connector runs continuously and interacts with two databases. During operation, database errors, transaction failures, and unexpected connection losses can occur.

If these situations are not handled properly, the connector could stop unexpectedly or leave the destination database in an inconsistent state.

In this project, we implement several error-handling mechanisms:

• Transaction rollback when a batch fails.
• Retry logic for selected transient transaction errors.
• Exponential backoff between retry attempts.
• Database connection recovery.
• Graceful shutdown when the application is interrupted.

Let's examine how these mechanisms work.

8.1 Transaction Rollback

As discussed in Section 7, the connector processes each batch within a destination database transaction.

If an error occurs while applying the CDC events or updating the checkpoint, the transaction must be rolled back.

The relevant code in processBatchWithRetry() is:

try {
    for (CDCEvent event : events) {

        applyChange(
                destinationConnection,
                event.operation(),
                event.employeeData()
        );
    }

    long lastEventId =
            events.get(events.size() - 1).eventId();

    updateCheckpoint(destinationConnection, lastEventId);

    destinationConnection.commit();

} catch (SQLException | RuntimeException e) {

    try {
        destinationConnection.rollback();
        transactionResolved = true;

    } catch (SQLException rollbackError) {

        e.addSuppressed(rollbackError);

        // Do not reuse a connection with an unresolved transaction.
        try {
            destinationConnection.close();
            connectionClosed = true;

        } catch (SQLException closeError) {
            e.addSuppressed(closeError);
        }

        throw e;
    }

    // Additional error handling follows...
}

The important points are:

1. All events in the batch are applied within one transaction.
2. The checkpoint is updated only after all events have been applied successfully.
3. The transaction is committed only after the checkpoint update succeeds.
4. If an error occurs, the connector attempts to roll back the transaction.

If the rollback itself fails, the connector closes the destination connection rather than reusing a connection whose transaction state is uncertain.

This approach helps prevent partially applied batches and inconsistent checkpoint updates.

8.2 Identifying Retryable Transaction Errors

Not every database error should be retried.

For example, a temporary transaction conflict may succeed when attempted again, whereas an invalid SQL statement or other permanent error is unlikely to succeed through repeated retries.

The connector uses SQLSTATE codes to identify retryable transaction errors.

Here is the implementation:

private static boolean isRetryableTransactionError(
        SQLException e) {

    String sqlState = e.getSQLState();

    if (sqlState == null) {
        return false;
    }

    return sqlState.startsWith("40")  // Transaction rollback
            || sqlState.equals("55P03"); // Lock not available
}

The connector considers the following errors retryable:

• SQLSTATE class 40: Transaction rollback errors, including serialization failures and deadlocks.
• SQLSTATE 55P03: Lock not available.

Other SQL errors are not automatically retried.

Connection failures are handled separately by the outer connection-recovery mechanism.

8.3 Implementing Retry Logic with Exponential Backoff

The connector retries a failed batch when the error is classified as a retryable transaction error.

The retry loop in processBatchWithRetry() is:

for (int attempt = 1; attempt <= MAX_RETRIES + 1; attempt++) {
    // Attempt to process the batch.
}

Notice that the loop uses MAX_RETRIES + 1.

If cdc.max.retries is configured as 3, the connector makes a maximum of four attempts:

• First attempt: Initial processing attempt.
• Second attempt: First retry.
• Third attempt: Second retry.
• Fourth attempt: Third retry.

The connector uses exponential backoff to increase the waiting time between retries.

The delay is calculated as follows:

long delay =
        INITIAL_RETRY_DELAY_MS * (1L << (attempt - 1));

With an initial retry delay of 1,000 milliseconds, the delays are:

Failed attempt	Waiting time
1	1 second
2	2 seconds
3	4 seconds

This approach avoids repeatedly retrying a failed transaction without allowing time for the underlying condition to clear.

Before retrying, the connector rolls back the failed transaction. It also checks whether the error is a connection failure or a retryable transaction error.

If the error is not retryable, or the maximum number of attempts has been reached, the exception is propagated to the caller.

8.4 Handling Database Connection Failures

Transaction retries are useful for temporary transaction errors, but they cannot solve a lost database connection.

For example, the PostgreSQL server might be restarted, become temporarily unavailable, or terminate an active connection.

The connector identifies connection failures using SQLSTATE codes.

The isConnectionFailure() method checks for:

• SQLSTATE class 08, which represents connection exceptions.
• 57P01, 57P02, and 57P03, which represent PostgreSQL server shutdown or unavailability conditions.

When a connection failure occurs during batch processing, the connector does not attempt to retry the transaction on the same connection. Instead, it propagates the exception to the outer connection-recovery loop.

The relevant logic in the main() method is:

} catch (SQLException e) {

    if (!running) {
        break;
    }

    if (!isConnectionFailure(e)) {
        System.err.println(
                "Non-connection database error. Stopping connector:"
        );
        e.printStackTrace();
        break;
    }

    System.err.println(
            "Database connection lost. Reconnecting in "
                    + RECONNECT_DELAY_MS + " ms..."
    );

    try {
        Thread.sleep(RECONNECT_DELAY_MS);

    } catch (InterruptedException interrupted) {
        Thread.currentThread().interrupt();
        running = false;
    }
}

If the exception is identified as a connection failure, the connector waits for the configured reconnect delay before attempting to establish new connections.

In this project, the reconnect delay is configured as:

cdc.reconnect.delay.ms=5000

Therefore, the connector waits five seconds before attempting to reconnect.

The database connections are managed using Java's try-with-resources statement. When the connection session exits, the resources are closed automatically.

The outer while (running) loop then attempts to establish new connections to both the source and destination databases.

8.5 Recovering the Processing Position After Reconnection

An important part of connection recovery is determining where processing should resume.

The connector does not rely on the last event ID held only in application memory.

Whenever a new database connection session is established, runContinuousPolling() retrieves the checkpoint again:

long lastProcessedEventId =
        getLastProcessedEventId(destinationConnection);

This ensures that the connector uses the last checkpoint committed in the destination database.

For example, suppose the connector successfully processes events up to event ID 30, and the corresponding checkpoint is committed.

If the source or destination connection is subsequently lost, the connector attempts to reconnect.

After establishing new connections, it retrieves the checkpoint value 30 and continues querying for events with IDs greater than 30.

Because the destination employee changes and checkpoint update are committed in the same transaction, the checkpoint does not advance independently of those changes.

8.6 Graceful Shutdown and Thread Interruption

The connector also handles application shutdown.

In the main() method, a JVM shutdown hook is registered:

Runtime.getRuntime().addShutdownHook(new Thread(() -> {
    running = false;
    System.out.println("\nShutdown requested...");
}));

When the application receives a shutdown request, the hook sets the running flag to false.

The continuous polling loop checks this flag:

while (running && !Thread.currentThread().isInterrupted()) {
    // Process available CDC events.
}

The connector also handles interruption during its polling sleep and retry delays.

When an InterruptedException occurs, the code restores the thread's interrupted status using:

Thread.currentThread().interrupt();

This preserves the interruption signal instead of silently consuming it.

8.7 Summary of Error Handling and Recovery

The following table summarizes the error-handling mechanisms implemented in the connector.

Situation	Connector behavior
Error while applying a batch	Roll back the destination transaction
Retryable transaction error	Roll back and retry with exponential backoff
Non-retryable SQL error	Propagate the exception and stop the connector
Database connection failure	Exit the current connection session and attempt reconnection
Rollback failure	Close the destination connection rather than reuse an unresolved transaction
Application shutdown	Set the running flag to false and stop the polling loop

These mechanisms provide a basic recovery strategy for the CDC pipeline. Transaction rollback protects destination consistency, retry logic handles selected temporary transaction errors, and connection recovery allows the connector to resume after database availability is restored.

The implementation is intentionally limited to the error conditions identified by its SQLSTATE checks. It does not automatically retry every possible database error.

9. Testing the CDC Pipeline

Now that we have implemented the PostgreSQL triggers and Java CDC connector, let's test the complete pipeline.

The objective is to verify that changes made to the source employees table are captured by PostgreSQL triggers, stored in the cdc_events table, and eventually applied to the destination database by the Java connector.

We will also verify batch processing, checkpoint management, and recovery behavior.

9.1 Starting the CDC Connector

Before testing, make sure both PostgreSQL databases are running and the required database objects have been created.

Navigate to the project directory:

cd Software-Engineering/cdc-demo

Set the database password environment variable:

export DB_PASSWORD=your_database_password

On Windows PowerShell, use:

$env:DB_PASSWORD="your_database_password"

Start the connector using Maven:

mvn clean compile exec:java -Dexec.mainClass="com.clockwise.cdc.CDCConnector"

The connector should establish connections to both databases and retrieve the current checkpoint.

You should see startup messages similar to the following:

Configuration loaded from application.properties
Starting CDC Connector...
CDC batch size: 3
Connected to source database: cdc_source
Connected to destination database: cdc_destination

Starting continuous CDC polling...
Last processed event ID: 0
Polling interval: 2000 ms
Press Ctrl+C to stop the connector.

The displayed checkpoint value depends on the previous processing history. The value 0 represents a newly initialized checkpoint.

9.2 Testing INSERT Operations

Let's first verify that an INSERT operation is captured and replicated.

Connect to the source database:

psql -U postgres -d cdc_source

Insert a new employee:

INSERT INTO employees (name, department, salary)
VALUES ('John Smith', 'Engineering', 85000.00);

The employee_changes_trigger automatically invokes the capture_employee_changes() function.

The function creates a CDC event containing:

• Operation: INSERT
• Record ID: The newly generated employee ID
• Changed data: The inserted employee information in JSONB format

Verify the event in the source database:

SELECT event_id, operation, record_id, changed_data, created_at
FROM cdc_events
ORDER BY event_id DESC
LIMIT 5;

You should find a new event with operation INSERT.

Next, connect to the destination database:

psql -U postgres -d cdc_destination

Verify that the employee has been replicated:

SELECT * FROM employees
ORDER BY id;

The inserted employee should now be present in the destination table.

9.3 Testing UPDATE Operations

Next, let's verify that changes to an existing employee are captured and replicated.

Execute the following statement in the source database:

UPDATE employees
SET salary = 95000.00
WHERE name = 'John Smith';

The trigger captures the UPDATE operation and stores the updated employee information in the cdc_events table.

Verify the newly generated event:

SELECT event_id, operation, record_id, changed_data
FROM cdc_events
ORDER BY event_id DESC
LIMIT 5;

You should find an event with operation UPDATE.

After the connector processes the event, query the destination database:

SELECT * FROM employees
WHERE name = 'John Smith';

The employee's salary should reflect the updated value.

9.4 Testing DELETE Operations

Now let's test the DELETE operation.

Execute the following statement in the source database:

DELETE FROM employees
WHERE name = 'John Smith';

The trigger captures the DELETE operation.

For a DELETE event, the trigger uses the OLD record to preserve the employee information that existed before deletion.

Verify the CDC event:

SELECT event_id, operation, record_id, changed_data
FROM cdc_events
ORDER BY event_id DESC
LIMIT 5;

You should find an event with operation DELETE.

After the connector processes the event, verify the destination database:

SELECT * FROM employees
WHERE name = 'John Smith';

The employee record should no longer exist in the destination table.

These three tests verify that the trigger-based CDC implementation captures all three basic database operations: INSERT, UPDATE, and DELETE.

9.5 Testing Batch Processing and Pagination

The next test verifies that the connector can process multiple CDC events using the configured batch size and fetch size.

For this demonstration, the configuration is:

cdc.batch.size=3
cdc.fetch.size=4

This means:

• The connector retrieves a maximum of four events in each source query.
• It applies up to three events in each destination transaction.

To generate multiple CDC events, insert several employees into the source database:

INSERT INTO employees (name, department, salary)
VALUES
    ('Alice Brown', 'Engineering', 90000.00),
    ('Bob Wilson', 'Finance', 78000.00),
    ('Carol Davis', 'HR', 72000.00),
    ('David Miller', 'Engineering', 95000.00),
    ('Emma Taylor', 'Finance', 82000.00),
    ('Frank Moore', 'HR', 75000.00),
    ('Grace Lee', 'Engineering', 98000.00),
    ('Henry Clark', 'Finance', 88000.00);

Each inserted row generates a separate CDC event.

With eight available events, the connector processes them as follows, assuming they are visible in event ID order:

Fetch operation	Events retrieved	Destination batches
First fetch	4 events	3 + 1
Second fetch	4 events	3 + 1

After processing the first four events, the connector fetches the next page using the updated event ID.

Once all available events have been processed, the next query returns no rows and the pagination loop ends.

Verify that the employees have been replicated:

SELECT * FROM employees
ORDER BY id;

This test demonstrates how pagination and batch processing work together to process more events than can fit into a single destination batch.

9.6 Verifying Checkpoint Management

The checkpoint table records the last successfully processed event ID.

Connect to the destination database and execute:

SELECT
    connector_name,
    last_processed_event_id,
    updated_at
FROM cdc_checkpoint;

The last_processed_event_id should advance as the connector successfully processes batches.

For example, if the last processed event ID is 42, the checkpoint table should contain:

connector_name	last_processed_event_id
employee-cdc-connector	42

The actual value depends on the events generated during testing.

This verifies that the connector updates the checkpoint after successfully applying a batch.

9.7 Testing Restart Recovery

Checkpoint management also allows the connector to resume processing after a restart.

To test this behavior:

1. Start the CDC connector.
2. Generate some employee changes in the source database.
3. Wait until the connector processes those events and updates the checkpoint.
4. Stop the connector using Ctrl+C.
5. Record the current checkpoint value from the destination database.
6. Start the connector again.

During startup, the connector calls getLastProcessedEventId() to retrieve the saved checkpoint.

The startup output should display the previously recorded event ID.

For example:

Starting continuous CDC polling...
Last processed event ID: 42

The connector then queries the source cdc_events table for events with IDs greater than the saved checkpoint.

This verifies that the connector resumes from its saved processing position instead of starting from the beginning.

9.8 Testing Database Connection Recovery

The connector also implements recovery when a database connection is lost.

To verify this behavior, the connection recovery test involves temporarily stopping PostgreSQL while the connector is running.

The general procedure is:

1. Start the CDC connector and allow it to establish connections to both databases.
2. Temporarily stop the PostgreSQL server.
3. Observe the connector's connection failure handling.
4. Restart PostgreSQL.
5. Allow the connector to establish new database connections.

When a connection failure is detected, the connector exits the current connection session and waits for the configured reconnect delay.

With the following configuration:

cdc.reconnect.delay.ms=5000

the connector waits five seconds before attempting to reconnect.

After successfully establishing new connections, it retrieves the checkpoint again and resumes processing from the saved position.

This verifies the connection recovery and checkpoint-based resumption behavior.

9.9 Summary of Testing

The project testing covered the following areas:

Test scenario	What it verifies
INSERT operation	New employee records are captured and replicated
UPDATE operation	Changes to existing employee records are captured and replicated
DELETE operation	Deleted employee records are removed from the destination
Batch processing	Multiple events are applied in configured batch sizes
Pagination	Additional events are retrieved through successive source queries
Checkpoint management	The last successfully processed event ID is persisted
Restart recovery	Processing resumes using the saved checkpoint
Connection recovery	The connector reconnects after database connection loss

The tests demonstrate the core functionality of the PostgreSQL trigger-based CDC pipeline, including change capture, replication, batch processing, checkpoint management, and connection recovery.

Note that the retry logic for transient transaction errors is implemented in the connector, but this testing summary does not claim that deadlocks or serialization failures were deliberately injected and tested.

10. Limitations and Future Improvements

The CDC pipeline developed in this tutorial demonstrates how PostgreSQL triggers and a Java application can be used to capture and replicate database changes.

It includes several important capabilities, including batch processing, pagination, checkpoint management, transaction handling, retry logic, and connection recovery.

However, the current implementation is intended as a learning and demonstration project. There are several limitations that should be considered before using this approach in a production environment.

Let's examine the main limitations and possible improvements.

10.1 Initial Data Synchronization

The current implementation captures changes that occur after the CDC trigger has been created.

It does not automatically copy existing employee records from the source database to the destination database when the connector starts for the first time.

For example, suppose the source database already contains 1,000 employee records before the CDC trigger is installed.

The trigger-based implementation captures subsequent INSERT, UPDATE, and DELETE operations, but it does not automatically replicate those existing 1,000 records.

As a result, the destination database may initially be incomplete.

Possible improvement:

Implement an initial snapshot mechanism that copies existing records from the source database to the destination database before processing ongoing CDC events.

A production implementation should also coordinate the snapshot with change capture so that changes occurring during the initial copy are not missed or applied incorrectly.

10.2 Event ID Ordering and Concurrent Transactions

One important limitation of the current implementation is the use of sequence-generated event_id values as the checkpoint position.

The cdc_events table uses BIGSERIAL to generate event IDs.

Although these IDs are generated in increasing allocation order, they do not necessarily represent transaction commit order.

Consider the following scenario:

1. Transaction T1 inserts a CDC event and receives event ID 100, but does not commit yet.
2. Transaction T2 inserts another CDC event and receives event ID 101.
3. Transaction T2 commits.
4. The Java connector reads event 101 and processes it.
5. The destination checkpoint advances to 101.
6. Transaction T1 commits afterward, making event 100 visible.

The connector queries for events using:

WHERE event_id > 101

As a result, event 100 is not selected.

This means that under concurrent source transactions, the current checkpoint strategy can potentially miss events.

Possible improvement:

Use a change-position mechanism that accounts for transaction commit ordering rather than relying solely on sequence allocation order.

For example, PostgreSQL logical decoding and Write-Ahead Log (WAL) positions can provide transaction-oriented change information.

Another approach would be to introduce appropriate source-side transaction serialization, although that can reduce concurrency and throughput.

This is an important consideration when designing a production CDC system.

10.3 CDC Event Table Growth

The current implementation continuously inserts captured changes into the cdc_events table.

However, it does not include an automatic cleanup or archival mechanism for previously processed events.

As the source database receives more changes, the CDC event table will continue to grow.

Over time, this can increase storage requirements and affect database maintenance and query performance.

Possible improvement:

Introduce a CDC event retention policy.

For example:

• Archive older CDC events for auditing or recovery purposes.
• Remove events that are no longer required after they have been safely processed.
• Perform cleanup in controlled batches.
• Retain sufficient history to support operational recovery requirements.

A production implementation should coordinate cleanup with connector checkpoints and any required recovery or audit retention period.

10.4 Polling-Based Change Detection

The Java connector uses continuous polling to check for new CDC events.

The polling interval is configurable:

cdc.polling.interval.ms=2000

This means the connector waits for the configured interval between polling cycles.

Although polling is relatively simple to implement, it has some limitations:

• Changes may not be replicated immediately.
• Frequent polling can generate unnecessary database queries when no new events are available.
• A longer polling interval can increase replication latency.

Possible improvement:

For systems requiring lower latency, consider an event-notification mechanism or a streaming-based CDC architecture.

For example, PostgreSQL supports logical decoding, which can be used to build a WAL-based CDC solution.

The appropriate approach depends on the required latency, throughput, operational complexity, and recovery requirements.

10.5 Schema Evolution

The current implementation is designed around a specific employee table structure:

employees
---------
id
name
department
salary

The Java connector also expects these fields when processing the JSON data stored in changed_data.

If the source table structure changes, the trigger function and Java connector may need corresponding updates.

For example, adding a new employee column would require reviewing how the trigger captures the row and how the Java application maps the JSON data into the destination table.

Possible improvement:

Introduce explicit schema versioning and a more flexible mapping mechanism.

This would help the CDC pipeline handle changes to source and destination schemas more systematically.

10.6 Monitoring and Operational Visibility

The current implementation provides console logging for important events, such as:

• Database connection establishment.
• CDC event processing.
• Batch commits.
• Checkpoint updates.
• Retry attempts.
• Connection failures and recovery.

However, it does not provide a dedicated monitoring dashboard or a comprehensive set of operational metrics.

Possible improvement:

Integrate a monitoring solution such as Prometheus and Grafana.

Useful metrics could include:

• Number of CDC events captured.
• Number of events successfully processed.
• Current checkpoint position.
• Number of pending CDC events.
• Batch processing duration.
• Replication latency.
• Number of transaction retries.
• Database reconnection attempts.

These metrics would help operators identify processing delays, database failures, and other operational problems.

10.7 Scalability and Throughput

The current connector processes CDC events sequentially using a single processing flow.

Batch processing reduces the number of destination transactions, but the implementation does not distribute processing across multiple workers.

For a high-volume system, a single connector may eventually become a throughput bottleneck.

Possible improvement:

Introduce controlled parallel processing where appropriate.

However, parallel processing requires careful consideration of event ordering, transaction boundaries, checkpoint management, and operations affecting the same employee record.

A production design should preserve the required ordering guarantees while improving throughput.

10.8 Summary of Limitations and Future Improvements

The following table summarizes the main limitations of the current implementation and possible improvements.

Area	Current limitation	Possible improvement
Initial synchronization	Existing source records are not automatically copied	Implement an initial snapshot
Event ordering	Sequence-generated IDs do not guarantee commit order	Use commit-aware change positions
Event retention	CDC events are not automatically archived or removed	Introduce retention and archival policies
Change detection	Polling introduces latency and repeated queries	Consider event notification or WAL-based CDC
Schema evolution	Connector expects a fixed employee structure	Introduce schema versioning and flexible mapping
Monitoring	Primarily console-based logging	Integrate Prometheus and Grafana
Scalability	Single processing flow	Consider controlled parallel processing

These improvements provide a possible roadmap for evolving the demonstration into a more robust CDC solution.

The current project intentionally focuses on understanding the fundamental concepts of trigger-based change capture, event processing, transaction management, and checkpoint-based recovery.

11. Conclusion

In this tutorial, we built a Change Data Capture (CDC) pipeline using PostgreSQL triggers and Java.

We started by creating separate source and destination PostgreSQL databases. We then implemented PostgreSQL trigger functions to capture INSERT, UPDATE, and DELETE operations and store the changes in a CDC events table.

Next, we developed a Java-based CDC connector using JDBC to continuously retrieve and process these events.

Along the way, we explored several important concepts:

• Trigger-based change capture: Using PostgreSQL triggers to capture database changes.
• Event processing: Reading CDC events and applying corresponding changes to the destination database.
• Batch processing and pagination: Processing multiple events efficiently.
• Checkpoint management: Tracking the last successfully processed event to support restart recovery.
• Transaction management: Applying destination changes and checkpoint updates within the same transaction.
• Error handling and recovery: Implementing transaction retries, exponential backoff, and database connection recovery.

We also discussed some limitations of the current implementation, including initial data synchronization, event ordering under concurrent transactions, event retention, and scalability.

The project demonstrates how fundamental database features and Java can be combined to build a functional CDC pipeline without relying on an external CDC framework.

Although this implementation is intended as a learning project rather than a production-ready CDC platform, it provides a practical foundation for understanding change capture, event-driven data synchronization, and reliable database processing.

Source Code

The complete source code, SQL scripts, and project configuration are available on GitHub:

PostgreSQL Trigger-Based CDC Demo – GitHub

Feel free to clone the repository, experiment with the implementation, and extend it with additional features.

Thank you for reading!