Repository: googlecloudplatform/dataflowtemplates
Issue: [Bug]: PubSubToMongoDB (V2) does not store data as UTF-8

### Related Template(s)

PubSubToMongoDB

### What happened?

The pipeline does not store special characters (e.g. ®) as UTF-8 in MongoDB, causing them to turn into ��.

I think the root cause is this line of code:
https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v2/pubsub-to-mongodb/src/main/java/com/google/cloud/teleport/v2/templates/PubSubToMongoDB.java#L471

Adjusting it like this seems to fix the issue:

```
messageObject = gson.fromJson(new String(pubsubMessage.getPayload(), StandardCharsets.UTF_8), JsonObject.class);
```


### Beam Version

Newer than 2.46.0

### Relevant log output

_No response_

Changed files:
v2/googlecloud-to-googlecloud/src/main/java/com/google/cloud/teleport/v2/templates/TextIOToBigQuery.java
v2/googlecloud-to-googlecloud/src/test/java/com/google/cloud/teleport/v2/templates/SpannerChangeStreamsToPubSubTest.java
v1/src/main/java/com/google/cloud/teleport/templates/common/ErrorConverters.java
v2/googlecloud-to-elasticsearch/src/main/java/com/google/cloud/teleport/v2/elasticsearch/transforms/FailedPubsubMessageToPubsubTopicFn.java
v2/pubsub-to-mongodb/src/main/java/com/google/cloud/teleport/v2/templates/PubSubToMongoDB.java
v2/googlecloud-to-elasticsearch/src/main/java/com/google/cloud/teleport/v2/elasticsearch/transforms/ProcessFailsafePubSubFn.java
v2/common/src/main/java/com/google/cloud/teleport/v2/transforms/ErrorConverters.java

diff --git a/v1/src/main/java/com/google/cloud/teleport/templates/common/ErrorConverters.java b/v1/src/main/java/com/google/cloud/teleport/templates/common/ErrorConverters.java
index 36a6fe29..656578b0 100644
--- a/v1/src/main/java/com/google/cloud/teleport/templates/common/ErrorConverters.java
+++ b/v1/src/main/java/com/google/cloud/teleport/templates/common/ErrorConverters.java
@@ -279,7 +279,7 @@ public class ErrorConverters {
       // Only set the payload if it's populated on the message.
       if (message.getPayload() != null) {
         failedRow
-            .set("payloadString", new String(message.getPayload()))
+            .set("payloadString", new String(message.getPayload(), StandardCharsets.UTF_8))
             .set("payloadBytes", message.getPayload());
       }
 
diff --git a/v2/common/src/main/java/com/google/cloud/teleport/v2/transforms/ErrorConverters.java b/v2/common/src/main/java/com/google/cloud/teleport/v2/transforms/ErrorConverters.java
index 10d5da95..3fa81373 100644
--- a/v2/common/src/main/java/com/google/cloud/teleport/v2/transforms/ErrorConverters.java
+++ b/v2/common/src/main/java/com/google/cloud/teleport/v2/transforms/ErrorConverters.java
@@ -331,7 +331,7 @@ public class ErrorConverters {
       PubsubMessage pubsubMessage = failsafeElement.getOriginalPayload();
       String message =
           pubsubMessage.getPayload().length > 0
-              ? new String(pubsubMessage.getPayload())
+              ? new String(pubsubMessage.getPayload(), StandardCharsets.UTF_8)
               : pubsubMessage.getAttributeMap().toString();
 
       // Format the timestamp for insertion
diff --git a/v2/googlecloud-to-elasticsearch/src/main/java/com/google/cloud/teleport/v2/elasticsearch/transforms/FailedPubsubMessageToPubsubTopicFn.java b/v2/googlecloud-to-elasticsearch/src/main/java/com/google/cloud/teleport/v2/elasticsearch/transforms/FailedPubsubMessageToPubsubTopicFn.java
index 821a878a..7ff5e8c5 100644
--- a/v2/googlecloud-to-elasticsearch/src/main/java/com/google/cloud/teleport/v2/elasticsearch/transforms/FailedPubsubMessageToPubsubTopicFn.java
+++ b/v2/googlecloud-to-elasticsearch/src/main/java/com/google/cloud/teleport/v2/elasticsearch/transforms/FailedPubsubMessageToPubsubTopicFn.java
@@ -51,7 +51,7 @@ public class FailedPubsubMessageToPubsubTopicFn
     PubsubMessage pubsubMessage = failsafeElement.getOriginalPayload();
     String message =
         pubsubMessage.getPayload().length > 0
-            ? new String(pubsubMessage.getPayload())
+            ? new String(pubsubMessage.getPayload(), StandardCharsets.UTF_8)
             : pubsubMessage.getAttributeMap().toString();
 
     // Format the timestamp for insertion
diff --git a/v2/googlecloud-to-elasticsearch/src/main/java/com/google/cloud/teleport/v2/elasticsearch/transforms/ProcessFailsafePubSubFn.java b/v2/googlecloud-to-elasticsearch/src/main/java/com/google/cloud/teleport/v2/elasticsearch/transforms/ProcessFailsafePubSubFn.java
index 6247dc10..2ac49e9e 100644
--- a/v2/googlecloud-to-elasticsearch/src/main/java/com/google/cloud/teleport/v2/elasticsearch/transforms/ProcessFailsafePubSubFn.java
+++ b/v2/googlecloud-to-elasticsearch/src/main/java/com/google/cloud/teleport/v2/elasticsearch/transforms/ProcessFailsafePubSubFn.java
@@ -20,6 +20,7 @@ import com.google.cloud.teleport.v2.values.FailsafeElement;
 import com.google.gson.Gson;
 import com.google.gson.JsonObject;
 import com.google.gson.JsonSyntaxException;
+import java.nio.charset.StandardCharsets;
 import org.apache.beam.sdk.io.gcp.pubsub.PubsubMessage;
 import org.apache.beam.sdk.metrics.Counter;
 import org.apache.beam.sdk.metrics.Metrics;
@@ -54,7 +55,9 @@ public class ProcessFailsafePubSubFn
 
     try {
       if (pubsubMessage.getPayload().length > 0) {
-        messageObject = gson.fromJson(new String(pubsubMessage.getPayload()), JsonObject.class);
+        messageObject =
+            gson.fromJson(
+                new String(pubsubMessage.getPayload(), StandardCharsets.UTF_8), JsonObject.class);
       }
 
       // If message attributes are present they will be serialized along with the message payload
diff --git a/v2/googlecloud-to-googlecloud/src/main/java/com/google/cloud/teleport/v2/templates/TextIOToBigQuery.java b/v2/googlecloud-to-googlecloud/src/main/java/com/google/cloud/teleport/v2/templates/TextIOToBigQuery.java
index 592e4b66..711dae83 100644
--- a/v2/googlecloud-to-googlecloud/src/main/java/com/google/cloud/teleport/v2/templates/TextIOToBigQuery.java
+++ b/v2/googlecloud-to-googlecloud/src/main/java/com/google/cloud/teleport/v2/templates/TextIOToBigQuery.java
@@ -30,6 +30,7 @@ import com.google.cloud.teleport.v2.utils.BigQueryIOUtils;
 import com.google.common.annotations.VisibleForTesting;
 import java.nio.channels.Channels;
 import java.nio.channels.ReadableByteChannel;
+import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.function.Supplier;
@@ -270,7 +271,8 @@ public class TextIOToBigQuery {
           FileSystems.open(FileSystems.matchNewResource(pathToJson, false));
       String json =
           new String(
-              StreamUtils.getBytesWithoutClosing(Channels.newInputStream(readableByteChannel)));
+              StreamUtils.getBytesWithoutClosing(Channels.newInputStream(readableByteChannel)),
+              StandardCharsets.UTF_8);
       return new JSONObject(json);
     } catch (Exception e) {
       throw new RuntimeException(e);
diff --git a/v2/googlecloud-to-googlecloud/src/test/java/com/google/cloud/teleport/v2/templates/SpannerChangeStreamsToPubSubTest.java b/v2/googlecloud-to-googlecloud/src/test/java/com/google/cloud/teleport/v2/templates/SpannerChangeStreamsToPubSubTest.java
index 342f0d3b..506754b7 100644
--- a/v2/googlecloud-to-googlecloud/src/test/java/com/google/cloud/teleport/v2/templates/SpannerChangeStreamsToPubSubTest.java
+++ b/v2/googlecloud-to-googlecloud/src/test/java/com/google/cloud/teleport/v2/templates/SpannerChangeStreamsToPubSubTest.java
@@ -28,6 +28,7 @@ import com.google.cloud.teleport.v2.spanner.SpannerServerResource;
 import com.google.cloud.teleport.v2.spanner.SpannerTestHelper;
 import com.google.cloud.teleport.v2.transforms.FileFormatFactorySpannerChangeStreamsToPubSub;
 import com.google.gson.Gson;
+import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Random;
@@ -705,7 +706,7 @@ public final class SpannerChangeStreamsToPubSubTest extends SpannerTestHelper {
 
     @Override
     public String apply(PubsubMessage message) {
-      return new String(message.getPayload());
+      return new String(message.getPayload(), StandardCharsets.UTF_8);
     }
   }
 }
diff --git a/v2/pubsub-to-mongodb/src/main/java/com/google/cloud/teleport/v2/templates/PubSubToMongoDB.java b/v2/pubsub-to-mongodb/src/main/java/com/google/cloud/teleport/v2/templates/PubSubToMongoDB.java
index 1e2db824..028275b2 100644
--- a/v2/pubsub-to-mongodb/src/main/java/com/google/cloud/teleport/v2/templates/PubSubToMongoDB.java
+++ b/v2/pubsub-to-mongodb/src/main/java/com/google/cloud/teleport/v2/templates/PubSubToMongoDB.java
@@ -468,7 +468,9 @@ public class PubSubToMongoDB {
 
       try {
         if (pubsubMessage.getPayload().length > 0) {
-          messageObject = gson.fromJson(new String(pubsubMessage.getPayload()), JsonObject.class);
+          messageObject =
+              gson.fromJson(
+                  new String(pubsubMessage.getPayload(), StandardCharsets.UTF_8), JsonObject.class);
         }
 
         // If message attributes are present they will be serialized along with the message payload