release-23.1: changefeedccl: pubsub sink refactor to batching sink

[blathers-crl]

Backport 1/1 commits from #100188 on behalf of @samiskin.

/cc @cockroachdb/release

---

Epic: https://cockroachlabs.atlassian.net/browse/CRDB-13237

This change is a followup to #99086 which moves the Pubsub sink to the batching sink framework.

The changes involve:

1. Moves the Pubsub code to match the SinkClient interface, moving to using the lower level v1 pubsub API that lets us publish batches manually
2. Removing the extra call to json.Marshal
3. Moving to using the pstest package for validating results in unit tests
4. Adding topic handling to the batching sink, where batches are created per-topic
5. Added a pubsub_sink_config since it can now handle Retry and Flush config settings
6. Added metrics even to the old pubsub for the purpose of comparing the two versions

At default settings, this resulted in a peak of 90k messages per second on a single node with throughput at 27.6% cpu usage, putting it at a similar level to kafka.

Running pubsub v2 across all of TPCC (nodes ran out of ranges at different speeds):

Running pubsub v1 (barely visible, 2k messages per second) followed by v2 on tpcc.order_line (in v2 only 2 nodes ended up having ranges assigned to them):

In the following graphs from the cloud console, where v1 was ran followed by v2, you can see how the main reason v1 was slow was that it wasn't able to batch different keys together.

Publish requests remained the same despite way more messages in v2

Release note (performance improvement): pubsub sink changefeeds can now support higher throughputs by enabling the changefeed.new_pubsub_sink_enabled cluster setting.

---

Release justification: Feature flagged (default off) and high impact to the pubsub sink throughput

[blathers-crl]

Thanks for opening a backport.

Please check the backport criteria before merging:

• Patches should only be created for serious issues or test-only changes.
• Patches should not break backwards-compatibility.
• Patches should change as little code as possible.
• Patches should not change on-disk formats or node communication protocols.
• Patches should not add new functionality.
• Patches must not add, edit, or otherwise modify cluster versions; or add version gates.

If some of the basic criteria cannot be satisfied, ensure that the exceptional criteria are satisfied within.

• There is a high priority need for the functionality that cannot wait until the next release and is difficult to address in another way.
• The new functionality is additive-only and only runs for clusters which have specifically “opted in” to it (e.g. by a cluster setting).
• New code is protected by a conditional check that is trivial to verify and ensures that it only runs for opt-in clusters.
• The PM and TL on the team that owns the changed code have signed off that the change obeys the above rules.

Add a brief release justification to the body of your PR to justify this backport.

Some other things to consider:

• What did we do to ensure that a user that doesn’t know & care about this backport, has no idea that it happened?
• Will this work in a cluster of mixed patch versions? Did we test that?
• If a user upgrades a patch version, uses this feature, and then downgrades, what happens?

[cockroach-teamcity]

This change is