bulker: fix for possible panic

jitsucom · Jan 28, 2024 · 443e249 · 443e249
1 parent 2186540
commit 443e249
Showing 1 changed file with 23 additions and 23 deletions.
diff --git a/bulkerapp/app/topic_manager.go b/bulkerapp/app/topic_manager.go
@@ -134,35 +134,35 @@ func (tm *TopicManager) Start() {
 }
 
 func (tm *TopicManager) LoadMetadata() {
-	topicsLastMessageDates := map[string]*time.Time{}
 	metadata, err := tm.kaftaAdminClient.GetMetadata(nil, true, tm.config.KafkaAdminMetadataTimeoutMs)
-	topicPartitionOffsets := make(map[kafka.TopicPartition]kafka.OffsetSpec)
-	for _, topic := range metadata.Topics {
-		t := topic.Topic
-		if !strings.HasPrefix(t, "__") {
-			for _, partition := range topic.Partitions {
-				topicPartitionOffsets[kafka.TopicPartition{Topic: &t, Partition: partition.ID}] = kafka.MaxTimestampOffsetSpec
-			}
-		}
-	}
-	start := time.Now()
-	res, err := tm.kaftaAdminClient.ListOffsets(context.Background(), topicPartitionOffsets)
 	if err != nil {
-		tm.Errorf("Error getting topic offsets: %v", err)
+		metrics.TopicManagerError("load_metadata_error").Inc()
+		tm.Errorf("Error getting metadata: %v", err)
 	} else {
-		for tp, offset := range res.ResultInfos {
-			if offset.Offset >= 0 && offset.Timestamp > 0 {
-				lastMessageDate := time.UnixMilli(offset.Timestamp)
-				topicsLastMessageDates[*tp.Topic] = &lastMessageDate
+		topicsLastMessageDates := map[string]*time.Time{}
+		topicPartitionOffsets := make(map[kafka.TopicPartition]kafka.OffsetSpec)
+		for _, topic := range metadata.Topics {
+			t := topic.Topic
+			if !strings.HasPrefix(t, "__") {
+				for _, partition := range topic.Partitions {
+					topicPartitionOffsets[kafka.TopicPartition{Topic: &t, Partition: partition.ID}] = kafka.MaxTimestampOffsetSpec
+				}
 			}
 		}
-		tm.Debugf("Got topic offsets for %d topics in %v", len(topicsLastMessageDates), time.Since(start))
-	}
+		start := time.Now()
+		res, err := tm.kaftaAdminClient.ListOffsets(context.Background(), topicPartitionOffsets)
+		if err != nil {
+			tm.Errorf("Error getting topic offsets: %v", err)
+		} else {
+			for tp, offset := range res.ResultInfos {
+				if offset.Offset >= 0 && offset.Timestamp > 0 {
+					lastMessageDate := time.UnixMilli(offset.Timestamp)
+					topicsLastMessageDates[*tp.Topic] = &lastMessageDate
+				}
+			}
+			tm.Debugf("Got topic offsets for %d topics in %v", len(topicsLastMessageDates), time.Since(start))
+		}
 
-	if err != nil {
-		metrics.TopicManagerError("load_metadata_error").Inc()
-		tm.Errorf("Error getting metadata: %v", err)
-	} else {
 		tm.processMetadata(metadata, topicsLastMessageDates)
 	}
 }