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,39 @@
+/**
+ * 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. 
+     * 
+     * @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. 
+     * 
+     * @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,26 @@ gradleLombokPluginVersion=3.2.0
 gradleSshPluginVersion=2.9.0
 guavaVersion=28.1-jre
 javaxServletApiVersion=4.0.0
-jacksonVersion=2.9.9.3
+jacksonVersion=2.10.3
 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
 
 # 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,39 @@
+/**
+ * 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;
+import lombok.SneakyThrows;
+
+/**
+ * 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() {
+    }
+
+    @SneakyThrows
+    public static String getGroupId(Type type, String... args) {
+        switch (type) {
+            case QualifiedStreamName:
+                Preconditions.checkNotNull(args);
+                Preconditions.checkArgument(args.length == 2);
+                return NameUtils.getScopedStreamName(args[0], args[1]);
+            default:
+                throw new IllegalArgumentException();
+        }
+    }
+
+    public enum Type {
+        QualifiedStreamName,
+    }
+}

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

@@ -0,0 +1,23 @@
+/**
+ * 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 java.nio.ByteBuffer;
+
+/**
+ * Codec interface that defines methods to encode and decoder data for a given codec type.
+ */
+public interface Codec {
+    String getCodecType();
+
+    ByteBuffer encode(ByteBuffer data);
+
+    ByteBuffer decode(ByteBuffer data);
+}

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

@@ -0,0 +1,129 @@
+/**
+ * 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 lombok.SneakyThrows;
+import org.xerial.snappy.Snappy;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.zip.GZIPInputStream;
+import java.util.zip.GZIPOutputStream;
+
+/**
+ * Factory class for creating codecs for codec types . 
+ */
+public class CodecFactory {
+    public static final String NONE = "";
+    public static final String MIME_GZIP = "application/x-gzip";
+    public static final String MIME_SNAPPY = "application/x-snappy-framed";
+
+    private static final Noop NOOP = new Noop();
+    private static final GZipCodec GZIP_COMPRESSOR = new GZipCodec();
+    private static final SnappyCodec SNAPPY_COMPRESSOR = new SnappyCodec();
+
+    public static Codec none() {
+        return NOOP;
+    }
+
+    public static Codec gzip() {
+        return GZIP_COMPRESSOR;
+    }
+
+    public static Codec snappy() {
+        return SNAPPY_COMPRESSOR;
+    }
+
+    private static class Noop implements Codec {
+        @Override
+        public String getCodecType() {
+            return NONE;
+        }
+
+        @Override
+        public ByteBuffer encode(ByteBuffer data) {
+            return data;
+        }
+
+        @Override
+        public ByteBuffer decode(ByteBuffer data) {
+            return data;
+        }
+    }
+
+    private static class GZipCodec implements Codec {
+        @Override
+        public String getCodecType() {
+            return MIME_GZIP;
+        }
+
+        @SneakyThrows(IOException.class)
+        @Override
+        public ByteBuffer encode(ByteBuffer data) {
+            byte[] array = new byte[data.remaining()];
+            data.get(array);
+
+            ByteArrayOutputStream bos = new ByteArrayOutputStream(array.length);
+            GZIPOutputStream gzipOS = new GZIPOutputStream(bos);
+            gzipOS.write(array);
+            gzipOS.close();
+            byte[] compressed = bos.toByteArray();
+            return ByteBuffer.wrap(compressed);
+        }
+
+        @SneakyThrows(IOException.class)
+        @Override
+        public ByteBuffer decode(ByteBuffer data) {
+            byte[] array = new byte[data.remaining()];
+            data.get(array);
+
+            ByteArrayInputStream bis = new ByteArrayInputStream(array);
+            ByteArrayOutputStream bos = new ByteArrayOutputStream();
+            GZIPInputStream gzipIS = new GZIPInputStream(bis);
+            byte[] buffer = new byte[1024];
+            int len;
+            while ((len = gzipIS.read(buffer)) != -1) {
+                bos.write(buffer, 0, len);
+            }
+            byte[] uncompressed = bos.toByteArray();
+            return ByteBuffer.wrap(uncompressed);
+        }
+    }
+
+    private static class SnappyCodec implements Codec {
+        @Override
+        public String getCodecType() {
+            return MIME_SNAPPY;
+        }
+
+        @SneakyThrows(IOException.class)
+        @Override
+        public ByteBuffer encode(ByteBuffer data) {
+            byte[] array = new byte[data.remaining()];
+            data.get(array);
+
+            byte[] compressed = Snappy.compress(array);
+            return ByteBuffer.wrap(compressed);
+        }
+
+        @SneakyThrows(IOException.class)
+        @Override
+        public ByteBuffer decode(ByteBuffer data) {
+            byte[] array = new byte[data.remaining()];
+            data.get(array);
+
+            byte[] uncompressed = Snappy.uncompress(array);
+            return ByteBuffer.wrap(uncompressed);
+        }
+    }
+
+}