bulker: disable consumers for stale topics

jitsucom · Jan 15, 2024 · 525fe5d · 525fe5d
1 parent 3a409df
commit 525fe5d
Showing 1 changed file with 1 addition and 1 deletion.
diff --git a/bulkerapp/app/topic_manager.go b/bulkerapp/app/topic_manager.go
@@ -316,7 +316,7 @@ func (tm *TopicManager) processMetadata(metadata *kafka.Metadata, lastMessageDat
 			}
 			for _, table := range tables {
 				topicId, _ := MakeTopicId(destination.Id(), "batch", table, false)
-				if !hasTopics || !dstTopics.Contains(topicId) {
+				if (!hasTopics || !dstTopics.Contains(topicId)) && !staleTopics.Contains(topicId) {
 					tm.Infof("Creating topic %s for destination %s", topicId, destination.Id())
 					err := tm.createDestinationTopic(topicId, nil)
 					if err != nil {