Hi All, I've pushed 2010 messages using the

SendAsync

method in the Pulsar Go client, flushing after every 1000 messages. While publishing, no errors are reported. However, when checking the stats, the message counter shows 2010 messages, but on the subscription, the message backlog is only 7 messages (as seen in internal stats). I'm trying to understand what could be causing this discrepancy. Can anyone help me figure out the issue?

#### Stats
{
  "msgRateIn" : 0.0,  
  "msgThroughputIn" : 0.0,
  "msgRateOut" : 0.0,
  "msgThroughputOut" : 0.0,  
  "bytesInCounter" : 41314,
  "msgInCounter" : 2010,  
  "bytesOutCounter" : 0,
  "msgOutCounter" : 0,  
  "averageMsgSize" : 0.0,
  "msgChunkPublished" : false,  
  "storageSize" : 41314,
  "backlogSize" : 41314,
  "publishRateLimitedTimes" : 0, 
  "earliestMsgPublishTimeInBacklogs" : 0,
  "offloadedStorageSize" : 0, 
  "lastOffloadLedgerId" : 0,
  "lastOffloadSuccessTimeStamp" : 0, 
  "lastOffloadFailureTimeStamp" : 0,
  "ongoingTxnCount" : 0, 
  "abortedTxnCount" : 0,
  "committedTxnCount" : 0,
  "publishers" : [ ],
  "waitingPublishers" : 0,  
  "subscriptions" : {
    "test_send_async" : {
      "msgRateOut" : 0.0,      
      "msgThroughputOut" : 0.0,
      "bytesOutCounter" : 0,
      "msgOutCounter" : 0,      
      "msgRateRedeliver" : 0.0,
      "messageAckRate" : 0.0,
      "chunkedMessageRate" : 0,
      "msgBacklog" : 7,
      "backlogSize" : 41314,
      "earliestMsgPublishTimeInBacklog" : 0,
      "msgBacklogNoDelayed" : 7,
      "blockedSubscriptionOnUnackedMsgs" : false,
      "msgDelayed" : 0,
      "unackedMessages" : 0,
      "type" : "None",
      "msgRateExpired" : 0.0,
      "totalMsgExpired" : 0,
      "lastExpireTimestamp" : 0,
      "lastConsumedFlowTimestamp" : 0,
      "lastConsumedTimestamp" : 0,
      "lastAckedTimestamp" : 0,
      "lastMarkDeleteAdvancedTimestamp" : 0,
      "consumers" : [ ],
      "isDurable" : true,
      "isReplicated" : false,
      "allowOutOfOrderDelivery" : false,
      "consumersAfterMarkDeletePosition" : { },
      "nonContiguousDeletedMessagesRanges" : 0,
      "nonContiguousDeletedMessagesRangesSerializedSize" : 0,
      "delayedMessageIndexSizeInBytes" : 0,
      "subscriptionProperties" : { },
      "filterProcessedMsgCount" : 0,
      "filterAcceptedMsgCount" : 0,
      "filterRejectedMsgCount" : 0,
      "filterRescheduledMsgCount" : 0,
      "durable" : true,
      "replicated" : false
    }
  },
  "replication" : { },
  "deduplicationStatus" : "Disabled",
  "nonContiguousDeletedMessagesRanges" : 0,
  "nonContiguousDeletedMessagesRangesSerializedSize" : 0,
  "delayedMessageIndexSizeInBytes" : 0,
  "compaction" : {
    "lastCompactionRemovedEventCount" : 0,
    "lastCompactionSucceedTimestamp" : 0,
    "lastCompactionFailedTimestamp" : 0,
    "lastCompactionDurationTimeInMills" : 0
  },
  "ownerBroker" : "ip-XXXX.ec2.internal:8010"
}

#### Internal Stats
{
  "entriesAddedCounter" : 7,
  "numberOfEntries" : 7,
  "totalSize" : 41314,
  "currentLedgerEntries" : 7,
  "currentLedgerSize" : 41314,
  "lastLedgerCreatedTimestamp" : "2024-02-21T13:31:29.879Z",
  "waitingCursorsCount" : 0,
  "pendingAddEntriesCount" : 0,
  "lastConfirmedEntry" : "2473:6",
  "state" : "LedgerOpened",
  "ledgers" : [ {
    "ledgerId" : 2473,
    "entries" : 0,
    "size" : 0,
    "offloaded" : false,
    "underReplicated" : false
  } ],
  "cursors" : {
    "test_send_async" : {
      "markDeletePosition" : "2473:-1",
      "readPosition" : "2473:-1",
      "waitingReadOp" : false,
      "pendingReadOps" : 0,
      "messagesConsumedCounter" : 0,
      "cursorLedger" : 2474,
      "cursorLedgerLastEntry" : 1,
      "individuallyDeletedMessages" : "[]",
      "lastLedgerSwitchTimestamp" : "2024-02-21T13:31:30.341Z",
      "state" : "Open",
      "active" : false,
      "numberOfEntriesSinceFirstNotAckedMessage" : 0,
      "totalNonContiguousDeletedMessagesRange" : 0,
      "subscriptionHavePendingRead" : false,
      "subscriptionHavePendingReplayRead" : false,
      "properties" : { }
    }
  },
  "schemaLedgers" : [ ],
  "compactedLedger" : {
    "ledgerId" : -1,
    "entries" : -1,
    "size" : -1,
    "offloaded" : false,
    "underReplicated" : false
  }
}

When sending messages asynchronously, you cannot assume that they are sent from the client to the broker until the callback receives a message id from each message that has been sent. This comment about the correct usage of the Pulsar Java client might be helpful: https://github.com/apache/pulsar/pull/22011#issuecomment-1936188709

@Lari Hotari I am calling flush method on producer to ensure all outstanding messages in the producer's internal buffer are acknowledged by the Pulsar broker

That might flush the outstanding messages, but you would still have to have logic to track in the callback that each message was actually sent. I assume that the flush won't return the error. (I don't know the pulsar-client-go well so I might be wrong)

@Lari Hotari: I was tracking callbacks, Haven't got any error in callbacks.

Do you get a message id for every sent message in the callback? Regarding "flush", I think that it only ensures that all queued messages on the client side have been written to the TCP/IP write buffer. There's no other guarantee. That's why checking that every single message is sent is necessary for consistency reasons. It might be possible to relax some checks in some cases since there's an ordering guarantee. If a later message has been sent, the ones before that could be assumed to be sent if no other errors have been received on the client.

@Lari Hotari: You are right, In callback I am not getting message IDs, but error is coming as nil. Anyways since it doesn't seems to be reliable will handle optimisation at our end.

You are right, In callback I am not getting message IDs, but error is coming as nil.

The reason for this is that the messages are still inflight and no timeout has occured. If you'd configure a strict timeout, the message sending could timeout. However, without transactions, you simply don't know if the message has been sent or not when you get a timeout.

Anyways since it doesn't seems to be reliable will handle optimisation at our end.

what's the optimization that you have in mind? When you use asynchronous sending, the application will have to implement a solution for backpressure. One common solution is to use a semaphore so that you limit the number of messages in flight. Until the callback returns a message id, the responsibility of message durability is on the messaging application and depending on the use case, there might have to be a solution where some state tracking is done for ensuring that the input gets successfully processed and results in expected messages being stored in Pulsar for further delivery.