Skip to content

[To dev/1.3] Subscription: unsubscribe completed topics when remove consumer config (#15660) #15696

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
SteveYurongSu merged 2 commits into from
Jun 16, 2025
Merged

[To dev/1.3] Subscription: unsubscribe completed topics when remove consumer config (#15660) #15696

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
5d8786e
Subscription: unsubscribe completed topics when remove consumer confi…
VGalaxies Jun 9, 2025
4bf5f84
Merge branch 'dev/1.3' into cp-15660
VGalaxies Jun 12, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 28 additions & 0 deletions ...anode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,6 +84,7 @@
import java.nio.ByteBuffer;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
Expand Down Expand Up @@ -160,7 +161,12 @@ public void handleExit() {
LOGGER.info(
"Subscription: remove consumer config {} when handling exit",
consumerConfigThreadLocal.get());
// we should not close the consumer here because it might reuse the previous consumption
// progress to continue consuming
// closeConsumer(consumerConfig);
// when handling exit, unsubscribe from topics that have already been completed as much as
// possible to release some resources (such as the underlying pipe) in a timely manner
unsubscribeCompleteTopics(consumerConfig);
consumerConfigThreadLocal.remove();
}
}
Expand Down Expand Up @@ -309,6 +315,8 @@ private TPipeSubscribeResp handlePipeSubscribeHeartbeatInternal(
// fetch topics should be unsubscribed
final List<String> topicNamesToUnsubscribe =
SubscriptionAgent.broker().fetchTopicNamesToUnsubscribe(consumerConfig, topics.keySet());
// here we did not immediately unsubscribe from topics in order to allow the client to perceive
// completed topics

return PipeSubscribeHeartbeatResp.toTPipeSubscribeResp(
RpcUtils.SUCCESS_STATUS, topics, endPoints, topicNamesToUnsubscribe);
Expand Down Expand Up @@ -682,6 +690,26 @@ private void closeConsumer(final ConsumerConfig consumerConfig) {
LOGGER.info("Subscription: consumer {} close successfully", consumerConfig);
}

private void unsubscribeCompleteTopics(final ConsumerConfig consumerConfig) {
// fetch subscribed topics
final Map<String, TopicConfig> topics =
SubscriptionAgent.topic()
.getTopicConfigs(
SubscriptionAgent.consumer()
.getTopicNamesSubscribedByConsumer(
consumerConfig.getConsumerGroupId(), consumerConfig.getConsumerId()));

// fetch topics should be unsubscribed
final List<String> topicNamesToUnsubscribe =
SubscriptionAgent.broker().fetchTopicNamesToUnsubscribe(consumerConfig, topics.keySet());

unsubscribe(consumerConfig, new HashSet<>(topicNamesToUnsubscribe));
LOGGER.info(
"Subscription: consumer {} unsubscribe {} (completed topics) successfully",
consumerConfig,
topicNamesToUnsubscribe);
}

//////////////////////////// consumer operations ////////////////////////////

private void createConsumer(final ConsumerConfig consumerConfig) throws SubscriptionException {
Expand Down
Loading