[GitHub] [pulsar] candlerb opened a new issue #5666: Python, Go APIs: producer.flush() not blocking

[GitBox <gi...@apache.org>]

candlerb opened a new issue #5666: Python, Go APIs: producer.flush() not blocking
URL: https://github.com/apache/pulsar/issues/5666
 
 
   **Describe the bug**
   `producer.flush()` returns before all message acknowledgement callbacks have been invoked
   
   **To Reproduce**
   
   ```
   from collections import defaultdict
   import pulsar
   import time
   
   NUM_MESSAGES = 500
   MESSAGE_SIZE = 1_000_000
   
   client = pulsar.Client('pulsar://localhost:6650')
   
   producer = client.create_producer('temp', producer_name='fred')
   
   # Ensure that stdout is flushed on each line
   old_print = print
   def print(*a, **b):
       old_print(*a, flush=True, **b)
   
   sent = 0
   bytes = 0
   results = defaultdict(lambda: 0)
   def ack(res, msg):
       global sent, bytes
       sent += 1
       bytes += len(msg.data())
       results[str(res)] += 1
   
   for i in range(NUM_MESSAGES):
       data = b"x" * MESSAGE_SIZE
       producer.send_async(data, callback=ack)
   
   print("Flushing...")
   producer.flush()
   print("Flush complete")
   for i in range(600):
       print("Sent: %d messages %d bytes" % (sent, bytes))
       if sent == NUM_MESSAGES:
           break
       time.sleep(0.1)
   else:
       print("Never got all the messages!")
   print("Results: %r" % dict(results))
   
   producer.close()
   client.close()
   ```
   
   Output:
   
   ```
   Flushing...
   Flush complete
   Sent: 150 messages 150000000 bytes
   Sent: 172 messages 172000000 bytes
   Sent: 180 messages 180000000 bytes
   Sent: 197 messages 197000000 bytes
   Sent: 215 messages 215000000 bytes
   Sent: 226 messages 226000000 bytes
   Sent: 226 messages 226000000 bytes
   Sent: 226 messages 226000000 bytes
   Sent: 226 messages 226000000 bytes
   Sent: 226 messages 226000000 bytes
   Sent: 226 messages 226000000 bytes
   Sent: 226 messages 226000000 bytes
   Sent: 226 messages 226000000 bytes
   Sent: 226 messages 226000000 bytes
   Sent: 228 messages 228000000 bytes
   Sent: 242 messages 242000000 bytes
   Sent: 258 messages 258000000 bytes
   Sent: 268 messages 268000000 bytes
   Sent: 269 messages 269000000 bytes
   Sent: 273 messages 273000000 bytes
   Sent: 279 messages 279000000 bytes
   Sent: 314 messages 314000000 bytes
   Sent: 331 messages 331000000 bytes
   Sent: 331 messages 331000000 bytes
   Sent: 341 messages 341000000 bytes
   Sent: 356 messages 356000000 bytes
   Sent: 367 messages 367000000 bytes
   Sent: 380 messages 380000000 bytes
   Sent: 396 messages 396000000 bytes
   Sent: 426 messages 426000000 bytes
   Sent: 438 messages 438000000 bytes
   Sent: 452 messages 452000000 bytes
   Sent: 470 messages 470000000 bytes
   Sent: 479 messages 479000000 bytes
   Sent: 480 messages 480000000 bytes
   Sent: 500 messages 500000000 bytes
   Results: {'Ok': 500}
   ```
   
   **Expected behavior**
   The [documentation](https://github.com/apache/pulsar/blob/master/pulsar-client-cpp/python/pulsar/__init__.py#L892) for "flush" says:
   
   > Flush all the messages buffered in the client and wait until all messages have been successfully persisted
   
   Therefore I was expecting that by the time `producer.flush()` returns, all messages would have been persisted *and* the related callbacks invoked.
   
   I guess it is possible that `flush()` has learned from the server that all messages have been persisted *before* the callbacks have been notified, but it seems weird.
   
   It's unclear if flush() is only intended to be used when [batching](https://pulsar.apache.org/docs/en/concepts-messaging/#batching) is enabled.  However even then, since batches can be delivered asynchronously in the background, I would still expect flush() to wait until *all* messages have been delivered, as the documentation says.
   
   **Additional context**
   Supercedes #5664 
   
   I can get the same behaviour in Go - #5664 has sample code.

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services