[Bug]: PubsubMessageWithTopicCoder.of() returns PubsubMessageWithAttributesAndMessageIdCoder

[stankiewicz]

What happened?

PubsubMessageWithTopicCoder should return PubsubMessageWithTopicCoder while it returns PubsubMessageWithAttributesAndMessageIdCoder

While investigating Dynamic Destinations on Direct runner I found out that PubsubMessageWithTopicCoder is never used and topic is lost and pipeline fails.

repro:

        final PCollection<PubsubMessage> recordsWithTopics =
                events.apply("Build Record", ParDo.of(new BuildRecord())).setCoder(
                    PubsubMessageWithTopicCoder.of());
        System.out.println(recordsWithTopics.getCoder());
        recordsWithTopics.apply("Send to consumers", PubsubIO.writeMessagesDynamic());

issue: KVCoder throws "cannot encode a null KV", difficult to debug.

workaround - use constructor:

        final PCollection<PubsubMessage> recordsWithTopics =
                events.apply("Build Record", ParDo.of(new BuildRecord())).setCoder(
                    new PubsubMessageWithTopicCoder());
        System.out.println(recordsWithTopics.getCoder());
        recordsWithTopics.apply("Send to consumers", PubsubIO.writeMessagesDynamic());

Issue Priority

Priority: 2 (default / most bugs should be filed as P2)

Issue Components

• [ ] Component: Python SDK
• [X] Component: Java SDK
• [ ] Component: Go SDK
• [ ] Component: Typescript SDK
• [ ] Component: IO connector
• [ ] Component: Beam YAML
• [ ] Component: Beam examples
• [ ] Component: Beam playground
• [ ] Component: Beam katas
• [ ] Component: Website
• [ ] Component: Spark Runner
• [ ] Component: Flink Runner
• [ ] Component: Samza Runner
• [ ] Component: Twister2 Runner
• [ ] Component: Hazelcast Jet Runner
• [ ] Component: Google Cloud Dataflow Runner

[stankiewicz]

.take-issue