Benchmark improvements

Author: BinaryIgor

README.md

@@ -2,13 +2,13 @@
 
 Events over SQL.
 
-Simple, Reliable, Fast.
+Simple. Reliable. Fast.
 
 Able to publish and consume thousands of events per second on a single Postgres instance.
 
-With sharding, it can easily support tens of thousands events per second for virtually endless scalability.
+With sharding, it can easily support tens and hundreds of thousands events per second for virtually endless scalability.
 
-For scalability details, see [benchmarks](/benchmarks/README.md).
+For proofs, see [benchmarks](/benchmarks/README.md).
 
 ## How it works
 

TODO.md

@@ -4,4 +4,5 @@
 * compact topics - unique key
 * join, aka streams
 * more elaborate definitions change support - especially around partitions growth & shrinkage
-* JavaDocs
+* JavaDocs
+* sharding - routing topics and consumers to a subset of shards based on some tag/pattern/function (public vs private topics use-case)

benchmarks/README.md

@@ -17,7 +17,7 @@ Some background and details:
 Defined in the `prepare_infra.py` script; sometimes resources are limited by `docker run` command, but essentially:
 * *benchmarks-app (consumer)* runs on 2 GB and 2 CPUs (AMD) machine
 * each *events-db* runs on 8 GB and 4 CPUs (AMD) machine
-* each *benchmarks-runner* runs alongside *events-db*, but is throttled to 2 GB memory and 2 CPUs 
+* each *benchmarks-runner* runs alongside *events-db*
 * there is a basic firewall and virtual private network (vpc) setup (`prepare_infra.py`), so that nobody is bothering us during benchmarks
 
 ## Requirements
@@ -52,6 +52,14 @@ We right now have four machines connected to each other by the vpc.
 To each we have access, using ssh public key authentication, as the `eventsql` user.
 Infrastructure is now ready, let's prepare apps.
 
+### EventSQL
+
+Before building apps, make sure that EventSQL is available on your machine, since it's their dependency. Just run:
+```
+mvn clean install
+```
+From the root repo directory.
+
 ### Apps
 
 Let's build `events-db` (from scripts dir again):

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

@@ -12,7 +12,6 @@
 
 import java.time.Duration;
 import java.util.*;
-import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
 
 @Component
@@ -23,7 +22,6 @@ public class EventsConsumer {
     private final EventSQL eventSQL;
     private final ObjectMapper objectMapper;
     private final EventsProperties eventsProperties;
-    private final AtomicBoolean accountCreatedBatchHandler = new AtomicBoolean(true);
     private final Map<String, Counter> accountHandledCountersByConsumer;
 
     public EventsConsumer(EventSQL eventSQL,
@@ -41,10 +39,6 @@ public EventsConsumer(EventSQL eventSQL,
                                 .register(meterRegistry)));
     }
 
-    public void batchAccountCreatedHandler(boolean batch) {
-        accountCreatedBatchHandler.set(batch);
-    }
-
     @PostConstruct
     void start() {
         eventSQL.registry()
@@ -56,13 +50,7 @@ void start() {
                     eventSQL.registry().registerConsumer(c);
                     eventSQL.consumers()
                             .startBatchConsumer(c.topic(), c.name(),
-                                    events -> {
-                                        if (accountCreatedBatchHandler.get()) {
-                                            handleAccountCreatedEvents(c.name(), events);
-                                        } else {
-                                            events.forEach(e -> handleAccountCreatedEvent(c.name(), e));
-                                        }
-                                    },
+                                    events -> handleAccountCreatedEvents(c.name(), events),
                                     EventSQLConsumers.ConsumptionConfig.of(10, 100));
                 });
     }
@@ -72,20 +60,6 @@ void stop() {
         eventSQL.consumers().stop(Duration.ofSeconds(5));
     }
 
-    private void handleAccountCreatedEvent(String consumer, Event event) {
-        try {
-            var accountCreated = objectMapper.readValue(event.value(), AccountCreated.class);
-            handleDelay();
-            accountHandledCountersByConsumer.get(consumer)
-                    .increment();
-        } catch (IllegalStateException e) {
-            throw new EventSQLConsumptionException(e, event);
-        } catch (Exception e) {
-            logger.error("Problem while handling AccountCreated event:", e);
-            throw new RuntimeException(e);
-        }
-    }
-
     private void handleAccountCreatedEvents(String consumer, List<Event> events) {
         try {
             var accountsCreated = new ArrayList<AccountCreated>(events.size());

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

@@ -12,9 +12,11 @@
 import java.time.Duration;
 import java.time.Instant;
 import java.time.LocalDateTime;
+import java.time.temporal.ChronoUnit;
 import java.util.*;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
+import java.util.stream.Stream;
 
 public class EventSQLBenchmarksRunner {
 
@@ -26,6 +28,7 @@ public class EventSQLBenchmarksRunner {
     static final EventSQLDialect SQL_DIALECT;
     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_BATCH_SIZE = envIntValueOrDefault("EVENTS_BATCH_SIZE", 1);
     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");
@@ -182,15 +185,27 @@ static <T> T executeQuery(DataSource source, String query, ResultSetMapper<T> re
     static void publishEvents(EventSQLPublisher publisher) throws Exception {
         var futures = new LinkedList<Future<?>>();
 
+        String eventsWerePublishedMessage;
+        if (EVENTS_BATCH_SIZE > 1) {
+            eventsWerePublishedMessage = "events were published - in batches of " + EVENTS_BATCH_SIZE;
+        } else {
+            eventsWerePublishedMessage = "events were published";
+        }
+
+        var publishRate = EVENTS_RATE / EVENTS_BATCH_SIZE;
+        var batchesToPublish = EVENTS_TO_PUBLISH / EVENTS_BATCH_SIZE;
         try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
-            for (var i = 0; i < EVENTS_TO_PUBLISH; i++) {
-                var result = executor.submit(() -> publishRandomEvent(publisher));
+            for (var i = 0; i < batchesToPublish; i++) {
+                var result = executor.submit(() -> publishRandomEventOrEventsBatch(publisher));
                 futures.add(result);
 
                 var publications = i + 1;
-                if (futures.size() >= EVENTS_RATE && publications < EVENTS_TO_PUBLISH) {
-                    System.out.printf("%s, %d/%d events were published, waiting 1s before next publications...%n",
-                            LocalDateTime.now(), publications, EVENTS_TO_PUBLISH);
+                if (futures.size() >= publishRate && publications < batchesToPublish) {
+                    System.out.printf("%s, %d/%d %s, waiting 1s before next publications...%n",
+                            LocalDateTime.now().truncatedTo(ChronoUnit.MILLIS),
+                            publications * EVENTS_BATCH_SIZE,
+                            EVENTS_TO_PUBLISH,
+                            eventsWerePublishedMessage);
                     Thread.sleep(1000);
                     awaitForFutures(futures);
                     futures.clear();
@@ -204,11 +219,18 @@ static void publishEvents(EventSQLPublisher publisher) throws Exception {
         }
     }
 
-    static void publishRandomEvent(EventSQLPublisher publisher) {
+    static void publishRandomEventOrEventsBatch(EventSQLPublisher publisher) {
         try {
             // make publication more evenly distributed in time
             Thread.sleep(RANDOM.nextInt(1000));
-            publisher.publish(nextEvent());
+            if (EVENTS_BATCH_SIZE > 1) {
+                var batch = Stream.generate(EventSQLBenchmarksRunner::nextEvent)
+                        .limit(EVENTS_BATCH_SIZE)
+                        .toList();
+                publisher.publishAll(batch);
+            } else {
+                publisher.publish(nextEvent());
+            }
         } catch (Exception e) {
             e.printStackTrace();
         }

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

@@ -25,7 +25,7 @@
 
 public class EventSQL {
 
-    public static final int DEFAULT_FLUSH_PUBLISH_BUFFER_SIZE = 500;
+    public static final int DEFAULT_FLUSH_PUBLISH_BUFFER_SIZE = 1000;
     // 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;