Merge branch 'master' into json-compatibility

Author: shiveshr

.travis.yml

@@ -10,7 +10,7 @@
 language: java
 install: true
 jdk:
-- openjdk8
+- oraclejdk11
 
 env:
   global:

charts/schema-registry/Chart.yaml

@@ -12,6 +12,6 @@ apiVersion: v2
 name: schema-registry
 description: Schema Registry Helm chart for Kubernetes
 type: application
-version: 0.0.1
-appVersion: 0.0.1
+version: 0.1.0
+appVersion: 0.1.0
 home: https://github.com/pravega/schema-registry

charts/schema-registry/README.md

@@ -49,14 +49,16 @@ The following table lists the configurable parameters of the schema registry cha
 | Parameter | Description | Default |
 | ----- | ----------- | ------ |
 | `replicas` | Number of replicas | `2` |
-| `image.repository` | Repository for schema registry image | `pravega/schema-registry` |
-| `image.tag` | Tag for schema registry image | `0.0.1` |
+| `image.repository` | Repository for schema registry image | `pravega/schemaregistry` |
+| `image.tag` | Tag for schema registry image | `0.1.0` |
 | `image.pullPolicy` | Pull policy for schema registry image | `IfNotPresent` |
 | `serviceAccount.create` | Whether to create a service account | `true` |
 | `serviceAccount.name` | Name of the service account | `schema-registry` |
 | `serviceAccount.annotations` | Annotations to be added to the service account | |
 | `service.type` | Schema registry service type | `LoadBalancer` |
 | `service.port` | Schema registry service port | `9092` |
+| `initContainer` | Configuration for the init container | `{}` |
+| `container` | Additional configuration for the schema registry container, to be provided if an init container has been configured | `{}` |
 | `ingress.enabled` | Whether to expose as an ingress resource | `false` |
 | `ingress.path` | Path for the ingress | `/` |
 | `ingress.annotations` | Annotations for the ingress | `{}` |

charts/schema-registry/templates/configmap.yaml

@@ -32,10 +32,17 @@ data:
   "DISABLE_BASIC_AUTH": "{{ .Values.authentication.disableBasicAuthentication }}"
   {{- end }}
   {{- if .Values.authentication.authorizationResourceQualifier }}
-  "AUTHORIZATION_RESOURCE_QUALIFIER": "{{ .Values.authentication.authorizationResourceQualifier }}"
+  "AUTHORIZATION_DOMAIN_RESOURCE_IDENTIFIER": "{{ .Values.authentication.authResourceDomain }}"
+  "AUTHORIZATION_NAMESPACE_RESOURCE_IDENTIFIER": "{{ .Values.authentication.authResourceNamespace }}"
+  "AUTHORIZATION_GROUP_RESOURCE_IDENTIFIER": "{{ .Values.authentication.authorizationResourceGroup }}"
   {{- end }}
   {{- if .Values.tls.enabled }}
   "TLS_CERT_FILE": "/etc/secret-volume/{{ .Values.tls.certFile }}"
   "TLS_KEY_FILE": "/etc/secret-volume/{{ .Values.tls.keyFile }}"
   "TLS_KEY_PASSWORD_FILE": "/etc/secret-volume/{{ .Values.tls.keyPasswordFile }}"
   {{- end }}
+  {{- if and .Values.container .Values.container.command }}
+  "PRE_START_CMD": "{{ .Values.container.command }}"
+  {{- else }}
+  "PRE_START_CMD": ""
+  {{- end }}

charts/schema-registry/templates/deployment.yaml

@@ -46,7 +46,7 @@ spec:
                 - sh
                 - -c
                 - netstat -ltn 2> /dev/null | grep {{ .Values.service.port }} || ss -ltn 2> /dev/null | grep {{ .Values.service.port }}
-          {{- if or .Values.authentication.enabled .Values.tls.enabled }}
+          {{- if or .Values.authentication.enabled .Values.tls.enabled (and .Values.container .Values.container.volume) }}
           volumeMounts:
           {{- if .Values.authentication.enabled }}
           - name: auth-passwd-secret
@@ -58,8 +58,29 @@ spec:
             mountPath: /etc/secret-volume
             readOnly: true
           {{- end }}
+          {{- if and .Values.container .Values.container.volume }}
+          - name: transfer-data
+            mountPath: {{ .Values.container.volume.mountPath }}
           {{- end }}
-      {{- if or .Values.authentication.enabled .Values.tls.enabled }}
+          {{- end }}
+      {{- if .Values.initContainer }}
+      initContainers:
+        - name: {{ .Values.initContainer.name }}
+          image: "{{ .Values.initContainer.image.repository }}:{{ .Values.initContainer.image.tag }}"
+          imagePullPolicy: {{ .Values.initContainer.image.pullPolicy }}
+          command:
+{{ toYaml .Values.initContainer.command | indent 12 }}
+          {{- if .Values.initContainer.resources }}
+          resources:
+{{ toYaml .Values.initContainer.command | indent 12 }}
+          {{- end }}
+          {{- if .Values.initContainer.volume }}
+          volumeMounts:
+          - name: transfer-data
+            mountPath: {{ .Values.initContainer.volume.mountPath }}
+          {{- end }}
+      {{- end }}
+      {{- if or .Values.authentication.enabled .Values.tls.enabled (and .Values.initContainer .Values.initContainer.volume) }}
       volumes:
       {{- if .Values.authentication.enabled }}
       - name: auth-passwd-secret
@@ -71,4 +92,8 @@ spec:
         secret:
           secretName: {{ .Values.tls.secret }}
       {{- end }}
+      {{- if and .Values.container .Values.container.volume }}
+      - name: transfer-data
+        emptyDir: {}
+      {{- end }}
       {{- end }}

charts/schema-registry/values.yaml

@@ -15,8 +15,8 @@
 replicas: 2
 
 image:
-  repository: pravega/schema-registry
-  tag: 0.0.1
+  repository: pravega/schemaregistry
+  tag: 0.1.0
   pullPolicy: IfNotPresent
 
 ## Service account name and whether to create it.
@@ -29,6 +29,26 @@ service:
   type: LoadBalancer
   port: 9092
 
+initContainer: {}
+  # name: ""
+  # image:
+  #   repository:
+  #   tag:
+  #   pullPolicy: IfNotPresent
+  ## provide a list containing the command that needs to be executed within the init container
+  ## e.g. ["/bin/sh", "-c", "cp /data/* /target"]
+  # command: []
+  # resources: {}
+  # volume:
+  #   mountPath: /target
+
+container: {}
+  ## just provide the command to be executed within the entrypoint
+  ## e.g. "cp /additional_libs/* /opt/schema-registry/lib/"
+  # command: ""
+  # volume:
+  #   mountPath: /additional_libs
+
 ## Ingress configuration.
 ingress:
   enabled: false
@@ -67,8 +87,10 @@ authentication:
   ## authentication is enabled
   passwordAuthSecret:
   userPasswordFile:
-  authorizationResourceQualifier: ""
   disableBasicAuthentication: false
+  authResourceDomain: "prn"
+  authNamespaceResourceIdentifier: "namespace"
+  authGroupResourceIdentifier: "group"
 
 tls:
   enabled: false

docker/pravega-schema-registry/scripts/entrypoint.sh

@@ -21,4 +21,8 @@ if [ ${WAIT_FOR} ];then
     ${SCRIPTS_DIR}/wait_for
 fi
 
+if [[ -n "$PRE_START_CMD" ]];then
+    $PRE_START_CMD
+fi
+
 exec /opt/schema-registry/bin/schema-registry

gradle.properties

@@ -47,8 +47,8 @@ gradleGitPluginVersion=2.2.0
 avroVersion=1.9.1
 avroProtobufVersion=1.7.7
 snappyVersion=1.1.7.3
-pravegaVersion=0.8.0-2623.279ac21-SNAPSHOT
-pravegaKeyCloakVersion=0.7.0
+pravegaVersion=0.9.0-2664.5410d55-SNAPSHOT
+pravegaKeyCloakVersion=0.8.0
 
 # Version and base tags can be overridden at build time
 schemaregistryVersion=0.2.0-SNAPSHOT

serializers/avro/src/main/java/io/pravega/schemaregistry/serializer/avro/impl/AvroDeserializer.java

@@ -9,8 +9,9 @@
  */
 package io.pravega.schemaregistry.serializer.avro.impl;
 
-import com.google.common.base.Charsets;
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
 import io.pravega.schemaregistry.serializer.avro.schemas.AvroSchema;
 import io.pravega.schemaregistry.client.SchemaRegistryClient;
 import io.pravega.schemaregistry.contract.data.SchemaInfo;
@@ -19,48 +20,59 @@
 import io.pravega.schemaregistry.serializer.shared.impl.SerializerConfig;
 import org.apache.avro.Schema;
 import org.apache.avro.io.BinaryDecoder;
+import org.apache.avro.io.DatumReader;
 import org.apache.avro.io.DecoderFactory;
 import org.apache.avro.reflect.ReflectDatumReader;
 import org.apache.avro.specific.SpecificDatumReader;
 import org.apache.avro.specific.SpecificRecordBase;
 
 import java.io.IOException;
 import java.io.InputStream;
+import java.nio.ByteBuffer;
 import java.util.concurrent.ConcurrentHashMap;
 
 class AvroDeserializer<T> extends AbstractDeserializer<T> {
-    private final AvroSchema<T> avroSchema;
-    private final ConcurrentHashMap<SchemaInfo, Schema> knownSchemas;
+    private final ConcurrentHashMap<ByteBuffer, DatumReader<T>> knownSchemaReaders;
+    private final boolean specific;
+    private final Schema readerSchema;
 
     AvroDeserializer(String groupId, SchemaRegistryClient client,
                      AvroSchema<T> schema,
                      SerializerConfig.Decoders decoder, EncodingCache encodingCache) {
         super(groupId, client, schema, false, decoder, encodingCache, true);
         Preconditions.checkNotNull(schema);
-        this.avroSchema = schema;
-        this.knownSchemas = new ConcurrentHashMap<>();
+        this.knownSchemaReaders = new ConcurrentHashMap<>();
+        specific = SpecificRecordBase.class.isAssignableFrom(schema.getTClass());
+        readerSchema = schema.getSchema();
+        ByteBuffer schemaData = schema.getSchemaInfo().getSchemaData();
+        knownSchemaReaders.put(schemaData, createDatumReader(readerSchema, readerSchema, specific));
     }
 
     @Override
     public final T deserialize(InputStream inputStream, SchemaInfo writerSchemaInfo, SchemaInfo readerSchemaInfo) throws IOException {
         Preconditions.checkNotNull(writerSchemaInfo);
-        Schema writerSchema;
-        if (knownSchemas.containsKey(writerSchemaInfo)) {
-            writerSchema = knownSchemas.get(writerSchemaInfo);
-        } else {
-            String schemaString = new String(writerSchemaInfo.getSchemaData().array(), Charsets.UTF_8);
-            writerSchema = new Schema.Parser().parse(schemaString);
-            knownSchemas.put(writerSchemaInfo, writerSchema);
-        }
-        Schema readerSchema = avroSchema.getSchema();
+        final ByteBuffer writerSchemaData = writerSchemaInfo.getSchemaData();
+        DatumReader<T> datumReader = knownSchemaReaders.computeIfAbsent(writerSchemaData, key -> {
+            Schema writerSchema = AvroSchema.from(writerSchemaInfo).getSchema();
+            return createDatumReader(writerSchema, this.readerSchema, specific);
+        });
         BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(inputStream, null);
-        
-        if (SpecificRecordBase.class.isAssignableFrom(avroSchema.getTClass())) {
-            SpecificDatumReader<T> datumReader = new SpecificDatumReader<>(writerSchema, readerSchema);
-            return datumReader.read(null, decoder);
+        return datumReader.read(null, decoder);
+    }
+
+    @VisibleForTesting
+     DatumReader<T> createDatumReader(Schema writerSchema, Schema readerSchema, boolean specific) {
+        DatumReader<T> datumReader;
+        if (specific) {
+            datumReader = new SpecificDatumReader<>(writerSchema, readerSchema);
         } else {
-            ReflectDatumReader<T> datumReader = new ReflectDatumReader<>(writerSchema, readerSchema);
-            return datumReader.read(null, decoder);
+            datumReader = new ReflectDatumReader<>(writerSchema, readerSchema);
         }
+        return datumReader;
+    }
+
+    @VisibleForTesting
+    ImmutableMap<ByteBuffer, DatumReader<T>> getKnownSchemaReaders() {
+        return ImmutableMap.copyOf(knownSchemaReaders);
     }
 }

serializers/avro/src/main/java/io/pravega/schemaregistry/serializer/avro/impl/AvroGenericDeserializer.java

@@ -9,41 +9,49 @@
  */
 package io.pravega.schemaregistry.serializer.avro.impl;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
-import io.pravega.schemaregistry.serializer.avro.schemas.AvroSchema;
+import com.google.common.collect.ImmutableMap;
 import io.pravega.schemaregistry.client.SchemaRegistryClient;
 import io.pravega.schemaregistry.contract.data.SchemaInfo;
+import io.pravega.schemaregistry.serializer.avro.schemas.AvroSchema;
 import io.pravega.schemaregistry.serializer.shared.impl.AbstractDeserializer;
 import io.pravega.schemaregistry.serializer.shared.impl.EncodingCache;
 import io.pravega.schemaregistry.serializer.shared.impl.SerializerConfig;
 import org.apache.avro.Schema;
 import org.apache.avro.generic.GenericDatumReader;
 import org.apache.avro.io.BinaryDecoder;
 import org.apache.avro.io.DecoderFactory;
-
+import org.apache.commons.lang3.tuple.Pair;
 import javax.annotation.Nullable;
 import java.io.IOException;
 import java.io.InputStream;
 import java.util.concurrent.ConcurrentHashMap;
 
 public class AvroGenericDeserializer extends AbstractDeserializer<Object> {
-    private final ConcurrentHashMap<SchemaInfo, Schema> knownSchemas;
+    private final ConcurrentHashMap<Pair<SchemaInfo, SchemaInfo>, GenericDatumReader<Object>> knownSchemaReaders;
 
     public AvroGenericDeserializer(String groupId, SchemaRegistryClient client, @Nullable AvroSchema<Object> schema,
                             SerializerConfig.Decoders decoder, EncodingCache encodingCache) {
         super(groupId, client, schema, false, decoder, encodingCache, true);
-        this.knownSchemas = new ConcurrentHashMap<>();
+        this.knownSchemaReaders = new ConcurrentHashMap<>();
     }
 
     @Override
     public final Object deserialize(InputStream inputStream, SchemaInfo writerSchemaInfo, SchemaInfo readerSchemaInfo) throws IOException {
         Preconditions.checkNotNull(writerSchemaInfo);
-        Schema writerSchema = knownSchemas.computeIfAbsent(writerSchemaInfo, x -> AvroSchema.from(x).getSchema());
-        Schema readerSchema = knownSchemas.computeIfAbsent(readerSchemaInfo, x -> AvroSchema.from(x).getSchema());
-        
-        GenericDatumReader<Object> genericDatumReader = new GenericDatumReader<>(writerSchema, readerSchema);
-        
+        final Pair<SchemaInfo, SchemaInfo> keyPair = Pair.of(writerSchemaInfo, readerSchemaInfo);
+        GenericDatumReader<Object> genericDatumReader = knownSchemaReaders.computeIfAbsent(keyPair, key -> {
+            Schema writerSchema = AvroSchema.from(writerSchemaInfo).getSchema();
+            Schema readerSchema = AvroSchema.from(readerSchemaInfo).getSchema();
+            return new GenericDatumReader<>(writerSchema, readerSchema);
+        });
         BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(inputStream, null);
         return genericDatumReader.read(null, decoder);
     }
+
+    @VisibleForTesting
+    ImmutableMap<Pair<SchemaInfo, SchemaInfo>, GenericDatumReader<Object>> getKnownSchemaReaders() {
+        return ImmutableMap.copyOf(knownSchemaReaders);
+    }
 }