release-22.2: changefeedccl: flush to sink before initial_scan_only completion

[blathers-crl]

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

/cc @cockroachdb/release

---

Resolves #90146

Previously if a changefeed was started with initial_scan_only, the job would be completed as soon as a resolved event was seen by the kv_feed even though we may not have flushed the messages to the sink yet. This could result in a huge difference where an initial_scan_only changefeed on a 100000 row table could complete successfully with a kafka topic only seeing <2000 messages. A similar issue of missing messages would also occur with schema_change_policy='stop', though the changefeed is marked failed in that case.

This change ensures the changefeed drains the eventProducer and sink upon a normal kvfeed exit.

This also uncovered a bug in the chunked_event_queue where empty() could return true even if there were many remaining messages due to the first chunk being empty but there still being future chunks.

Release note (bug fix): initial_scan_only changefeeds now ensure that all messages have successfully flushed to the sink prior to completion instead of potentially missing messages.

---

Release justification: high priority and very low risk bug fix

[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