Merge pull request #113 from zalando-nakadi/feature/batching-sb1

Batch event submission (#48)

Author: ePaul

nakadi-producer-spring-boot-starter/pom.xml

@@ -10,7 +10,7 @@
     <parent>
         <groupId>org.zalando</groupId>
         <artifactId>nakadi-producer-reactor</artifactId>
-        <version>4.2.0</version>
+        <version>4.3.0</version>
     </parent>
 
     <artifactId>nakadi-producer-spring-boot-starter</artifactId>

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

@@ -60,8 +60,8 @@ public void eventsShouldNotBeSentTwiceWhenLockExpiresDuringTransmission() {
         Collection<EventLog> eventLogsLockedSecond = eventTransmissionService.lockSomeEvents();
 
         // when both job instances try to send their locked events
-        eventLogsLockedFirst.forEach(eventTransmissionService::sendEvent);
-        eventLogsLockedSecond.forEach(eventTransmissionService::sendEvent);
+        eventTransmissionService.sendEvents(eventLogsLockedFirst);
+        eventTransmissionService.sendEvents(eventLogsLockedSecond);
 
         // Then the event should have been sent only once.
         List<String> value = nakadiClient.getSentEvents(MY_EVENT_TYPE);

nakadi-producer/pom.xml

@@ -10,7 +10,7 @@
     <parent>
         <groupId>org.zalando</groupId>
         <artifactId>nakadi-producer-reactor</artifactId>
-        <version>4.2.0</version>
+        <version>4.3.0</version>
     </parent>
 
     <artifactId>nakadi-producer</artifactId>

nakadi-producer/src/main/java/org/zalando/nakadiproducer/transmission/MockNakadiPublishingClient.java

@@ -28,7 +28,7 @@ public MockNakadiPublishingClient(ObjectMapper objectMapper) {
     }
 
     @Override
-    public synchronized void publish(String eventType, List<?> nakadiEvents) {
+    public synchronized void publish(String eventType, List<?> nakadiEvents) throws Exception {
         nakadiEvents.stream().map(this::transformToJson).forEach(e -> sentEvents.add(eventType, e));
     }
 

nakadi-producer/src/main/java/org/zalando/nakadiproducer/transmission/impl/EventBatcher.java

@@ -0,0 +1,88 @@
+package org.zalando.nakadiproducer.transmission.impl;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.AllArgsConstructor;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.ToString;
+import lombok.extern.slf4j.Slf4j;
+import org.zalando.nakadiproducer.eventlog.impl.EventLog;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.function.Consumer;
+
+@Slf4j
+public class EventBatcher {
+
+    private static final long NAKADI_BATCH_SIZE_LIMIT_IN_BYTES = 50000000;
+    private final ObjectMapper objectMapper;
+    private final Consumer<List<BatchItem>> publisher;
+
+    private List<BatchItem> batch;
+    private long aggregatedBatchSize;
+
+    public EventBatcher(ObjectMapper objectMapper, Consumer<List<BatchItem>> publisher) {
+        this.objectMapper = objectMapper;
+        this.publisher = publisher;
+
+        this.batch = new ArrayList<>();
+        this.aggregatedBatchSize = 0;
+    }
+
+    /**
+     * Pushes one event to be published. It will be either published right now, or with some other events,
+     * latest when calling {@link #finish()}.
+     * @param eventLogEntry The event log entry for this event.
+     * @param nakadiEvent The Nakadi form of the event.
+     */
+    public void pushEvent(EventLog eventLogEntry, NakadiEvent nakadiEvent) {
+        long eventSize;
+
+        try {
+            eventSize = objectMapper.writeValueAsBytes(nakadiEvent).length;
+        } catch (Exception e) {
+            log.error("Could not serialize event {} of type {}, skipping it.", eventLogEntry.getId(), eventLogEntry.getEventType(), e);
+            return;
+        }
+
+
+        if (!batch.isEmpty() &&
+                (hasAnotherEventType(batch, eventLogEntry) || batchWouldBecomeTooBig(aggregatedBatchSize, eventSize))) {
+            this.publisher.accept(batch);
+
+            batch = new ArrayList<>();
+            aggregatedBatchSize = 0;
+        }
+
+        batch.add(new BatchItem(eventLogEntry, nakadiEvent));
+        aggregatedBatchSize += eventSize;
+    }
+
+    /**
+     * Publishes all events which were pushed and not yet published.
+     */
+    public void finish() {
+        if (!batch.isEmpty()) {
+            this.publisher.accept(batch);
+        }
+    }
+
+    private boolean hasAnotherEventType(List<BatchItem> batch, EventLog event) {
+        return !event.getEventType().equals(batch.get(0).getEventLogEntry().getEventType());
+    }
+
+    private boolean batchWouldBecomeTooBig(long aggregatedBatchSize, long eventSize) {
+        return aggregatedBatchSize + eventSize > 0.8 * NAKADI_BATCH_SIZE_LIMIT_IN_BYTES;
+    }
+
+    @AllArgsConstructor
+    @Getter
+    @EqualsAndHashCode
+    @ToString
+    protected static class BatchItem {
+        EventLog eventLogEntry;
+        NakadiEvent nakadiEvent;
+    }
+
+}

nakadi-producer/src/main/java/org/zalando/nakadiproducer/transmission/impl/EventTransmissionService.java

@@ -3,22 +3,27 @@
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import lombok.extern.slf4j.Slf4j;
+import org.zalando.fahrschein.EventPublishingException;
+import org.zalando.fahrschein.domain.BatchItemResponse;
 import org.zalando.nakadiproducer.eventlog.impl.EventLog;
 import org.zalando.nakadiproducer.eventlog.impl.EventLogRepository;
 import org.zalando.nakadiproducer.transmission.NakadiPublishingClient;
+import org.zalando.nakadiproducer.transmission.impl.EventBatcher.BatchItem;
 
 import javax.transaction.Transactional;
+
 import java.io.IOException;
-import java.io.UncheckedIOException;
 import java.time.Clock;
 import java.time.Instant;
+import java.util.Arrays;
 import java.util.Collection;
-import java.util.HashMap;
 import java.util.LinkedHashMap;
+import java.util.List;
 import java.util.UUID;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import static java.time.temporal.ChronoUnit.MINUTES;
-import static java.util.Collections.singletonList;
 
 @Slf4j
 public class EventTransmissionService {
@@ -44,20 +49,72 @@ public Collection<EventLog> lockSomeEvents() {
     }
 
     @Transactional
-    public void sendEvent(EventLog eventLog) {
-        if (lockNearlyExpired(eventLog)) {
-            // to avoid that two instances process this event, we skip it
-            return;
+    public void sendEvents(Collection<EventLog> events) {
+        EventBatcher batcher = new EventBatcher(objectMapper, this::publishBatch);
+
+        for (EventLog event : events) {
+            if (lockNearlyExpired(event)) {
+                // to avoid that two instances process this event, we skip it
+                continue;
+            }
+
+            NakadiEvent nakadiEvent;
+
+            try {
+                nakadiEvent = mapToNakadiEvent(event);
+            } catch (Exception e) {
+                log.error("Could not serialize event {} of type {}, skipping it.", event.getId(), event.getEventType(), e);
+                continue;
+            }
+
+            batcher.pushEvent(event, nakadiEvent);
         }
 
+        batcher.finish();
+    }
+
+    /**
+     * Publishes a list of events.
+     * All of the events in this list need to be destined for the same event type.
+     */
+    private void publishBatch(List<BatchItem> batch) {
         try {
-            nakadiPublishingClient.publish(eventLog.getEventType(), singletonList(mapToNakadiEvent(eventLog)));
-            log.info("Event {} locked by {} was successfully transmitted to nakadi", eventLog.getId(), eventLog.getLockedBy());
-            eventLogRepository.delete(eventLog);
+            this.tryToPublishBatch(batch);
         } catch (Exception e) {
-            log.error("Event {} locked by {} could not be transmitted to nakadi: {}", eventLog.getId(), eventLog.getLockedBy(), e.getMessage());
+            log.error("Could not send {} events of type {}, skipping them.", batch.size(), batch.get(0).getEventLogEntry().getEventType(), e);
+        }
+    }
+
+    /**
+     * Tries to publish a set of events (all of which need to belong to the same event type).
+     * The successful ones will be deleted from the database.
+     */
+    private void tryToPublishBatch(List<BatchItem> batch) throws Exception {
+        Stream<EventLog> successfulEvents;
+        String eventType = batch.get(0).getEventLogEntry().getEventType();
+        try {
+            nakadiPublishingClient.publish(
+                    eventType,
+                    batch.stream()
+                            .map(BatchItem::getNakadiEvent)
+                            .collect(Collectors.toList())
+            );
+            successfulEvents = batch.stream().map(BatchItem::getEventLogEntry);
+            log.info("Sent {} events of type {}.", batch.size(), eventType);
+        } catch (EventPublishingException e) {
+            log.error("{} out of {} events of type {} failed to be sent.", e.getResponses().length, batch.size(), eventType);
+            List<String> failedEids = collectEids(e);
+            successfulEvents =
+                    batch.stream()
+                            .map(BatchItem::getEventLogEntry)
+                            .filter(rawEvent -> !failedEids.contains(convertToUUID(rawEvent.getId())));
         }
 
+        successfulEvents.forEach(eventLogRepository::delete);
+    }
+
+    private List<String> collectEids(EventPublishingException e) {
+        return Arrays.stream(e.getResponses()).map(BatchItemResponse::getEid).collect(Collectors.toList());
     }
 
     private boolean lockNearlyExpired(EventLog eventLog) {
@@ -67,7 +124,7 @@ private boolean lockNearlyExpired(EventLog eventLog) {
         return now().isAfter(eventLog.getLockedUntil().minus(1, MINUTES));
     }
 
-    public NakadiEvent mapToNakadiEvent(final EventLog event) {
+    private NakadiEvent mapToNakadiEvent(final EventLog event) throws IOException {
         final NakadiEvent nakadiEvent = new NakadiEvent();
 
         final NakadiMetadata metadata = new NakadiMetadata();
@@ -76,13 +133,7 @@ public NakadiEvent mapToNakadiEvent(final EventLog event) {
         metadata.setFlowId(event.getFlowId());
         nakadiEvent.setMetadata(metadata);
 
-        HashMap<String, Object> payloadDTO;
-        try {
-            payloadDTO = objectMapper.readValue(event.getEventBodyData(), new TypeReference<LinkedHashMap<String, Object>>() { });
-        } catch (IOException e) {
-            log.error("An error occurred at JSON deserialization", e);
-            throw new UncheckedIOException(e);
-        }
+        LinkedHashMap<String, Object> payloadDTO = objectMapper.readValue(event.getEventBodyData(), new TypeReference<LinkedHashMap<String, Object>>() { });
 
         nakadiEvent.setData(payloadDTO);
 
@@ -106,5 +157,4 @@ private String convertToUUID(final int number) {
         return new UUID(0, number).toString();
     }
 
-
 }

nakadi-producer/src/main/java/org/zalando/nakadiproducer/transmission/impl/EventTransmitter.java

@@ -8,6 +8,6 @@ public EventTransmitter(EventTransmissionService eventTransmissionService) {
     }
 
     public void sendEvents() {
-        eventTransmissionService.lockSomeEvents().forEach(eventTransmissionService::sendEvent);
+        eventTransmissionService.sendEvents(eventTransmissionService.lockSomeEvents());
     }
 }

nakadi-producer/src/test/java/org/zalando/nakadiproducer/eventlog/impl/EventLogWriterTest.java

@@ -13,7 +13,7 @@
 import org.mockito.ArgumentCaptor;
 import org.mockito.Captor;
 import org.mockito.Mock;
-import org.mockito.runners.MockitoJUnitRunner;
+import org.mockito.junit.MockitoJUnitRunner;
 import org.zalando.nakadiproducer.flowid.FlowIdComponent;
 import org.zalando.nakadiproducer.util.Fixture;
 import org.zalando.nakadiproducer.util.MockPayload;

nakadi-producer/src/test/java/org/zalando/nakadiproducer/transmission/MockNakadiPublishingClientTest.java

@@ -27,15 +27,15 @@ public void returnsEmptyResultIfNoEventsHaveBeenSent() {
     }
 
     @Test
-    public void returnsOnlyThoseEventsOfTheGivenType() {
+    public void returnsOnlyThoseEventsOfTheGivenType() throws Exception {
         mockNakadiPublishingClient.publish(MY_EVENT_TYPE, singletonList(new Event("anEvent")));
         mockNakadiPublishingClient.publish(OTHER_EVENT_TYPE, singletonList(new Event("anotherEvent")));
 
         assertThat(mockNakadiPublishingClient.getSentEvents(MY_EVENT_TYPE), contains("{\"attribute\":\"anEvent\"}"));
     }
 
     @Test
-    public void concatenatesSubsequentlyPublishedEventLists() {
+    public void concatenatesSubsequentlyPublishedEventLists() throws Exception {
         mockNakadiPublishingClient.publish(MY_EVENT_TYPE,
             asList(new Event("event1"), new Event("event2"))
         );
@@ -55,7 +55,7 @@ public void concatenatesSubsequentlyPublishedEventLists() {
     }
 
     @Test
-    public void deletesAllEventsOnClear() {
+    public void deletesAllEventsOnClear() throws Exception {
         mockNakadiPublishingClient.publish(MY_EVENT_TYPE, singletonList(new Event("event1")));
         mockNakadiPublishingClient.clearSentEvents();