[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
• 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