Fix thread not waking up when there is still data to be sent

[gqmelo]

When producing messages quickly without waiting for the future of previous requests, there could be some situations when the last batch was not sent.

That seemed to be more frequent with larger messages (~100KiB), but apparently it could happen to any message when linger_ms is 0. Not sure if it could happen when it is non-zero though.

The reason is that BrokerConnection.send_pending_requests_v2 would fill the internal buffer with the bytes from a request and try to send it.

kafka-python/kafka/conn.py

Line 1071 in 512d0a0

self._send_buffer = self._protocol.send_bytes()

If it couldn't send it completely for some reason, it would try to send again in the next call to send_pending_requests_v2.

But if between those 2 calls, BrokerConnection.send was called, new data would be appended to self._protocol: KafkaProtocol:

kafka-python/kafka/conn.py

Line 1015 in 512d0a0

correlation_id = self._protocol.send_request(request)

but the second call to send_pending_requests_v2 wouldn't check if any new data was available and would return False:

kafka-python/kafka/conn.py

Line 1070 in 512d0a0

if not self._send_buffer:

This would tell KafkaClient._poll that all pending data was sent, which would make the client not listen to socked write readiness anymore:

kafka-python/kafka/client_async.py

Lines 744 to 748 in 512d0a0

	if conn.send_pending_requests_v2():
	# If send is complete, we dont need to track write readiness
	# for this socket anymore
	if key.events ^ selectors.EVENT_WRITE:
	self._selector.modify(

[gqmelo]

This is where the test would fail before the fix.