Refinements (#154)

* README improvements
* removal of unneeded empty lines
* EventLogWriter.fireBusinessEvents now takes `Collection<?>` instead of `Collection<Object>`, same as the other plural fire methods.
* rename variable to be more clear
* add javadoc
* extract some code out of a loop
* add tests for transmission service

Author: ePaul

README.md

@@ -241,8 +241,8 @@ for sending events to a compacted event type.
 
 In some cases, like when there usually are large time gaps between producing events for the same compaction key,
 the risk of getting events for the same key out-of-order is small.
-For these cases, you just can define a bean of type [`CompactionKeyExtractor`](nakadi-producer/src/main/java/org/zalando/nakadiproducer/eventlog/CompactionKeyExtractor.java) , and then all events of that event
-type will get sent with a compaction key.
+For these cases, you just can define a bean of type [`CompactionKeyExtractor`](nakadi-producer/src/main/java/org/zalando/nakadiproducer/eventlog/CompactionKeyExtractor.java),
+and then all events of that event type will be sent with a compaction key.
 
 ```java
 @Configuration
@@ -261,7 +261,6 @@ matches the payload object (in undefined order) will be used.
 There are also some more factory methods with different signatures for more special cases, and you can also write
 your own implementation (but for the usual cases, the one shown here should be enough).
 
-
 ### Event snapshots (optional)
 
 A Snapshot event is a special type of data change event (data operation) defined by Nakadi.
@@ -300,7 +299,7 @@ your `application.properties` includes
 management.endpoints.web.exposure.include=snapshot-event-creation,your-other-endpoints,...`
 ```
 and if one or more Spring Beans implement the `org.zalando.nakadiproducer.snapshots.SnapshotEventGenerator` interface.
-(Note that this will automatically together with the compaction key feature mentioned above,
+(Note that this will automatically work together with the compaction key feature mentioned above,
  if you have registered a compaction key extractor matching the type of the data objects in your snapshots.)
 
 The optional filter specifier of the trigger request will be passed as a string parameter to the

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

@@ -39,8 +39,6 @@ public class EndToEndTestIT extends BaseMockedExternalCommunicationIT {
     @Autowired
     private MockNakadiPublishingClient nakadiClient;
 
-
-
     @BeforeEach
     @AfterEach
     public void clearNakadiEvents() {
@@ -92,7 +90,6 @@ public void compactionKeyIsNotInvented() throws IOException {
                 is(empty()));
     }
 
-
     @Test
     public void businessEventsShouldBeSubmittedToNakadi() throws IOException {
         MockPayload payload = Fixture.mockPayload(1, CODE);

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

@@ -231,5 +231,5 @@ public interface EventLogWriter {
    *            parameter)
    */
     @Transactional
-    void fireBusinessEvents(String eventType, Collection<Object> payloads);
+    void fireBusinessEvents(String eventType, Collection<?> payloads);
 }

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

@@ -30,7 +30,7 @@ public class EventLogWriterImpl implements EventLogWriter {
     private final ObjectMapper objectMapper;
     private final FlowIdComponent flowIdComponent;
 
-    private final Map<String, CompactionKeyExtractor> extractors;
+    private final Map<String, CompactionKeyExtractor> extractorsByEventType;
 
     public EventLogWriterImpl(EventLogRepository eventLogRepository,
                               ObjectMapper objectMapper,
@@ -39,12 +39,21 @@ public EventLogWriterImpl(EventLogRepository eventLogRepository,
         this.eventLogRepository = eventLogRepository;
         this.objectMapper = objectMapper;
         this.flowIdComponent = flowIdComponent;
-        this.extractors = keyExtractors.stream()
+        this.extractorsByEventType = keyExtractors.stream()
                 .collect(groupingBy(
                         CompactionKeyExtractor::getEventType,
                         collectingAndThen(toList(), EventLogWriterImpl::joinCompactors)));
     }
 
+    /**
+     * Helper function (used in the constructor) to join a (non-empty) list of compaction key extractors
+     * (for the same event type) into a single one. If that list has just one element, it is returned.
+     * Otherwise, a new extractor is created whose retrieval method will just iterate through all the
+     * extractors, ask them for the key and returns any that is non-empty.
+     *
+     * @param list a list of extractors, non-empty.
+     * @return a single extractor based on the list which will return a key when any of the extractors returns one.
+     */
     private static CompactionKeyExtractor joinCompactors(List<CompactionKeyExtractor> list) {
         Preconditions.checkArgument(!list.isEmpty());
         if(list.size() == 1) {
@@ -59,7 +68,6 @@ private static CompactionKeyExtractor joinCompactors(List<CompactionKeyExtractor
         }
     }
 
-
     @Override
     @Transactional
     public void fireCreateEvent(final String eventType, final String dataType, final Object data) {
@@ -121,13 +129,13 @@ public void fireBusinessEvent(final String eventType, Object payload) {
 
     @Override
     @Transactional
-    public void fireBusinessEvents(final String eventType, final Collection<Object> payload) {
+    public void fireBusinessEvents(final String eventType, final Collection<?> payload) {
         final Collection<EventLog> eventLogs = createBusinessEventLogs(eventType, payload);
         eventLogRepository.persist(eventLogs);
     }
 
     private Collection<EventLog> createBusinessEventLogs(final String eventType,
-                                                     final Collection<Object> eventPayloads) {
+                                                     final Collection<?> eventPayloads) {
         CompactionKeyExtractor extractor = getKeyExtractorFor(eventType);
         return eventPayloads.stream()
                 .map(payload -> createEventLog(eventType, payload,
@@ -142,10 +150,11 @@ private Collection<EventLog> createDataEventLogs(
             final Collection<?> data
     ) {
         CompactionKeyExtractor extractor = getKeyExtractorFor(eventType);
+        String dataOp = eventDataOperation.toString();
         return data.stream()
                 .map(payload -> createEventLog(
                         eventType,
-                        new DataChangeEventEnvelope(eventDataOperation.toString(), dataType, payload),
+                        new DataChangeEventEnvelope(dataOp, dataType, payload),
                         extractor.getKeyOrNull(payload)))
                 .collect(toList());
     }
@@ -161,7 +170,7 @@ private String getCompactionKeyFor(String eventType, Object payload) {
     }
 
     private CompactionKeyExtractor getKeyExtractorFor(String eventType) {
-        return extractors.getOrDefault(eventType, NOOP_EXTRACTOR);
+        return extractorsByEventType.getOrDefault(eventType, NOOP_EXTRACTOR);
     }
 
     private EventLog createEventLog(final String eventType, final Object eventPayload, String compactionKey) {

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

@@ -110,7 +110,6 @@ public void threeCompactionExtractors() {
         assertThat(compactionKeys, contains(equalTo("Hello"), equalTo("World"), equalTo("List?")));
     }
 
-
     private List<String> getPersistedCompactionKeys() {
         verify(eventLogRepository).persist(eventLogsCapture.capture());
         Collection<EventLog> eventLogs = eventLogsCapture.getValue();

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

@@ -81,6 +81,31 @@ public void testWithoutFlowId() throws JsonProcessingException {
         assertThat(read(events.get(0), "$.metadata"), not(hasKey("flow_id")));
     }
 
+    @Test
+    public void testWithCompactionKey() throws JsonProcessingException {
+        String compactionKey = "XYZ";
+        String payloadString = mapper.writeValueAsString(Fixture.mockPayload(42, "bla"));
+        EventLog ev = new EventLog(27, "type", payloadString, null, now(), now(), null, now().plus(5, MINUTES), compactionKey);
+
+        service.sendEvents(singletonList(ev));
+
+        List<String> events = publishingClient.getSentEvents("type");
+        assertThat(events, hasSize(1));
+        assertThat(read(events.get(0), "$.metadata.partition_compaction_key"), is(compactionKey));
+    }
+
+    @Test
+    public void testWithoutCompactionKey() throws JsonProcessingException {
+        String payloadString = mapper.writeValueAsString(Fixture.mockPayload(42, "bla"));
+        EventLog ev = new EventLog(27, "type", payloadString, null, now(), now(), null, now().plus(5, MINUTES), null);
+
+        service.sendEvents(singletonList(ev));
+
+        List<String> events = publishingClient.getSentEvents("type");
+        assertThat(events, hasSize(1));
+        assertThat(read(events.get(0), "$.metadata"), not(hasKey("partition_compaction_key")));
+    }
+
     @Test
     public void testErrorInPayloadDeserializationIsHandledGracefully() throws IOException {
         String payloadString = mapper.writeValueAsString(Fixture.mockPayload(42, "bla"));