[BUG] Cannot acknowledge messages that come in without a topicName in MessageID

[banum2-ppb]

Description

Depending on the producer (fs-client, java, KOP etc) sometimes messages come in without a topicName inside MessageID

Acknowledging this type of message breaks in
await _subConsumers[messageId.Topic].Acknowledge(messageId, cancellationToken).ConfigureAwait(false);
because messageId.Topic == ""

the consumer already has the topicName and I think the Acknowledge should create the topicName based on the other fields if the topicName is missing.

Reproduction Steps

.

Expected behavior

.

Actual behavior

.

Regression?

No response

Known Workarounds

Recreating the messageID is a work-around for this issue.

var ackM = new MessageId(message.MessageId.LedgerId, 
                message.MessageId.EntryId, 
                message.MessageId.Partition,
                message.MessageId.BatchIndex, 
                $"{_consumer.Topic}-partition-{message.MessageId.Partition}");
await _consumer!.Acknowledge(ackM);

Configuration

No response

Other information

No response

[blankensteiner]

Hi @banum2-ppb
When the topic is empty, could this be because the message is part of a batch?

[banum2-ppb]

no, batching is disabled entirely. Official java library / dotpulsar sends it with the topicName set, fs-client / kafka KOP do not