API improvements

Author: BinaryIgor

README.md

@@ -28,7 +28,7 @@ CREATE TABLE event (
   key TEXT,
   value BYTEA NOT NULL,
   created_at TIMESTAMP NOT NULL DEFAULT NOW(),
-  metadata JSONB NOT NULL,
+  metadata JSON NOT NULL,
   PRIMARY KEY (topic, id)
 ) PARTITION BY LIST (topic);
 

benchmarks/events-db/init_db.sql

@@ -19,7 +19,7 @@ CREATE TABLE event (
   key TEXT,
   value BYTEA NOT NULL,
   created_at TIMESTAMP NOT NULL DEFAULT NOW(),
-  metadata JSONB NOT NULL,
+  metadata JSON NOT NULL,
   PRIMARY KEY (topic, id)
 ) PARTITION BY LIST (topic);
 

benchmarks/runner/Dockerfile

@@ -1,5 +1,5 @@
 FROM eclipse-temurin:21-alpine
 
-COPY target/eventsql-benchmarks-runer-jar-with-dependencies.jar /app.jar
+COPY target/eventsql-benchmarks-runner-jar-with-dependencies.jar /app.jar
 
 ENTRYPOINT ["java", "-jar", "/app.jar"]

benchmarks/runner/pom.xml

@@ -51,7 +51,7 @@
                             <goal>single</goal>
                         </goals>
                         <configuration>
-                            <finalName>eventsql-benchmarks-runer</finalName>
+                            <finalName>eventsql-benchmarks-runner</finalName>
                             <archive>
                                 <manifest>
                                     <mainClass>com.binaryigor.eventsql.benchmarks.EventSQLBenchmarksRunner</mainClass>

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

@@ -61,7 +61,7 @@ public static void main(String[] args) throws Exception {
 
         var start = System.currentTimeMillis();
 
-        publishEvents(eventSQL.publisher(), topicDefinition);
+        publishEvents(eventSQL.publisher());
         var publicationDuration = Duration.ofMillis(System.currentTimeMillis() - start);
 
         printDelimiter();
@@ -94,11 +94,6 @@ static String envValueOrDefault(String key, String defaultValue) {
         return System.getenv().getOrDefault(key, defaultValue);
     }
 
-    static String envValueOrThrow(String key) {
-        return Optional.ofNullable(System.getenv().get(key))
-                .orElseThrow(() -> new RuntimeException("%s env variable is required but was not supplied!".formatted(key)));
-    }
-
     static int envIntValueOrDefault(String key, int defaultValue) {
         return Integer.parseInt(envValueOrDefault(key, String.valueOf(defaultValue)));
     }
@@ -173,12 +168,12 @@ static <T> T executeQuery(DataSource source, String query, ResultSetMapper<T> re
         }
     }
 
-    static void publishEvents(EventSQLPublisher publisher, TopicDefinition topicDefinition) throws Exception {
+    static void publishEvents(EventSQLPublisher publisher) throws Exception {
         var futures = new LinkedList<Future<?>>();
 
         try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
             for (var i = 0; i < EVENTS_TO_PUBLISH; i++) {
-                var result = executor.submit(() -> publishRandomEvent(publisher, topicDefinition));
+                var result = executor.submit(() -> publishRandomEvent(publisher));
                 futures.add(result);
 
                 var publications = i + 1;
@@ -198,20 +193,18 @@ static void publishEvents(EventSQLPublisher publisher, TopicDefinition topicDefi
         }
     }
 
-    static void publishRandomEvent(EventSQLPublisher publisher, TopicDefinition topicDefinition) {
+    static void publishRandomEvent(EventSQLPublisher publisher) {
         try {
             // make publication more evenly distributed in time
             Thread.sleep(RANDOM.nextInt(1000));
-            var partition = RANDOM.nextInt(topicDefinition.partitions());
-            var event = nextEvent(partition);
-            publisher.publish(event);
+            publisher.publish(nextEvent());
         } catch (Exception e) {
             e.printStackTrace();
         }
     }
 
-    static EventPublication nextEvent(int partition) {
-        return new EventPublication(TEST_TOPIC, partition, accountCreatedEventJson().getBytes(StandardCharsets.UTF_8));
+    static EventPublication nextEvent() {
+        return new EventPublication(TEST_TOPIC, accountCreatedEventJson().getBytes(StandardCharsets.UTF_8));
     }
 
     static String accountCreatedEventJson() {

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

@@ -3,27 +3,10 @@
 import java.util.Map;
 
 public record EventPublication(String topic,
-                               int partition,
                                String key,
                                byte[] value,
                                Map<String, String> metadata) {
 
-    public EventPublication(String topic, int partition, String key, byte[] value) {
-        this(topic, partition, key, value, Map.of());
-    }
-
-    public EventPublication(String topic, int partition, byte[] value, Map<String, String> metadata) {
-        this(topic, partition, null, value, metadata);
-    }
-
-    public EventPublication(String topic, int partition, byte[] value) {
-        this(topic, partition, value, Map.of());
-    }
-
-    public EventPublication(String topic, String key, byte[] value, Map<String, String> metadata) {
-        this(topic, -1, key, value, metadata);
-    }
-
     public EventPublication(String topic, String key, byte[] value) {
         this(topic, key, value, Map.of());
     }
@@ -35,8 +18,4 @@ public EventPublication(String topic, byte[] value, Map<String, String> metadata
     public EventPublication(String topic, byte[] value) {
         this(topic, value, Map.of());
     }
-
-    public EventPublication withPartition(int partition) {
-        return new EventPublication(topic, partition, key, value, metadata);
-    }
 }

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

@@ -1,7 +1,7 @@
 package com.binaryigor.eventsql;
 
-import com.binaryigor.eventsql.internal.EventSQLOps;
 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;
@@ -18,7 +18,8 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
-import java.util.Optional;
+
+import static java.util.Collections.unmodifiableList;
 
 public class EventSQL {
 
@@ -27,34 +28,20 @@ public class EventSQL {
     private final EventSQLConsumers consumers;
 
     public EventSQL(DataSource dataSource, SQLDialect sqlDialect) {
-        this(dataSource, sqlDialect, Clock.systemUTC(), Optional.empty());
+        this(dataSource, sqlDialect, Clock.systemUTC());
     }
 
     public EventSQL(DataSource dataSource, SQLDialect sqlDialect, Clock clock) {
-        this(dataSource, sqlDialect, clock, Optional.empty());
-    }
-
-    public EventSQL(DataSource dataSource,
-                    SQLDialect sqlDialect,
-                    Clock clock,
-                    Optional<EventSQLConsumers.DLTEventFactory> dltEventFactory) {
-        this(List.of(dataSource), sqlDialect, clock, dltEventFactory);
+        this(List.of(dataSource), sqlDialect, clock);
     }
 
     public EventSQL(Collection<DataSource> dataSources, SQLDialect sqlDialect) {
-        this(dataSources, sqlDialect, Clock.systemUTC(), Optional.empty());
+        this(dataSources, sqlDialect, Clock.systemUTC());
     }
 
     public EventSQL(Collection<DataSource> dataSources,
                     SQLDialect sqlDialect,
                     Clock clock) {
-        this(dataSources, sqlDialect, clock, Optional.empty());
-    }
-
-    public EventSQL(Collection<DataSource> dataSources,
-                    SQLDialect sqlDialect,
-                    Clock clock,
-                    Optional<EventSQLConsumers.DLTEventFactory> dltEventFactory) {
         if (dataSources.isEmpty()) {
             throw new IllegalArgumentException("At least one data source is required");
         }
@@ -72,13 +59,12 @@ public EventSQL(Collection<DataSource> dataSources,
 
             var topicRepository = new SqlTopicRepository(transactions);
             var consumerRepository = new SqlConsumerRepository(transactions);
-            var eventRepository = new SqlEventRepository(transactions);
+            var eventRepository = new SqlEventRepository(transactions, sqlDialect);
 
             var registry = new DefaultEventSQLRegistry(topicRepository, eventRepository, consumerRepository, transactions);
 
             var topicDefinitionsCache = new TopicDefinitionsCache(topicRepository);
             var ops = new EventSQLOps(topicDefinitionsCache, transactions, consumerRepository, eventRepository, clock);
-            dltEventFactory.ifPresent(ops::configureDLTEventFactory);
 
             registryList.add(registry);
             publisherList.add(ops);
@@ -91,9 +77,9 @@ public EventSQL(Collection<DataSource> dataSources,
             publisher = publisherList.getFirst();
             consumers = consumersList.getFirst();
         } else {
-            registry = new ShardedEventSQLRegistry(registryList);
-            publisher = new ShardedEventSQLPublisher(publisherList);
-            consumers = new ShardedEventSQLConsumers(consumersList);
+            registry = new ShardedEventSQLRegistry(unmodifiableList(registryList));
+            publisher = new ShardedEventSQLPublisher(unmodifiableList(publisherList));
+            consumers = new ShardedEventSQLConsumers(unmodifiableList(consumersList));
         }
     }
 

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

@@ -7,4 +7,8 @@ public interface EventSQLPublisher {
     void publish(EventPublication publication);
 
     void publishAll(Collection<EventPublication> publications);
+
+    void configurePartitioner(Partitioner partitioner);
+
+    Partitioner partitioner();
 }

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

@@ -0,0 +1,5 @@
+package com.binaryigor.eventsql;
+
+public interface Partitioner {
+    int partition(EventPublication publication, int topicPartitions);
+}

src/main/java/com/binaryigor/eventsql/internal/DefaultDLTEventFactory.java

@@ -33,8 +33,6 @@ public Optional<EventPublication> create(EventSQLConsumptionException exception,
         metadata.put("exceptionType", cause.getClass().getName());
         metadata.put("exceptionMessage", Optional.ofNullable(cause.getMessage()).orElse(exception.getMessage()));
 
-        // TODO: maybe warning, strange situation
-        var partition = dltTopicDefinitionOpt.get().partitions() < event.partition() ? -1 : dltTopicDefinitionOpt.get().partitions();
-        return Optional.of(new EventPublication(dltTopic, partition, event.key(), event.value(), metadata));
+        return Optional.of(new EventPublication(dltTopic, event.key(), event.value(), metadata));
     }
 }