Merge pull request #24 from dylanmei/decompose_configuration

Decompose and test connector configuration

src/main/java/com/nordstrom/kafka/connect/auth/AWSAssumeRoleCredentialsProvider.java

@@ -83,15 +83,15 @@ private void verifyNotNullOrEmpty(final String field, final String fieldName) {
     }
   }
 
-  String getExternalId() {
+  public String getExternalId() {
     return this.externalId;
   }
 
-  String getRoleArn() {
+  public String getRoleArn() {
     return this.roleArn;
   }
 
-  String getSessionName() {
+  public String getSessionName() {
     return this.sessionName;
   }
 }

src/main/java/com/nordstrom/kafka/connect/formatters/JsonPayloadFormatter.java

@@ -18,12 +18,6 @@
 import static java.util.Collections.emptyMap;
 
 public class JsonPayloadFormatter implements PayloadFormatter, Configurable {
-  enum SchemaVisibility {
-    ALL,
-    MIN,
-    NONE
-  }
-
   private final ObjectWriter recordWriter = new ObjectMapper().writerFor(Payload.class);
   private final ObjectWriter recordsWriter = new ObjectMapper().writerFor(Payload[].class);
   private final JsonConverter converter = new JsonConverter();
@@ -44,8 +38,8 @@ public JsonPayloadFormatter() {
 
   @Override
   public void configure(Map<String, ?> configs) {
-    keySchemaVisibility = configureSchemaVisibility(configs, "formatter.key.schema.visibility");
-    valueSchemaVisibility = configureSchemaVisibility(configs, "formatter.value.schema.visibility");
+    keySchemaVisibility = configureSchemaVisibility(configs, "key.schema.visibility");
+    valueSchemaVisibility = configureSchemaVisibility(configs, "value.schema.visibility");
   }
 
   private SchemaVisibility configureSchemaVisibility(final Map<String, ?> configs, final String key) {
@@ -68,6 +62,14 @@ private SchemaVisibility configureSchemaVisibility(final Map<String, ?> configs,
     return viz;
   }
 
+  public SchemaVisibility getKeySchemaVisibility() {
+    return keySchemaVisibility;
+  }
+
+  public SchemaVisibility getValueSchemaVisibility() {
+    return valueSchemaVisibility;
+  }
+
   public String format(final SinkRecord record) {
     try {
       return recordWriter.writeValueAsString(recordToPayload(record));

src/main/java/com/nordstrom/kafka/connect/formatters/SchemaVisibility.java

@@ -0,0 +1,7 @@
+package com.nordstrom.kafka.connect.formatters;
+
+public enum SchemaVisibility {
+  ALL,
+  MIN,
+  NONE
+}