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

@@ -61,7 +61,7 @@ private abstract static class CrossLanguageConfiguration {
     String streamName;
     String awsAccessKey;
     String awsSecretKey;
-    Region region;
+    String region;
     @Nullable String serviceEndpoint;
 
     public void setStreamName(String streamName) {
@@ -77,7 +77,7 @@ public void setAwsSecretKey(String awsSecretKey) {
     }
 
     public void setRegion(String region) {
-      this.region = Region.of(region);
+      this.region = region;
     }
 
     public void setServiceEndpoint(@Nullable String serviceEndpoint) {
@@ -120,7 +120,7 @@ public PTransform<PCollection<byte[]>, KinesisIO.Write.Result> buildExternal(
               .withClientConfiguration(
                   ClientConfiguration.builder()
                       .credentialsProvider(provider)
-                      .region(configuration.region)
+                      .region(Regions.fromName(configuration.region))
                       .endpoint(endpoint)
                       .build())
               .withPartitioner(p -> configuration.partitionKey)
@@ -230,7 +230,7 @@ public PTransform<PBegin, PCollection<byte[]>> buildExternal(
               .withClientConfiguration(
                   ClientConfiguration.builder()
                       .credentialsProvider(provider)
-                      .region(configuration.region)
+                      .region(Regions.fromName(configuration.region))
                       .endpoint(endpoint)
                       .build());