Describe the bug
Creating a source connector will make pulsar always wait for the previous message to be acknowledged before producing the next message. This is caused by this future.join().

To Reproduce
Steps to reproduce the behavior:

1. Run this test souce connector with --processing-guarantees EFFECTIVELY_ONCE
2. Run the same source connector with --processing-guarantees ATLEAST_ONCE
3. Read the logs to see how much higher the throughput is with ATLEAST_ONCE. For me the throughput with EFFECTIVELY_ONCE was only 90 ack/s and over 300 000 ack/s with ATLEAST_ONCE.

Expected behavior
Throughput with --processing-guarantees EFFECTIVELY_ONCE should be almost as high as with --processing-guarantees ATLEAST_ONCE.

Desktop (please complete the following information):

• OS: Ubuntu 18.04
• pulsar 2.5.0 standalone

Additional context
I haven't tested but I think that this issue should affect regular pulsar functions as well.

I have asked @merlimat why the future.join() is required, but he didn't remember why it was added and asked me to open an issue.