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/AWSSchemaRegistryClient.java

@@ -42,11 +42,15 @@
 import software.amazon.awssdk.services.glue.model.DataFormat;
 import software.amazon.awssdk.services.glue.model.GetSchemaByDefinitionRequest;
 import software.amazon.awssdk.services.glue.model.GetSchemaByDefinitionResponse;
+import software.amazon.awssdk.services.glue.model.GetSchemaRequest;
+import software.amazon.awssdk.services.glue.model.GetSchemaResponse;
 import software.amazon.awssdk.services.glue.model.GetSchemaVersionRequest;
 import software.amazon.awssdk.services.glue.model.GetSchemaVersionResponse;
 import software.amazon.awssdk.services.glue.model.GetTagsRequest;
 import software.amazon.awssdk.services.glue.model.GetTagsResponse;
 import software.amazon.awssdk.services.glue.model.GlueRequest;
+import software.amazon.awssdk.services.glue.model.ListSchemaVersionsRequest;
+import software.amazon.awssdk.services.glue.model.ListSchemaVersionsResponse;
 import software.amazon.awssdk.services.glue.model.MetadataKeyValuePair;
 import software.amazon.awssdk.services.glue.model.PutSchemaVersionMetadataRequest;
 import software.amazon.awssdk.services.glue.model.PutSchemaVersionMetadataResponse;
@@ -56,9 +60,12 @@
 import software.amazon.awssdk.services.glue.model.RegisterSchemaVersionResponse;
 import software.amazon.awssdk.services.glue.model.RegistryId;
 import software.amazon.awssdk.services.glue.model.SchemaId;
+import software.amazon.awssdk.services.glue.model.SchemaVersionListItem;
 
 import java.net.URI;
 import java.net.URISyntaxException;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Map;
 import java.util.StringJoiner;
 import java.util.UUID;
@@ -180,7 +187,7 @@ public GetSchemaVersionResponse getSchemaVersionResponse(@NonNull String schemaV
         return schemaVersionResponse;
     }
 
-    private GetSchemaVersionRequest getSchemaVersionRequest(String schemaVersionId) {
+    public GetSchemaVersionRequest getSchemaVersionRequest(String schemaVersionId) {
         GetSchemaVersionRequest getSchemaVersionRequest = GetSchemaVersionRequest.builder()
                 .schemaVersionId(schemaVersionId).build();
         return getSchemaVersionRequest;
@@ -193,6 +200,43 @@ private void validateSchemaVersionResponse(GetSchemaVersionResponse schemaVersio
         }
     }
 
+    /**
+     * Get Schema by passing the schema id.
+     * @param schemaId Schema Id
+     * @return          schema returns the schema corresponding to the
+     *                  schema id passed and null in case service is not able to found the
+     *                  schema corresponding to schema id.
+     * @throws AWSSchemaRegistryException on any error while fetching the schema
+     */
+    public GetSchemaResponse getSchemaResponse(@NonNull SchemaId schemaId)
+            throws AWSSchemaRegistryException {
+        GetSchemaResponse schemaResponse = null;
+
+        try {
+            schemaResponse = client.getSchema(getSchemaRequest(schemaId));
+            validateSchemaResponse(schemaResponse, schemaId);
+        } catch (Exception e) {
+            String errorMessage = String.format("Failed to get schema Id = %s", schemaId);
+            throw new AWSSchemaRegistryException(errorMessage, e);
+        }
+
+        return schemaResponse;
+    }
+
+    private GetSchemaRequest getSchemaRequest(SchemaId schemaId) {
+        GetSchemaRequest getSchemaRequest = GetSchemaRequest.builder()
+                .schemaId(schemaId)
+                .build();
+        return getSchemaRequest;
+    }
+
+    private void validateSchemaResponse(GetSchemaResponse schemaResponse, SchemaId schemaId) {
+        if (schemaResponse == null) {
+            String message = String.format("Schema is not present for the schema id = %s", schemaId);
+            throw new AWSSchemaRegistryException(message);
+        }
+    }
+
     private UUID returnSchemaVersionIdIfAvailable(GetSchemaByDefinitionResponse response) {
         if (response.schemaVersionId() != null
                 && response.statusAsString().equals(AWSSchemaRegistryConstants.SchemaVersionStatus.AVAILABLE.toString())) {
@@ -266,6 +310,53 @@ public UUID createSchema(String schemaName,
         return schemaVersionId;
     }
 
+    /**
+     * List all versions of the schema
+     * @param schemaName       Schema Name
+     * @return                 List of schema versions
+     * @throws AWSSchemaRegistryException on any error during the registration and fetching of schema version
+     */
+    public List<SchemaVersionListItem> getSchemaVersions(String schemaName) {
+        ListSchemaVersionsRequest listSchemaVersionsRequest = getListSchemaVersionsRequest(schemaName);
+        List<SchemaVersionListItem> schemaVersionList = new ArrayList<>();
+        boolean done = false;
+        try {
+            while (!done) {
+                //Get list of schema versions from source registry
+                ListSchemaVersionsResponse listSchemaVersionsResponse = client.listSchemaVersions(listSchemaVersionsRequest);
+                schemaVersionList = listSchemaVersionsResponse.schemas();
+
+                //Keep paginating till the end
+                if (listSchemaVersionsResponse.nextToken() == null) {
+                    done = true;
+                }
+
+                //Create the request object to get next set of results using the nextToken
+                listSchemaVersionsRequest = getListSchemaVersionsRequest(schemaName, listSchemaVersionsResponse.nextToken());
+            }
+        } catch (Exception e) {
+            String errorMessage = String.format("Failed to get schema version for schema = %s", schemaName);
+            throw new AWSSchemaRegistryException(errorMessage, e);
+        }
+
+        return schemaVersionList;
+    }
+
+    private ListSchemaVersionsRequest getListSchemaVersionsRequest(String schemaName, String nextToken) {
+        return ListSchemaVersionsRequest
+                .builder()
+                .nextToken(nextToken)
+                .schemaId(SchemaId.builder().schemaName(schemaName).registryName(glueSchemaRegistryConfiguration.getRegistryName()).build())
+                .build();
+    }
+
+    private ListSchemaVersionsRequest getListSchemaVersionsRequest(String schemaName) {
+        return ListSchemaVersionsRequest
+                .builder()
+                .schemaId(SchemaId.builder().schemaName(schemaName).registryName(glueSchemaRegistryConfiguration.getRegistryName()).build())
+                .build();
+    }
+
     /**
      * Register the schema and return schema version Id once it is available.
      * @param schemaDefinition Schema Definition

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

@@ -323,7 +323,7 @@ private void validateAndSetJacksonDeserializationFeatures(Map<String, ?> configs
         }
     }
 
-    private boolean isPresent(Map<String, ?> configs,
+    public boolean isPresent(Map<String, ?> configs,
                               String key) {
         if (!GlueSchemaRegistryUtils.getInstance()
                 .checkIfPresentInMap(configs, key)) {

cross-region-replication-converter/.gitignore

@@ -0,0 +1,40 @@
+target/
+!.mvn/wrapper/maven-wrapper.jar
+!**/src/main/**/target/
+!**/src/test/**/target/
+
+resources/
+
+### IntelliJ IDEA ###
+.idea/modules.xml
+.idea/jarRepositories.xml
+.idea/compiler.xml
+.idea/libraries/
+*.iws
+*.iml
+*.ipr
+
+### Eclipse ###
+.apt_generated
+.classpath
+.factorypath
+.project
+.settings
+.springBeans
+.sts4-cache
+
+### NetBeans ###
+/nbproject/private/
+/nbbuild/
+/dist/
+/nbdist/
+/.nb-gradle/
+build/
+!**/src/main/**/build/
+!**/src/test/**/build/
+
+### VS Code ###
+.vscode/
+
+### Mac OS ###
+.DS_Store