Producer respects Context passed to Send() and SendAsync() when appying backpressure

[dferstay]

Previously, the Producer ignored the context passed to Send() and
SendAsync().

Now, the Producer respects the context in the case where the
ProducerOptions.MaxPendingMessages limit is breached. In this case, the
producer will block until a permit for sending the message is available or
until the context expires, whichever comes first.

Failures to send messages due to context expiration are communicated to
callers via the existing TimeoutError error code.

Signed-off-by: Daniel Ferstay [email protected]

Motivation

When the Producer is applying backpressure to the caller it will block attempting to acquire a send permit.

Before this change change, the producer would block indefinitely as a send permit must be acquired before the send timeout can be checked by the internal runEventsLoop go-routine.

With this change, callers can control the duration they are willing to wait for a send permit by using the context passed to the Send() and SendAsync() methods.

Modifications

• Modify Semaphore.Acquire() to accept a context and block until acquisition is complete or the passed context is expired, whichever comes first
• Modify partitionProducer.internalSendAsync() to propagate the calling context when attempting to acquire the publish semaphore, and fail the request if the acquire is unsuccessful.

Verifying this change

• additional unit tests for the Sempahore
• additional producer_test that verifies publish on context expiration while applying backpressure

[cckellogg]

/pulsarbot run-failure-checks