Sink Connector should publish all outstanding messages on flush

[jordanbull]

When flush is called with a non-full batch, the publisher client may wait up to maxDelayThresholdMs before publishing the remaining messages introducing unnecessary tail latency on every flush. This change will force those messages to be published at the start of the flush call.

[google-cla]

Thanks for your pull request. It looks like this may be your first contribution to a Google open source project (if not, look below for help). Before we can look at your pull request, you'll need to sign a Contributor License Agreement (CLA).

📝 Please visit https://cla.developers.google.com/ to sign.

Once you've signed (or fixed any issues), please reply here with @googlebot I signed it! and we'll verify it.

---

What to do if you already signed the CLA

Individual signers

• It's possible we don't have your GitHub username or you're using a different email address on your commit. Check your existing CLA data and verify that your email is set on your git commits.

Corporate signers

• Your company has a Point of Contact who decides which employees are authorized to participate. Ask your POC to be added to the group of authorized contributors. If you don't know who your Point of Contact is, direct the Google project maintainer to go/cla#troubleshoot (Public version).
• The email used to register you as an authorized contributor must be the email used for the Git commit. Check your existing CLA data and verify that your email is set on your git commits.
• The email used to register you as an authorized contributor must also be attached to your GitHub account.

ℹ️ Googlers: Go here for more info.

[jordanbull]

@googlebot I signed it!

[samarthsingal]

Can you elaborate why this is needed? As implemented the flush() call just processes the results of the publishing and the publisher does it own batching. flush() does not cause any publishing. With this, batching will be difficult to predict since it will at both flush time and when maxDelayThresholdMs elapses. If tail latency is a concern, have you tried just using a lower setting for maxDelayThresholdMs?

[jordanbull]

Given Connect executes both the flush/commit and polls/puts in a synchronous loop, there will be no new messages arriving while waiting for flush to complete. Flush currently is awaiting completion of every message put in the producer anyways. Since there aren't any new messages coming, I figured we might as well send the batch immediately rather than wait for the time to elapse given the number of messages in that batch isn't going to change.

[samarthsingal]

My understanding was that poll/flush cycle was async and the sink would keep polling and delivering more messages regardless of whether the previous put() call had committed. That is my reading from https://github.com/a0x8o/kafka/blob/93185ee0de2349c7b8413f48611f3ce1e10dc951/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java#L215-L232- in each iteration() call it first tries to commit and then poll for new messages.

Is it documented that this is synchronous?

[jordanbull]

My understanding was that poll/flush cycle was async and the sink would keep polling and delivering more messages regardless of whether the previous put() call had committed. That is my reading from https://github.com/a0x8o/kafka/blob/93185ee0de2349c7b8413f48611f3ce1e10dc951/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java#L215-L232- in each iteration() call it first tries to commit and then poll for new messages.

Is it documented that this is synchronous?

Unfortunately, I'm unable to find documentation that explicitly states that (maybe worth contributing to the java docs in connect), but as I read it, both the WorkerSinkTask code and the flush method of the PubSubSinkTask already depend on this fact for correctness.

IE PubSubSinkTask.flush would have a race in awaiting the future results and clearing them at https://github.com/GoogleCloudPlatform/pubsub/blob/master/kafka-connector/src/main/java/com/google/pubsub/kafka/sink/CloudPubSubSinkTask.java#L369

Similarly in WorkerSinkTask polling adds offsets to the un-synchronized currentOffsets map and then the commit call clears that same map https://github.com/apache/kafka/blob/trunk/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java#L571