[Bug]: Default PubsubMessage coder will drop message id and orderingKey

[TheNeuralBit]

What happened?

We register multiple coders for PubsubMessage:

beam/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubCoderProviderRegistrar.java

Lines 32 to 40 in 9b2f87d

	public List<CoderProvider> getCoderProviders() {
	return ImmutableList.of(
	CoderProviders.forCoder(
	TypeDescriptor.of(PubsubMessage.class), PubsubMessageWithAttributesCoder.of()),
	CoderProviders.forCoder(
	TypeDescriptor.of(PubsubMessage.class), PubsubMessageWithMessageIdCoder.of()),
	CoderProviders.forCoder(
	TypeDescriptor.of(PubsubMessage.class),
	PubsubMessageWithAttributesAndMessageIdCoder.of()));

But in practice everything but the first one is a no-op. We will never use them (see #22216 (comment) for a detailed explanation). This means that in cases where the coder for a PCollection<PubsubMessage> is inferred, we will use PubsubMessageWithAttributesCoder, which does not handle messageID or orderingKey.

Issue Priority

Priority: 2

Issue Component

Component: io-java-gcp

[TheNeuralBit]

Marking this P1 since it could lead to data loss (in rare cases). We might mitigate by making PubsubMessageWithAttributesCoder (and other PubsubMessage coders) raise an error if the unsupported fields are non-null.

A better long-term fix would be to default to a coder that handles all fields, but this could break update compatibility so we'd need to be careful.

[apilloud]

@egalpin @TheNeuralBit Is this fixed by #22216?

[egalpin]

@apilloud It's introduced by #22216 rather than fixed. Part of the complexity, at least for me, is the multiple coders registered against the same class in PubsubCoderProviderRegistrar. Since all pubsub messages of type org.apache.beam.sdk.io.gcp.pubsub.PubsubMessage appear to use the first coder in the registrar by default, any messages with a superset of properties compared to PubsubMessageWithAttributesCoder will have those properties nullified silently.

Assuming that the "superset" coder which supports all the fields were the default, what would the drawback be in the case where, say, the messageId would never be used? It would mean more bytes to represent the PubsubMessage, but are there other drawbacks? Otherwise it seems like it might be easiest to make the "superset" coder the default, and view other coders as data-size optimizations that users could opt in to.

I do like the idea of raising an error in the coders in the case where unsupported fields are non-null, I think this could be an effective way to defend against data loss that could be implemented relatively quickly.

[TheNeuralBit]

are there other drawbacks?

The only other drawback I'm aware of is that we'd likely break update compatibility for pipelines that were using the default coder. It might be good to discuss this on the dev list to weigh our options here. Would you want to start that thread @egalpin?