kgo: add min fetch bytes option

Previously, this was hardcoded to be 1. May as well add it as an option.

pkg/kgo/config.go

@@ -95,6 +95,7 @@ type cfg struct {
 
 	// ***CONSUMER SECTION***
 	maxWait        int32
+	minBytes       int32
 	maxBytes       int32
 	maxPartBytes   int32
 	resetOffset    Offset
@@ -196,6 +197,7 @@ func defaultCfg() cfg {
 		partitioner:         StickyKeyPartitioner(nil), // default to how Kafka partitions
 
 		maxWait:        5000,
+		minBytes:       1,
 		maxBytes:       50 << 20,
 		maxPartBytes:   10 << 20,
 		resetOffset:    NewOffset().AtStart(),
@@ -662,6 +664,18 @@ func FetchMaxBytes(b int32) ConsumerOpt {
 	return consumerOpt{func(cfg *cfg) { cfg.maxBytes = b }}
 }
 
+// FetchMinBYtes sets the minimum amount of bytes a broker will try to send
+// during a fetch, overriding the default 1 byte.
+//
+// With the default of 1, data is sent as soon as it is available. By bumping
+// this, the broker will try to wait for more data, which may improve server
+// throughput at the expense of added latency.
+//
+// This corresponds to the Java fetch.min.bytes setting.
+func FetchMinBytes(b int32) ConsumerOpt {
+	return consumerOpt{func(cfg *cfg) { cfg.minBytes = b }}
+}
+
 // FetchMaxPartitionBytes sets the maximum amount of bytes that will be
 // consumed for a single partition in a fetch request, overriding the default
 // 10MiB. Note that if a single batch is larger than this number, that batch

pkg/kgo/source.go

@@ -345,6 +345,7 @@ func (s *source) unuseAll(reqOffsets map[string]map[int32]*seqOffsetFrom) {
 func (s *source) createReq() (req *fetchRequest, again bool) {
 	req = &fetchRequest{
 		maxWait:        s.cl.cfg.maxWait,
+		minBytes:       s.cl.cfg.minBytes,
 		maxBytes:       s.cl.cfg.maxBytes,
 		maxPartBytes:   s.cl.cfg.maxPartBytes,
 		rack:           s.cl.cfg.rack,
@@ -1045,6 +1046,7 @@ func v1MessageToRecord(
 type fetchRequest struct {
 	version      int16
 	maxWait      int32
+	minBytes     int32
 	maxBytes     int32
 	maxPartBytes int32
 	rack         string
@@ -1094,7 +1096,7 @@ func (f *fetchRequest) AppendTo(dst []byte) []byte {
 		Version:        f.version,
 		ReplicaID:      -1,
 		MaxWaitMillis:  f.maxWait,
-		MinBytes:       1,
+		MinBytes:       f.minBytes,
 		MaxBytes:       f.maxBytes,
 		IsolationLevel: f.isolationLevel,
 		SessionID:      f.session.id,