Buffer and lock idea

Author: BinaryIgor

README.md

@@ -12,7 +12,7 @@ For scalability details, see [benchmarks](/benchmarks/README.md).
 
 ## How it works
 
-We just need to have three tables (postgres syntax):
+We just need to have a few tables (postgres syntax):
 
 ```sql
 CREATE TABLE topic (
@@ -21,28 +21,51 @@ CREATE TABLE topic (
   created_at TIMESTAMP NOT NULL DEFAULT NOW()
 );
 
+CREATE TABLE consumer (
+  topic TEXT NOT NULL,
+  name TEXT NOT NULL,
+  partition SMALLINT NOT NULL,
+  first_event_id BIGINT,
+  last_event_id BIGINT,
+  last_consumption_at TIMESTAMP,
+  consumed_events BIGINT NOT NULL,
+  created_at TIMESTAMP NOT NULL DEFAULT NOW(),
+  PRIMARY KEY (topic, name, partition)
+);
+
 CREATE TABLE event (
   topic TEXT NOT NULL,
   id BIGSERIAL NOT NULL,
   partition SMALLINT NOT NULL,
   key TEXT,
   value BYTEA NOT NULL,
+  buffered_at TIMESTAMP NOT NULL,
   created_at TIMESTAMP NOT NULL DEFAULT NOW(),
   metadata JSON NOT NULL,
   PRIMARY KEY (topic, id)
 ) PARTITION BY LIST (topic);
 
-CREATE TABLE consumer (
+-- Same schema as event, just not partitioned. --
+-- It is used to handle eventual consistency of auto increment; --
+-- there is no guarantee that record of id 2 is visible after id 1 record. --
+-- Events are first inserted to the event_buffer; --
+-- they are then moved to event table in bulk, by a single, serialized writer; --
+-- because there is only one writer, it fixes eventual consistency issue --
+CREATE TABLE event_buffer (
   topic TEXT NOT NULL,
-  name TEXT NOT NULL,
+  id BIGSERIAL PRIMARY KEY,
   partition SMALLINT NOT NULL,
-  first_event_id BIGINT,
-  last_event_id BIGINT,
-  last_consumption_at TIMESTAMP,
-  consumed_events BIGINT NOT NULL,
+  key TEXT,
+  value BYTEA NOT NULL,
   created_at TIMESTAMP NOT NULL DEFAULT NOW(),
-  PRIMARY KEY (topic, name, partition)
+  metadata JSON NOT NULL
+);
+-- Used to lock single event_buffer to event writer; --
+-- there cannot be more than one record of this table! --
+CREATE TABLE event_buffer_lock (
+  created_at TIMESTAMP NOT NULL DEFAULT NOW()
 );
+INSERT INTO event_buffer_lock VALUES (DEFAULT);
 ```
 
 To consume messages, we just need to periodically (every one to a few seconds) do:
@@ -56,9 +79,6 @@ FOR UPDATE SKIP LOCKED;
 
 SELECT * FROM event
 WHERE topic = :topic AND (:last_event_id IS NULL OR id > :last_event_id)
-  -- eventual consistency of auto increment; there is no guarantee that record of id 2 is visible after id 1 record --
-  -- in the implementation, we set insert statements timeout to '250 ms' so it is safe --
-  AND created_at < (NOW() - interval '333 ms')
 ORDER BY id LIMIT :limit;
 
 (process events)
@@ -74,7 +94,7 @@ COMMIT;
 Optionally, to increase throughput & concurrency, we might have a partitioned topic and consumers (-1 partition standing
 for not partitioned topic/consumer/event).
 
-Distribution of partitioned events is the sole responsibility of the publisher.
+Distribution of partitioned events is the sole responsibility of a publisher.
 
 Consumption of such events per partition (0 in an example) might look like this:
 
@@ -87,9 +107,6 @@ FOR UPDATE SKIP LOCKED;
 
 SELECT * FROM event
 WHERE topic = :topic AND partition = 0 AND (:last_event_id IS NULL OR id > :last_event_id)
-  -- eventual consistency of auto increment; there is no guarantee that record of id 2 is visible after id 1 record --
-  -- in the implementation, we set insert statements timeout to '250 ms' so it is safe --
-  AND created_at < (NOW() - interval '333 ms')
 ORDER BY id LIMIT :limit;
 
 (process events)
@@ -107,20 +124,20 @@ definition has. It's a rather acceptable tradeoff and easy to enforce at the lib
 
 ## How to use it
 
-`EventSQL` is an entrypoint to the whole library. It requires single data source properties or a list of
+`EventSQL` is an entrypoint to the whole library. It requires standard Java `javax.sql.DataSource` or a list of
 them:
 
 ```java
+
 import com.binaryigor.eventsql.EventSQL;
-// EventSQL.Dialect is a dialect of your events backend - POSTGRES, MYSQL, MARIADB and so on;
+// dialect of your events backend - POSTGRES, MYSQL, MARIADB and so on;
 // as of now, only POSTGRES has fully tested support;
 // should also work with others but some things - event table partition management for example - works only with Postgres, for others it must be managed manually
-var eventSQL = EventSQL.of(new EventSQL.DataSourceProperties(EventSQL.Dialect.POSTGRES, "dbUrl", "dbUsername", "dbPassword"));
-ver shardedEventSQL = EventSQL.of(
-  List.of(
-    new EventSQL.DataSourceProperties(EventSQL.Dialect.POSTGRES, "db0Url", "db0Username", "db0Password"),
-    new EventSQL.DataSourceProperties(EventSQL.Dialect.POSTGRES, "db1Url", "db1Username", "db1Password"),
-    new EventSQL.DataSourceProperties(EventSQL.Dialect.POSTGRES, "db2Url", "db2Username", "db2Password")));
+import com.binaryigor.eventsql.EventSQLDialect;
+import javax.sql.DataSource;
+
+var eventSQL = new EventSQL(dataSource, EventSQLDialect.POSTGRES);
+ver shardedEventSQL = new EventSQL(dataSources, EventSQLDialect.POSTGRES);
 ```
 
 Sharded version works in the same vain - it just assumes that topics and consumers are hosted on multiple dbs.
@@ -231,7 +248,7 @@ var consumers = eventSQL.consumers();
 consumers.startConsumer("txt_topic", "single-consumer", event -> {
   // handle single event
 });
-// with more frequent polling - by default it is 1 second
+// with more frequent polling - by default it is 0.5 second
 consumers.startConsumer("txt_topic", "single-consumer-customized", event -> {
   // handle single event
 }, Duration.ofMillis(100));

benchmarks/app/build_and_package.bash

@@ -46,7 +46,7 @@ export run_cmd="docker run -d \\
   -e DB0_URL=\"$DB0_URL\" -e DB0_ENABLED=\"$DB0_ENABLED\" \\
   -e DB1_URL=\"$DB1_URL\" -e DB1_ENABLED=\"$DB1_ENABLED\" \\
   -e DB2_URL=\"$DB2_URL\" -e DB2_ENABLED=\"$DB2_ENABLED\" \\
-  --network host --restart unless-stopped \\
+  --network host \\
   --name $app $tagged_image"
 
 cd ..

benchmarks/app/src/main/java/com/binaryigor/eventsql/benchmarks/EventSQLBenchmarksApp.java

@@ -1,11 +1,16 @@
 package com.binaryigor.eventsql.benchmarks;
 
 import com.binaryigor.eventsql.EventSQL;
+import com.binaryigor.eventsql.EventSQLDialect;
+import com.zaxxer.hikari.HikariConfig;
+import com.zaxxer.hikari.HikariDataSource;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.annotation.Bean;
 
+import javax.sql.DataSource;
+
 @SpringBootApplication
 @EnableConfigurationProperties(EventsProperties.class)
 public class EventSQLBenchmarksApp {
@@ -16,11 +21,20 @@ public static void main(String[] args) {
 
     @Bean
     EventSQL eventSQL(EventsProperties eventsProperties) {
-        var dataSourceProperties = eventsProperties.dataSources().stream()
+        var dataSources = eventsProperties.dataSources().stream()
                 .filter(EventsProperties.DataSourceProperties::enabled)
-                .map(ps -> new EventSQL.DataSourceProperties(EventSQL.Dialect.POSTGRES,
-                        ps.url(), ps.username(), ps.password()))
+                .map(this::dataSource)
                 .toList();
-        return EventSQL.of(dataSourceProperties);
+        return new EventSQL(dataSources, EventSQLDialect.POSTGRES);
+    }
+
+    private DataSource dataSource(EventsProperties.DataSourceProperties properties) {
+        var config = new HikariConfig();
+        config.setJdbcUrl(properties.url());
+        config.setUsername(properties.username());
+        config.setPassword(properties.password());
+        config.setMinimumIdle(properties.connections());
+        config.setMaximumPoolSize(properties.connections());
+        return new HikariDataSource(config);
     }
 }

benchmarks/events-db/build_and_package.bash

@@ -33,7 +33,7 @@ export tag=$tag
 export run_cmd="docker run -d \\
   -e \"POSTGRES_USER=postgres\" -e \"POSTGRES_PASSWORD=postgres\" \\
    --memory ${memory_limit} --cpus ${cpus_limit} \
-  --network host ${volume} --restart unless-stopped --name $app $tagged_image"
+  --network host ${volume} --name $app $tagged_image"
 
 cd ..
 envsubst '${app} ${tag}' < scripts/template_load_and_run_app.bash > "$app/dist/load_and_run_app.bash"

benchmarks/events-db/init_db.sql

@@ -12,25 +12,41 @@ CREATE TABLE topic (
   created_at TIMESTAMP NOT NULL DEFAULT NOW()
 );
 
+CREATE TABLE consumer (
+  topic TEXT NOT NULL,
+  name TEXT NOT NULL,
+  partition SMALLINT NOT NULL,
+  first_event_id BIGINT,
+  last_event_id BIGINT,
+  last_consumption_at TIMESTAMP,
+  consumed_events BIGINT NOT NULL,
+  created_at TIMESTAMP NOT NULL DEFAULT NOW(),
+  PRIMARY KEY (topic, name, partition)
+);
+
 CREATE TABLE event (
   topic TEXT NOT NULL,
   id BIGSERIAL NOT NULL,
   partition SMALLINT NOT NULL,
   key TEXT,
   value BYTEA NOT NULL,
+  buffered_at TIMESTAMP NOT NULL,
   created_at TIMESTAMP NOT NULL DEFAULT NOW(),
   metadata JSON NOT NULL,
   PRIMARY KEY (topic, id)
 ) PARTITION BY LIST (topic);
 
-CREATE TABLE consumer (
+CREATE TABLE event_buffer (
   topic TEXT NOT NULL,
-  name TEXT NOT NULL,
+  id BIGSERIAL PRIMARY KEY,
   partition SMALLINT NOT NULL,
-  first_event_id BIGINT,
-  last_event_id BIGINT,
-  last_consumption_at TIMESTAMP,
-  consumed_events BIGINT NOT NULL,
+  key TEXT,
+  value BYTEA NOT NULL,
   created_at TIMESTAMP NOT NULL DEFAULT NOW(),
-  PRIMARY KEY (topic, name, partition)
+  metadata JSON NOT NULL
+);
+
+CREATE TABLE event_buffer_lock (
+  created_at TIMESTAMP NOT NULL DEFAULT NOW()
 );
+INSERT INTO event_buffer_lock VALUES (DEFAULT);

benchmarks/runner/src/main/java/com/binaryigor/eventsql/benchmarks/EventSQLBenchmarksRunner.java

@@ -23,10 +23,10 @@ public class EventSQLBenchmarksRunner {
     static final String DB_USERNAME = envValueOrDefault("DB_URL", "events");
     static final String DB_PASSWORD = envValueOrDefault("DB_PASSWORD", "events");
     static final int DATA_SOURCE_POOL_SIZE = envIntValueOrDefault("DATA_SOURCE_POOL_SIZE", 50);
-    static final EventSQL.Dialect SQL_DIALECT = EventSQL.Dialect.valueOf(envValueOrDefault("SQL_DIALECT", "POSTGRES"));
+    static final EventSQLDialect SQL_DIALECT = EventSQLDialect.valueOf(envValueOrDefault("SQL_DIALECT", "POSTGRES"));
     static final int RUNNER_INSTANCES = envIntValueOrDefault("RUNNER_INSTANCES", 1);
     static final int EVENTS_TO_PUBLISH = envIntValueOrDefault("EVENTS_TO_PUBLISH", 60_000);
-    static final int EVENTS_RATE = envIntValueOrDefault("EVENTS_RATE", 1000);
+    static final int EVENTS_RATE = envIntValueOrDefault("EVENTS_RATE", 1_000);
     static final String TEST_TOPIC = envValueOrDefault("TEST_TOPIC", "account_created");
     static final String TEST_CONSUMER = envValueOrDefault("TEST_CONSUMER", "benchmarks-consumer");
 
@@ -39,7 +39,7 @@ public static void main(String[] args) throws Exception {
 
         var dataSource = dataSource(DB_URL, DB_USERNAME, DB_PASSWORD);
 
-        var eventSQL = EventSQL.of(new EventSQL.DataSourceProperties(SQL_DIALECT, DB_URL, DB_USERNAME, DB_PASSWORD, DATA_SOURCE_POOL_SIZE));
+        var eventSQL = new EventSQL(dataSource, SQL_DIALECT);
 
         printDelimiter();
 
@@ -60,6 +60,7 @@ public static void main(String[] args) throws Exception {
         var start = System.currentTimeMillis();
 
         publishEvents(eventSQL.publisher());
+        waitForPublishBufferToEmpty(dataSource);
         var publicationDuration = Duration.ofMillis(System.currentTimeMillis() - start);
 
         printDelimiter();
@@ -230,6 +231,25 @@ static void awaitForFutures(List<Future<?>> futures) {
         });
     }
 
+    static void waitForPublishBufferToEmpty(DataSource dataSource) {
+        while (true) {
+            try {
+                var bufferSize = executeQuery(dataSource, "SELECT COUNT(*) FROM event_buffer", r -> {
+                    if (r.next()) {
+                        return r.getInt(1);
+                    }
+                    return 0;
+                });
+                if (bufferSize == 0) {
+                    break;
+                }
+                Thread.sleep(250);
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+    }
+
     static void waitForConsumers(DataSource dataSource, ConsumerDefinition consumerDefinition) throws Exception {
         var eventsStats = eventTableStats(dataSource);
         while (true) {
@@ -258,7 +278,7 @@ static void waitForConsumers(DataSource dataSource, ConsumerDefinition consumerD
                 break;
             }
 
-            Thread.sleep(1000);
+            Thread.sleep(500);
         }
     }
 

pom.xml

@@ -35,11 +35,6 @@
             <artifactId>jooq</artifactId>
             <version>${jooq.version}</version>
         </dependency>
-        <dependency>
-            <groupId>com.zaxxer</groupId>
-            <artifactId>HikariCP</artifactId>
-            <version>${hikari.version}</version>
-        </dependency>
 
         <dependency>
             <groupId>org.junit.jupiter</groupId>
@@ -71,6 +66,12 @@
             <version>${testcontainers.version}</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>com.zaxxer</groupId>
+            <artifactId>HikariCP</artifactId>
+            <version>${hikari.version}</version>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.postgresql</groupId>
             <artifactId>postgresql</artifactId>