PostgreSQL auditing

You know from my blogs that I am a PostgreSQL addict. If there is one thing that made my DBA days easier, was the mighty PGAudit extension. With pgaudit we can have detailed session and/or object audit logging.

Could expand more on why audit logs are essential under certain circumstances like government, financial, or ISO certification audits but my focus would be more on operations.

Imagine wanting to perform certain database migrations. You can switch traffic to a read replica which you shall promote, but this has the drawback of failed writes. How could you identify which time of day statistically has mainly reads and little or no writes? This is where the aggregation of the PGAudit logs would give you the answer.

Another example is cpu spikes building up during the day and you want to check the query patterns and see any correlation.

Furtermore you want to have a list of read queries that could be cached. By having PGAudit enabled you can identify the frequency of those queries and then decide how you can maximize the caching impact by picking the right queries.

Could go on and on. Overall PGAudit can do wonders.

 

 

So let’s get started.

We need to have a PostgreSQL installation with the extension enabled.
Debian already has a package for PGAudit available. In other cases you need to build the package on your own. Instructions can be found on the official guide.

We use a debian based docker image thus apt-get will do the work

FROM postgres:17
USER root
RUN apt-get update; apt-get install postgresql-17-pgaudit -y
USER postgres

Since PGAudit is installed we can create a custom postgresql configuration enabling it.

listen_addresses = '*'
port = 5432
max_connections = 20
shared_buffers = 128MB
temp_buffers = 8MB
work_mem = 4MB
max_wal_senders = 3
shared_preload_libraries = 'pgaudit'

pgaudit.log = 'all'  
pgaudit.log_catalog = 'off'
pgaudit.log_parameter = 'off'
pgaudit.log_statement_once = 'on'

Let’s put them all together into a docker compose file

version: '3.1'
 
services:
  postgres:
    build: ./
    restart: always
    environment:
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: postgres
    volumes:
      - ./postgresql.conf:/etc/postgresql/postgresql.conf
    command:
      - "-c"
      - "config_file=/etc/postgresql/postgresql.conf"
    ports:
      - 5432:5432

To run issue

 

docker compose up

Docker Compose V2 is out there with many good features, you can find more about it on the book I authored:
A Developer’s Essential Guide to Docker Compose
.

No we can execute some queries and see how they are logged.

$ docker compose exec -it postgres bash
# psql
postgres=# SELECT 1;
 ?column? 
----------
        1
(1 row)

If we check the container logs, we shall see an audit trail of the queries we executed.

postgres-1  | 2025-09-18 23:33:05.166 GMT [271] LOG:  AUDIT: SESSION,1,1,READ,SELECT,,,SELECT 1,<not logged>

Now this looks amazing however there are certain things to take into consideration.

Costs

Audit logs are expensive because of their volume. Before enabling them on production you need to make sure they are stored in a cost effective form of storage.

Resources

They do consume resources from your PostgreSQL instance. Size up your resources carefully.

Caution on Parameters

Do not log the parameters and if you do, understand that you log the actual data contained in the database. Logging parameters and storing audit logs in a widely access storage likely results in a data leak. Avoid logging the parameters. If you do so have a good reason and also a proper place to store them that satisfies the security needs.

Prevent Data Leaks

If your application is written badly and the queries do not use parameterized statements the audit logs will lead to dataleaks. All queries should be parameterized and should avoid any hardcoded string inside.

We have this table

CREATE TABLE table_name (
    id SERIAL PRIMARY KEY,
    sensitive_field TEXT NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

This is a bad query


SELECT *FROM table_name WHERE sensitive_field='data-leak';

Leads to leaking the sensitive field value to the terminal


postgres-1 | 2025-09-18 23:44:53.953 GMT [271] STATEMENT: SELECT *FROM table_name WHERE sensitive_field='data-leak';

Instead you can use a parameterized query, just like the following jdbc example

             PreparedStatement stmt = conn.prepareStatement("SELECT * FROM table_name WHERE sensitive_field = ?")) {

            // Parameterized query - safe from SQL injection
            stmt.setString(1, "data-leak");

In the logs the information is not leaked.


postgres-1  | 2025-09-19 07:20:39.531 GMT [63] LOG:  AUDIT: SESSION,3,1,READ,SELECT,,,SELECT * FROM table_name WHERE sensitive_field = $1,<not logged>

 

That’s it. Let’s enjoy PostgreSQL that amazing piece of technology and handle with care.

PostgreSQL driver host Fail-over

Previously we setup BiDirectional replication for PostgreSQL.

You might have various legitimate reasons to do so and probably will get to that in another blog.
Overall picking this type of replication can be influenced a lot by the nature of your application, the need for active active dr scenarios and even cases of migration .

Since we have replication in place it would be great to examine the case of an outage and how we can utilise the native fail-over functionality of the PostgreSQL drivers.
We shall change the conflict resolution strategy to last_update_wins. This way between two simultaneous updates in each database the update with the max commit timestamp will be the one chosen one.

listen_addresses = '*'
port = 5432
max_connections = 20
shared_buffers = 128MB
temp_buffers = 8MB
work_mem = 4MB
wal_level = logical
max_wal_senders = 3
track_commit_timestamp = on
shared_preload_libraries = 'pglogical'
pglogical.conflict_resolution = 'last_update_wins'

We need to spin up the compose services with the new changes:

docker compose up

Docker Compose V2 is out there with many good features, you can find more about it on the book I authored:
A Developer’s Essential Guide to Docker Compose
.

Take note that based on the programming language and the driver, this functionality might not always be available. The concept is that when you configure the connection pool to establish connection to the database you can configure two hosts. The first host will be the primary one and the secondary host will be the one to fail-over once the primary host gets offline. The fail-over can be interchangeable, essentially the driver tries to find the first available host.

Python and the driver psycopg2 offer this functionality. We shall implement an app using the flask api. The app will give two endpoints, one for fetching an employee’s salary and one to increment the salary by 1:

from flask import Flask 
from psycopg2.pool import SimpleConnectionPool

app = Flask(__name__)

postgreSQL_pool = SimpleConnectionPool(1, 20, user="postgres",
                                       password="postgres",
                                       host="localhost,localhost",
                                       port="5432,5431",
                                       database="postgres",
                                       options="-c search_path=test_schema")


@app.route('/employee/<employee_id>/salary/increment', methods=['POST'])
def increment_salary(employee_id):
    conn = postgreSQL_pool.getconn()
    cur = conn.cursor()
    cur.execute("""
        UPDATE employee SET salary=salary + %s WHERE id = %s;
        """, (1, employee_id))
    conn.commit()
    cur.close()
    postgreSQL_pool.putconn(conn)
    return '', 204


@app.route('/employee/<employee_id>/salary')
def index(employee_id):
    conn = postgreSQL_pool.getconn()
    cur = conn.cursor()
    cur.execute("""
        SELECT salary FROM employee WHERE id=%s;
        """, employee_id)
    salary = cur.fetchone()[0]
    cur.close()
    postgreSQL_pool.putconn(conn)
    return str(salary), 200

Let’s example the SimpleConnectionPool, we can see two hosts separated with a comma (it’s localhost since it’s our local docker compose running) and on the port section the respective host ports are separated by comma.

We can run the app

flash run

And on another terminal issue the calls using curl

$ curl -X POST http://localhost:5000/employee/1/salary/increment
$ curl http://localhost:5000/employee/1/salary

Overall the salary will increase and we should see that on the get request.

Now let’s shut down one database

docker compose stop postgres-b

The first call after this operation will be a failed one, however the connection will be reinitialized and point to the secondary host.

% curl http://localhost:5000/employee/1/salary
<!doctype html>
<html lang=en>
<title>500 Internal Server Error</title>
<h1>Internal Server Error</h1>
<p>The server encountered an internal error and was unable to complete your request. Either the server is overloaded or there is an error in the application.</p>
%  curl http://localhost:5000/employee/1/salary
1254.23                              

The same functionality applies for other drivers. Take for example the Java driver configuration on a spring boot application.

spring.datasource.url=jdbc:postgresql://localhost:5432,localhost:5431/postgres?currentSchema=test_schema
spring.datasource.username=postgres
spring.datasource.password=postgres

On the jdbc url we add two hosts comma delimited localhost:5432,localhost:5431

Then we can implement an application with the same functionality.

package com.egkatzioura.psqlfailover.repository;

import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Repository;


@Repository
public class EmployeeRepository {

    private final JdbcTemplate jdbcTemplate;

    public EmployeeRepository(JdbcTemplate jdbcTemplate) {
        this.jdbcTemplate = jdbcTemplate;
    }


    public void incrementSalary(Long employeeId, float increment) {
        jdbcTemplate.update("UPDATE employee SET salary=salary+? WHERE id=?",increment, employeeId);
    }

    public Float fetchSalary(Long employeeId) {
        return jdbcTemplate.queryForObject("SELECT salary FROM employee WHERE id=?",new Object[]{employeeId},Float.class);
    }
}

package com.egkatzioura.psqlfailover.controller;

import com.egkatzioura.psqlfailover.repository.EmployeeRepository;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class EmployeeController {

    private final EmployeeRepository employeeRepository;

    public EmployeeController(EmployeeRepository employeeRepository) {
        this.employeeRepository = employeeRepository;
    }

    @PostMapping("/employee/{id}/salary/increment")
    public void incrementSalary(@PathVariable Long id) {
        employeeRepository.incrementSalary(id,1f);
    }

    @GetMapping("/employee/{id}/salary")
    public Float fetchSalary(@PathVariable Long id) {
        return employeeRepository.fetchSalary(id);
    }
}

Thanks to the replication the changes should have reached the other database. You can start and restart the compose services in a round robin fashion. The changes will be replicated and thus every time there is a fail-over the data will be there.

While we start and stop the databases docker compose stop postgres-b, we can issue requests using curl:

$ curl -X POST http://localhost:8080/employee/1/salary/increment
$ curl http://localhost:8080/employee/1/salary

Eventually the java driver handles the fail-over even more gracefully. Instead of failing on the first request during the fail-over instead it will fisr to connect to the other host and give back the results.

That’s it. You setup BiDirectional replication on PostgreSQL and you managed to take advantage of the driver capabilities to fail-over to different hosts. Hope you had some fun!

PostgreSQL BiDirectional Replication

As you can understand from my previous blogs I am really into PostgreSQL.

Previously we run Debezium in Embedded mode. Behind the scenes Debezium consumes the changes that were committed to the transaction log. This happens by utilising the logical decoding feature of PostgreSQL.

In this blog we shall focus on replication and more specific bidirectional replication. To achieve bidirectional replication in PostgreSQL we need the module pglogical. You might wonder the deference between logical decoding and pglogical. Essentially logical decoding has its origins from PgLocigal. View PgLocial as a more featureful module while logical decoding is embedded to a PostgreSQL distribution.

We will create a custom PostgreSQL Docker image and install PgLogical.

# Use the official PostgreSQL image as base
FROM postgres:15
USER root
RUN apt-get update; apt-get install postgresql-15-pglogical -y
USER postgres

Also we need to have a PostgreSQL configuration that will enable PgLogical replication and the conflict resolution.

listen_addresses = '*'
port = 5432
max_connections = 20
shared_buffers = 128MB
temp_buffers = 8MB
work_mem = 4MB
wal_level = logical
max_wal_senders = 3
track_commit_timestamp = on
shared_preload_libraries = 'pglogical'
pglogical.conflict_resolution = 'first_update_wins'

Let’s break this down. We added pglogical and we enabled track_commit_timestamp. By enabling this parameter PostgreSQL tracks the commit time of transactions. This will be crucial for the conflict resolution strategy.
Now let’s see the conflict resolution. We selected ‘first_update_wins’, therefore in case of two transactions operating on the same row, the transaction that finished first will be the one to be considered.

Bidirectional replication is setup upon a table. Since we use Docker we shall provide an initialization script to PostgreSQL.

The script will:

  • Enable pglogical
  • Create the table
  • Add a target node
  • Insert the row we shall run tests upon
#!/bin/bash
set -e

psql -v ON_ERROR_STOP=1 --username "$POSTGRES_USER" --dbname "$POSTGRES_DB" &amp;lt;&amp;lt;-EOSQL
  ALTER SYSTEM RESET shared_preload_libraries;
  CREATE EXTENSION pglogical;

  create schema test_schema;
  create table test_schema.employee(
          id  SERIAL PRIMARY KEY,
          firstname   TEXT    NOT NULL,
          lastname    TEXT    NOT NULL,
          email       TEXT    not null,
          age         INT     NOT NULL,
          salary         real,
          unique(email)
      );

  SELECT pglogical.create_node(
      node_name := '$TARGET',
      dsn := 'host=$TARGET port=5432 dbname=$POSTGRES_DB user=$POSTGRES_USER password=$POSTGRES_PASSWORD');

  SELECT pglogical.replication_set_add_table('default', 'test_schema.employee', true);

  insert into test_schema.employee (id,firstname,lastname,email,age,salary) values (1,'John','Doe 1','john1@doe.com',18,1234.23);


EOSQL

Let’s create the instances now using docker compose.

version: '3.1'

services:
  postgres-a:
    build: ./pglogicalimage
    restart: always
    environment:
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: postgres
      TARGET: postgres-b
    volumes:
      - ./config/postgresql.conf:/etc/postgresql/postgresql.conf
      - ./init:/docker-entrypoint-initdb.d
    command:
      - "-c"
      - "config_file=/etc/postgresql/postgresql.conf"
    ports:
      - 5431:5432
  postgres-b:
    build: ./pglogicalimage
    restart: always
    environment:
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: postgres
      TARGET: postgres-a
    volumes:
      - ./config/postgresql.conf:/etc/postgresql/postgresql.conf
      - ./init:/docker-entrypoint-initdb.d
    command:
      - "-c"
      - "config_file=/etc/postgresql/postgresql.conf"
    ports:
      - 5432:5432

We can get our instances up and running by issuing

docker compose up

Docker Compose V2 is out there with many good features, you can find more about it on the book I authored:
A Developer’s Essential Guide to Docker Compose
.

Since both instances are up and running we need to enable the replication. Therefore we shall subscribe the nodes to each other.

Execute on the first node

SELECT pglogical.create_subscription(
  subscription_name := 'postgres_b',
  provider_dsn := 'host=postgres-b port=5432 dbname=postgres user=postgres password=postgres',
  synchronize_data := false,
  forward_origins := '{}' );

Execute at the second node

SELECT pglogical.create_subscription(
  subscription_name := 'postgres_a',
  provider_dsn := 'host=postgres-a port=5432 dbname=postgres user=postgres password=postgres',
  synchronize_data := false,
  forward_origins := '{}' );

You can use any PostgreSQL client that suits you. Alternatively you can just use the psql client that comes packaged with the Docker Images.
For example:

Login to the first node

docker compose exec postgres-a psql  --username postgres --dbname postgres

Login to the second node

docker compose exec postgres-b psql  --username postgres --dbname postgres

Let’s see how conflict resolution will work now.

On the first node we shall run the following snippet

BEGIN;
UPDATE test_schema.employee SET lastname='first wins';

#before committing start transaction on postgres-b

COMMIT;

Don’t press commit immediately, instead take the time and before you commit the transaction start the following transaction on the second node.

BEGIN;
UPDATE test_schema.employee SET lastname='second looses';

#make sure transaction on node postgres-a is committed first.

COMMIT;

This transaction will be committed after the transaction that takes places in postgres-a.

Let’s check the logs on postgres-a-1

postgres-a-1  | 2024-05-01 07:10:45.128 GMT [70] LOG:  CONFLICT: remote UPDATE on relation test_schema.employee (local index employee_pkey). Resolution: keep_local.
postgres-a-1  | 2024-05-01 07:10:45.128 GMT [70] DETAIL:  existing local tuple {id[int4]:1 firstname[text]:John lastname[text]:first wins email[text]:john1@doe.com age[int4]:18 salary[float4]:1234.23} xid=748,origin=0,timestamp=2024-05-01 07:10:42.269227+00; remote tuple {id[int4]:1 firstname[text]:John lastname[text]:second looses email[text]:john1@doe.com age[int4]:18 salary[float4]:1234.23} in xact origin=1,timestamp=2024-05-01 07:10:45.125791+00,commit_lsn=0/16181C0
postgres-a-1  | 2024-05-01 07:10:45.128 GMT [70] CONTEXT:  apply UPDATE from remote relation test_schema.employee in commit before 0/16181C0, xid 747 committed at 2024-05-01 07:10:45.125791+00 (action #2) from node replorigin 1

The transaction that took place on postgres-a finished first. Postgres-a received the replication data from the transaction of node postgres-b. A comparison was issued on the commit timestamp, because the commit timestamp of the transaction on postgres-a was earlier the resolution was to keep the local changes.

We can see the reverse on postgres-b


postgres-b-1 | 2024-05-01 07:10:45.127 GMT [81] LOG: CONFLICT: remote UPDATE on relation test_schema.employee (local index employee_pkey). Resolution: apply_remote.
postgres-b-1 | 2024-05-01 07:10:45.127 GMT [81] DETAIL: existing local tuple {id[int4]:1 firstname[text]:John lastname[text]:second looses email[text]:john1@doe.com age[int4]:18 salary[float4]:1234.23} xid=747,origin=0,timestamp=2024-05-01 07:10:45.125791+00; remote tuple {id[int4]:1 firstname[text]:John lastname[text]:first wins email[text]:john1@doe.com age[int4]:18 salary[float4]:1234.23} in xact origin=1,timestamp=2024-05-01 07:10:42.269227+00,commit_lsn=0/1618488
postgres-b-1 | 2024-05-01 07:10:45.127 GMT [81] CONTEXT: apply UPDATE from remote relation test_schema.employee in commit before 0/1618488, xid 748 committed at 2024-05-01 07:10:42.269227+00 (action #2) from node replorigin 1

Let’s check the result in the database.

postgres=# SELECT*FROM test_schema.employee;
 id | firstname |  lastname  |     email     | age | salary  
----+-----------+------------+---------------+-----+---------
  1 | John      | first wins | john1@doe.com |  18 | 1234.23

As expected the first transaction is the one that stayed.
To wrap it up:

  • We started two transactions in parallel
  • We changed the same row
  • We accepted the changes of the transaction that finished first

That’s it. Hope you had some fun and now you have another tool for your needs. In the next blog we shall examine PostgreSQL’s driver capabilities and how we can configure an automated failover to another instance.

Debezium in Embedded mode

In a previous blog we setup a Debezium server reading events from a from a PostgresQL database. Then we streamed those changes to a Redis instance through a Redis stream.

We might get the impression that in order to run Debezium we need to have two extra components running in our infrastructure:

  • A standalone Debezium server instance
  • A software component with streaming capabilities and various integrations, such as Redis or Kafka

This is not always the case since Debezium can run in embedded mode. By running in embedded mode you use Debezium in order to read directly from a database’s transaction log. It is up to you how you are gonna handle the entries retrieved. The process reading the entries from the transaction log can reside on any Java application thus there is no need for a standalone deployment.

Apart from the number of components reduced, the other benefit is that we can alter the entries as we read them from the database and take action in our application. Sometimes we might just need a subset of the capabilities offered.

Let’s use the same PotsgreSQL configurations we used previously

listen_addresses = '*'
port = 5432
max_connections = 20
shared_buffers = 128MB
temp_buffers = 8MB
work_mem = 4MB
wal_level = logical
max_wal_senders = 3

Also we shall create an initialization script for the table we want to focus

#!/bin/bash
set -e
 
psql -v ON_ERROR_STOP=1 --username "$POSTGRES_USER" --dbname "$POSTGRES_DB" <<-EOSQL
  create schema test_schema;
  create table test_schema.employee(
          id  SERIAL PRIMARY KEY,
          firstname   TEXT    NOT NULL,
          lastname    TEXT    NOT NULL,
          email       TEXT    not null,
          age         INT     NOT NULL,
          salary         real,
          unique(email)
      );
EOSQL

Our Docker Compose file will look like this

version: '3.1'
 
services:

  postgres:
    image: postgres
    restart: always
    environment:
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: postgres
    volumes:
      - ./postgresql.conf:/etc/postgresql/postgresql.conf
      - ./init:/docker-entrypoint-initdb.d
    command:
      - "-c"
      - "config_file=/etc/postgresql/postgresql.conf"
    ports:
      - 5432:5432

The configuration files we created are mounted to the PostgreSQL Docker container. Docker Compose V2 is out there with many good features, you can find more about it on the book I authored:
A Developer’s Essential Guide to Docker Compose
.

Provided we run docker compose up, a postgresql server with a schema and a table will be up and running. Also that server will have logical decoding enabled and Debezium shall be able to track changes on that table through the transaction log.
We have everything needed to proceed on building our application.

First let’s add the dependencies needed:

 
    <properties>
        <maven.compiler.source>17</maven.compiler.source>
        <maven.compiler.target>17</maven.compiler.target>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <version.debezium>2.3.1.Final</version.debezium>
        <logback-core.version>1.4.12</logback-core.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>io.debezium</groupId>
            <artifactId>debezium-api</artifactId>
            <version>${version.debezium}</version>
        </dependency>
        <dependency>
            <groupId>io.debezium</groupId>
            <artifactId>debezium-embedded</artifactId>
            <version>${version.debezium}</version>
        </dependency>
        <dependency>
            <groupId>io.debezium</groupId>
            <artifactId>debezium-connector-postgres</artifactId>
            <version>${version.debezium}</version>
        </dependency>
        <dependency>
            <groupId>io.debezium</groupId>
            <artifactId>debezium-storage-jdbc</artifactId>
            <version>${version.debezium}</version>
        </dependency>
        <dependency>
            <groupId>ch.qos.logback</groupId>
            <artifactId>logback-classic</artifactId>
            <version>${logback-core.version}</version>
        </dependency>
        <dependency>
            <groupId>ch.qos.logback</groupId>
            <artifactId>logback-core</artifactId>
            <version>${logback-core.version}</version>
        </dependency>
    </dependencies>

We also need to create the Debezium embedded properties:

name=embedded-debezium-connector
connector.class=io.debezium.connector.postgresql.PostgresConnector
offset.storage=org.apache.kafka.connect.storage.FileOffsetBackingStore
offset.flush.interval.ms=60000
database.hostname=127.0.0.1
database.port=5432
database.user=postgres
database.password=postgres
database.dbname=postgres
database.server.name==embedded-debezium
debezium.source.plugin.name=pgoutput
plugin.name=pgoutput
database.server.id=1234
topic.prefix=embedded-debezium
schema.include.list=test_schema
table.include.list=test_schema.employee

Apart from establishing the connection towards the PostgresQL Database we also decided to store the offset in a file. By using the offset in Debezium we keep track of the progress we do on processing the events.

On each change that happens on the table test_schema.employee we shall receive an event. Once we receive that event our codebase should handle it.
To handle the events we need to create a DebeziumEngine.ChangeConsumer. The ChangeConsumer will consume the events emitted.

package com.egkatzioura;

import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.RecordChangeEvent;
import org.apache.kafka.connect.source.SourceRecord;

import java.util.List;

public class CustomChangeConsumer implements DebeziumEngine.ChangeConsumer<RecordChangeEvent<SourceRecord>> {

    @Override
    public void handleBatch(List<RecordChangeEvent<SourceRecord>> records, DebeziumEngine.RecordCommitter<RecordChangeEvent<SourceRecord>> committer) throws InterruptedException {
        for(RecordChangeEvent<SourceRecord> record: records) {
            System.out.println(record.record().toString());
        }
    }

}

Every incoming event will be printed on the console.

Now we can add our main class where we setup the engine.

package com.egkatzioura;

import io.debezium.embedded.Connect;
import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.format.ChangeEventFormat;

import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.util.Properties;

public class Application {

    public static void main(String[] args) throws IOException {
        Properties properties = new Properties();

        try(final InputStream stream = Application.class.getClassLoader().getResourceAsStream("embedded_debezium.properties")) {
            properties.load(stream);
        }
        properties.put("offset.storage.file.filename",new File("offset.dat").getAbsolutePath());

        var engine = DebeziumEngine.create(ChangeEventFormat.of(Connect.class))
                .using(properties)
                .notifying(new CustomChangeConsumer())
                .build();
        engine.run();

    }

}

Provided our application is running as well as the PostgresQL database we configured previously, we can start inserting data

docker exec -it debezium-embedded-postgres-1 psql postgres postgres
psql (15.3 (Debian 15.3-1.pgdg120+1))
Type "help" for help.

postgres=# insert into test_schema.employee (firstname,lastname,email,age,salary) values ('John','Doe 1','john1@doe.com',18,1234.23);

Also we can see the change on the console

SourceRecord{sourcePartition={server=embedded-debezium}, sourceOffset={last_snapshot_record=true, lsn=22518160, txId=743, ts_usec=1705916606794160, snapshot=true}} ConnectRecord{topic='embedded-debezium.test_schema.employee', kafkaPartition=null, key=Struct{id=1}, keySchema=Schema{embedded-debezium.test_schema.employee.Key:STRUCT}, value=Struct{after=Struct{id=1,firstname=John,lastname=Doe 1,email=john1@doe.com,age=18,salary=1234.23},source=Struct{version=2.3.1.Final,connector=postgresql,name=embedded-debezium,ts_ms=1705916606794,snapshot=last,db=postgres,sequence=[null,"22518160"],schema=test_schema,table=employee,txId=743,lsn=22518160},op=r,ts_ms=1705916606890}, valueSchema=Schema{embedded-debezium.test_schema.employee.Envelope:STRUCT}, timestamp=null, headers=ConnectHeaders(headers=)}

We did it. We managed to run Debezium through a Java application without the need of a standalone Debezium server running or a streaming component. You can find the code on GitHub.

Debezium Server with PostgreSQL and Redis Stream

Debezium is a great tool for capturing the row level changes that happen on a Database and stream those changes to a broker of our choice.

Since this functionality stays in the boundaries of a Database, it helps on keeping our applications simple. For example there in no need for an application to emit events on any database interactions. Debezium will monitor the row changes and will emit the events. Based on the broker solution used with Debezium a consumer can subscribe to the broker thus receive the changes.

PostgreSQL being a popular SQL database, it is supported by Debezium.

Our goal would be to listen to PostgreSQL changes and stream them to a Redis stream through a Debezium Server. It is common to use Debizum with Kafka, in case where Kafka is not present in a team’s Tech stack we can use other brokers.

In our case we would keep things lightweight by using Redis Streams.

Redis will be setup without any extra configurations.

In order to use PostgreSQL with Debezium it is essentials to alter the configuration on postgreSQL.

The configuration we shall use on postgreSQL will be the following

listen_addresses = '*'
port = 5432
max_connections = 20
shared_buffers = 128MB
temp_buffers = 8MB
work_mem = 4MB
wal_level = logical
max_wal_senders = 3

As we can see we use the logical_decoding from PostgreSQL.
From the documentation:

Logical decoding is the process of extracting all persistent changes to a database’s tables into a coherent, easy to understand format which can be interpreted without detailed knowledge of the database’s internal state.

In PostgreSQL, logical decoding is implemented by decoding the contents of the write-ahead log, which describe changes on a storage level, into an application-specific form such as a stream of tuples or SQL statements.

We will also create a namespace and a table for PostgreSQL. The namespace and the table will be the ones to listen for changes.

#!/bin/bash
set -e

psql -v ON_ERROR_STOP=1 --username "$POSTGRES_USER" --dbname "$POSTGRES_DB" <<-EOSQL
  create schema test_schema;
  create table test_schema.employee(
          id  SERIAL PRIMARY KEY,
          firstname   TEXT    NOT NULL,
          lastname    TEXT    NOT NULL,
          email       TEXT    not null,
          age         INT     NOT NULL,
          salary         real,
          unique(email)
      );
EOSQL

This is the table we used in a previous PostgreSQL example.

Debezium will have to be able to interact with the PostgreSQL server as well as the the redis server.
The configuration should be the following.

debezium.sink.type=redis
debezium.sink.redis.address=redis:6379
debezium.source.connector.class=io.debezium.connector.postgresql.PostgresConnector
debezium.source.offset.storage.file.filename=data/offsets.dat
debezium.source.offset.flush.interval.ms=0
debezium.source.database.hostname=postgres
debezium.source.database.port=5432
debezium.source.database.user=postgres
debezium.source.database.password=postgres
debezium.source.database.dbname=postgres
debezium.source.database.server.name=tutorial
debezium.source.schema.whitelist=test_schema
debezium.source.plugin.name=pgoutput

By examining the configuration we can see that we have the necessary information for Debezium to communicate to the PostgreSQL database, also we specify the schema that we created previously. Therefore only changes from that schema will be streamed. We can also make things more restrictive for example whitelisting tables.

Since this demo will involve three different software Components docker compose will come in handy.

version: '3.1'

services:
  redis:
    image: redis
    ports:
      - 6379:6379
    depends_on:
      - postgres
  postgres:
    image: postgres
    restart: always
    environment:
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: postgres
    volumes:
      - ./postgresql.conf:/etc/postgresql/postgresql.conf
      - ./init:/docker-entrypoint-initdb.d
    command:
      - "-c"
      - "config_file=/etc/postgresql/postgresql.conf"
    ports:
      - 5432:5432
  debezium:
    image: debezium/server
    volumes:
      - ./conf:/debezium/conf
      - ./data:/debezium/data
    depends_on:
      - redis

By using Compose we were able to spin up three different software components on the same network. This helps the components to interact with each other by using the dns names of the services as specified on Compose. Also the configuration files we created previously are mounted to the Docker containers. Docker Compose V2 is out there with many good features, you can find more about it on the book I authored
A Developer’s Essential Guide to Docker Compose
.

In order to get the stack running we shall execute the following command

$ docker compose up

Since it is up and running, we can now start listening for events.

We shall login to Redis and start listen for any possible database updates.

$ docker exec -it debezium-example-redis-1 redis-cli
> xread block 1000000 streams tutorial.test_schema.employee $

This will make it possible to block until we receive one event from the stream.
If we examine the stream name we should see the pattern of {server-name}.{schema}.{table}. This would allow consumers to subscribe only to the changes of interest.

Onwards we will make an entry.

$ docker exec -it debezium-example-postgres-1 psql postgres postgres
> insert into test_schema.employee (firstname,lastname,email,age,salary) values ('John','Doe 1','john1@doe.com',18,1234.23);
> \q

If we check the redis session we should see that we received an event from the Redis stream

127.0.0.1:6379> xread block 1000000 streams tutorial.test_schema.employee $
1) 1) "tutorial.test_schema.employee"
   2) 1) 1) "1663796657336-0"
         2) 1) "{\"schema\":{\"type\":\"struct\",\"fields\":[{\"type\":\"int32\",\"optional\":false,\"default\":0,\"field\":\"id\"}],\"optional\":false,\"name\":\"tutorial.test_schema.employee.Key\"},\"payload\":{\"id\":1}}"
            2) "{\"schema\":{\"type\":\"struct\",\"fields\":[{\"type\":\"struct\",\"fields\":[{\"type\":\"int32\",\"optional\":false,\"default\":0,\"field\":\"id\"},{\"type\":\"string\",\"optional\":false,\"field\":\"firstname\"},{\"type\":\"string\",\"optional\":false,\"field\":\"lastname\"},{\"type\":\"string\",\"optional\":false,\"field\":\"email\"},{\"type\":\"int32\",\"optional\":false,\"field\":\"age\"},{\"type\":\"float\",\"optional\":true,\"field\":\"salary\"}],\"optional\":true,\"name\":\"tutorial.test_schema.employee.Value\",\"field\":\"before\"},{\"type\":\"struct\",\"fields\":[{\"type\":\"int32\",\"optional\":false,\"default\":0,\"field\":\"id\"},{\"type\":\"string\",\"optional\":false,\"field\":\"firstname\"},{\"type\":\"string\",\"optional\":false,\"field\":\"lastname\"},{\"type\":\"string\",\"optional\":false,\"field\":\"email\"},{\"type\":\"int32\",\"optional\":false,\"field\":\"age\"},{\"type\":\"float\",\"optional\":true,\"field\":\"salary\"}],\"optional\":true,\"name\":\"tutorial.test_schema.employee.Value\",\"field\":\"after\"},{\"type\":\"struct\",\"fields\":[{\"type\":\"string\",\"optional\":false,\"field\":\"version\"},{\"type\":\"string\",\"optional\":false,\"field\":\"connector\"},{\"type\":\"string\",\"optional\":false,\"field\":\"name\"},{\"type\":\"int64\",\"optional\":false,\"field\":\"ts_ms\"},{\"type\":\"string\",\"optional\":true,\"name\":\"io.debezium.data.Enum\",\"version\":1,\"parameters\":{\"allowed\":\"true,last,false,incremental\"},\"default\":\"false\",\"field\":\"snapshot\"},{\"type\":\"string\",\"optional\":false,\"field\":\"db\"},{\"type\":\"string\",\"optional\":true,\"field\":\"sequence\"},{\"type\":\"string\",\"optional\":false,\"field\":\"schema\"},{\"type\":\"string\",\"optional\":false,\"field\":\"table\"},{\"type\":\"int64\",\"optional\":true,\"field\":\"txId\"},{\"type\":\"int64\",\"optional\":true,\"field\":\"lsn\"},{\"type\":\"int64\",\"optional\":true,\"field\":\"xmin\"}],\"optional\":false,\"name\":\"io.debezium.connector.postgresql.Source\",\"field\":\"source\"},{\"type\":\"string\",\"optional\":false,\"field\":\"op\"},{\"type\":\"int64\",\"optional\":true,\"field\":\"ts_ms\"},{\"type\":\"struct\",\"fields\":[{\"type\":\"string\",\"optional\":false,\"field\":\"id\"},{\"type\":\"int64\",\"optional\":false,\"field\":\"total_order\"},{\"type\":\"int64\",\"optional\":false,\"field\":\"data_collection_order\"}],\"optional\":true,\"field\":\"transaction\"}],\"optional\":false,\"name\":\"tutorial.test_schema.employee.Envelope\"},\"payload\":{\"before\":null,\"after\":{\"id\":1,\"firstname\":\"John\",\"lastname\":\"Doe 1\",\"email\":\"john1@doe.com\",\"age\":18,\"salary\":1234.23},\"source\":{\"version\":\"1.9.5.Final\",\"connector\":\"postgresql\",\"name\":\"tutorial\",\"ts_ms\":1663796656393,\"snapshot\":\"false\",\"db\":\"postgres\",\"sequence\":\"[null,\\\"24289128\\\"]\",\"schema\":\"test_schema\",\"table\":\"employee\",\"txId\":738,\"lsn\":24289128,\"xmin\":null},\"op\":\"c\",\"ts_ms\":1663796657106,\"transaction\":null}}"
(10.17s)
127.0.0.1:6379> 

How cool is that? We can now start streaming our databases changes to the broker of our choice.

You can find the source code on GitHub.

Run a docker PostgreSQL instance locally for Testing

Running a PostgreSQL instance ad-hoc for testing is not always as bootstraping as it should be. This blog will run a PostgreSQL instance that connects to your workstation’s network and instead of using one of the popular tools like dbeaver we shall use the client that comes with the instance. Also we shall run a bootstrap script to have some data pre-inserted.

Let’s get started by running the instance. On purpose I will use another port. On scenarios of multiple instances running in your workstation, port collisions are likely. The workaround would be to choose port 5433.

docker run --rm --name test-instance -e POSTGRES_PASSWORD=password -p 5433:5432 postgres

This will run PostgreSQL and you shall be able to connect to port 5433. On a CTRL-C the instance will be stopped and destroyed.

Now instead of using an external tool to connect let’s use the instance itself, it comes with psql pre-installed.

docker exec -it test-instance /bin/bash
> psql postgres postgres
postgres=# \h
Available help:
  ABORT                            ALTER TRIGGER                    CREATE RULE                      DROP GROUP                       LISTEN
  ALTER AGGREGATE                  ALTER TYPE                       CREATE SCHEMA                    DROP INDEX                       LOAD
  ALTER COLLATION                  ALTER USER                       CREATE SEQUENCE                  DROP LANGUAGE                    LOCK
.....
postgres=# \q

The instance works and connections from the outside are possible.

Next step would be to bootstrap a db initialization script.

#!/bin/bash
set -e
 
psql -v ON_ERROR_STOP=1 --username postgres --dbname postgres <<-EOSQL
    create schema test_schema;
 
    create table test_schema.employee(
        id  SERIAL PRIMARY KEY,
        firstname   TEXT    NOT NULL,
        lastname    TEXT    NOT NULL,
        email       TEXT    not null,
        age         INT     NOT NULL,
        salary         real,
        unique(email)
    );
 
    insert into test_schema.employee (firstname,lastname,email,age,salary)
    values ('John','Doe 1','john1@doe.com',18,1234.23);

EOSQL

Supposing the file with the script is called init_db.sh

Let’s run the command with the initialization schema mounted.

docker run --rm --name test-instance -v /path/to/init_db.sh:/docker-entrypoint-initdb.d/init-db-script.sh -e POSTGRES_PASSWORD=password -p 5433:5432 postgres

And let’s check the results.

docker exec -it test-instance /bin/bash
>psql postgres postgres
postgres=# SELECT*FROM test_schema.employee;
 id | firstname | lastname |     email     | age | salary
----+-----------+----------+---------------+-----+---------
  1 | John      | Doe 1    | john1@doe.com |  18 | 1234.23
(1 row)

That’s it! You created a Postgresql database through docker, you did connect to it also you added a bootstrap script with data.

Read replicas and Spring Data Part 4: Configuring the read repository

Previously we set up two EntityManagers in the same application. One for the reads and one for the writes. Now it’s time to create our read repository.

The read only repository will use the secondary read only EntityManager.

In order to make it a read only repository, it is essential not to have any save and persist actions.

package com.gkatzioura.springdatareadreplica.repository;

import java.util.List;

import org.springframework.data.repository.Repository;

import com.gkatzioura.springdatareadreplica.config.ReadOnlyRepository;
import com.gkatzioura.springdatareadreplica.entity.Employee;

/**
 * This is a read only repository
 */
public interface ReadEmployeeRepository extends Repository {

    List findAll();

}

Our next task would be to create this repository with the read database entity manager.
This means that all repositories shall be created using the default entity manager except from the read only repositories.

I would create an Annotation first. This annotation will declare my repository as Read only. Also I will use this annotation for the scanning operation so that the appropriate EntityManager will be used.

package com.gkatzioura.springdatareadreplica.config;

import java.lang.annotation.Documented;
import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;

@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.TYPE})
@Documented
public @interface ReadOnlyRepository {
}

Now I know that spring boot removes the need for annotations and does repository creation in an automated way however our case is a peculiar one.

By making some adjustments our read only repository will look like this

package com.gkatzioura.springdatareadreplica.repository;

import java.util.List;

import org.springframework.data.repository.Repository;

import com.gkatzioura.springdatareadreplica.config.ReadOnlyRepository;
import com.gkatzioura.springdatareadreplica.entity.Employee;

/**
 * This is a read only repository
 */
@ReadOnlyRepository
public interface ReadEmployeeRepository extends Repository {

    List findAll();

}

And now it’s time to work with our repository scanning. All the repositories will be injected with the main EntityManager except from the ones annotated with the @ReadOnlyRepository annotation.

package com.gkatzioura.springdatareadreplica.config;

import javax.sql.DataSource;

import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.jdbc.DataSourceBuilder;
import org.springframework.boot.orm.jpa.EntityManagerFactoryBuilder;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.data.jpa.repository.config.EnableJpaRepositories;
import org.springframework.orm.jpa.LocalContainerEntityManagerFactoryBean;

@Configuration
@EnableJpaRepositories(
        basePackages = "com.gkatzioura",
        excludeFilters = @ComponentScan.Filter(ReadOnlyRepository.class),
        entityManagerFactoryRef = "entityManagerFactory"
)
public class PrimaryEntityManagerConfiguration {

    @Value("${spring.datasource.username}")
    private String username;

    @Value("${spring.datasource.password}")
    private String password;

    @Value("${spring.datasource.url}")
    private String url;

    @Bean
    @Primary
    public DataSource dataSource() throws Exception {
        return DataSourceBuilder.create()
                                .url(url)
                                .username(username)
                                .password(password)
                                .driverClassName("org.postgresql.Driver")
                                .build();
    }

    @Bean
    @Primary
    public LocalContainerEntityManagerFactoryBean entityManagerFactory(
            EntityManagerFactoryBuilder builder,
            @Qualifier("dataSource") DataSource dataSource) {
        return builder.dataSource(dataSource)
                      .packages("com.gkatzioura.springdatareadreplica")
                      .persistenceUnit("main")
                      .build();
    }

}

Also we will add the configuration for the read only repositories.

package com.gkatzioura.springdatareadreplica.config;

import javax.sql.DataSource;

import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.jdbc.DataSourceBuilder;
import org.springframework.boot.orm.jpa.EntityManagerFactoryBuilder;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.jpa.repository.config.EnableJpaRepositories;
import org.springframework.orm.jpa.LocalContainerEntityManagerFactoryBean;

@Configuration
@EnableJpaRepositories(
        basePackages = "com.gkatzioura",
        includeFilters= @ComponentScan.Filter(ReadOnlyRepository.class),
        entityManagerFactoryRef = "readEntityManagerFactory"
)
public class ReadOnlyEntityManagerConfiguration {

    @Value("${spring.datasource.username}")
    private String username;

    @Value("${spring.datasource.password}")
    private String password;

    @Value("${spring.datasource.readUrl}")
    private String readUrl;

    @Bean
    public DataSource readDataSource() throws Exception {
        return DataSourceBuilder.create()
                                .url(readUrl)
                                .username(username)
                                .password(password)
                                .driverClassName("org.postgresql.Driver")
                                .build();
    }

    @Bean
    public LocalContainerEntityManagerFactoryBean readEntityManagerFactory(
            EntityManagerFactoryBuilder builder,
            @Qualifier("readDataSource") DataSource dataSource) {
        return builder.dataSource(dataSource)
                      .packages("com.gkatzioura.springdatareadreplica")
                      .persistenceUnit("read")
                      .build();
    }

}

The secondary entity manager will be injected only to the repositories that only have the @ReadOnlyRepository annotation.

And to show this let’s make some changes to our controller.

package com.gkatzioura.springdatareadreplica.controller;

import java.util.List;

import org.springframework.http.HttpStatus;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.ResponseStatus;
import org.springframework.web.bind.annotation.RestController;

import com.gkatzioura.springdatareadreplica.entity.Employee;
import com.gkatzioura.springdatareadreplica.repository.EmployeeRepository;
import com.gkatzioura.springdatareadreplica.repository.ReadEmployeeRepository;

@RestController
public class EmployeeContoller {

    private final EmployeeRepository employeeRepository;
    private final ReadEmployeeRepository readEmployeeRepository;

    public EmployeeContoller(EmployeeRepository employeeRepository,
                             ReadEmployeeRepository readEmployeeRepository) {
        this.employeeRepository = employeeRepository;
        this.readEmployeeRepository = readEmployeeRepository;
    }

    @GetMapping("/employee")
    public List getEmployees() {
        return employeeRepository.findAll();
    }

    @GetMapping("/employee/read")
    public List getEmployeesRead() {
        return readEmployeeRepository.findAll();
    }

    @PostMapping("/employee")
    @ResponseStatus(HttpStatus.CREATED)
    public void addEmployee(@RequestBody Employee employee) {
        employeeRepository.save(employee);
    }

}

As you add employees to the system the read only repository will keep fetching the old employees while the main repository will fetch all of them including the recently persisted.

Read replicas and Spring Data Part 3: Configuring two entity managers

Our previous setup works as expected. What we shall do now is to get one step further and configure two separate entity managers without affecting the functionality we achieved previously.

The first step would be to set the default entity manager configuration to a primary one.
This is the first step

package com.gkatzioura.springdatareadreplica.config;

import javax.sql.DataSource;

import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.jdbc.DataSourceBuilder;
import org.springframework.boot.orm.jpa.EntityManagerFactoryBuilder;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.data.jpa.repository.config.EnableJpaRepositories;
import org.springframework.orm.jpa.LocalContainerEntityManagerFactoryBean;

@Configuration
public class PrimaryEntityManagerConfiguration {

    @Value("${spring.datasource.username}")
    private String username;

    @Value("${spring.datasource.password}")
    private String password;

    @Value("${spring.datasource.url}")
    private String url;

    @Bean
    @Primary
    public DataSource dataSource() throws Exception {
        return DataSourceBuilder.create()
                                .url(url)
                                .username(username)
                                .password(password)
                                .driverClassName("org.postgresql.Driver")
                                .build();
    }

    @Bean
    @Primary
    public LocalContainerEntityManagerFactoryBean entityManagerFactory(
            EntityManagerFactoryBuilder builder,
            @Qualifier("dataSource") DataSource dataSource) {
        return builder.dataSource(dataSource)
                      .packages("com.gkatzioura.springdatareadreplica")
                      .persistenceUnit("main")
                      .build();
    }

}

If you run your application with this configuration it will run just like our application previously.
Now it is time to configure the read only entity manager.

package com.gkatzioura.springdatareadreplica.config;

import javax.sql.DataSource;

import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.jdbc.DataSourceBuilder;
import org.springframework.boot.orm.jpa.EntityManagerFactoryBuilder;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Primary;
import org.springframework.orm.jpa.LocalContainerEntityManagerFactoryBean;

@Configuration
public class ReadOnlyEntityManagerConfiguration {

    @Value("${spring.datasource.username}")
    private String username;

    @Value("${spring.datasource.password}")
    private String password;

    @Value("${spring.datasource.readUrl}")
    private String readUrl;

    @Bean
    public DataSource readDataSource() throws Exception {
        return DataSourceBuilder.create()
                                .url(readUrl)
                                .username(username)
                                .password(password)
                                .driverClassName("org.postgresql.Driver")
                                .build();
    }

    @Bean
    public LocalContainerEntityManagerFactoryBean readEntityManagerFactory(
            EntityManagerFactoryBuilder builder,
            @Qualifier("readDataSource") DataSource dataSource) {
        return builder.dataSource(dataSource)
                      .packages("com.gkatzioura.springdatareadreplica")
                      .persistenceUnit("read")
                      .build();
    }

}

Also I will add a method to a controller in order to save the models.

package com.gkatzioura.springdatareadreplica.controller;

import java.util.List;

import org.springframework.http.HttpStatus;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.ResponseStatus;
import org.springframework.web.bind.annotation.RestController;

import com.gkatzioura.springdatareadreplica.entity.Employee;
import com.gkatzioura.springdatareadreplica.repository.EmployeeRepository;

@RestController
public class EmployeeContoller {

    private final EmployeeRepository employeeRepository;

    public EmployeeContoller(EmployeeRepository employeeRepository) {
        this.employeeRepository = employeeRepository;
    }

    @GetMapping("/employee")
    public List<Employee> getEmployees() {
        return employeeRepository.findAll();
    }

    @PostMapping("/employee")
    @ResponseStatus(HttpStatus.CREATED)
    public void addEmployee(@RequestBody Employee employee) {
        employeeRepository.save(employee);
    }

}

If you do try to add the an employee using the controller and then query the read database you shall see that no entry is being added at all.

So we have our primary entity manager up and running and we also have a secondary one. The secondary one is not used yet. The next blog focuses on putting the secondary read only entity manager in use.

Read replicas and Spring Data Part 2: Configuring the base project

In our previous post we set up multiple PostgreSQL instances with the same data.
Our next step would be to configure our spring project by using the both servers.

As stated previously we shall use some of the code taken from the Spring Boot JPA post, since we use exactly the same database.

This shall be our gradle build file

plugins {
	id 'org.springframework.boot' version '2.1.9.RELEASE'
	id 'io.spring.dependency-management' version '1.0.8.RELEASE'
	id 'java'
}

group = 'com.gkatzioura'
version = '0.0.1-SNAPSHOT'
sourceCompatibility = '1.8'

repositories {
	mavenCentral()
}

dependencies {
	implementation 'org.springframework.boot:spring-boot-starter-data-jpa'
	implementation 'org.springframework.boot:spring-boot-starter-web'
	implementation "org.postgresql:postgresql:42.2.8"
	testImplementation 'org.springframework.boot:spring-boot-starter-test'
}

Now let’s proceed on creating the model based on the table created on the previous blog.

package com.gkatzioura.springdatareadreplica.entity;

import javax.persistence.Column;
import javax.persistence.Entity;
import javax.persistence.GeneratedValue;
import javax.persistence.GenerationType;
import javax.persistence.Id;
import javax.persistence.Table;

@Entity
@Table(name = "employee", catalog="spring_data_jpa_example")
public class Employee {

    @Id
    @Column(name = "id")
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;

    @Column(name = "firstname")
    private String firstName;

    @Column(name = "lastname")
    private String lastname;

    @Column(name = "email")
    private String email;

    @Column(name = "age")
    private Integer age;

    @Column(name = "salary")
    private Integer salary;

    public Long getId() {
        return id;
    }

    public void setId(Long id) {
        this.id = id;
    }

    public String getFirstName() {
        return firstName;
    }

    public void setFirstName(String firstName) {
        this.firstName = firstName;
    }

    public String getLastname() {
        return lastname;
    }

    public void setLastname(String lastname) {
        this.lastname = lastname;
    }

    public String getEmail() {
        return email;
    }

    public void setEmail(String email) {
        this.email = email;
    }

    public Integer getAge() {
        return age;
    }

    public void setAge(Integer age) {
        this.age = age;
    }

    public Integer getSalary() {
        return salary;
    }

    public void setSalary(Integer salary) {
        this.salary = salary;
    }

}

And the next step is to create a spring data repository.

package com.gkatzioura.springdatareadreplica.repository;

import org.springframework.data.jpa.repository.JpaRepository;
import com.gkatzioura.springdatareadreplica.entity.Employee;

public interface EmployeeRepository extends JpaRepository<Employee,Long> {
}

Also we are going to add a controller.

package com.gkatzioura.springdatareadreplica.controller;

import java.util.List;

import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import com.gkatzioura.springdatareadreplica.entity.Employee;
import com.gkatzioura.springdatareadreplica.repository.EmployeeRepository;

@RestController
public class EmployeeContoller {

    private final EmployeeRepository employeeRepository;

    public EmployeeContoller(EmployeeRepository employeeRepository) {
        this.employeeRepository = employeeRepository;
    }

    @RequestMapping("/employee")
    public List<Employee> getEmployees() {
        return employeeRepository.findAll();
    }

}

All that it takes is to just add the right properties in you application.yaml

spring:
  datasource:
    platform: postgres
    driverClassName: org.postgresql.Driver
    username: db-user
    password: your-password
    url: jdbc:postgresql://127.0.0.2:5432/postgres

Spring boot has made it possible nowadays not to bother with any JPA configurations.

This is all you need in order to run the application. Once your application is running just try to fetch the employees.

curl http://localhost:8080/employee

As you have seen we did not do any JPA configuration. Since Spring Boot 2 specifying the database url is sufficient for the auto configuration to kick in and do all this configuration for you.

However in our case we want to have multiple datasource and entity manager configurations. In the next post we shall configure the entity managers for our application.

Spring Data with JPA and @NamedQueries

If you use Spring Data and @NamedQuery annotations at your JPA entity you can easily use them in a more convenient way using the spring data repository.

On a previous blog we created a spring data project using spring boot and docker. We will use the pretty same project and enhance our repository’s functionality.

We will implement a named query that will fetch employees only if their Last Name has as many characters as the ones specified.

package com.gkatzioura.springdata.jpa.persistence.entity;

import javax.persistence.*;

/**
 * Created by gkatzioura on 6/2/16.
 */
@Entity
@Table(name = "employee", schema="spring_data_jpa_example")
@NamedQuery(name = "Employee.fetchByLastNameLength",
        query = "SELECT e FROM Employee e WHERE CHAR_LENGTH(e.lastname) =:length "
)
public class Employee {

    @Id
    @Column(name = "id")
    @GeneratedValue(strategy = GenerationType.SEQUENCE)
    private Long id;

    @Column(name = "firstname")
    private String firstName;

    @Column(name = "lastname")
    private String lastname;

    @Column(name = "email")
    private String email;

    @Column(name = "age")
    private Integer age;

    @Column(name = "salary")
    private Integer salary;

    public Long getId() {
        return id;
    }

    public void setId(Long id) {
        this.id = id;
    }

    public String getFirstName() {
        return firstName;
    }

    public void setFirstName(String firstName) {
        this.firstName = firstName;
    }

    public String getLastname() {
        return lastname;
    }

    public void setLastname(String lastname) {
        this.lastname = lastname;
    }

    public String getEmail() {
        return email;
    }

    public void setEmail(String email) {
        this.email = email;
    }

    public Integer getAge() {
        return age;
    }

    public void setAge(Integer age) {
        this.age = age;
    }

    public Integer getSalary() {
        return salary;
    }

    public void setSalary(Integer salary) {
        this.salary = salary;
    }
}

Pay extra attention to the query name and the convention we follow @{EntityName}.{queryName}.
Then we will add the method to our spring data repository.

package com.gkatzioura.springdata.jpa.persistence.repository;

import com.gkatzioura.springdata.jpa.persistence.entity.Employee;
import org.springframework.data.jpa.repository.JpaRepository;
import org.springframework.data.repository.query.Param;
import org.springframework.stereotype.Repository;

import java.util.List;

/**
 * Created by gkatzioura on 6/2/16.
 */
@Repository
public interface EmployeeRepository extends JpaRepository<Employee,Long>, EmployeeRepositoryCustom {

    List<Employee> fetchByLastNameLength(@Param("length") Long length);
}

And last but not least add some functionality to our controller.

package com.gkatzioura.springdata.jpa.controller;

import com.gkatzioura.springdata.jpa.persistence.entity.Employee;
import com.gkatzioura.springdata.jpa.persistence.repository.EmployeeRepository;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.util.List;

/**
 * Created by gkatzioura on 6/2/16.
 */
@RestController
public class TestController {

    @Autowired
    private EmployeeRepository employeeRepository;

    @RequestMapping("/employee")
    public List<Employee> getTest() {

        return employeeRepository.findAll();
    }

    @RequestMapping("/employee/filter")
    public List<Employee> getFiltered(String firstName,@RequestParam(defaultValue = "0") Double bonusAmount) {

        return employeeRepository.getFirstNamesLikeAndBonusBigger(firstName,bonusAmount);
    }

    @RequestMapping("/employee/lastnameLength")
    public List<Employee> fetchByLength(Long length) {
        return employeeRepository.fetchByLastNameLength(length);
    }

}

You can find the source code on github.