This repository was archived by the owner on Jan 24, 2024. It is now read-only.
Close offset consumers when channel is inactive#425
Merged
jiazhai merged 4 commits intostreamnative:masterfrom Apr 6, 2021
Merged
Close offset consumers when channel is inactive#425jiazhai merged 4 commits intostreamnative:masterfrom
jiazhai merged 4 commits intostreamnative:masterfrom
Conversation
BewareMyPower
changed the title
Collaborator
Author
|
Mark it as WIP since some tests failed with unknown reason, and integration tests all failed. |
BewareMyPower
changed the title
jiazhai
approved these changes
Apr 6, 2021
jiazhai
pushed a commit
that referenced
this pull request
Apr 8, 2021
### Motivation `testDeleteClosedTopics` is very easy to fail in CI environment because there's a race condition. #425 closes a group's offset consumers when the channel becomes inactive. However, before Kafka consumer closes, it sends a `COMMIT_OFFSET` request to commit offsets, which is the behavior of the default `enable.auto.commit=true` config. Currently, KoP acknowledges the offset's associated message id in the callback of `GroupMetadataManager#storeGroup`, see `GroupCoordinator#handleCommitOffsets`: ```java result.whenCompleteAsync((ignore, e) ->{ if (e == null){ offsetAcker.ackOffsets(groupId, offsetMetadata); } }); ``` So it's asynchronous with `handleCommitOffsets` itself. There's a possibility that after the channel was closed and `OffsetAcker`'s consumer was removed and closed, `OffsetAcker#ackOffsets` would be invoked again, and `getConsumer` would be invoked so that a new offset consumer would be created to the topic. Here're some related logs in CI tests as the evidence. > 15:49:07.884 [pulsar-client-io-40-1:org.apache.pulsar.client.impl.ConsumerImpl@698] INFO org.apache.pulsar.client.impl.ConsumerImpl - [persistent://public/default/test-delete-closed-topics-partition-0][sub-2] Subscribing to topic on cnx [id: 0xb7 fd0eb7, L:/127.0.0.1:50468 - R:localhost/127.0.0.1:15002], consumerId 2 > 15:49:07.980 [TestNG-method=testDeleteClosedTopics-1:org.apache.kafka.clients.consumer.KafkaConsumer@2152] DEBUG org.apache.kafka.clients.consumer.KafkaConsumer - [Consumer clientId=consumer-2, groupId=sub-2] Kafka consumer has been closed > 15:49:07.986 [pulsar-client-io-40-1:org.apache.pulsar.client.impl.ConsumerImpl@698] INFO org.apache.pulsar.client.impl.ConsumerImpl - [persistent://public/default/test-delete-closed-topics-partition-0][sub-2] Subscribing to topic on cnx [id: 0xb7fd0eb7, L:/127.0.0.1:50468 - R:localhost/127.0.0.1:15002], consumerId 3 > ... > org.apache.pulsar.broker.service.BrokerServiceException$TopicBusyException: Topic has 1 connected producers/consumers We can see after the Kafka consumer was closed, a new offset consumer was created again. ### Modifications `ackOffsets` does the acknowledgement asynchronously but may triggers the creation of an offset consumer, so this PR makes `OffsetAcker#ackOffsets` be called before `GroupMetadataManager#storeGroup` and it prevents an offset consumer being created after channel is inactive. Besides, it adds some condition checks to the test. In CI environment, sometimes topics cannot be deleted by 404 (`Partitioned topic does not exist`).
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
Sign up for free
to subscribe to this conversation on GitHub.
Already have an account?
Sign in.
2 participants
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.
Fixes #229
Motivation
The consumers of
OffsetAckerare used to commit offsets only. However, they would never be closed. Therefore the topics that are consumed by Kafka clients cannot be deleted because topic has active producers/subscriptions.Modifications
SYNC_GROUPrequest, which will callOffsetAcker#addOffsetsTrackerto add offset consumers if success.OffsetAcker#getConsumer(called byackOffsets) will recreate the consumer if it doesn't exist.