code: func (c *connection) Fetch(req *proto.FetchReq) (*proto.FetchResp, error) {
  	b, err := c.sendRequest(req)
  	if err != nil {
  		return nil, err
  	}
  
  	resp, err := proto.ReadVersionedFetchResp(bytes.NewReader(b), req.GetVersion())
  	if err != nil {
  		return nil, err
  	}
  
  	// Compressed messages are returned in full batches for efficiency
  	// (the broker doesn't need to decompress).
  	// This means that it's possible to get some leading messages
  	// with a smaller offset than requested. Trim those.
  	for ti := range resp.Topics {
  		topic := &resp.Topics[ti]
  		reqTopic := &req.Topics[ti]
  		for pi := range topic.Partitions {
  			partition := &topic.Partitions[pi]
  			requestedOffset := reqTopic.Partitions[pi].FetchOffset
  			i := 0
  			if partition.MessageVersion < 2 {
  				for _, msg := range partition.Messages {
  					if msg.Offset >= requestedOffset {
  						break
  					}
  					i++
  				}
  				partition.Messages = partition.Messages[i:]
  			} else {
  				firstOffset := partition.RecordBatch.FirstOffset
  				for _, rec := range partition.RecordBatch.Records {
  					if firstOffset+rec.OffsetDelta >= requestedOffset {
  						break
  					}
  					i++
  				}
  				partition.RecordBatch.Records = partition.RecordBatch.Records[i:]
  			}
  		}
  	}
  	return resp, nil
  }
docs: // Fetch sends given fetch request to kafka node and returns related response.
  // Calling this method on closed connection will always return ErrClosed.