bulker topic_manager: support bulker sharding

jitsucom · Mar 5, 2024 · 25a661a · 25a661a
1 parent 57b10db
commit 25a661a
Showing 1 changed file with 1 addition and 1 deletion.
diff --git a/bulkerapp/app/topic_manager.go b/bulkerapp/app/topic_manager.go
@@ -187,12 +187,12 @@ func (tm *TopicManager) processMetadata(metadata *kafka.Metadata, nonEmptyTopics
 	staleTopics := utils.NewSet[string]()
 
 	for topic, topicMetadata := range metadata.Topics {
+		allTopics.Put(topic)
 		hash := utils.HashStringInt(topic)
 		topicShardNum := hash % uint32(tm.config.ShardsCount)
 		if int(topicShardNum) != tm.shardNumber {
 			continue
 		}
-		allTopics.Put(topic)
 		if tm.abandonedTopics.Contains(topic) {
 			abandonedTopicsCount++
 			continue