fix(kafka): Fixes partition selection in distributors (#14242)

grafana · Sep 24, 2024 · 3f47233 · 3f47233
1 parent b6e9945
commit 3f47233
Showing 1 changed file with 5 additions and 6 deletions.
diff --git a/pkg/distributor/distributor.go b/pkg/distributor/distributor.go
@@ -901,12 +901,11 @@ func (d *Distributor) sendStreamToKafka(ctx context.Context, stream KeyedStream,
 	if len(stream.Stream.Entries) == 0 {
 		return nil
 	}
-	/*	partitionID, err := d.partitionRing.PartitionRing().ActivePartitionForKey(stream.HashKey)
-		if err != nil {
-			d.kafkaAppends.WithLabelValues("kafka", "fail").Inc()
-			return fmt.Errorf("failed to find active partition for stream: %w", err)
-		}*/
-	partitionID := int32(0)
+	partitionID, err := d.partitionRing.PartitionRing().ActivePartitionForKey(stream.HashKey)
+	if err != nil {
+		d.kafkaAppends.WithLabelValues("kafka", "fail").Inc()
+		return fmt.Errorf("failed to find active partition for stream: %w", err)
+	}
 
 	startTime := time.Now()