Message loss with message delivery not guaranteed

[satishviswanathan]

Description

I have a pulsar client created with a one producer created for a topic and ran a load tests and I could see that few of the messages are lost. In the current implementation is message delivery guaranteed with a publisher confirmation ?

Pulsar broker :

Current version of pulsar is: 4.1.0

Broker Config

broker.conf.txt

Library Version: 4.3.2

Client and Producer code

builder = PulsarClient.Builder()
.Authentication(AuthenticationFactory.Token(TOKEN))
.CheckCertificateRevocation(true)
.ExceptionHandler(new CustomExceptionHandler())
.KeepAliveInterval(new TimeSpan(0, 0, 30))
.RetryInterval(new TimeSpan(0, 0, 2))
.CloseInactiveConnectionsInterval(new TimeSpan(0, 0, 30));
builder.ConnectionSecurity(EncryptionPolicy.EnforceEncrypted)
.ServiceUrl(new Uri($"pulsar+ssl://{pulsarHostName}:6651"))
.TrustedCertificateAuthority(Certificate.GenerateCertificate(certificateOption.Value.Pulsar.Ca))
.VerifyCertificateAuthority(true)
.VerifyCertificateName(false);

_client = builder.Build();
_producer = _client.NewProducer(Schema.ByteArray)
.ProducerName($"{@event}-{Environment.MachineName}")

 .StateChangedHandler(producerStateChanged =>
 {
     var topicName = producerStateChanged.Producer.Topic;
     var state = producerStateChanged.ProducerState;

     _logger.LogDebug(new ApplicationLogMessage
     {
         Message = $"The producer for topic '{topicName}' changed state to '{state}'"
     });
 })
 .Topic($"persistent://global/seed/Test)
 .Create();

Publisher code

var messageId = await producer.Send(new MessageMetadata(), byteArray, CancellationToken.None).ConfigureAwait(false);

Reproduction Steps

The publisher code is exposed via an http end point.

Run a load test

Number of users : 10
Number of iterations : 20

Expected behavior

Expected: 200 Messages to be published.

Actual behavior

There are few messages loss with EntryId and LedgerId returned as negative values

Regression?

No response

Known Workarounds

No response

Configuration

No response

Other information

No response

[blankensteiner]

Calling 'producer.Send' gives you a ValueTask<MessageId>, and this is not completed until the broker has confirmed receiving the message and giving it a message ID. So if Send is completed and you get a message ID, but the message is not on the topic, then it sounds like an issue with the broker (or your configuration of the broker and/or topic).

[satishviswanathan]

Thank you @blankensteiner , I got the MessageId but the EntryId and LedgeId returned within the MessageId are -ve values. So does that mean it's a broker issue or library wasn't able to get a confirmation.

I have update the Original thread with pulsar version and broker config we use in K8s our cluster

[blankensteiner]

When you send a message using DotPulsar it will send a "CommandSend" to the broker. The broker will then send either a "CommandSendReceipt" or "CommandSendError" in response to that. If DotPulsar gets a CommandSendError, you would get an exception; if DotPulsar gets a CommandSendReceipt, you get the message ID. Since you are getting a message ID, it seems that the broker is telling DotPulsar that it has received and stored the message.

[satishviswanathan]

Thank you @blankensteiner for your help. MessageId is returned but what does this negative values below mean ?

"partitionKey": "none",
"partition": -1,
"ledgerId": -9223372036854800000.0,
"batchIndex": -1,
"eventType": "TopicIntegrationEvent",
"entryId": -9223372036854800000.0,
"dataSize": 1161,
"eventName": "TopicIntegrationEvent"

[blankensteiner]

A MessageId consists of ledgerId, entryId, partition, batchIndex, and topic. Where did you get "partitionKey", "eventType", "dataSize", and "eventName" from? and where is the topic?

[satishviswanathan]

Internally we were logging other params , yes as you mentioned we are getting only the following properties in messageid.

"partition": -1,
"ledgerId": -9223372036854800000.0,
"batchIndex": -1,
"entryId": -9223372036854800000.0,

during the load test we got few of them returned with -ve values and found those were the missing ones. Does that mean publisher couldn't confirm or something missing from our end ?

Load Test

Service which exposes an end point to publish a message to a topic. Heave 2 instances of this services running and the load to these instances are load balanced in a round robin fashion. So with this setup we have 2 producers actively publishing into this topic who's subscription is keyshared.

[blankensteiner]

I would suggest talking with the Pulsar community (maybe on slack). There must be someone with knowledge of the protocol and broker that can answer this. From DotPulsar's point of view, with the current implementation, we expect the broker to have successfully stored the message if CommandSendReceipt is received as a response to CommandSend. Maybe this has to change, but I think it would be weird to not use CommandSendError to signal that something is not right.

[satishviswanathan]

Thank you. Do you see anything wrong with client implementation code that I have shared ?

In the pulsar conf I updated the flag brokerDeduplicationEnabled as false and after that I'm not seeing the -ve values returned. Does that mean the SequenceId uniqueness is not guaranteed per producer.

during the load tests my application didn’t have any restarts but just wanted to check if my observations were right.

Scenario

While creating the producer there is an option to set the InitialSequenceId() if that is not set then I guess it defaulted as 0. So If my application restarts then the sequenceId will be reset as 0 and will there be a chance of generating a sequenceId which was generated in the past? Should the producer track and save the sequenceId's so that the InitialSequenceId() is set with the latest value whenever the producer restarts?

[blankensteiner]

Thank you. Do you see anything wrong with client implementation code that I have shared ?

No, I do not.

In the pulsar conf I updated the flag brokerDeduplicationEnabled as false and after that I'm not seeing the -ve values returned. Does that mean the SequenceId uniqueness is not guaranteed per producer.

I guess so, but you would have to check with the Pulsar community (it is not DotPulsar specific).

Should the producer track and save the sequenceId's so that the InitialSequenceId() is set with the latest value whenever the producer restarts?

Yes.

[satishviswanathan]

Thank you. Do you see anything wrong with client implementation code that I have shared ?

No, I do not.

In the pulsar conf I updated the flag brokerDeduplicationEnabled as false and after that I'm not seeing the -ve values returned. Does that mean the SequenceId uniqueness is not guaranteed per producer.

I guess so, but you would have to check with the Pulsar community (it is not DotPulsar specific).

Should the producer track and save the sequenceId's so that the InitialSequenceId() is set with the latest value whenever the producer restarts?

Yes.

By looking at the dotpulsar library code I thought the sequenceId is generate by the library and not by pulsar. So you suspect pulsar is not honoring producer name and sequenceId combination uniqueness ?

[blankensteiner]

The end-user, creating the producer, has to keep track of the sequenceId. I don't know what I suspect, which is why someone with more knowledge of the protocol/broker may be more helpful. Could be that the messageIds you describe are signaling that the message is a duplicate.

[praveengopalan]

We have initiated a new thread in the Pulsar Slack channel:
https://apache-pulsar.slack.com/archives/C5Z4T36F7/p1759732652521889

According to the response generated by the AI assistant (available here: https://apache-pulsar.slack.com/archives/C07CA4STH99/p1759732676588439
), this appears to be a known issue in Pulsar, particularly when using non-Java clients or batching. It has been identified as a client-side batching issue, for which the Java client already includes a fix.

Could you please confirm if a similar fix is planned or expected for the DotPulsar as well?

[satishviswanathan]

@blankensteiner - I guess Pulsar java client had this issue addressed in v3.1.0.

https://pulsar.apache.org/release-notes/versioned/pulsar-3.1.0/

[blankensteiner]

Hi @satishviswanathan and @praveengopalan
A PR is most welcome, otherwise I need a detailed description of what needs to change.