Ingest storage: configure BrokerMaxReadBytes on Kafka reader (#7148)

* Ingest storage: configure BrokerMaxReadBytes on Kafka reader Signed-off-by: Marco Pracucci <[email protected]> * Update pkg/storage/ingest/reader.go Co-authored-by: Dimitar Dimitrov <[email protected]> --------- Signed-off-by: Marco Pracucci <[email protected]> Co-authored-by: Dimitar Dimitrov <[email protected]>
grafana · Jan 17, 2024 · 5c37e4e · 5c37e4e
1 parent ac3cb58
commit 5c37e4e
Showing 1 changed file with 8 additions and 1 deletion.
diff --git a/pkg/storage/ingest/reader.go b/pkg/storage/ingest/reader.go
@@ -265,15 +265,22 @@ func (r *PartitionReader) recordFetchesMetrics(fetches kgo.Fetches) {
 }
 
 func (r *PartitionReader) newKafkaReader(at kgo.Offset) (*kgo.Client, error) {
+	const fetchMaxBytes = 100_000_000
+
 	opts := append(
 		commonKafkaClientOptions(r.kafkaCfg, r.metrics.kprom, r.logger),
 		kgo.ConsumePartitions(map[string]map[int32]kgo.Offset{
 			r.kafkaCfg.Topic: {r.partitionID: at},
 		}),
 		kgo.FetchMinBytes(1),
-		kgo.FetchMaxBytes(100_000_000),
+		kgo.FetchMaxBytes(fetchMaxBytes),
 		kgo.FetchMaxWait(5*time.Second),
 		kgo.FetchMaxPartitionBytes(50_000_000),
+
+		// BrokerMaxReadBytes sets the maximum response size that can be read from
+		// Kafka. This is a safety measure to avoid OOMing on invalid responses.
+		// franz-go recommendation is to set it 2x FetchMaxBytes.
+		kgo.BrokerMaxReadBytes(2*fetchMaxBytes),
 	)
 	client, err := kgo.NewClient(opts...)
 	if err != nil {