Issue 63: Add precondition checks to service model objects (#65)

* Issue 63: Add precondition checks

Signed-off-by: Shivesh Ranjan <[email protected]>

build.gradle

@@ -191,7 +191,7 @@ project('contract') {
         compile group: 'org.glassfish.jersey.media', name: 'jersey-media-json-jackson', version: jerseyVersion
         compile group: 'javax.xml.bind', name: 'jaxb-api', version: jaxbVersion
         compile group: 'org.glassfish.jaxb', name: 'jaxb-runtime', version: jaxbVersion
-
+        testCompile group: 'io.pravega', name: 'pravega-test-testcommon', version: pravegaVersion
     }
 
     javadoc {

contract/src/main/java/io/pravega/schemaregistry/contract/data/CodecType.java

@@ -14,32 +14,38 @@
 import io.pravega.common.ObjectBuilder;
 import lombok.Builder;
 import lombok.Data;
+import lombok.NonNull;
 
 /**
  * Encapsulates properties of a codecType.  
  */
 @Data
 @Builder
 public class CodecType {
+    public static final CodecType NONE = new CodecType("");
+    private static final int MAX_PROPERTIES_SIZE = 921600; // 900 kb
+
     /**
      * Name that identifies the codec type. Users could typically use the mime type name for the encoding.   
      */
-    private final String name;
+    private @NonNull final String name;
     /**
      * User defined key value strings that users can use to add any additional metadata to the codecType. 
      * This can be used to share additional information with the decoder about how to decode, for example, if codecType was
      * for encryption, the additional information could include algorithm and other params required for decryption.
      * This is opaque to the service and stored with the codecType when the codec is registered for a group and delivered
      * with encoding information. 
      */
-    private final ImmutableMap<String, String> properties;
+    private @NonNull final ImmutableMap<String, String> properties;
 
     public CodecType(String name) {
         this(name, ImmutableMap.of());
     }
 
-    public CodecType(String name, ImmutableMap<String, String> properties) {
-        Preconditions.checkArgument(name != null);
+    public CodecType(@NonNull String name, @NonNull ImmutableMap<String, String> properties) {
+        Preconditions.checkArgument(properties.entrySet().stream().mapToInt(x -> x.getKey().length() + x.getValue().length())
+                                  .reduce(0, Integer::sum) < MAX_PROPERTIES_SIZE,
+                "Invalid properties, make sure that total size of properties map is less than 900 kb.");
         this.name = name;
         this.properties = properties;
     }

contract/src/main/java/io/pravega/schemaregistry/contract/data/Compatibility.java

@@ -13,6 +13,7 @@
 import io.pravega.common.ObjectBuilder;
 import lombok.Builder;
 import lombok.Data;
+import lombok.NonNull;
 
 /**
  * Defines different Compatibility policy options for schema evolution for schemas within a group.
@@ -26,15 +27,16 @@ public class Compatibility {
     /**
      * Enum that defines the Type of compatibility policy.
      */
-    private final Type type;
+    private @NonNull final Type type;
     private final BackwardAndForward backwardAndForward;
 
     private Compatibility(Type type) {
         this(type, null);
     }
 
-    private Compatibility(Type type, BackwardAndForward backwardAndForward) {
-        Preconditions.checkArgument(!type.equals(Type.Advanced) || backwardAndForward != null);
+    private Compatibility(@NonNull Type type, BackwardAndForward backwardAndForward) {
+        Preconditions.checkArgument(!type.equals(Type.Advanced) || backwardAndForward != null, 
+                "For advanced type, At lease one of backward or forward policy should be supplied.");
         this.type = type;
         this.backwardAndForward = backwardAndForward;
     }

contract/src/main/java/io/pravega/schemaregistry/contract/data/EncodingInfo.java

@@ -10,6 +10,7 @@
 package io.pravega.schemaregistry.contract.data;
 
 import lombok.Data;
+import lombok.NonNull;
 
 /**
  * Encoding Info describes the details of encoding for each event payload. Each combination of schema version and codec type
@@ -21,13 +22,13 @@ public class EncodingInfo {
     /**
      * Version of the schema which is used in encoding the data. 
      */
-    private final VersionInfo versionInfo;
+    private @NonNull final VersionInfo versionInfo;
     /**
      * Actual schema which is used in encoding the data. 
      */
-    private final SchemaInfo schemaInfo;
+    private @NonNull final SchemaInfo schemaInfo;
     /**
      * Codec type which is used in encoding the data. 
      */
-    private final CodecType codecType;
+    private @NonNull final CodecType codecType;
 }

contract/src/main/java/io/pravega/schemaregistry/contract/data/GroupHistoryRecord.java

@@ -10,6 +10,7 @@
 package io.pravega.schemaregistry.contract.data;
 
 import lombok.Data;
+import lombok.NonNull;
 
 /**
  * Describes changes to the group and the compatibility {@link GroupHistoryRecord#compatibility} that were 
@@ -24,15 +25,15 @@ public class GroupHistoryRecord {
     /**
      * Schema information object for the schemaInfo that was added to the group.
      */
-    private final SchemaInfo schemaInfo;
+    private @NonNull final SchemaInfo schemaInfo;
     /**
      * Version information object that uniquely identifies the schemaInfo in the group. 
      */
-    private final VersionInfo versionInfo;
+    private @NonNull final VersionInfo versionInfo;
     /**
      * Compatibility applied at the time when the schemaInfo was registered. 
      */
-    private final Compatibility compatibility;
+    private @NonNull final Compatibility compatibility;
     /**
      * Service's Time when the schemaInfo was registered. 
      */

contract/src/main/java/io/pravega/schemaregistry/contract/data/GroupProperties.java

@@ -9,9 +9,11 @@
  */
 package io.pravega.schemaregistry.contract.data;
 
+import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableMap;
 import lombok.Builder;
 import lombok.Data;
+import lombok.NonNull;
 
 /**
  * Different configuration choices for a group. 
@@ -35,11 +37,11 @@ public class GroupProperties {
     /**
      * Serialization format allowed for the group. 
      */
-    private final SerializationFormat serializationFormat;
+    private @NonNull final SerializationFormat serializationFormat;
     /**
      * Compatibility to be applied for the group. 
      */
-    private final Compatibility compatibility;
+    private @NonNull final Compatibility compatibility;
     /**
      * Flag to indicate whether multiple types of schemas can be added to the group or not. If set to false, all schemas
      * added to the group should have the same {@link SchemaInfo#type}.
@@ -48,13 +50,17 @@ public class GroupProperties {
     /**
      * User defined key value strings for any metadata they want to associate with the group. 
      */
-    private final ImmutableMap<String, String> properties;
+    private @NonNull final ImmutableMap<String, String> properties;
 
     public GroupProperties(SerializationFormat serializationFormat, Compatibility compatibility, boolean allowMultipleTypes) {
         this(serializationFormat, compatibility, allowMultipleTypes, ImmutableMap.of());
     }
 
-    public GroupProperties(SerializationFormat serializationFormat, Compatibility compatibility, boolean allowMultipleTypes, ImmutableMap<String, String> properties) {
+    public GroupProperties(@NonNull SerializationFormat serializationFormat, @NonNull Compatibility compatibility, boolean allowMultipleTypes,
+                           @NonNull ImmutableMap<String, String> properties) {
+        Preconditions.checkArgument(properties != null && properties.size() <= 100 &&
+                        properties.entrySet().stream().allMatch(x -> x.getKey().length() <= 200 && x.getValue().length() <= 200),
+                "Invalid properties, make sure each key and value are less than or equal to 200 bytes and there are no more than 100 entries.");
         this.serializationFormat = serializationFormat;
         this.compatibility = compatibility;
         this.allowMultipleTypes = allowMultipleTypes;

contract/src/main/java/io/pravega/schemaregistry/contract/data/SchemaInfo.java

@@ -14,6 +14,7 @@
 import io.pravega.common.ObjectBuilder;
 import lombok.Builder;
 import lombok.Data;
+import lombok.NonNull;
 
 import java.nio.ByteBuffer;
 
@@ -31,26 +32,30 @@
 @Data
 @Builder
 public class SchemaInfo {
+    public static final int ONE_MB = 1024 * 1024;
     /**
      * Identifies the object type that is represented by the schema. 
      */
-    private final String type;
+    private @NonNull final String type;
     /**
      * Serialization format that this schema is intended to be used for. 
      */
-    private final SerializationFormat serializationFormat;
+    private @NonNull final SerializationFormat serializationFormat;
     /**
      * Schema as an array of 8-bit unsigned bytes. 
      */
-    private final ByteBuffer schemaData;
+    private @NonNull final ByteBuffer schemaData;
     /**
      * User defined key value strings that users can use to add any additional metadata to the schema. 
      */
-    private final ImmutableMap<String, String> properties;
+    private @NonNull final ImmutableMap<String, String> properties;
 
-    public SchemaInfo(String type, SerializationFormat serializationFormat, ByteBuffer schemaData, ImmutableMap<String, String> properties) {
-        Preconditions.checkArgument(type != null);
-        Preconditions.checkArgument(serializationFormat != SerializationFormat.Any);
+    public SchemaInfo(@NonNull String type, @NonNull SerializationFormat serializationFormat, @NonNull ByteBuffer schemaData, @NonNull ImmutableMap<String, String> properties) {
+        Preconditions.checkArgument(serializationFormat != SerializationFormat.Any, "Invalid Serialization Format.");
+        Preconditions.checkArgument(schemaData.remaining() <= 8 * ONE_MB, "Invalid schema binary.");
+        Preconditions.checkArgument(properties.size() <= 100 && 
+                        properties.entrySet().stream().allMatch(x -> x.getKey().length() <= 200 && x.getValue().length() <= 200),
+                "Invalid properties, make sure each key and value are less than or equal to 200 bytes and there are no more than 100 entries.");
         this.type = type;
         this.serializationFormat = serializationFormat;
         this.schemaData = schemaData;

contract/src/main/java/io/pravega/schemaregistry/contract/data/SchemaWithVersion.java

@@ -12,6 +12,7 @@
 import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Data;
+import lombok.NonNull;
 
 /**
  * Object that encapsulates schemaInfo with its associated version. 
@@ -23,9 +24,9 @@ public class SchemaWithVersion {
     /**
      * Schema Information object. 
      */
-    private final SchemaInfo schemaInfo;
+    private @NonNull final SchemaInfo schemaInfo;
     /**
      * Version information object that identifies the corresponding schema object. 
      */
-    private final VersionInfo versionInfo;
+    private @NonNull final VersionInfo versionInfo;
 }

contract/src/main/java/io/pravega/schemaregistry/contract/data/SerializationFormat.java

@@ -51,9 +51,9 @@ public static SerializationFormat custom(String fullTypeName) {
     /**
      * Method to create a serialization format with a full name.
      * 
-     * @param fullTypeName Custom type name. 
+     * @param fullTypeName Full type name. 
      * @param format Serialization format.
-     * @return {@link SerializationFormat#Custom} with supplied custom type name. 
+     * @return {@link SerializationFormat#fullTypeName} with supplied custom type name. 
      */
     public static SerializationFormat withName(SerializationFormat format, String fullTypeName) {
         Preconditions.checkArgument(format != null);

contract/src/main/java/io/pravega/schemaregistry/contract/data/VersionInfo.java

@@ -13,6 +13,7 @@
 import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Data;
+import lombok.NonNull;
 
 /**
  * Version information object that encapsulates properties that uniquely identify a specific version of a schema within a group.
@@ -34,7 +35,7 @@ public class VersionInfo {
      * Object type which is declared in the corresponding {@link SchemaInfo#type} for the schemainfo that is identified 
      * by this version info. 
      */
-    private final String type;
+    private @NonNull final String type;
     /**
      * A version number that identifies the position of schema among other schemas in the group that share the same 'type'.
      */

contract/src/test/java/io/pravega/schemaregistry/contract/transform/ContractTest.java

@@ -0,0 +1,104 @@
+/**
+ * Copyright (c) Dell Inc., or its subsidiaries. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.pravega.schemaregistry.contract.transform;
+
+import com.google.common.collect.ImmutableMap;
+import io.pravega.common.Exceptions;
+import io.pravega.schemaregistry.contract.data.CodecType;
+import io.pravega.schemaregistry.contract.data.Compatibility;
+import io.pravega.schemaregistry.contract.data.GroupProperties;
+import io.pravega.schemaregistry.contract.data.SchemaInfo;
+import io.pravega.schemaregistry.contract.data.SerializationFormat;
+import io.pravega.test.common.AssertExtensions;
+import org.junit.Test;
+
+import java.nio.ByteBuffer;
+import java.util.Base64;
+import java.util.Random;
+
+public class ContractTest {
+    private static final Random RANDOM = new Random();
+
+    @Test
+    public void testObjectCreation() {
+        ByteBuffer wrap = ByteBuffer.wrap(new byte[0]);
+        ImmutableMap.Builder<String, String> builder = new ImmutableMap.Builder<>();
+        for (int i = 0; i <= 100; i++) {
+            builder.put(Integer.toString(i), "a");
+        }
+        ImmutableMap<String, String> mapOf101Entries = builder.build();
+        builder = new ImmutableMap.Builder<>();
+        String bigString = getBigString(201);
+
+        builder.put(bigString, bigString);
+        ImmutableMap<String, String> mapOfBigSizedEntries = builder.build();
+
+        // schemainfo
+        AssertExtensions.assertThrows("Null type not allowed", 
+                () -> new SchemaInfo(null, SerializationFormat.Avro, wrap, ImmutableMap.of()),
+            e -> Exceptions.unwrap(e) instanceof NullPointerException);
+        AssertExtensions.assertThrows("Null format not allowed",
+                () -> new SchemaInfo("", null, wrap, ImmutableMap.of()),
+                e -> Exceptions.unwrap(e) instanceof NullPointerException);
+        AssertExtensions.assertThrows("SerializationFormat `Any` not allowed",
+                () -> new SchemaInfo("", SerializationFormat.Any, wrap, ImmutableMap.of()),
+                e -> Exceptions.unwrap(e) instanceof IllegalArgumentException);
+        AssertExtensions.assertThrows("Null data not allowed",
+                () -> new SchemaInfo("", SerializationFormat.Avro, null, ImmutableMap.of()),
+                e -> Exceptions.unwrap(e) instanceof NullPointerException);
+        AssertExtensions.assertThrows("Null properties not allowed",
+                () -> new SchemaInfo("", SerializationFormat.Avro, wrap, null),
+                e -> Exceptions.unwrap(e) instanceof NullPointerException);
+        AssertExtensions.assertThrows("Schema binary max size - 8mb",
+                () -> new SchemaInfo("", SerializationFormat.Avro, ByteBuffer.wrap(new byte[9 * 1024 * 1024]), ImmutableMap.of()),
+                e -> Exceptions.unwrap(e) instanceof IllegalArgumentException);
+        AssertExtensions.assertThrows("Schema properties less than 200 bytes each",
+                () -> new SchemaInfo("", SerializationFormat.Avro, wrap, mapOfBigSizedEntries),
+                e -> Exceptions.unwrap(e) instanceof IllegalArgumentException);
+        AssertExtensions.assertThrows("Max properties allowed is 100",
+                () -> new SchemaInfo("", SerializationFormat.Avro, wrap, mapOf101Entries),
+                e -> Exceptions.unwrap(e) instanceof IllegalArgumentException);
+
+        // group properties
+        AssertExtensions.assertThrows("Null format not allowed",
+                () -> new GroupProperties(null, Compatibility.backward(), false),
+                e -> Exceptions.unwrap(e) instanceof NullPointerException);
+        AssertExtensions.assertThrows("Null compatibility not allowed",
+                () -> new GroupProperties(SerializationFormat.Any, null, false),
+                e -> Exceptions.unwrap(e) instanceof NullPointerException);
+        AssertExtensions.assertThrows("Null properties not allowed",
+                () -> new GroupProperties(SerializationFormat.Any, Compatibility.backward(), false, null),
+                e -> Exceptions.unwrap(e) instanceof NullPointerException);
+        AssertExtensions.assertThrows("Group properties less than 200 bytes each",
+                () -> new GroupProperties(SerializationFormat.Avro, Compatibility.backward(), false, mapOfBigSizedEntries),
+                e -> Exceptions.unwrap(e) instanceof IllegalArgumentException);
+        AssertExtensions.assertThrows("Max properties allowed is 100",
+                () -> new GroupProperties(SerializationFormat.Avro, Compatibility.backward(), false, mapOf101Entries),
+                e -> Exceptions.unwrap(e) instanceof IllegalArgumentException);
+
+        // codec type
+        AssertExtensions.assertThrows("Null name not allowed",
+                () -> new CodecType(null, ImmutableMap.of()),
+                e -> Exceptions.unwrap(e) instanceof NullPointerException);
+        AssertExtensions.assertThrows("Codec properties less than 900 kb overall",
+                () -> new CodecType("", ImmutableMap.of(getBigString(1024 * 1024), "a")),
+                e -> Exceptions.unwrap(e) instanceof IllegalArgumentException);
+        AssertExtensions.assertThrows("Codec properties less than 900 kb overall",
+                () -> new CodecType("", ImmutableMap.of("a", getBigString(1024 * 1024))),
+                e -> Exceptions.unwrap(e) instanceof IllegalArgumentException);
+
+    }
+
+    private String getBigString(int size) {
+        byte[] array = new byte[size];
+        RANDOM.nextBytes(array);
+        return Base64.getEncoder().encodeToString(array);
+    }
+}