[controller] Support dangling topic check and deletion in topic cleanup service. #974
Conversation
This file contains 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
haoxu07
force-pushed
the
danglingTopicDeletion
branch
2 times, most recently
from
a53fd79
to
ae70258
Compare
haoxu07
force-pushed
the
danglingTopicDeletion
branch
from
ae70258
to
3ef9f3a
Compare
lluwm
reviewed
May 9, 2024
Comment on lines
+467
to
+471
| int versionNum = Version.parseVersionFromKafkaTopicName(pubSubTopic.getName()); | ||
| if (!store.containsVersion(versionNum)) { | ||
| LOGGER.info("Will remove dangling version topic {}", pubSubTopic); | ||
| topicsTobeCleanup.add(pubSubTopic); | ||
| } |
There was a problem hiding this comment.
We submitted a change recently to preserve VT data when DIV fatal issue happens. Can we double check that it won't be considered as dandling topics thus got deleted here?
adamxchen
reviewed
May 23, 2024
| @@ -74,13 +82,17 @@ public class TopicCleanupService extends AbstractVeniceService { | |||
| private boolean isRTTopicDeletionBlocked = false; | |||
| private boolean isLeaderControllerOfControllerCluster = false; | |||
| private long refreshQueueCycle = Time.MS_PER_MINUTE; | |||
| private Optional<PubSubAdminAdapter> apacheKafkaAdminAdapter; | |||
There was a problem hiding this comment.
Is there any benefit of using Optional<>? If there's not much value, we can use null if it's absent.
| && System.currentTimeMillis() - danglingTopicCleanupIntervalMs > recentDanglingTopicCleanupTime) { | ||
| List<PubSubTopic> pubSubTopics = collectDanglingTopics(topicsWithRetention); | ||
| if (!pubSubTopics.isEmpty()) { | ||
| LOGGER.info("Find topic discrepancy: topics not in kafka but in xinfra {}", pubSubTopics); |
There was a problem hiding this comment.
we should not expose the proprietary pubsub system name in OSS. And IMO it's more intuitive to say
LOGGER.warn("Find topic discrepancy: {} is not present among all pubsub systems"
| String storeName = pubSubTopic.getStoreName(); | ||
| String clusterDiscovered = admin.discoverCluster(storeName).getFirst(); | ||
| Store store = admin.getStore(clusterDiscovered, storeName); | ||
| LOGGER.info("Find topic discrepancy case: {}", pubSubTopic); |
There was a problem hiding this comment.
I'd suggest for these bad but not the end-of-world case, we use WARN. It's not good to have discrepancy among different pubsub ystems.
| @@ -217,10 +248,12 @@ public void testCleanupVeniceTopics() throws ExecutionException { | |||
| verify(topicManager, never()).ensureTopicIsDeletedAndBlockWithRetry(getPubSubTopic(storeName2, "_rt")); | |||
| // Delete should be blocked by remote VT | |||
| verify(topicManager, never()).ensureTopicIsDeletedAndBlockWithRetry(getPubSubTopic(storeName3, "_rt")); | |||
| verify(topicCleanupServiceStats, atLeastOnce()).recordDeletableTopicsCount(5); | |||
| verify(topicCleanupServiceStats, atLeastOnce()).recordDeletableTopicsCount(6); | |||
There was a problem hiding this comment.
Does the existing test case has covered all 3 cases?
1. Real-time topic, if store is not hybrid, that topic will be deleted.
2. Version topic, if store does not contain that version, the topic will be deleted.
3. If there is no store available for that topic, the topic will be deleted.
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.
Summary, imperative, start upper case, don't end with a period
Add dangling topic checking functionality,
TopicCleanupServicewill compare topics listed from kafka and topics from other type of pub sub client, if there is topic could not be found by kafka client but can be founded by other type of pub sub client, topic will be deleted by such cases:Add a new config to enable this feature:
CONTROLLER_DANGLING_TOPIC_CLEAN_UP_INTERVAL_SECOND. If its value is >0 and admin client class is not kafka, the feature will be turned on. Otherwise, it will be disabled.How was this PR tested?
CI, integration test will be in internal repo with other type of pub sub system.
Does this PR introduce any user-facing changes?