feat(plugin-server): Preserve distinct ID locality on overflow rerouting

[tkaemming]

Problem

See #20945 for details.

Changes

See #20945 for details. This change fixes the (known) issue with that implementation.

Does this work well for both Cloud and self-hosted?

Yes.

How did you test this code?

Updated existing tests, as before.

This also is based on a couple of changes that prevent the issue we saw earlier, making this change itself safer:

• fix(plugin-server): Better limit Kafka message key types to avoid problematic values #21289 protects against the problematic undefined message key value by preventing its use when type checking
• test(plugin-server): Improve the robustness of overflow handling tests #21315 improved the existing tests so that they call the producer rather than mocking it, which would have identified the issue where promises associated with producing messages with undefined keys never settled, blocking the consumer loop indefinitely

This patch can be applied to this PR to demonstrate the bad behavior when running tests:

diff --git a/plugin-server/src/main/ingestion-queues/batch-processing/each-batch-ingestion.ts b/plugin-server/src/main/ingestion-queues/batch-processing/each-batch-ingestion.ts
index 313e0cf083..e659c5fa1c 100644
--- a/plugin-server/src/main/ingestion-queues/batch-processing/each-batch-ingestion.ts
+++ b/plugin-server/src/main/ingestion-queues/batch-processing/each-batch-ingestion.ts
@@ -269,7 +269,7 @@ async function emitToOverflow(queue: IngestionConsumer, kafkaMessages: Message[]
                 // ``message.key`` should not be undefined here, but in the
                 // (extremely) unlikely event that it is, set it to ``null``
                 // instead as that behavior is safer.
-                key: useRandomPartitioning ? null : message.key ?? null,
+                key: undefined,
                 headers: message.headers,
                 waitForAck: true,
             })