Issue 40: Serializers Implementation for Schema Registry

build.gradle

@@ -195,6 +195,31 @@ def getProjectVersion() {
     return ver
 }
 
+project('serializers') {
+    dependencies {
+        compile project(':common')
+        compile project(':client')
+        compile group: 'org.apache.avro', name: 'avro', version: avroVersion
+        compile group: 'com.google.protobuf', name: 'protobuf-java', version: protobufProtocVersion
+        compile group: 'com.google.protobuf', name: 'protobuf-java-util', version: protobufUtilVersion
+        compile group: 'io.pravega', name: 'pravega-client', version: pravegaVersion
+        compile group: 'org.xerial.snappy', name: 'snappy-java', version: snappyVersion
+        compile group: 'com.fasterxml.jackson.module', name: 'jackson-module-jsonSchema', version: jacksonVersion
+        testCompile group: 'org.slf4j', name: 'log4j-over-slf4j', version: slf4jApiVersion
+        testCompile group: 'ch.qos.logback', name: 'logback-classic', version: qosLogbackVersion
+        testCompile group: 'io.pravega', name: 'pravega-test-testcommon', version: pravegaVersion
+    }
+
+    javadoc {
+        title = "Serializers"
+        dependsOn delombok
+        source = delombok.outputDir
+        failOnError = false
+        exclude "**/impl/**";
+        options.addBooleanOption("Xdoclint:all,-reference", true)
+    }
+}
+
 apply plugin: 'distribution'
 
 task sourceCopy(type: Copy) {
@@ -297,3 +322,4 @@ class DockerPushTask extends Exec {
         }
     }
 }
+

checkstyle/spotbugs-exclude.xml

@@ -5,6 +5,9 @@
     <Match> <!-- generated code -->
         <Package name="io.pravega.schemaregistry.test.integrationtest.generated" />
     </Match>
+    <Match> <!-- generated code -->
+        <Package name="io.pravega.schemaregistry.testobjs.generated" />
+    </Match>
     <Match> <!-- does not work well with futures -->
         <Bug pattern="NP_NONNULL_PARAM_VIOLATION" />
     </Match>

common/src/main/java/io/pravega/schemaregistry/common/NameUtil.java

@@ -0,0 +1,45 @@
+/**
+ * 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.common;
+
+public class NameUtil {
+    /**
+     * Extracts the name from the fully qualified type name. Name represents the last token after ".". 
+     * If the qualified name does not contain "." then the name is same as qualified name. 
+     * 
+     * Example: io.pravega.MyObject will return MyObject
+     * Example: MyObject will return MyObject
+     * 
+     * @param qualifiedName qualified name to extract name from. 
+     * @return extracted name. 
+     */
+    public static String extractName(String qualifiedName) {
+        int nameStart = qualifiedName.lastIndexOf(".");
+        return qualifiedName.substring(nameStart + 1);
+    }
+
+    /**
+     * Extracts name and the prefix qualifier before the name. Name represents the last token after ".".
+     * Qualifier is the prefix before the name. 
+     * If the qualified name does not contain "." then the name is same as qualified name and qualifier is empty string.
+     * 
+     * Example: io.pravega.MyObject will return ["MyObject", "io.pravega"]
+     * Example: MyObject will return ["MyObject", ""]
+     * 
+     * @param qualifiedName qualified name to extract tokens from. 
+     * @return an array containing name at index 0 and qualifier at index 1. 
+     */
+    public static String[] extractNameAndQualifier(String qualifiedName) {
+        int nameStart = qualifiedName.lastIndexOf(".");
+        String name = qualifiedName.substring(nameStart + 1);
+        String pckg = nameStart < 0 ? "" : qualifiedName.substring(0, nameStart);
+        return new String[]{name, pckg};
+    }
+}

gradle.properties

@@ -26,24 +26,27 @@ gradleLombokPluginVersion=3.2.0
 gradleSshPluginVersion=2.9.0
 guavaVersion=28.1-jre
 javaxServletApiVersion=4.0.0
-jacksonVersion=2.9.9.3
+jacksonVersion=2.11.1
 javaxwsrsApiVersion=2.1
 jaxbVersion=2.3.0
 javaxAnnotationVersion=1.3.2
-jerseyVersion=2.29
+jerseyVersion=2.30
 junitVersion=4.12
 lombokVersion=1.18.4
 mockitoVersion=2.23.0
 jacocoVersion=0.8.2
 protobufGradlePlugin=0.8.9
-protobufProtocVersion=3.5.1
+protobufProtocVersion=3.11.4
+protobufUtilVersion=3.11.4
 qosLogbackVersion=1.2.3
 swaggerJersey2JaxrsVersion=1.5.22
 slf4jApiVersion=1.7.25
 gradleGitPluginVersion=2.2.0
 avroVersion=1.9.1
 avroProtobufVersion=1.7.7
-pravegaVersion=0.6.2
+snappyVersion=1.1.7.3
+pravegaVersion=0.7.0
+pravegaKeyCloakVersion=0.7.0
 
 # Version and base tags can be overridden at build time
 schemaregistryVersion=0.0.1-SNAPSHOT

serializers/src/main/java/io/pravega/schemaregistry/GroupIdGenerator.java

@@ -0,0 +1,37 @@
+/**
+ * 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;
+
+import com.google.common.base.Preconditions;
+import io.pravega.shared.NameUtils;
+
+/**
+ * Defines strategies for generating groupId for stream. 
+ * Currently there is only one naming strategy that uses streams fully qualified scoped stream name.
+ */
+public class GroupIdGenerator {
+    private GroupIdGenerator() {
+    }
+
+    public static String getGroupId(Scheme scheme, String... args) {
+        switch (scheme) {
+            case QualifiedStreamName:
+                Preconditions.checkNotNull(args);
+                Preconditions.checkArgument(args.length == 2, "Both scope and stream name should be supplied.");
+                return NameUtils.getScopedStreamName(args[0], args[1]);
+            default:
+                throw new IllegalArgumentException("Unknown Group id generation schema.");
+        }
+    }
+
+    public enum Scheme {
+        QualifiedStreamName,
+    }
+}

serializers/src/main/java/io/pravega/schemaregistry/codec/Codec.java

@@ -0,0 +1,51 @@
+/**
+ * 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.codec;
+
+import io.pravega.schemaregistry.contract.data.CodecType;
+import io.pravega.schemaregistry.contract.data.EncodingInfo;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+/**
+ * Codec interface that defines methods to encode and decoder data for a given codec type.
+ */
+public interface Codec {
+    /**
+     * Codec Type object that contains a string name identifying the Codec Type. 
+     * This name should be same as the codecType that is registered for the group in schema registry service. 
+     * The serializers will use this codec to encode the data and deserializers will find
+     * the decoder for the encoded data from {@link EncodingInfo#getCodecType()}
+     * 
+     * @return Name of the codec. 
+     */
+    CodecType getCodecType();
+
+    /**
+     * Implementation should encode the remaining bytes in the buffer and return a new ByteBuffer that includes
+     * the encoded data at its current position. 
+     * 
+     * @param data ByteBuffer to encode. 
+     * @return encoded ByteBuffer with position set to the start of encoded data. 
+     * @throws IOException IOException can be thrown while reading from or writing to byte buffers.
+     */
+    ByteBuffer encode(ByteBuffer data) throws IOException;
+
+    /**
+     * Implementation should decode the remaining bytes in the buffer and return a new ByteBuffer that includes
+     * the decoded data at its current position. 
+     *
+     * @param data encoded ByteBuffer to decode. 
+     * @return decoded ByteBuffer with position set to the start of decoded data. 
+     * @throws IOException can be thrown while reading from or writing to byte buffers.
+     */
+    ByteBuffer decode(ByteBuffer data) throws IOException;
+}

serializers/src/main/java/io/pravega/schemaregistry/schemas/AvroSchema.java

@@ -0,0 +1,141 @@
+/**
+ * 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.schemas;
+
+import com.google.common.base.Charsets;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
+import io.pravega.schemaregistry.contract.data.SchemaInfo;
+import io.pravega.schemaregistry.contract.data.SerializationFormat;
+import lombok.Getter;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.reflect.ReflectData;
+import org.apache.avro.specific.SpecificData;
+import org.apache.avro.specific.SpecificRecordBase;
+
+import java.nio.ByteBuffer;
+
+/**
+ * Container class for Avro Schema.
+ *
+ * @param <T> Type of element. 
+ */
+public class AvroSchema<T> implements Schema<T> {
+    @Getter
+    private final org.apache.avro.Schema schema;
+    private final SchemaInfo schemaInfo;
+    @Getter
+    private final Class<T> tClass;
+
+    private AvroSchema(org.apache.avro.Schema schema, Class<T> tClass) {
+        this.schema = schema;
+        this.schemaInfo = new SchemaInfo(schema.getFullName(),
+                SerializationFormat.Avro, getSchemaBytes(), ImmutableMap.of());
+        this.tClass = tClass;
+    }
+
+    private AvroSchema(SchemaInfo schemaInfo) {
+        String schemaString = new String(schemaInfo.getSchemaData().array(), Charsets.UTF_8);
+        this.schema = new org.apache.avro.Schema.Parser().parse(schemaString);
+        this.schemaInfo = schemaInfo;
+        this.tClass = null;
+    }
+
+    /**
+     * Method to create a typed AvroSchema for the given class. It extracts the avro schema from the class.
+     * For Avro generated classes, the schema is retrieved from the class. 
+     * For POJOs the schema is extracted using avro's {@link ReflectData}. 
+     *
+     * @param tClass Class whose object's schema is used.
+     * @param <T> Type of the Java class. 
+     * @return {@link AvroSchema} with generic type T that extracts and captures the avro schema. 
+     */
+    public static <T> AvroSchema<T> of(Class<T> tClass) {
+        org.apache.avro.Schema schema;
+        if (SpecificRecordBase.class.isAssignableFrom(tClass)) {
+            schema = SpecificData.get().getSchema(tClass);
+        } else {
+            schema = ReflectData.get().getSchema(tClass);
+        }
+        return new AvroSchema<>(schema, tClass);
+    }
+
+    /**
+     * Method to create a typed AvroSchema of type {@link Object} from the given schema. 
+     * This schema can be used to express any non record schema.
+     *
+     * @param schema Schema to use. 
+     * @return Returns an AvroSchema with {@link GenericRecord} type. 
+     */
+    public static AvroSchema<Object> of(org.apache.avro.Schema schema) {
+        return new AvroSchema<>(schema, Object.class);
+    }
+
+    /**
+     * Method to create a typed AvroSchema of type {@link GenericRecord} from the given schema. 
+     *
+     * @param schema Schema to use. 
+     * @return Returns an AvroSchema with {@link GenericRecord} type. 
+     */
+    public static AvroSchema<GenericRecord> ofRecord(org.apache.avro.Schema schema) {
+        Preconditions.checkArgument(schema.getType().equals(org.apache.avro.Schema.Type.RECORD));
+        return new AvroSchema<>(schema, GenericRecord.class);
+    }
+
+    /**
+     * It is same as {@link #of(Class)} except that it generates an AvroSchema typed as T. 
+     *
+     * This is useful for supplying a map of Avro schemas for multiplexed serializers and deserializers. 
+     *
+     * @param tDerived Class whose schema should be used.
+     * @param tBase Base class for the typed AvroSchema object.
+     * @param <T> Type of base class. 
+     * @return Returns an AvroSchema with T type. 
+     */
+    public static <T> AvroSchema<T> ofBaseType(Class<? extends T> tDerived, Class<T> tBase) {
+        Preconditions.checkArgument(tBase.isAssignableFrom(tDerived));
+
+        return new AvroSchema<>(ReflectData.get().getSchema(tDerived), tBase);
+    }
+
+    /**
+     * It is same as {@link #of(Class)} except that it generates an AvroSchema typed as {@link SpecificRecordBase}. 
+     *
+     * This is useful for supplying a map of Avro schemas for multiplexed serializers and deserializers. 
+     *
+     * @param tClass Class whose schema should be used.
+     * @param <T> Type of class whose schema is to be used. 
+     * @return Returns an AvroSchema with {@link SpecificRecordBase} type. 
+     */
+    public static <T extends SpecificRecordBase> AvroSchema<SpecificRecordBase> ofSpecificRecord(Class<T> tClass) {
+        Preconditions.checkArgument(SpecificRecordBase.class.isAssignableFrom(tClass));
+
+        return new AvroSchema<>(SpecificData.get().getSchema(tClass), SpecificRecordBase.class);
+    }
+
+    /**
+     * Method to create a typed AvroSchema of type {@link GenericRecord} from schema info. 
+     *
+     * @param schemaInfo Schema info object that has schema data in binary form.  
+     * @return Returns an AvroSchema with {@link GenericRecord} type. 
+     */
+    public static AvroSchema<Object> from(SchemaInfo schemaInfo) {
+        return new AvroSchema<>(schemaInfo);
+    }
+
+    private ByteBuffer getSchemaBytes() {
+        return ByteBuffer.wrap(schema.toString().getBytes(Charsets.UTF_8));
+    }
+
+    @Override
+    public SchemaInfo getSchemaInfo() {
+        return schemaInfo;
+    }
+}