Allow configuration to be serialized

Author: damccorm

sdks/java/io/amazon-web-services2/src/main/java/org/apache/beam/sdk/io/aws2/kinesis/KinesisTransformRegistrar.java

@@ -104,14 +104,14 @@ public PTransform<PCollection<byte[]>, KinesisIO.Write.Result> buildExternal(
           AwsBasicCredentials.create(configuration.awsAccessKey, configuration.awsSecretKey);
       StaticCredentialsProvider provider = StaticCredentialsProvider.create(creds);
       SerializableFunction<byte[], byte[]> serializer = v -> v;
-      @Nullable URI endpoint;
+      @Nullable URI endpoint = null;
       if (configuration.serviceEndpoint != null) {
         try {
-          endpoint = configuration.serviceEndpoint;
+          endpoint = new URI(configuration.serviceEndpoint);
         }
         catch (URISyntaxException ex) {
           throw new RuntimeException(
-              String.format("Service endpoint must be URI format, got: %s", serviceEndpoint));
+              String.format("Service endpoint must be URI format, got: %s", configuration.serviceEndpoint));
         }
       }
       KinesisIO.Write<byte[]> writeTransform =
@@ -214,14 +214,14 @@ public PTransform<PBegin, PCollection<byte[]>> buildExternal(
       AwsBasicCredentials creds =
           AwsBasicCredentials.create(configuration.awsAccessKey, configuration.awsSecretKey);
       StaticCredentialsProvider provider = StaticCredentialsProvider.create(creds);
-      @Nullable URI endpoint;
+      @Nullable URI endpoint = null;
       if (configuration.serviceEndpoint != null) {
         try {
-          endpoint = configuration.serviceEndpoint;
+          endpoint = new URI(configuration.serviceEndpoint);
         }
         catch (URISyntaxException ex) {
           throw new RuntimeException(
-              String.format("Service endpoint must be URI format, got: %s", serviceEndpoint));
+              String.format("Service endpoint must be URI format, got: %s", configuration.serviceEndpoint));
         }
       }
       KinesisIO.Read readTransform =