Add producer.flush() before cleanup to ensure all buffered messages are sent

The current implementation calls `producer.close()` immediately in the cleanup method without first flushing the producer's internal buffer. Messages can accumulate in the buffer after `sendAsync()` returns, and closing without flushing may result in message loss. Should call producer.flush() before closing to ensure all pending messages are immediately transmitted to the Pulsar broker.