Merge pull request #5 from evgeniy-khist/postgresq-listen-notify

Use PostgreSQL Listen/Notify instead of DB polling. Resolves #3

Author: eugene-khyst

README.md

@@ -21,7 +21,8 @@
   - [Synchronously updating projections](#4-6)
   - [Asynchronously sending integration events to a message broker](#4-7)
     - [Reliable transactional outbox with PostgreSQL](#4-7-1)
-    - [Database polling alternatives](#4-7-2)
+    - [Database polling](#4-7-2)
+    - [Database polling alternative](#4-7-3)
   - [Adding new asynchronous event handlers](#4-8)
   - [Class diagrams](#4-9)
     - [Class diagram of the domain model](#4-9-1)
@@ -272,7 +273,7 @@ INSERT INTO ES_AGGREGATE_SNAPSHOT (AGGREGATE_ID, VERSION, JSON_DATA)
 VALUES (:aggregateId, :version, :jsonObj::json)
 ```
 
-Snapshotting for an aggregate type can be disabled and configured in the `application.yml`
+Snapshotting for an aggregate type can be disabled and configured in the [`application.yml`](src/main/resources/application.yml)
 
 ```yaml
 event-sourcing:
@@ -417,33 +418,63 @@ the events it created won't be read by the event subscription processor until tr
 
 ![PostgreSQL reliable transactional outbox](img/postgresql-reliable-outbox.svg)
 
-Event subscription processing can be disabled and configured in the `application.yml`
+#### <a id="4-7-2"></a>Database polling
+
+To get new events from the `ES_EVENT` table, the application has to poll the database.
+The shorter the polling period, the shorter the delay between persisting a new event and processing it by the subscription. 
+But the lag is inevitable. If the polling period is 1 second, then the lag is at most 1 second.
+
+The polling mechanism implementation [ScheduledEventSubscriptionProcessor](src/main/java/com/example/eventsourcing/service/ScheduledEventSubscriptionProcessor.java)
+uses a Spring annotation [@Scheduled](https://docs.spring.io/spring-framework/docs/current/javadoc-api/org/springframework/scheduling/annotation/Scheduled.html) to poll database with a fixed period.
+
+The polling event subscription processing can be enabled and configured in the [`application.yml`](src/main/resources/application.yml)
 
 ```yaml
 event-sourcing:
-  subscriptions:
-    enabled: true
+  subscriptions: polling # Enable database polling subscription processing
+  polling-subscriptions:
     polling-initial-delay: PT1S
     polling-interval: PT1S
 ```
 
-#### <a id="4-7-2"></a>Database polling alternatives
+#### <a id="4-7-3"></a>Database polling alternative
 
-PostgreSQL `LISTEN`/`NOTIFY` functionality can be used instead of polling.
+To reduce the lag associated with database polling, the polling period can be set to a very low value, 
+such as 1 second.
+But this means that there will be 3600 database queries per hour and 86400 per day, even if there are no new events. 
 
-A key limitation of the PostgreSQL JDBC driver is that it cannot receive asynchronous notifications 
+PostgreSQL `LISTEN` and `NOTIFY` feature can be used instead of polling.
+This mechanism allows for sending asynchronous notifications across database connections.
+Notifications are not sent directly from the application, 
+but via the database [trigger](src/main/resources/db/migration/V2__notify_trigger.sql) on a table.
+
+To use this functionality an unshared [PgConnection](https://jdbc.postgresql.org/documentation/publicapi/org/postgresql/jdbc/PgConnection.html) 
+which remains open is required.
+The long-lived dedicated JDBC `Connection` for receiving notifications has to be created using the `DriverManager` API, 
+instead of getting from a pooled `DataSource`.
+
+PostgreSQL JDBC driver can't receive asynchronous notifications 
 and must poll the backend to check if any notifications were issued. 
-A timeout can be given to the poll function, 
+A timeout can be given to the poll function `getNotifications(int timeoutMillis)`, 
 but then the execution of statements from other threads will block.
+When `timeoutMillis` = 0, blocks forever or until at least one notification has been received. 
+It means that notification is delivered almost immediately, without a lag.
+If more than one notification is about to be received, these will be returned in one batch.
+
+This solution  significantly reduces the number of issued queries 
+and also solves the lag problem that the polling solution suffers from.
 
-Thus, the creation of a long-lived dedicated JDBC `Connection` for receiving notifications is required.
-This connection should not be obtained from a pooled DataSources. 
-Instead, a dedicated `Connection` has to be created using the `DriverManager` API.
+The Listen/Notify mechanism implementation [PostgresChannelEventSubscriptionProcessor](src/main/java/com/example/eventsourcing/service/PostgresChannelEventSubscriptionProcessor.java)
+is inspired by the Spring Integration class [PostgresChannelMessageTableSubscriber](https://github.com/spring-projects/spring-integration/blob/v6.0.0/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriber.java).
+
+The Listen/Notify event subscription processing can be enabled in the [`application.yml`](src/main/resources/application.yml)
+
+```yaml
+event-sourcing:
+  subscriptions: postgres-channel # Enable Listen/Notify event subscription processing
+```
 
-In practice, implementations based on PostgreSQL `LISTEN`/`NOTIFY` are quite complex. 
-For example,
-[PostgresChannelMessageTableSubscriber](https://github.com/spring-projects/spring-integration/blob/v6.0.0/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriber.java)
-from the Spring Integration.
+This mechanism is used by default as more efficient.
 
 ### <a id="4-8"></a>Adding new asynchronous event handlers
 
@@ -520,7 +551,7 @@ Using PostgreSQL as an event store has a lot of advantages, but there are also d
    
 7. Run E2E tests and see the output
     ```bash
-    E2E_TESTING=true; ./gradlew clean test -i
+    E2E_TESTING=true ./gradlew clean test -i
     ```
 
 8. Explore the database using the Adminer database management tool at http://localhost:8181.

src/main/java/com/example/eventsourcing/PostgreSqlEventSourcingApplication.java renamed to src/main/java/com/example/eventsourcing/PostgresEventSourcingApplication.java

@@ -8,9 +8,9 @@
 @SpringBootApplication
 @EnableScheduling
 @EnableAsync
-public class PostgreSqlEventSourcingApplication {
+public class PostgresEventSourcingApplication {
 
     public static void main(String[] args) {
-        SpringApplication.run(PostgreSqlEventSourcingApplication.class, args);
+        SpringApplication.run(PostgresEventSourcingApplication.class, args);
     }
 }

src/main/java/com/example/eventsourcing/service/PostgresChannelEventSubscriptionProcessor.java

@@ -0,0 +1,147 @@
+package com.example.eventsourcing.service;
+
+import com.example.eventsourcing.service.event.AsyncEventHandler;
+import jakarta.annotation.PostConstruct;
+import jakarta.annotation.PreDestroy;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.postgresql.PGNotification;
+import org.postgresql.jdbc.PgConnection;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.autoconfigure.jdbc.DataSourceProperties;
+import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
+import org.springframework.stereotype.Component;
+
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.List;
+import java.util.concurrent.*;
+
+@Component
+@ConditionalOnProperty(name = "event-sourcing.subscriptions", havingValue = "postgres-channel")
+@RequiredArgsConstructor
+@Slf4j
+public class PostgresChannelEventSubscriptionProcessor {
+
+    private final List<AsyncEventHandler> eventHandlers;
+    private final EventSubscriptionProcessor eventSubscriptionProcessor;
+    private final DataSourceProperties dataSourceProperties;
+    private final ExecutorService executor = newExecutor();
+    private CountDownLatch latch = new CountDownLatch(0);
+    private Future<?> future = CompletableFuture.completedFuture(null);
+    private volatile PgConnection connection;
+
+    private static ExecutorService newExecutor() {
+        CustomizableThreadFactory threadFactory =
+                new CustomizableThreadFactory("postgres-channel-event-subscription-");
+        threadFactory.setDaemon(true);
+        return Executors.newSingleThreadExecutor(threadFactory);
+    }
+
+    @PostConstruct
+    public synchronized void start() {
+        if (this.latch.getCount() > 0) {
+            return;
+        }
+        this.latch = new CountDownLatch(1);
+        this.future = executor.submit(() -> {
+            try {
+                while (isActive()) {
+                    try {
+                        PgConnection conn = getPgConnection();
+                        try (Statement stmt = conn.createStatement()) {
+                            stmt.execute("LISTEN channel_event_notify");
+                        } catch (Exception ex) {
+                            try {
+                                conn.close();
+                            } catch (Exception suppressed) {
+                                ex.addSuppressed(suppressed);
+                            }
+                            throw ex;
+                        }
+
+                        this.eventHandlers.forEach(this::processNewEvents);
+
+                        try {
+                            this.connection = conn;
+                            while (isActive()) {
+                                PGNotification[] notifications = conn.getNotifications(0);
+                                // Unfortunately, there is no good way of interrupting a notification
+                                // poll but by closing its connection.
+                                if (!isActive()) {
+                                    return;
+                                }
+                                if (notifications != null) {
+                                    for (PGNotification notification : notifications) {
+                                        String parameter = notification.getParameter();
+                                        eventHandlers.stream()
+                                                .filter(eventHandler -> eventHandler.getAggregateType().toString().equals(parameter))
+                                                .forEach(this::processNewEvents);
+                                    }
+                                }
+                            }
+                        } finally {
+                            conn.close();
+                        }
+                    } catch (Exception e) {
+                        // The getNotifications method does not throw a meaningful message on interruption.
+                        // Therefore, we do not log an error, unless it occurred while active.
+                        if (isActive()) {
+                            log.error("Failed to poll notifications from Postgres database", e);
+                        }
+                    }
+                }
+            } finally {
+                this.latch.countDown();
+            }
+        });
+    }
+
+    private PgConnection getPgConnection() throws SQLException {
+        return DriverManager.getConnection(
+                dataSourceProperties.determineUrl(),
+                dataSourceProperties.determineUsername(),
+                dataSourceProperties.determinePassword()
+        ).unwrap(PgConnection.class);
+    }
+
+    private boolean isActive() {
+        if (Thread.interrupted()) {
+            Thread.currentThread().interrupt();
+            return false;
+        }
+        return true;
+    }
+
+    private void processNewEvents(AsyncEventHandler eventHandler) {
+        try {
+            eventSubscriptionProcessor.processNewEvents(eventHandler);
+        } catch (Exception e) {
+            log.warn("Failed to handle new events for subscription %s"
+                    .formatted(eventHandler.getSubscriptionName()), e);
+        }
+    }
+
+    @PreDestroy
+    public synchronized void stop() {
+        if (this.future.isDone()) {
+            return;
+        }
+        this.future.cancel(true);
+        PgConnection conn = this.connection;
+        if (conn != null) {
+            try {
+                conn.close();
+            } catch (SQLException ignored) {
+            }
+        }
+        try {
+            if (!this.latch.await(5, TimeUnit.SECONDS)) {
+                throw new IllegalStateException(
+                        "Failed to stop %s".formatted(PostgresChannelEventSubscriptionProcessor.class.getName()));
+            }
+        } catch (InterruptedException ignored) {
+        }
+    }
+}

src/main/java/com/example/eventsourcing/service/ScheduledEventSubscriptionProcessor.java

@@ -10,7 +10,7 @@
 import java.util.List;
 
 @Component
-@ConditionalOnProperty(name = "event-sourcing.subscriptions.enabled", havingValue = "true")
+@ConditionalOnProperty(name = "event-sourcing.subscriptions", havingValue = "polling")
 @RequiredArgsConstructor
 @Slf4j
 public class ScheduledEventSubscriptionProcessor {
@@ -19,8 +19,8 @@ public class ScheduledEventSubscriptionProcessor {
     private final EventSubscriptionProcessor eventSubscriptionProcessor;
 
     @Scheduled(
-            fixedDelayString = "${event-sourcing.subscriptions.polling-interval}",
-            initialDelayString = "${event-sourcing.subscriptions.polling-initial-delay}"
+            fixedDelayString = "${event-sourcing.polling-subscriptions.polling-interval}",
+            initialDelayString = "${event-sourcing.polling-subscriptions.polling-initial-delay}"
     )
     public void processNewEvents() {
         eventHandlers.forEach(this::processNewEvents);

src/main/resources/application.yml

@@ -17,7 +17,8 @@ event-sourcing:
       enabled: true
       # Create a snapshot on every nth event
       nth-event: 10
-  subscriptions:
-    enabled: true
+  # Available subscription: polling, postgres-channel
+  subscriptions: postgres-channel
+  polling-subscriptions:
     polling-initial-delay: PT1S
     polling-interval: PT1S

src/main/resources/db/migration/V1__init.sql

@@ -4,7 +4,7 @@ CREATE TABLE IF NOT EXISTS ES_AGGREGATE (
   AGGREGATE_TYPE  TEXT     NOT NULL
 );
 
-CREATE INDEX IF NOT EXISTS IDX_ES_AGGREGATE_AGGREGATE_TYPE on ES_AGGREGATE (AGGREGATE_TYPE);
+CREATE INDEX IF NOT EXISTS IDX_ES_AGGREGATE_AGGREGATE_TYPE ON ES_AGGREGATE (AGGREGATE_TYPE);
 
 CREATE TABLE IF NOT EXISTS ES_EVENT (
   ID              BIGSERIAL  PRIMARY KEY,
@@ -16,9 +16,9 @@ CREATE TABLE IF NOT EXISTS ES_EVENT (
   UNIQUE (AGGREGATE_ID, VERSION)
 );
 
-CREATE INDEX IF NOT EXISTS IDX_ES_EVENT_TRANSACTION_ID_ID on ES_EVENT (TRANSACTION_ID, ID);
-CREATE INDEX IF NOT EXISTS IDX_ES_EVENT_AGGREGATE_ID on ES_EVENT (AGGREGATE_ID);
-CREATE INDEX IF NOT EXISTS IDX_ES_EVENT_VERSION on ES_EVENT (VERSION);
+CREATE INDEX IF NOT EXISTS IDX_ES_EVENT_TRANSACTION_ID_ID ON ES_EVENT (TRANSACTION_ID, ID);
+CREATE INDEX IF NOT EXISTS IDX_ES_EVENT_AGGREGATE_ID ON ES_EVENT (AGGREGATE_ID);
+CREATE INDEX IF NOT EXISTS IDX_ES_EVENT_VERSION ON ES_EVENT (VERSION);
 
 CREATE TABLE IF NOT EXISTS ES_AGGREGATE_SNAPSHOT (
   AGGREGATE_ID  UUID     NOT NULL REFERENCES ES_AGGREGATE (ID),
@@ -27,8 +27,8 @@ CREATE TABLE IF NOT EXISTS ES_AGGREGATE_SNAPSHOT (
   PRIMARY KEY (AGGREGATE_ID, VERSION)
 );
 
-CREATE INDEX IF NOT EXISTS IDX_ES_AGGREGATE_SNAPSHOT_AGGREGATE_ID on ES_AGGREGATE_SNAPSHOT (AGGREGATE_ID);
-CREATE INDEX IF NOT EXISTS IDX_ES_AGGREGATE_SNAPSHOT_VERSION on ES_AGGREGATE_SNAPSHOT (VERSION);
+CREATE INDEX IF NOT EXISTS IDX_ES_AGGREGATE_SNAPSHOT_AGGREGATE_ID ON ES_AGGREGATE_SNAPSHOT (AGGREGATE_ID);
+CREATE INDEX IF NOT EXISTS IDX_ES_AGGREGATE_SNAPSHOT_VERSION ON ES_AGGREGATE_SNAPSHOT (VERSION);
 
 CREATE TABLE IF NOT EXISTS ES_EVENT_SUBSCRIPTION (
   SUBSCRIPTION_NAME    TEXT    PRIMARY KEY,
@@ -57,4 +57,4 @@ CREATE TABLE IF NOT EXISTS RM_ORDER_ROUTE (
   PRIMARY KEY (ORDER_ID, ADDRESS)
 );
 
-CREATE INDEX IF NOT EXISTS IDX_RM_ORDER_ROUTE_ORDER_ID on RM_ORDER_ROUTE (ORDER_ID);
+CREATE INDEX IF NOT EXISTS IDX_RM_ORDER_ROUTE_ORDER_ID ON RM_ORDER_ROUTE (ORDER_ID);

src/main/resources/db/migration/V2__notify_trigger.sql

@@ -0,0 +1,17 @@
+CREATE OR REPLACE FUNCTION CHANNEL_EVENT_NOTIFY_FCT()
+RETURNS TRIGGER AS
+  $BODY$
+  DECLARE
+    aggregate_type  TEXT;
+  BEGIN
+    SELECT a.AGGREGATE_TYPE INTO aggregate_type FROM ES_AGGREGATE a WHERE a.ID = NEW.AGGREGATE_ID;
+    PERFORM pg_notify('channel_event_notify', aggregate_type);
+    RETURN NEW;
+  END;
+  $BODY$
+  LANGUAGE PLPGSQL;
+
+CREATE OR REPLACE TRIGGER CHANNEL_EVENT_NOTIFY_TRG
+  AFTER INSERT ON ES_EVENT
+  FOR EACH ROW
+  EXECUTE PROCEDURE CHANNEL_EVENT_NOTIFY_FCT();

src/test/java/com/example/eventsourcing/ListenNotifyPostgresEventSourcingApplicationTests.java

@@ -0,0 +1,7 @@
+package com.example.eventsourcing;
+
+import org.springframework.test.context.TestPropertySource;
+
+@TestPropertySource(properties = "event-sourcing.subscriptions=postgres-channel")
+class ListenNotifyPostgresEventSourcingApplicationTests extends PostgresEventSourcingApplicationTests {
+}

src/test/java/com/example/eventsourcing/OrderTestScript.java

@@ -39,9 +39,8 @@ public class OrderTestScript {
     private final String kafkaBrokers;
     private final ObjectMapper objectMapper = new ObjectMapper();
     private final BasicJsonTester jsonTester = new BasicJsonTester(getClass());
-
-    @SneakyThrows
-    public void execute() {
+    
+    public void execute() throws JsonProcessingException {
         log.info("Place a new order");
         UUID orderId = placeOrder("""
                 {
@@ -193,6 +192,7 @@ public void execute() {
         log.info("Print integration events");
         var kafkaConsumer = createKafkaConsumer(KafkaTopicsConfig.TOPIC_ORDER_EVENTS);
         List<String> kafkaRecordValues = getKafkaRecords(kafkaConsumer, Duration.ofSeconds(10), 23);
+        kafkaConsumer.unsubscribe();
         assertThat(kafkaRecordValues)
                 .hasSizeGreaterThanOrEqualTo(23);
 

src/test/java/com/example/eventsourcing/PollingPostgresEventSourcingApplicationTests.java

@@ -0,0 +1,7 @@
+package com.example.eventsourcing;
+
+import org.springframework.test.context.TestPropertySource;
+
+@TestPropertySource(properties = "event-sourcing.subscriptions=polling")
+class PollingPostgresEventSourcingApplicationTests extends PostgresEventSourcingApplicationTests {
+}