[fix][client]Producer stuck or geo-replication stuck due to wrong value of message.numMessagesInBatch

[poorbarcode]

Motivation

Background

• Replication uses producer to copy messages, which will use the original message metadata(including message.numMessagesInBatch).
• Producer uses a int variable messagesCount to cache how many messages is in pending sending state.
• Producer will resent all pending messages after reconnections, if messagesCount.

Root cause: Producer.resendMessages was skipped due to the following issue

• the client side issue causes message.metadata.numMessagesInBatch being 0, such as send_batch could send entries whose batch size is 0 when the topic is partitioned streamnative/pulsar-rs#376
• the internal producer will copy message metadata from the original messages, which also copied a 0 value message.metadata.numMessagesInBatch.
• since message.metadata.numMessagesInBatch can be 0 due to client-side bugs, producer.pendingMessages.messagesCount will also be 0, and lead to that the producer assumed there is no messages need to be resent. see also https://github.com/apache/pulsar/blob/master/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java#L1808
• the internal producer of replicators went to stuck state.

---

The issue we encountered: there are many pending requests, but producer.messagesCount is 0, which leaded to skipping to resend messages.

Modifications

Producer.resendMessage should only care about whether there is pending requests in producer.pendingMessages, It should not care about how many messages in packages.

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

Matching PR in forked repository

PR in forked repository: x

[codecov-commenter]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 74.44%. Comparing base (fbab357) to head (42a0485).
⚠️ Report is 2 commits behind head on master.

Additional details and impacted files

@@              Coverage Diff              @@
##             master   #25106       +/-   ##
=============================================
+ Coverage     30.84%   74.44%   +43.60%     
- Complexity       51     2346     +2295     
=============================================
  Files          1840     1899       +59     
  Lines        145469   149656     +4187     
  Branches      16907    17393      +486     
=============================================
+ Hits          44866   111418    +66552     
+ Misses        93599    29358    -64241     
- Partials       7004     8880     +1876     

Flag	Coverage Δ
inttests	26.30% <100.00%> (-0.06%)	⬇️
systests	23.03% <50.00%> (+0.04%)	⬆️
unittests	73.97% <100.00%> (?)

Flags with carried forward coverage won't be shown. Click here to find out more.

Files with missing lines	Coverage Δ
...va/org/apache/pulsar/client/impl/ProducerImpl.java	84.34% <100.00%> (+34.04%)	⬆️

... and 1489 files with indirect coverage changes

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.