For non-retriable errors returned from the backend, the client library currently sets future.set_exception() for all outstanding futures:

before it tries to change its state to paused by calling the _batch_done_callback():

This results in race conditions where the callback associated with the future is invoked before the client enters paused state. Which causes issues if the callback attempts to do a resume_publish() before the client is in the paused state.