Add serializer

apache · Dec 19, 2024 · a9eed7d · a9eed7d
1 parent 90c13e9
commit a9eed7d
Showing 1 changed file with 3 additions and 1 deletion.
diff --git a/...ervices2/src/main/java/org/apache/beam/sdk/io/aws2/kinesis/KinesisTransformRegistrar.java b/...ervices2/src/main/java/org/apache/beam/sdk/io/aws2/kinesis/KinesisTransformRegistrar.java
@@ -109,6 +109,7 @@ public PTransform<PCollection<byte[]>, KinesisIO.Write.Result> buildExternal(
       AwsBasicCredentials creds =
           AwsBasicCredentials.create(configuration.awsAccessKey, configuration.awsSecretKey);
       StaticCredentialsProvider provider = StaticCredentialsProvider.create(creds);
+      SerializableFunction<byte[], byte[]> serializer = v -> v;
       KinesisIO.Write<byte[]> writeTransform =
           KinesisIO.<byte[]>write()
               .withStreamName(configuration.streamName)
@@ -118,7 +119,8 @@ public PTransform<PCollection<byte[]>, KinesisIO.Write.Result> buildExternal(
                       .region(configuration.region)
                       .endpoint(configuration.serviceEndpoint)
                       .build())
-              .withPartitioner(p -> configuration.partitionKey);
+              .withPartitioner(p -> configuration.partitionKey)
+              .withSerializer(serializer);
 
       return writeTransform;
     }