Implement PubsubRowToMessage transform

[damondouglas]

This PR addresses #21412 with a PubsubRowToMessage transform.

Validation logic

Prior to processing Row inputs, it throws errors from Schema validation according to the following fields, where in the Row input:

attributes - refers to the expected Map<String, String>
timestamp - refers to the expected ReadableDateTime
payload - refers to either the expected FieldType.BYTES or TypeName.ROW
user fields - refers to any fields that are present in the Row input that are not attributes, timestamp, or payload fields
PayloadSerializer - refers to the interface responsible for serializing a Row to byte[]

NOTE: ... in the table below indicates that the value in the cell is irrelevant for the case

attributes	timestamp	payload	user fields	PayloadSerializer	reason
not Map<String, String>	...	...	...	...	attributes are expected to match the same type as a PubsubMessage
...	not DATETIME	...	...	...	the timestamp is expected to be derived from the Schema type
...	...	not BYTES or ROW	...	...	we expect no other type but BYTES or ROW
...	...	exists	exists	...	providing the payload and user fields is incompatible
...	...	BYTES	...	not null	providing a PayloadSerializer in the context of payload BYTES is incompatible
...	...	ROW	...	null	PayloadSerializer is required to serialize payload ROW input
...	...	...	exist	null	PayloadSerializer is required to serialize a Row with user fields

Timestamp logic

The following summarizes the logic when populating the stringified timestamp of the resulting PubsubMessage's attributes, where in the Row input:

attributes - refers to the expected Map<String, String>
timestamp - refers to the expected ReadableDateTime
targetTimestampKey - refers to the optional configuration parameter
sourceTimestampKey - refers to the expected timestamp field name of the Row input

NOTE: ... in the table below indicates that the value in the cell is irrelevant for the case

user provided timestamp in attributes	timestamp	targetTimestampKey	result PubsubMessage attributes key	result PubsubMessage attributes value	reason
exists	...	...	sourceTimestampKey	value from Instant.now()	we only use the Row's timestamp field as input
...	exists	...	sourceTimestampKey	value from Row input	we apply the stringified timestamp value to the PubsubMessage attributes using the sourceTimestampKey as its key
...	...	exists	targetTimestampKey	...	we use the configured key instead of the default sourceTimestampKey
...	absent	...	...	value from Instant.now()	when Row input lacks the timestamp we generate from Instant.now()

Payload logic

The following summarizes the logic when populating the byte[] payload to a PubsubMessage, where in the Row input:

payload - refers to either the expected FieldType.BYTES or TypeName.ROW
user fields - refers to any fields that are present in the Row input that are not attributes, timestamp, or payload fields

payload	user fields	PubsubMessage byte[]	reason
BYTES	absent	raw byte[] value from payload Field	no serialization is needed since the Row input already has the byte[] payload
ROW	absent	serialized Row output from payload Field	serialization is applied using the PayloadSerializer
absent	present	serialized Row output from user fields	serialization is applied to the Row using the PayloadSerializer resulting from user fields after removing the timestamp and attributes fields

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI.

[github-actions]

Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control

[johnjcasey]

LGTM