Merge pull request #892 from zalando/ordering-key-fields

Implementation of ordering key fields

CHANGELOG.md

@@ -6,6 +6,9 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 
 ## [Unreleased]
 
+### Added
+- Extended event type's definition to support ordering_key_field attribute
+
 ### Removed
 - Removed high-level API feature flag
 

docs/_data/nakadi-event-bus-api.yaml

@@ -2468,6 +2468,35 @@ definitions:
       authorization:
         $ref: '#/definitions/EventTypeAuthorization'
 
+      ordering_key_fields:
+        type: array
+        description: |
+          This is only an informational field. The events are delivered to consumers in the order they were published.
+          No reordering is done by Nakadi.
+
+          This field is useful in case the producer wants to communicate the complete order accross all the events
+          published to all partitions. This is the case when there is an incremental generator on the producer side,
+          for example.
+
+          It differs from `partition_key_fields` in the sense that it's not used for partitioning (known as sharding in
+          some systems). The order indicated by `ordering_key_fields` can also differ from the order the events are in
+          each partition, in case of out-of-order submission.
+
+          In most cases, this would have just a single item (the path of the field
+          by which this is to be ordered), but can have multiple items, in which case
+          those are considered as a compound key, with lexicographic ordering (first
+          item is most significant).
+        items:
+          type: string
+          description: |
+            Indicates a single ordering field. This is a dot separated string, which is applied
+            onto the whole event object, including the contained metadata and data (in
+            case of a data change event) objects.
+
+            The field must be present in the schema. This field can be modified at any moment, but event type owners are
+            expected to notify consumer in advance about the change.
+          example: "data.incremental_counter"
+
       audience:
         type: string
         x-extensible-enum:

src/acceptance-test/java/org/zalando/nakadi/webservice/UserJourneyAT.java

@@ -4,6 +4,7 @@
 import com.google.common.base.Charsets;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Lists;
 import com.google.common.io.Resources;
 import com.jayway.restassured.RestAssured;
 import com.jayway.restassured.response.Header;
@@ -90,9 +91,11 @@ public void userJourneyM1() throws InterruptedException, IOException {
                 .body("owning_application", equalTo(owningApp))
                 .body("category", equalTo("undefined"))
                 .body("audience", equalTo("external_public"))
+                .body("ordering_key_fields", equalTo(Lists.newArrayList("foo", "bar.baz")))
                 .body("schema.type", equalTo("json_schema"))
-                .body("schema.schema", equalTo("{\"type\": \"object\", \"properties\": " +
-                        "{\"foo\": {\"type\": \"string\"}}, \"required\": [\"foo\"]}"));
+                .body("schema.schema", equalTo("{\"type\": \"object\", \"properties\": {\"foo\": " +
+                        "{\"type\": \"string\"}, \"bar\": {\"type\": \"object\", \"properties\": " +
+                        "{\"baz\": {\"type\": \"string\"}}}}, \"required\": [\"foo\"]}"));
 
         // list event types
         jsonRequestSpec()

src/main/java/org/zalando/nakadi/domain/EventTypeBase.java

@@ -18,6 +18,7 @@
 public class EventTypeBase {
 
     private static final List<String> EMPTY_PARTITION_KEY_FIELDS = ImmutableList.of();
+    private static final List<String> EMPTY_ORDERING_KEY_FIELDS = ImmutableList.of();
 
     @NotNull
     @Pattern(regexp = "[a-zA-Z][-0-9a-zA-Z_]*(\\.[0-9a-zA-Z][-0-9a-zA-Z_]*)*", message = "format not allowed")
@@ -41,6 +42,9 @@ public class EventTypeBase {
     @Nullable
     private List<String> partitionKeyFields;
 
+    @Nullable
+    private List<String> orderingKeyFields;
+
     @Valid
     @NotNull
     private EventTypeSchemaBase schema;
@@ -106,6 +110,7 @@ public EventTypeBase(final EventTypeBase eventType) {
         this.setCompatibilityMode(eventType.getCompatibilityMode());
         this.setAuthorization(eventType.getAuthorization());
         this.setAudience(eventType.getAudience());
+        this.setOrderingKeyFields(eventType.getOrderingKeyFields());
     }
 
     public String getName() {
@@ -168,6 +173,14 @@ public void setPartitionKeyFields(final List<String> partitionKeyFields) {
         this.partitionKeyFields = partitionKeyFields;
     }
 
+    public List<String> getOrderingKeyFields() {
+        return unmodifiableList(orderingKeyFields != null ? orderingKeyFields : EMPTY_ORDERING_KEY_FIELDS);
+    }
+
+    public void setOrderingKeyFields(@Nullable final List<String> orderingKeyFields) {
+        this.orderingKeyFields = orderingKeyFields;
+    }
+
     public List<EnrichmentStrategyDescriptor> getEnrichmentStrategies() {
         return enrichmentStrategies;
     }

src/main/java/org/zalando/nakadi/service/EventTypeService.java

@@ -486,6 +486,8 @@ private void validateSchema(final EventTypeBase eventType) throws InvalidEventTy
                 validatePartitionKeys(schema, eventType);
             }
 
+            validateOrderingKeys(schema, eventType);
+
             if (eventType.getCompatibilityMode() == CompatibilityMode.COMPATIBLE) {
                 validateJsonSchemaConstraints(schemaAsJson);
             }
@@ -516,6 +518,16 @@ private void validatePartitionKeys(final Schema schema, final EventTypeBase even
         }
     }
 
+    private void validateOrderingKeys(final Schema schema, final EventTypeBase eventType)
+            throws InvalidEventTypeException, JSONException, SchemaException {
+        final List<String> absentFields = eventType.getOrderingKeyFields().stream()
+                .filter(field -> !schema.definesProperty(convertToJSONPointer(field)))
+                .collect(Collectors.toList());
+        if (!absentFields.isEmpty()) {
+            throw new InvalidEventTypeException("ordering_key_fields " + absentFields + " absent in schema");
+        }
+    }
+
     private String convertToJSONPointer(final String value) {
         return value.replaceAll("\\.", "/");
     }

src/test/resources/sample-event-type-update.json

@@ -6,7 +6,7 @@
   "schema": {
     "version": "1.0.0",
     "type": "json_schema",
-    "schema": "{\"type\": \"object\", \"properties\": {\"foo\": {\"type\": \"string\"}}, \"required\": [\"foo\"]}"
+    "schema": "{\"type\": \"object\", \"properties\": {\"foo\": {\"type\": \"string\"}, \"bar\": {\"type\": \"object\", \"properties\": {\"baz\": {\"type\": \"string\"}}}}, \"required\": [\"foo\"]}"
   },
   "options" : {
     "retention_time": 86400000

src/test/resources/sample-event-type.json

@@ -5,7 +5,8 @@
   "compatibility_mode": "compatible",
   "schema": {
     "type": "json_schema",
-    "schema": "{\"type\": \"object\", \"properties\": {\"foo\": {\"type\": \"string\"}}, \"required\": [\"foo\"]}"
+    "schema": "{\"type\": \"object\", \"properties\": {\"foo\": {\"type\": \"string\"}, \"bar\": {\"type\": \"object\", \"properties\": {\"baz\": {\"type\": \"string\"}}}}, \"required\": [\"foo\"]}"
   },
-  "audience": "external_public"
+  "audience": "external_public",
+  "ordering_key_fields": ["foo", "bar.baz"]
 }