Dataflow runner specialization of PubsubIO should validate messages

[damccorm]

Currently invalid messages, such as those with attributes exceeding the maximum size are processed by the bundle successfully but fail to commit.

Throwing an exception when trying to write such a message directly would increase visibility as well as allowing users to catch and handle such exceptions.

Imported from Jira BEAM-12942. Original Jira may contain additional context.
Reported by: scwhittle.

[damccorm]

Unable to assign user @scwhittle. If able, self-assign, otherwise tag @damccorm so that he can assign you. Because of GitHub's spam prevention system, your activity is required to enable assignment in this repo.

[scwhittle]

@damccorm Hi Danny, can you assign this to me? I don't seem to be able to though I have committed to the repo before.

[damccorm]

One of the unfortunate downsides of moving to issues is that non-committers can't self-assign/triage issues that they didn't create by default. I think that's a workflow we'll need to support going forward though, I'll look into it. For the moment, I'd recommend commenting/declaring intent to work on any issues. Once #21719 is in, you should be able to self assign and close issues with chat-op commands

[scwhittle]

.take-issue

[scwhittle]

.close-issue