Merge pull request #6 from BinaryIgor/improvements

Improvements

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,26 +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,
-  last_event_id BIGINT,
-  last_consumption_at TIMESTAMP,
+  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:
@@ -69,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:
 
@@ -105,14 +130,14 @@ them:
 ```java
 
 import com.binaryigor.eventsql.EventSQL;
-import javax.sql.DataSource;
 // 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
-import org.jooq.SQLDialect;
+import com.binaryigor.eventsql.EventSQLDialect;
+import javax.sql.DataSource;
 
-var eventSQL = new EventSQL(dataSource, SQLDialect.POSTGRES);
-ver shardedEventSQL = new EventSQL(dataSources, SQLDialect.POSTGRES);
+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.
@@ -223,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,9 +1,9 @@
 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.jooq.SQLDialect;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
@@ -25,7 +25,7 @@ EventSQL eventSQL(EventsProperties eventsProperties) {
                 .filter(EventsProperties.DataSourceProperties::enabled)
                 .map(this::dataSource)
                 .toList();
-        return new EventSQL(dataSources, SQLDialect.POSTGRES);
+        return new EventSQL(dataSources, EventSQLDialect.POSTGRES);
     }
 
     private DataSource dataSource(EventsProperties.DataSourceProperties properties) {

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,23 +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,
-  last_event_id BIGINT,
-  last_consumption_at TIMESTAMP,
+  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/build_and_package.bash

@@ -31,7 +31,7 @@ export app=$app
 export tag=$tag
 export run_cmd="docker run -d \\
   -e EVENTS_RATE -e EVENTS_TO_PUBLISH -e RUNNER_INSTANCES \\
-  --memory 2G --cpus 2 \\
+  --memory 2G --cpus 4 \\
   --network host \\
   --name $app $tagged_image"
 

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

@@ -3,14 +3,12 @@
 import com.binaryigor.eventsql.*;
 import com.zaxxer.hikari.HikariConfig;
 import com.zaxxer.hikari.HikariDataSource;
-import org.jooq.SQLDialect;
 
 import javax.sql.DataSource;
 import java.math.BigDecimal;
 import java.math.RoundingMode;
 import java.nio.charset.StandardCharsets;
 import java.sql.ResultSet;
-import java.time.Clock;
 import java.time.Duration;
 import java.time.Instant;
 import java.time.LocalDateTime;
@@ -24,11 +22,11 @@ public class EventSQLBenchmarksRunner {
     static final String DB_URL = envValueOrDefault("DB_URL", "jdbc:postgresql://localhost:5432/events");
     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", 25);
-    static final SQLDialect SQL_DIALECT = SQLDialect.valueOf(envValueOrDefault("SQL_DIALECT", "POSTGRES"));
+    static final int DATA_SOURCE_POOL_SIZE = envIntValueOrDefault("DATA_SOURCE_POOL_SIZE", 50);
+    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", 10_000);
-    static final int EVENTS_RATE = envIntValueOrDefault("EVENTS_RATE", 1000);
+    static final int EVENTS_TO_PUBLISH = envIntValueOrDefault("EVENTS_TO_PUBLISH", 60_000);
+    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");
 
@@ -41,7 +39,7 @@ public static void main(String[] args) throws Exception {
 
         var dataSource = dataSource(DB_URL, DB_USERNAME, DB_PASSWORD);
 
-        var eventSQL = new EventSQL(dataSource, SQL_DIALECT, Clock.systemUTC());
+        var eventSQL = new EventSQL(dataSource, SQL_DIALECT);
 
         printDelimiter();
 
@@ -62,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();
@@ -232,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) {
@@ -260,7 +278,7 @@ static void waitForConsumers(DataSource dataSource, ConsumerDefinition consumerD
                 break;
             }
 
-            Thread.sleep(1000);
+            Thread.sleep(500);
         }
     }
 
@@ -277,7 +295,6 @@ record EventTableStats(long lastId,
                 throw new IllegalArgumentException("lastIdsPerPartition cannot be null or empty");
             }
         }
-
     }
 
     record ConsumerTableStats(Map<Integer, Long> lastIdsPerPartition) {

src/main/java/com/binaryigor/eventsql/EventSQL.java

@@ -1,20 +1,22 @@
 package com.binaryigor.eventsql;
 
+import com.binaryigor.eventsql.internal.DefaultEventSQLConsumers;
+import com.binaryigor.eventsql.internal.DefaultEventSQLPublisher;
 import com.binaryigor.eventsql.internal.DefaultEventSQLRegistry;
-import com.binaryigor.eventsql.internal.EventSQLOps;
 import com.binaryigor.eventsql.internal.TopicDefinitionsCache;
 import com.binaryigor.eventsql.internal.sharded.ShardedEventSQLConsumers;
 import com.binaryigor.eventsql.internal.sharded.ShardedEventSQLPublisher;
 import com.binaryigor.eventsql.internal.sharded.ShardedEventSQLRegistry;
-import com.binaryigor.eventsql.internal.sql.SqlConsumerRepository;
-import com.binaryigor.eventsql.internal.sql.SqlEventRepository;
-import com.binaryigor.eventsql.internal.sql.SqlTopicRepository;
-import com.binaryigor.eventsql.internal.sql.SqlTransactions;
+import com.binaryigor.eventsql.internal.sql.SQLConsumerRepository;
+import com.binaryigor.eventsql.internal.sql.SQLEventRepository;
+import com.binaryigor.eventsql.internal.sql.SQLTopicRepository;
+import com.binaryigor.eventsql.internal.sql.SQLTransactions;
 import org.jooq.SQLDialect;
 import org.jooq.impl.DSL;
 
 import javax.sql.DataSource;
 import java.time.Clock;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
@@ -23,25 +25,31 @@
 
 public class EventSQL {
 
+    public static final int DEFAULT_FLUSH_PUBLISH_BUFFER_SIZE = 500;
+    // delay is applied only if there is no more records to flush; see DefaultEventSQLPublisher.startFlushPublishBufferThread()
+    public static final Duration DEFAULT_FLUSH_PUBLISH_BUFFER_DELAY = Duration.ofMillis(250);
     private final EventSQLRegistry registry;
     private final EventSQLPublisher publisher;
     private final EventSQLConsumers consumers;
 
-    public EventSQL(DataSource dataSource, SQLDialect sqlDialect) {
+    public EventSQL(DataSource dataSource, EventSQLDialect sqlDialect) {
         this(dataSource, sqlDialect, Clock.systemUTC());
     }
 
-    public EventSQL(DataSource dataSource, SQLDialect sqlDialect, Clock clock) {
+    public EventSQL(DataSource dataSource, EventSQLDialect sqlDialect, Clock clock) {
         this(List.of(dataSource), sqlDialect, clock);
     }
 
-    public EventSQL(Collection<DataSource> dataSources, SQLDialect sqlDialect) {
+    public EventSQL(Collection<DataSource> dataSources, EventSQLDialect sqlDialect) {
         this(dataSources, sqlDialect, Clock.systemUTC());
     }
 
-    public EventSQL(Collection<DataSource> dataSources,
-                    SQLDialect sqlDialect,
-                    Clock clock) {
+    public EventSQL(Collection<DataSource> dataSources, EventSQLDialect sqlDialect, Clock clock) {
+        this(dataSources, sqlDialect, clock, DEFAULT_FLUSH_PUBLISH_BUFFER_SIZE, DEFAULT_FLUSH_PUBLISH_BUFFER_DELAY);
+    }
+
+    public EventSQL(Collection<DataSource> dataSources, EventSQLDialect sqlDialect, Clock clock,
+                    int flushPublishBufferSize, Duration flushPublishBufferDelay) {
         if (dataSources.isEmpty()) {
             throw new IllegalArgumentException("At least one data source is required");
         }
@@ -53,25 +61,29 @@ public EventSQL(Collection<DataSource> dataSources,
         System.setProperty("org.jooq.no-logo", "true");
         System.setProperty("org.jooq.no-tips", "true");
 
+        var jooqDialect = SQLDialect.valueOf(sqlDialect.name());
+
         dataSources.forEach(dataSource -> {
-            var dslContext = DSL.using(dataSource, sqlDialect);
-            var transactions = new SqlTransactions(dslContext);
+            var dslContext = DSL.using(dataSource, jooqDialect);
+            var transactions = new SQLTransactions(dslContext);
 
-            var topicRepository = new SqlTopicRepository(transactions);
-            var consumerRepository = new SqlConsumerRepository(transactions);
-            var eventRepository = new SqlEventRepository(transactions, sqlDialect);
+            var topicRepository = new SQLTopicRepository(transactions);
+            var consumerRepository = new SQLConsumerRepository(transactions);
+            var eventRepository = new SQLEventRepository(transactions, transactions, sqlDialect);
 
             var registry = new DefaultEventSQLRegistry(topicRepository, eventRepository, consumerRepository, transactions);
 
             var topicDefinitionsCache = new TopicDefinitionsCache(topicRepository);
-            var ops = new EventSQLOps(topicDefinitionsCache, transactions, consumerRepository, eventRepository, clock);
+            var publisher = new DefaultEventSQLPublisher(topicDefinitionsCache, transactions,
+                    eventRepository, flushPublishBufferSize, flushPublishBufferDelay);
+            var consumers = new DefaultEventSQLConsumers(topicDefinitionsCache, transactions,
+                    consumerRepository, eventRepository, publisher, clock);
 
             registryList.add(registry);
-            publisherList.add(ops);
-            consumersList.add(ops);
+            publisherList.add(publisher);
+            consumersList.add(consumers);
         });
 
-
         if (dataSources.size() == 1) {
             registry = registryList.getFirst();
             publisher = publisherList.getFirst();