#191: add main logic for EidGeneratorStrategy

nakadi-producer-spring-boot-starter/src/main/java/org/zalando/nakadiproducer/NakadiProducerAutoConfiguration.java

@@ -27,7 +27,9 @@
 import org.zalando.fahrschein.http.api.RequestFactory;
 import org.zalando.fahrschein.http.simple.SimpleRequestFactory;
 import org.zalando.nakadiproducer.eventlog.CompactionKeyExtractor;
+import org.zalando.nakadiproducer.eventlog.EidGeneratorStrategy;
 import org.zalando.nakadiproducer.eventlog.EventLogWriter;
+import org.zalando.nakadiproducer.eventlog.impl.EventLogMapper;
 import org.zalando.nakadiproducer.eventlog.impl.EventLogRepository;
 import org.zalando.nakadiproducer.eventlog.impl.EventLogRepositoryImpl;
 import org.zalando.nakadiproducer.eventlog.impl.EventLogWriterImpl;
@@ -121,6 +123,12 @@ public FlowIdComponent flowIdComponent() {
         }
     }
 
+    @Bean
+    @ConditionalOnMissingBean(EidGeneratorStrategy.class)
+    public EidGeneratorStrategy eidGeneratorStrategy() {
+        return EidGeneratorStrategy.noop();
+    }
+
     @Bean
     @ConditionalOnMissingBean
     public SnapshotEventCreationEndpoint snapshotEventCreationEndpoint(
@@ -139,9 +147,9 @@ public SnapshotCreationService snapshotCreationService(
     }
 
     @Bean
-    public EventLogWriter eventLogWriter(EventLogRepository eventLogRepository, ObjectMapper objectMapper,
-                                         FlowIdComponent flowIdComponent, List<CompactionKeyExtractor> extractorList) {
-        return new EventLogWriterImpl(eventLogRepository, objectMapper, flowIdComponent, extractorList);
+    public EventLogWriter eventLogWriter(EventLogRepository eventLogRepository, EventLogMapper eventLogMapper,
+                                         List<CompactionKeyExtractor> extractorList) {
+        return new EventLogWriterImpl(eventLogRepository, eventLogMapper, extractorList);
     }
 
     @Bean
@@ -176,4 +184,10 @@ public EventTransmissionService eventTransmissionService(
     public FlywayMigrator flywayMigrator() {
         return new FlywayMigrator();
     }
+
+    @Bean
+    public EventLogMapper eventLogMapper(ObjectMapper objectMapper, FlowIdComponent flowIdComponent,
+                                         EidGeneratorStrategy eidGeneratorStrategy) {
+        return new EventLogMapper(objectMapper, flowIdComponent, eidGeneratorStrategy);
+    }
 }

nakadi-producer-spring-boot-starter/src/main/java/org/zalando/nakadiproducer/eventlog/impl/EventLogRepositoryImpl.java

@@ -96,16 +96,18 @@ public void persist(Collection<EventLog> eventLogs) {
               namedParameterMap.addValue("lockedBy", eventLog.getLockedBy());
               namedParameterMap.addValue("lockedUntil", eventLog.getLockedUntil());
               namedParameterMap.addValue("compactionKey", eventLog.getCompactionKey());
+              namedParameterMap.addValue("eid", eventLog.getEid());
               return namedParameterMap;
             })
             .toArray(MapSqlParameterSource[]::new);
 
       jdbcTemplate.batchUpdate(
           "INSERT INTO " +
               "    nakadi_events.event_log " +
-              "    (event_type, event_body_data, flow_id, created, last_modified, locked_by, locked_until, compaction_key)" +
+              "    (event_type, event_body_data, flow_id, created, last_modified, locked_by, locked_until, compaction_key, eid)" +
               "VALUES " +
-              "    (:eventType, :eventBodyData, :flowId, :created, :lastModified, :lockedBy, :lockedUntil, :compactionKey)",
+              "    (:eventType, :eventBodyData, :flowId, :created, :lastModified, :lockedBy, :lockedUntil, :compactionKey, " +
+              "     COALESCE(:eid, CAST(LPAD(TO_HEX(CAST(CURRVAL('nakadi_events.event_log_id_seq') as BIGINT)), 32, '0') AS UUID)))",
           namedParameterMaps
       );
     }

nakadi-producer-spring-boot-starter/src/main/resources/db_nakadiproducer/migrations/V2/V2133546886.1.3__nakadi_event_eid.sql

@@ -0,0 +1 @@
+ALTER TABLE nakadi_events.event_log ADD COLUMN eid UUID;

nakadi-producer-spring-boot-starter/src/test/java/org/zalando/nakadiproducer/config/EmbeddedDataSourceConfig.java

@@ -9,6 +9,9 @@
 import java.io.IOException;
 
 import javax.sql.DataSource;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.TransactionDefinition;
+import org.springframework.transaction.support.TransactionTemplate;
 
 @Configuration
 public class EmbeddedDataSourceConfig {

nakadi-producer-spring-boot-starter/src/test/java/org/zalando/nakadiproducer/eventlog/impl/EventLogRepositoryIT.java

@@ -1,21 +1,27 @@
 package org.zalando.nakadiproducer.eventlog.impl;
 
 import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasSize;
+import static org.hamcrest.Matchers.samePropertyValuesAs;
 import static org.hamcrest.core.Is.is;
 
-import javax.transaction.Transactional;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.jdbc.core.BeanPropertyRowMapper;
 import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.transaction.annotation.Transactional;
+import org.springframework.transaction.support.TransactionTemplate;
 import org.zalando.nakadiproducer.BaseMockedExternalCommunicationIT;
 
 import java.util.List;
 
-@Transactional
 public class EventLogRepositoryIT extends BaseMockedExternalCommunicationIT {
 
     @Autowired
@@ -24,6 +30,9 @@ public class EventLogRepositoryIT extends BaseMockedExternalCommunicationIT {
     @Autowired
     private JdbcTemplate jdbcTemplate;
 
+    @Autowired
+    private TransactionTemplate transactionTemplate;
+
     private static final String WAREHOUSE_EVENT_BODY_DATA =
             ("{'self':'http://WAREHOUSE_DOMAIN',"
             + "'code':'WH-DE-EF',"
@@ -47,22 +56,18 @@ public class EventLogRepositoryIT extends BaseMockedExternalCommunicationIT {
     @BeforeEach
     public void setUp() {
         eventLogRepository.deleteAll();
-
-        persistTestEvent("FLOW_ID");
+        jdbcTemplate.execute("alter sequence nakadi_events.event_log_id_seq restart with 1");
     }
 
     private void persistTestEvent(String flowId) {
-        final EventLog eventLog = EventLog.builder()
-                .eventBodyData(WAREHOUSE_EVENT_BODY_DATA)
-                .eventType(WAREHOUSE_EVENT_TYPE)
-                .compactionKey(COMPACTION_KEY)
-                .flowId(flowId)
-                .build();
+        final EventLog eventLog = buildEventLog(flowId);
         eventLogRepository.persist(eventLog);
     }
 
     @Test
+    @Transactional
     public void testDeleteMultipleEvents() {
+        persistTestEvent("FLOW_ID");
         persistTestEvent("second_Flow-ID");
         persistTestEvent("third flow-ID");
         persistTestEvent("fourth flow-ID");
@@ -77,8 +82,7 @@ public void testDeleteMultipleEvents() {
 
         List<EventLog> remaining = findAllEventsInDB();
         assertThat(remaining, hasSize(1));
-        assertThat(remaining.get(0).getId(), is(notDeleted.getId()));
-        assertThat(remaining.get(0).getFlowId(), is(notDeleted.getFlowId()));
+        assertThat(remaining.get(0), equalTo(notDeleted));
     }
 
     private List<EventLog> findAllEventsInDB() {
@@ -88,18 +92,140 @@ private List<EventLog> findAllEventsInDB() {
     }
 
     @Test
+    @Transactional
     public void testFindEventInRepositoryById() {
+        persistTestEvent("FLOW_ID");
         Integer id = jdbcTemplate.queryForObject(
             "SELECT id FROM nakadi_events.event_log WHERE flow_id = 'FLOW_ID'",
             Integer.class);
         final EventLog eventLog = eventLogRepository.findOne(id);
         compareWithPersistedEvent(eventLog);
     }
 
+    @Test
+    @Transactional
+    public void testInsertEventWithDefaultEid() {
+        persistTestEvent("FLOW_ID");
+        EventLog actual = findAllEventsInDB().get(0);
+
+        EventLog expected = buildEventLog("FLOW_ID", 1, buildEid(1));
+        assertEvent(actual, expected);
+    }
+
+    @Test
+    @Transactional
+    public void testInsertEventWithDefinedEid() {
+        EventLog testEvent = buildEventLog("flow-id", 1, UUID.randomUUID());
+        eventLogRepository.persist(testEvent);
+
+        EventLog actual = findAllEventsInDB().get(0);
+        assertEvent(actual, testEvent);
+    }
+
+    @Test
+    @Transactional
+    public void testInsertEventWithNegativeId() {
+        jdbcTemplate.execute("ALTER SEQUENCE nakadi_events.event_log_id_seq " +
+            " MINVALUE " + Integer.MIN_VALUE +
+            " START " + Integer.MIN_VALUE +
+            " RESTART " + Integer.MIN_VALUE);
+        persistTestEvent("FLOW_ID");
+        EventLog actual = findAllEventsInDB().get(0);
+
+        EventLog expected = buildEventLog("FLOW_ID", Integer.MIN_VALUE, buildEid(Integer.MIN_VALUE));
+        assertEvent(actual, expected);
+    }
+
+    /**
+     * This test checks that the default eid is generated correctly when multiple transactions are running in parallel.
+     * The test creates three events in two parallel transactions.
+     * Execution order of the transactions is the following:
+     * -- / T1 starts -- Event1 ----------------------------------------- Event3 --- T1 ends -/--
+     * -------------------------- / T2 starts --- Event2 --- T2 ends -/--------------------------
+     */
+    @Test
+    public void testInsertEventWithDefaultEidWithParallelTransactions() throws Exception {
+        EventLog firstExpected = buildEventLog("first flow-id", 1, buildEid(1));
+        EventLog secondExpected = buildEventLog("second flow-id", 2, buildEid(2));
+        EventLog thirdExpected = buildEventLog("third flow-id", 3, buildEid(3));
+
+        CountDownLatch latchInsideTransaction = new CountDownLatch(1);
+        CountDownLatch latchOutsideTransaction = new CountDownLatch(1);
+
+        CompletableFuture<Void> future = CompletableFuture.runAsync(() ->
+            transactionTemplate.executeWithoutResult(
+                (s) -> {
+                    // We persist first event with default eid in the start of first transaction
+                    // It should be first event in the table.
+                    persistTestEvent("first flow-id");
+
+                    latchInsideTransaction.countDown();
+                    // Waiting for the second transaction to complete
+                    try {
+                        latchOutsideTransaction.await(2, TimeUnit.SECONDS);
+                    } catch (Exception e) {
+                        throw new RuntimeException(e);
+                    }
+
+                    // We persist second event in the end of first transaction.
+                    // It should be third event in the table.
+                    persistTestEvent("third flow-id");
+                }
+            )
+        );
+
+        // Waiting for the first transaction to start
+        latchInsideTransaction.await(1, TimeUnit.SECONDS);
+
+        // We persist third event in the second transaction.
+        transactionTemplate.executeWithoutResult(
+            (s) -> persistTestEvent("second flow-id")
+        );
+        // We check that the first and third events haven't visible yet
+        List<EventLog> events = findAllEventsInDB();
+        assertThat(events, hasSize(1));
+        assertEvent(events.get(0), secondExpected);
+
+        // Waiting for the first transaction to complete
+        latchOutsideTransaction.countDown();
+        future.get(1, TimeUnit.SECONDS);
+
+        // We check that the all three events are in the table
+        events = findAllEventsInDB();
+        assertThat(events, hasSize(3));
+        assertEvent(events.get(0), firstExpected);
+        assertEvent(events.get(1), secondExpected);
+        assertEvent(events.get(2), thirdExpected);
+    }
+
+    private void assertEvent(EventLog actual, EventLog expected) {
+        assertThat(actual,
+            samePropertyValuesAs(expected, "created", "lastModified")
+        );
+    }
+
     private void compareWithPersistedEvent(final EventLog eventLog) {
         assertThat(eventLog.getEventBodyData(), is(WAREHOUSE_EVENT_BODY_DATA));
         assertThat(eventLog.getEventType(), is(WAREHOUSE_EVENT_TYPE));
         assertThat(eventLog.getCompactionKey(), is(COMPACTION_KEY));
     }
 
+    private EventLog buildEventLog(String flowId) {
+        return buildEventLog(flowId, null, null);
+    }
+
+    private EventLog buildEventLog(String flowId, Integer id, UUID eid) {
+        return EventLog.builder()
+            .id(id)
+            .eventBodyData(WAREHOUSE_EVENT_BODY_DATA)
+            .eventType(WAREHOUSE_EVENT_TYPE)
+            .compactionKey(COMPACTION_KEY)
+            .flowId(flowId)
+            .eid(eid)
+            .build();
+    }
+
+    private UUID buildEid(int id) {
+        return new UUID(0, id);
+    }
 }

nakadi-producer/src/main/java/org/zalando/nakadiproducer/eventlog/EidGeneratorStrategy.java

@@ -0,0 +1,22 @@
+package org.zalando.nakadiproducer.eventlog;
+
+import java.util.UUID;
+import org.zalando.nakadiproducer.eventlog.impl.EventLog;
+
+/**
+ * Strategy for generating EIDs.
+ * <p>
+ * EID - is unique identifier for nakadi events and we expect that implementation will generate unique EID for each event.
+ */
+public interface EidGeneratorStrategy {
+
+    static EidGeneratorStrategy noop() {
+        return (EventLog eventLog) -> null;
+    }
+
+    static EidGeneratorStrategy random() {
+        return (EventLog eventLog) -> UUID.randomUUID();
+    }
+
+    UUID generateEid(EventLog eventLog);
+}

nakadi-producer/src/main/java/org/zalando/nakadiproducer/eventlog/impl/DataChangeEventEnvelope.java

@@ -1,13 +1,15 @@
 package org.zalando.nakadiproducer.eventlog.impl;
 
 import lombok.AllArgsConstructor;
+import lombok.EqualsAndHashCode;
 import lombok.Getter;
 import lombok.Setter;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
 
 @Getter
 @Setter
+@EqualsAndHashCode
 @AllArgsConstructor
 class DataChangeEventEnvelope {
     @JsonProperty("data_op")

nakadi-producer/src/main/java/org/zalando/nakadiproducer/eventlog/impl/EventLog.java

@@ -1,22 +1,34 @@
 package org.zalando.nakadiproducer.eventlog.impl;
 
+import java.util.UUID;
 import lombok.AllArgsConstructor;
 import lombok.Builder;
-import lombok.Getter;
+import lombok.Data;
 import lombok.NoArgsConstructor;
-import lombok.Setter;
-import lombok.ToString;
 
 import java.time.Instant;
 
-@ToString
-@Getter
-@Setter
+@Data
 @NoArgsConstructor
 @AllArgsConstructor
 @Builder(toBuilder = true)
 public class EventLog {
 
+    public EventLog(Integer id, String eventType, String eventBodyData, String flowId,
+                    Instant created,
+                    Instant lastModified, String lockedBy, Instant lockedUntil,
+                    String compactionKey) {
+        this.id = id;
+        this.eventType = eventType;
+        this.eventBodyData = eventBodyData;
+        this.flowId = flowId;
+        this.created = created;
+        this.lastModified = lastModified;
+        this.lockedBy = lockedBy;
+        this.lockedUntil = lockedUntil;
+        this.compactionKey = compactionKey;
+    }
+
     private Integer id;
     private String eventType;
     private String eventBodyData;
@@ -26,4 +38,18 @@ public class EventLog {
     private String lockedBy;
     private Instant lockedUntil;
     private String compactionKey;
+    private UUID eid;
+
+    /**
+     * This is only needed for backward compatibility.
+     *
+     * <p>For instance 213 will be converted to "00000000-0000-0000-0000-0000000000d5"</p>
+     */
+    public UUID getEid() {
+        if (eid == null) {
+            return new UUID(0, id);
+        }
+
+        return eid;
+    }
 }