bulker: automatically create topics for special destination types: ba…

…ckup and metrics
jitsucom · Dec 21, 2023 · 813fe75 · 813fe75
1 parent 3693c0f
commit 813fe75
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 2 deletions.
diff --git a/bulkerapp/app/configuration_source.go b/bulkerapp/app/configuration_source.go
@@ -23,6 +23,7 @@ type DestinationConfig struct {
 	UsesBulker          bool      `mapstructure:"usesBulker" json:"usesBulker"`
 	bulker.Config       `mapstructure:",squash"`
 	bulker.StreamConfig `mapstructure:",squash"`
+	Special             string `mapstructure:"special" json:"special"`
 }
 
 func (dc *DestinationConfig) Id() string {

diff --git a/bulkerapp/app/topic_manager.go b/bulkerapp/app/topic_manager.go
@@ -249,17 +249,34 @@ func (tm *TopicManager) processMetadata(metadata *kafka.Metadata) {
 		}
 	}
 	for _, destination := range tm.repository.GetDestinations() {
-		dstTopics, ok := tm.consumedTopics[destination.Id()]
+		dstTopics, hasTopics := tm.consumedTopics[destination.Id()]
 		for mode, config := range tm.requiredDestinationTopics {
 			topicId, _ := MakeTopicId(destination.Id(), mode, allTablesToken, false)
-			if !ok || !dstTopics.Contains(topicId) {
+			if !hasTopics || !dstTopics.Contains(topicId) {
 				//tm.Debugf("Creating topic %s for destination %s", topicId, destination.Id())
 				err := tm.createDestinationTopic(topicId, config)
 				if err != nil {
 					tm.Errorf("Failed to create topic %s for destination %s: %v", topicId, destination.Id(), err)
 				}
 			}
 		}
+		if destination.config.Special == "backup" || destination.config.Special == "metrics" {
+			// create predefined tables for special kind of destinations: backup and metrics
+			tables := []string{destination.config.Special}
+			if destination.config.Special == "metrics" {
+				tables = append(tables, "active_incoming")
+			}
+			for _, table := range tables {
+				topicId, _ := MakeTopicId(destination.Id(), "batch", table, false)
+				if !hasTopics || !dstTopics.Contains(topicId) {
+					tm.Infof("Creating topic %s for destination %s", topicId, destination.Id())
+					err := tm.createDestinationTopic(topicId, nil)
+					if err != nil {
+						tm.Errorf("Failed to create topic %s for destination %s: %v", topicId, destination.Id(), err)
+					}
+				}
+			}
+		}
 	}
 	tm.allTopics = allTopics
 	err := tm.ensureTopic(tm.config.KafkaDestinationsTopicName, tm.config.KafkaDestinationsTopicPartitions,