[kafka] Produce silently drops writes and hangs the response when two topic names alias one bucket #4963

Search before asking

• I searched in the issues and found nothing similar.

Fluss version

main (development)

Please describe the bug 🐞

GatewayKafkaProduceBackend.AppendBatch keys both the append entries and the per-partition result futures by TableBucket:

batch.entriesPerBucket.put(bucket, converted.records());
batch.results.put(bucket, result);

A single Produce request can address the same Fluss table through two topic names: a bare name (t, resolved through kafka.default.database) and the qualified name (fluss.t). Both resolve to the same TableBucket, and the later entry overwrites the earlier one:

1. the first name's records are never appended, while the converted Arrow memory is freed at batch close — silent data loss;
2. the first name's result future is never completed, so the whole Produce response hangs until the client's delivery.timeout.ms;
3. the client then retries the request, duplicating the second name's already-written records.

This is reachable with a real producer. GatewayKafkaMetadataBackend resolves bare names for explicit Metadata requests, and MetadataHandler returns a dedicated alias topic ID for bare names so that one client can cache both names at the same time — using both spellings against one producer is a supported configuration,
e.g. while migrating from bare to qualified names. ProduceHandler.validateRequest only rejects duplicate topic names and per-topic duplicate partition IDs, so t partition 0 plus fluss.t partition 0 passes validation.

Solution

register the result futures with putIfAbsent and complete the later name with INVALID_REQUEST before authorization/transcode

Are you willing to submit a PR?

• I'm willing to submit a PR!