Glue Schema Registry schema replication converter

.gitignore

@@ -10,4 +10,5 @@
 **/*.project
 **/*.classpath
 **/*.factorypath
-**/dependency-reduced-pom.xml
+**/dependency-reduced-pom.xml
+/integration-tests/*

common/src/main/java/com/amazonaws/services/schemaregistry/common/AWSSerializerInput.java

@@ -19,6 +19,7 @@
 import lombok.EqualsAndHashCode;
 import lombok.Getter;
 import lombok.ToString;
+import software.amazon.awssdk.services.glue.model.Compatibility;
 
 /**
  * Encapsulates general inputs for serializer
@@ -38,8 +39,11 @@ public class AWSSerializerInput {
     @Getter
     private String transportName;
 
+    @Getter
+    private Compatibility compatibility;
+
     @Builder
-    public AWSSerializerInput(String schemaDefinition, String schemaName, String dataFormat, String transportName) {
+    public AWSSerializerInput(String schemaDefinition, String schemaName, String dataFormat, String transportName, Compatibility compatibility) {
         this.schemaDefinition = schemaDefinition;
 
         if (transportName != null) {
@@ -55,5 +59,11 @@ public AWSSerializerInput(String schemaDefinition, String schemaName, String dat
         }
 
         this.dataFormat = dataFormat;
+
+        if (compatibility != null) {
+            this.compatibility = compatibility;
+        } else {
+            this.compatibility = Compatibility.BACKWARD;
+        }
     }
 }

common/src/main/java/com/amazonaws/services/schemaregistry/common/SchemaByDefinitionFetcher.java

@@ -10,9 +10,9 @@
 import lombok.NonNull;
 import lombok.RequiredArgsConstructor;
 import lombok.SneakyThrows;
+import software.amazon.awssdk.services.glue.model.Compatibility;
 
-import java.util.Map;
-import java.util.UUID;
+import java.util.*;
 import java.util.concurrent.TimeUnit;
 
 /**
@@ -29,6 +29,10 @@ public class SchemaByDefinitionFetcher {
     @VisibleForTesting
     protected final LoadingCache<Schema, UUID> schemaDefinitionToVersionCache;
 
+    @NonNull
+    @VisibleForTesting
+    protected final LoadingCache<SchemaV2, UUID> schemaDefinitionToVersionCacheV2;
+
     public SchemaByDefinitionFetcher(
         final AWSSchemaRegistryClient awsSchemaRegistryClient,
         final GlueSchemaRegistryConfiguration glueSchemaRegistryConfiguration) {
@@ -39,6 +43,11 @@ public SchemaByDefinitionFetcher(
             .maximumSize(glueSchemaRegistryConfiguration.getCacheSize())
             .refreshAfterWrite(glueSchemaRegistryConfiguration.getTimeToLiveMillis(), TimeUnit.MILLISECONDS)
             .build(new SchemaDefinitionToVersionCache());
+
+        this.schemaDefinitionToVersionCacheV2 = CacheBuilder.newBuilder()
+                .maximumSize(glueSchemaRegistryConfiguration.getCacheSize())
+                .refreshAfterWrite(glueSchemaRegistryConfiguration.getTimeToLiveMillis(), TimeUnit.MILLISECONDS)
+                .build(new SchemaDefinitionToVersionCacheV2());
     }
 
     /**
@@ -103,6 +112,80 @@ public UUID getORRegisterSchemaVersionId(
         return schemaVersionId;
     }
 
+    /**
+     * Get Schema Version ID by following below steps :
+     * <p>
+     * 1) If schema version id exists in registry then get it from registry
+     * 2) If schema version id does not exist in registry
+     * then if auto registration is enabled
+     * then if schema exists but version doesn't exist
+     * then
+     * 2.1) Register schema version
+     * else if schema does not exist
+     * then
+     * 2.2) create schema and register schema version
+     *
+     * @param schemaDefinition Schema Definition
+     * @param schemaName       Schema Name
+     * @param dataFormat       Data Format
+     * @param metadata         metadata for schema version
+     * @return Schema Version ID
+     * @throws AWSSchemaRegistryException on any error while fetching the schema version ID
+     */
+    @SneakyThrows
+    public UUID getORRegisterSchemaVersionIdV2(
+            @NonNull String schemaDefinition,
+            @NonNull String schemaName,
+            @NonNull String dataFormat,
+            @NonNull Compatibility compatibility,
+            @NonNull Map<String, String> metadata) throws AWSSchemaRegistryException {
+        UUID schemaVersionId;
+        Map<SchemaV2, UUID> schemaWithVersionId;
+        final SchemaV2 schema = new SchemaV2(schemaDefinition, dataFormat, schemaName, compatibility);
+
+        try {
+            return schemaDefinitionToVersionCacheV2.get(schema);
+        } catch (Exception ex) {
+            Throwable schemaRegistryException = ex.getCause();
+            String exceptionCauseMessage = schemaRegistryException.getCause().getMessage();
+
+            if (exceptionCauseMessage.contains(AWSSchemaRegistryConstants.SCHEMA_VERSION_NOT_FOUND_MSG)) {
+                if (!glueSchemaRegistryConfiguration.isSchemaAutoRegistrationEnabled()) {
+                    throw new AWSSchemaRegistryException(AWSSchemaRegistryConstants.AUTO_REGISTRATION_IS_DISABLED_MSG,
+                            schemaRegistryException);
+                }
+                schemaVersionId =
+                        awsSchemaRegistryClient.registerSchemaVersion(schemaDefinition, schemaName, dataFormat, metadata);
+                schemaDefinitionToVersionCacheV2.put(schema, schemaVersionId);
+            } else if (exceptionCauseMessage.contains(AWSSchemaRegistryConstants.SCHEMA_NOT_FOUND_MSG)) {
+                if (!glueSchemaRegistryConfiguration.isSchemaAutoRegistrationEnabled()) {
+                    throw new AWSSchemaRegistryException(AWSSchemaRegistryConstants.AUTO_REGISTRATION_IS_DISABLED_MSG,
+                            schemaRegistryException);
+                }
+
+                //When schema is not created in the target, create the schema in target and
+                // register all the existing schema version from the source schema to the target in the same order.
+                schemaWithVersionId =
+                        awsSchemaRegistryClient.createSchemaV2(schemaName, dataFormat, schemaDefinition, compatibility, metadata);
+
+                //Cache all the schema versions for a Glue Schema Registry schema
+                schemaWithVersionId.entrySet()
+                                .stream()
+                                .forEach(item -> {
+                                    schemaDefinitionToVersionCacheV2.put(item.getKey(), item.getValue());
+                                });
+            } else {
+                String msg =
+                        String.format(
+                                "Exception occurred while fetching or registering schema definition = %s, schema name = %s ",
+                                schemaDefinition, schemaName);
+                throw new AWSSchemaRegistryException(msg, schemaRegistryException);
+            }
+        }
+        schemaVersionId = schemaDefinitionToVersionCacheV2.get(schema);
+        return schemaVersionId;
+    }
+
     @RequiredArgsConstructor
     private class SchemaDefinitionToVersionCache extends CacheLoader<Schema, UUID> {
         @Override
@@ -111,4 +194,13 @@ public UUID load(Schema schema) {
                 schema.getSchemaDefinition(), schema.getSchemaName(), schema.getDataFormat());
         }
     }
+
+    @RequiredArgsConstructor
+    private class SchemaDefinitionToVersionCacheV2 extends CacheLoader<SchemaV2, UUID> {
+        @Override
+        public UUID load(SchemaV2 schema) {
+            return awsSchemaRegistryClient.getSchemaVersionIdByDefinition(
+                    schema.getSchemaDefinition(), schema.getSchemaName(), schema.getDataFormat());
+        }
+    }
 }

common/src/main/java/com/amazonaws/services/schemaregistry/common/SchemaV2.java

@@ -0,0 +1,47 @@
+/*
+ * Copyright 2020 Amazon.com, Inc. or its affiliates.
+ * 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
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.amazonaws.services.schemaregistry.common;
+
+import lombok.AllArgsConstructor;
+import lombok.Value;
+import software.amazon.awssdk.services.glue.model.Compatibility;
+
+/**
+ * Schema entity represents a schema and it's properties stored in Glue Schema Registry.
+ */
+@AllArgsConstructor
+@Value
+public class SchemaV2 {
+    /**
+     * Schema Definition contains the string representation of schema version stored during registration.
+     */
+    private String schemaDefinition;
+
+    /**
+     * Data Format represents the string notation of data format used during registration of schea. Ex: Avro, JSON etc.
+     */
+    private String dataFormat;
+
+    /**
+     * Schema Name represents name of the schema under which the schema version was registered.
+     */
+    private String schemaName;
+
+    /**
+     * Compatibility mode refers to the compatibility settings of the schema.
+     */
+    private Compatibility compatibilityMode;
+}

common/src/main/java/com/amazonaws/services/schemaregistry/common/configs/GlueSchemaRegistryConfiguration.java

@@ -46,11 +46,20 @@ public class GlueSchemaRegistryConfiguration {
     private AWSSchemaRegistryConstants.COMPRESSION compressionType = AWSSchemaRegistryConstants.COMPRESSION.NONE;
     private String endPoint;
     private String region;
+    // TODO: Remove configs that are not useful non replication use-cases
+    // https://github.com/awslabs/aws-glue-schema-registry/issues/292
+    private String sourceEndPoint;
+    private String sourceRegion;
+    private String targetEndPoint;
+    private String targetRegion;
     private long timeToLiveMillis = 24 * 60 * 60 * 1000L;
     private int cacheSize = 200;
     private AvroRecordType avroRecordType;
     private ProtobufMessageType protobufMessageType;
     private String registryName;
+    private String sourceRegistryName;
+    private String targetRegistryName;
+    private int replicateSchemaVersionCount;
     private Compatibility compatibilitySetting;
     private String description;
     private boolean schemaAutoRegistrationEnabled = false;
@@ -89,8 +98,15 @@ private void buildConfigs(Map<String, ?> configs) {
 
     private void buildSchemaRegistryConfigs(Map<String, ?> configs) {
         validateAndSetAWSRegion(configs);
+        validateAndSetAWSSourceRegion(configs);
+        validateAndSetAWSTargetRegion(configs);
         validateAndSetAWSEndpoint(configs);
+        validateAndSetAWSSourceEndpoint(configs);
+        validateAndSetAWSTargetEndpoint(configs);
         validateAndSetRegistryName(configs);
+        validateAndSetSourceRegistryName(configs);
+        validateAndSetTargetRegistryName(configs);
+        validateAndSetReplicateSchemaVersionCount(configs);
         validateAndSetDescription(configs);
         validateAndSetAvroRecordType(configs);
         validateAndSetProtobufMessageType(configs);
@@ -164,6 +180,20 @@ private void validateAndSetAWSRegion(Map<String, ?> configs) {
         }
     }
 
+    private void validateAndSetAWSSourceRegion(Map<String, ?> configs) {
+        if (isPresent(configs, AWSSchemaRegistryConstants.AWS_SOURCE_REGION)) {
+            this.sourceRegion = String.valueOf(configs.get(AWSSchemaRegistryConstants.AWS_SOURCE_REGION));
+        }
+    }
+
+    private void validateAndSetAWSTargetRegion(Map<String, ?> configs) {
+        if (isPresent(configs, AWSSchemaRegistryConstants.AWS_TARGET_REGION)) {
+            this.targetRegion = String.valueOf(configs.get(AWSSchemaRegistryConstants.AWS_TARGET_REGION));
+        } else {
+            this.targetRegion = this.region;
+        }
+    }
+
     private void validateAndSetCompatibility(Map<String, ?> configs) {
         if (isPresent(configs, AWSSchemaRegistryConstants.COMPATIBILITY_SETTING)) {
             this.compatibilitySetting = Compatibility.fromValue(
@@ -190,21 +220,57 @@ private void validateAndSetRegistryName(Map<String, ?> configs) {
         }
     }
 
+    private void validateAndSetSourceRegistryName(Map<String, ?> configs) {
+        if (isPresent(configs, AWSSchemaRegistryConstants.SOURCE_REGISTRY_NAME)) {
+            this.sourceRegistryName = String.valueOf(configs.get(AWSSchemaRegistryConstants.SOURCE_REGISTRY_NAME));
+        }
+    }
+
+    private void validateAndSetTargetRegistryName(Map<String, ?> configs) {
+        if (isPresent(configs, AWSSchemaRegistryConstants.TARGET_REGISTRY_NAME)) {
+            this.targetRegistryName = String.valueOf(configs.get(AWSSchemaRegistryConstants.TARGET_REGISTRY_NAME));
+        } else {
+            this.targetRegistryName = this.registryName;
+        }
+    }
+
+    private void validateAndSetReplicateSchemaVersionCount(Map<String, ?> configs) {
+        if (isPresent(configs, AWSSchemaRegistryConstants.REPLICATE_SCHEMA_VERSION_COUNT)) {
+            this.replicateSchemaVersionCount = Integer.valueOf(configs.get(AWSSchemaRegistryConstants.REPLICATE_SCHEMA_VERSION_COUNT).toString());
+        } else {
+            this.replicateSchemaVersionCount = AWSSchemaRegistryConstants.DEFAULT_REPLICATE_SCHEMA_VERSION_COUNT;
+        }
+    }
+
     private void validateAndSetAWSEndpoint(Map<String, ?> configs) {
         if (isPresent(configs, AWSSchemaRegistryConstants.AWS_ENDPOINT)) {
             this.endPoint = String.valueOf(configs.get(AWSSchemaRegistryConstants.AWS_ENDPOINT));
         }
     }
 
+    private void validateAndSetAWSSourceEndpoint(Map<String, ?> configs) {
+        if (isPresent(configs, AWSSchemaRegistryConstants.AWS_SOURCE_ENDPOINT)) {
+            this.sourceEndPoint = String.valueOf(configs.get(AWSSchemaRegistryConstants.AWS_SOURCE_ENDPOINT));
+        }
+    }
+
+    private void validateAndSetAWSTargetEndpoint(Map<String, ?> configs) {
+        if (isPresent(configs, AWSSchemaRegistryConstants.AWS_TARGET_ENDPOINT)) {
+            this.targetEndPoint = String.valueOf(configs.get(AWSSchemaRegistryConstants.AWS_TARGET_ENDPOINT));
+        } else {
+            this.targetEndPoint = String.valueOf(configs.get(AWSSchemaRegistryConstants.AWS_ENDPOINT));
+        }
+    }
+
     private void validateAndSetProxyUrl(Map<String, ?> configs) {
         if (isPresent(configs, AWSSchemaRegistryConstants.PROXY_URL)) {
-    		String value = (String) configs.get(AWSSchemaRegistryConstants.PROXY_URL);
-    		try {
-    			this.proxyUrl = URI.create(value);
-    		} catch (IllegalArgumentException e) {
-        		String message = String.format("Proxy URL property is not a valid URL: %s", value);
-        		throw new AWSSchemaRegistryException(message, e);
-        	}
+                String value = (String) configs.get(AWSSchemaRegistryConstants.PROXY_URL);
+                try {
+                        this.proxyUrl = URI.create(value);
+                } catch (IllegalArgumentException e) {
+                        String message = String.format("Proxy URL property is not a valid URL: %s", value);
+                        throw new AWSSchemaRegistryException(message, e);
+                }
         }
     }
 

common/src/main/java/com/amazonaws/services/schemaregistry/utils/AWSSchemaRegistryConstants.java

@@ -15,6 +15,7 @@
 
 package com.amazonaws.services.schemaregistry.utils;
 
+import software.amazon.awssdk.services.glue.endpoints.internal.Value;
 import software.amazon.awssdk.services.glue.model.Compatibility;
 
 public final class AWSSchemaRegistryConstants {
@@ -30,6 +31,30 @@ public final class AWSSchemaRegistryConstants {
      * AWS region to use while initializing the client for service.
      */
     public static final String AWS_REGION = "region";
+    /**
+     * AWS source endpoint to use while initializing the client for service.
+     */
+    public static final String AWS_SOURCE_ENDPOINT = "source.endpoint";
+    /**
+     * AWS source region to use while initializing the client for service.
+     */
+    public static final String AWS_SOURCE_REGION = "source.region";
+    /**
+     * AWS target endpoint to use while initializing the client for service.
+     */
+    public static final String AWS_TARGET_ENDPOINT = "target.endpoint";
+    /**
+     * AWS target region to use while initializing the client for service.
+     */
+    public static final String AWS_TARGET_REGION = "target.region";
+    /**
+     * Number of schema versions to replicate from source to target
+     */
+    public static final String REPLICATE_SCHEMA_VERSION_COUNT = "replicateSchemaVersionCount";
+    /**
+     * Default number of schema versions to replicate from source to target
+     */
+    public static final Integer DEFAULT_REPLICATE_SCHEMA_VERSION_COUNT = 10;
     /**
      * Header Version Byte.
      */
@@ -104,6 +129,14 @@ public final class AWSSchemaRegistryConstants {
      * Default registry name if not passed by the client.
      */
     public static final String DEFAULT_REGISTRY_NAME = "default-registry";
+    /**
+     * Source Registry Name.
+     */
+    public static final String SOURCE_REGISTRY_NAME = "source.registry.name";
+    /**
+     * Target Registry Name.
+     */
+    public static final String TARGET_REGISTRY_NAME = "target.registry.name";
     /**
      * Compatibility setting, will be helpful at the time of schema evolution.
      */