PubsubMessageWithTopicCoder.of() is returning wrong coder (apache#31619)

* PubsubMessageWithTopicCoder was never used * refactor
Polber · Jul 19, 2024 · c0f6ac9 · c0f6ac9
1 parent be62163
commit c0f6ac9
Show file tree

Hide file tree

Showing 4 changed files with 7 additions and 7 deletions.
diff --git a/...va/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java b/...va/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java
@@ -1488,7 +1488,7 @@ public PDone expand(PCollection<T> input) {
                   .get(BAD_RECORD_TAG)
                   .setCoder(BadRecord.getCoder(input.getPipeline())));
       PCollection<PubsubMessage> pubsubMessages =
-          pubsubMessageTuple.get(pubsubMessageTupleTag).setCoder(new PubsubMessageWithTopicCoder());
+          pubsubMessageTuple.get(pubsubMessageTupleTag).setCoder(PubsubMessageWithTopicCoder.of());
       switch (input.isBounded()) {
         case BOUNDED:
           pubsubMessages.apply(

diff --git a/...platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubMessageWithTopicCoder.java b/...platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubMessageWithTopicCoder.java
@@ -45,8 +45,8 @@ public static Coder<PubsubMessage> of(TypeDescriptor<PubsubMessage> ignored) {
     return of();
   }
 
-  public static PubsubMessageWithAttributesAndMessageIdCoder of() {
-    return new PubsubMessageWithAttributesAndMessageIdCoder();
+  public static PubsubMessageWithTopicCoder of() {
+    return new PubsubMessageWithTopicCoder();
   }
 
   @Override

diff --git a/...o/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIOTest.java b/...o/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIOTest.java
@@ -732,7 +732,7 @@ public void testWriteMalformedMessagesWithErrorHandler() throws Exception {
       PCollection<PubsubMessage> messages =
           pipeline.apply(
               Create.timestamped(ImmutableList.of(pubsubMsg, failingPubsubMsg))
-                  .withCoder(new PubsubMessageWithTopicCoder()));
+                  .withCoder(PubsubMessageWithTopicCoder.of()));
       messages.setIsBoundedInternal(PCollection.IsBounded.BOUNDED);
       ErrorHandler<BadRecord, PCollection<Long>> badRecordErrorHandler =
           pipeline.registerBadRecordErrorHandler(new ErrorSinkTransform());
@@ -882,7 +882,7 @@ public void testDynamicTopics(boolean isBounded) throws IOException {
 
       PCollection<PubsubMessage> messages =
           pipeline.apply(
-              Create.timestamped(pubsubMessages).withCoder(new PubsubMessageWithTopicCoder()));
+              Create.timestamped(pubsubMessages).withCoder(PubsubMessageWithTopicCoder.of()));
       if (!isBounded) {
         messages = messages.setIsBoundedInternal(PCollection.IsBounded.UNBOUNDED);
       }
@@ -919,7 +919,7 @@ public void testBigMessageBounded() throws IOException {
       PCollection<PubsubMessage> messages =
           pipeline.apply(
               Create.timestamped(ImmutableList.of(pubsubMsg))
-                  .withCoder(new PubsubMessageWithTopicCoder()));
+                  .withCoder(PubsubMessageWithTopicCoder.of()));
       messages.setIsBoundedInternal(PCollection.IsBounded.BOUNDED);
       messages.apply(PubsubIO.writeMessagesDynamic().withClientFactory(factory));
       pipeline.run();

diff --git a/...oud-platform/src/test/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubUnboundedSinkTest.java b/...oud-platform/src/test/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubUnboundedSinkTest.java
@@ -223,7 +223,7 @@ public void testDynamicTopics() throws IOException {
                           Instant.ofEpochMilli(o.getTimestampMsSinceEpoch())))
               .collect(Collectors.toList());
 
-      p.apply(Create.timestamped(pubsubMessages).withCoder(new PubsubMessageWithTopicCoder()))
+      p.apply(Create.timestamped(pubsubMessages).withCoder(PubsubMessageWithTopicCoder.of()))
           .apply(sink);
       p.run();
     }