Fix: KafkaClient.closeConsumer should not block
#73
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
felixschlegel
force-pushed
the
fs-kafka-consumer-close-stop-blocking
branch
2 times, most recently
from
5b17640 to
9e0f40d
Compare
FranzBusch
requested changes
Jun 29, 2023
| guard result == RD_KAFKA_RESP_ERR_NO_ERROR else { | ||
| throw KafkaError.rdKafkaError(wrapping: result) | ||
| } | ||
| try await client.commitSync(message) |
There was a problem hiding this comment.
The underlying APIs don't support cancellation. Let's call this out in the docs of this method.
| case finished | ||
| /// | ||
| /// - Parameter client: Client used for handling the connection to the Kafka cluster. | ||
| case finished(client: KafkaClient) |
There was a problem hiding this comment.
That is a bit weird. We should make sure that the end state actually doesn't retain the client so that we can deinit it. Maybe we need a new state here.
Motivation: Currently our invocation to `rd_kafka_commit` inside of `KafkaCosumer.commitSync` is blocking a cooperative thread. This PR aims to make `KafkaCosumer.commitSync` non-blocking by using the callback-based commit API. Modifications: * move `commitSync` logic to `KafkaClient` * replace the blocking invocation to [rd_kafka_commit](https://docs.confluent.io/platform/current/clients/librdkafka/html/rdkafka_8h.html#ab96539928328f14c3c9177ea0c896c87) with a callback-based invocation to [rd_kafka_commit_queue](https://docs.confluent.io/platform/current/clients/librdkafka/html/rdkafka_8h.html#af76a6a73baa9c2621536e3f6882a3c1a) which is then wrapped inside a `withAsyncThrowingContinuation` statement
Motivation: [rd_kakfa_consumer_close](https://docs.confluent.io/platform/current/clients/librdkafka/html/rdkafka_8h.html#a37b54d329e12d745889defe96e7d043d) was blocking. This PR proposes using the [rd_kakfa_consumer_close_queue](https://docs.confluent.io/platform/current/clients/librdkafka/html/rdkafka_8h.html#a9dd5c18bdfed81c8847b259f0a8d498d) API which is non-blocking and served through the normal poll loop. We now Modifications: * `KafkaClient.consumerClose`: use `rd_kakfa_consumer_close_queue` in favour of `rd_kakfa_consumer_close` * create a new variable `KafkaClient.isConsumerClosed` that indicates if the poll loop needs to continue polling or if it can stop running * updated state management in `KafkaConsumer` to accomodate for polling when the `KafkaConsumer` is in the process of closing Result: Calling `KafkaClient.consumerClose` is not blocking anymore.
felixschlegel
force-pushed
the
fs-kafka-consumer-close-stop-blocking
branch
from
9e0f40d to
dd902ba
Compare
Modifications: * introduce new `KafkaConsumer.StateMachine.State` `.finishing` to avoid retaining `client` in state `.finished` * rename `KafkaConsumer.shutdownGracefully` to `KafkaConsumer.triggerGracefulShutdown` * add note that `KafkaConsumer.commitSync` does not support `Task` cancellation
felixschlegel
force-pushed
the
fs-kafka-consumer-close-stop-blocking
branch
from
a0cb8d8 to
af22ded
Compare
FranzBusch
approved these changes
Jun 30, 2023
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Motivation:
rd_kakfa_consumer_close
was blocking. This PR proposes using the
rd_kakfa_consumer_close_queue
API which is non-blocking and served through the normal poll loop.
Modifications:
KafkaClient.consumerClose: userd_kakfa_consumer_close_queueinfavour of
rd_kakfa_consumer_closeKafkaClient.isConsumerClosedthat indicatesif the poll loop needs to continue polling or if it can stop running
KafkaConsumerto accomodate for pollingwhen the
KafkaConsumeris in the process of closingResult:
Calling
KafkaClient.consumerCloseis not blocking anymore