sarama消费kafka协议版本问题

func (r *responseHeader) decode(pd packetDecoder) (err error) {
    r.length, err = pd.getInt32()
    if err != nil {
        return err
    }
    if r.length <= 4 || r.length > MaxResponseSize {
        return PacketDecodingError{fmt.Sprintf("message of length %d too large or too small", r.length)}
    }

    r.correlationID, err = pd.getInt32()
    return err
}

Fetch struct {
            // The minimum number of message bytes to fetch in a request - the broker
            // will wait until at least this many are available. The default is 1,
            // as 0 causes the consumer to spin when no messages are available.
            // Equivalent to the JVM's `fetch.min.bytes`.
            Min int32
            // The default number of message bytes to fetch from the broker in each
            // request (default 32768). This should be larger than the majority of
            // your messages, or else the consumer will spend a lot of time
            // negotiating sizes and not actually consuming. Similar to the JVM's
            // `fetch.message.max.bytes`.
            Default int32
            // The maximum number of message bytes to fetch from the broker in a
            // single request. Messages larger than this will return
            // ErrMessageTooLarge and will not be consumable, so you must be sure
            // this is at least as large as your largest message. Defaults to 0
            // (no limit). Similar to the JVM's `fetch.message.max.bytes`. The
            // global `sarama.MaxResponseSize` still applies.
            Max int32
        }

func (bc *brokerConsumer) fetchNewMessages() (*FetchResponse, error) {
    request := &FetchRequest{
        MinBytes:    bc.consumer.conf.Consumer.Fetch.Min,
        MaxWaitTime: int32(bc.consumer.conf.Consumer.MaxWaitTime / time.Millisecond),
    }
    if bc.consumer.conf.Version.IsAtLeast(V0_10_0_0) {
        request.Version = 2
    }
    if bc.consumer.conf.Version.IsAtLeast(V0_10_1_0) {
        request.Version = 3
        request.MaxBytes = MaxResponseSize
    }
    if bc.consumer.conf.Version.IsAtLeast(V0_11_0_0) {
        request.Version = 4
        request.Isolation = ReadUncommitted // We don't support yet transactions.
    }

    for child := range bc.subscriptions {
        request.AddBlock(child.topic, child.partition, child.offset, child.fetchSize)
    }

    return bc.broker.Fetch(request)
}

func (child *partitionConsumer) parseResponse(response *FetchResponse) ([]*ConsumerMessage, error) {
    ........
    if nRecs == 0 {
        partialTrailingMessage, err := block.Records.isPartial()
        if err != nil {
            return nil, err
        }
        // We got no messages. If we got a trailing one then we need to ask for more data.
        // Otherwise we just poll again and wait for one to be produced...
        if partialTrailingMessage {
            if child.conf.Consumer.Fetch.Max > 0 && child.fetchSize == child.conf.Consumer.Fetch.Max {
                // we can't ask for more data, we've hit the configured limit
                child.sendError(ErrMessageTooLarge)
                child.offset++ // skip this one so we can keep processing future messages
            } else {
                child.fetchSize *= 2
                if child.conf.Consumer.Fetch.Max > 0 && child.fetchSize > child.conf.Consumer.Fetch.Max {
                    child.fetchSize = child.conf.Consumer.Fetch.Max
                }
            }
        }

        return nil, nil
    }

    // we got messages, reset our fetch size in case it was increased for a previous request
    child.fetchSize = child.conf.Consumer.Fetch.Default
    ........
}

Fetch Request (Version: 1) => replica_id max_wait_time min_bytes [topics] 
  replica_id => INT32
  max_wait_time => INT32
  min_bytes => INT32
  topics => topic [partitions] 
    topic => STRING
    partitions => partition fetch_offset partition_max_bytes 
      partition => INT32
      fetch_offset => INT64
      partition_max_bytes => INT32

Fetch Request (Version: 3) => replica_id max_wait_time min_bytes max_bytes [topics]
  replica_id => INT32 
  max_wait_time => INT32
  min_bytes => INT32
  max_bytes => INT32
  topics => topic [partitions]
    topic => STRING
    partitions => partition fetch_offset max_bytes
      partition => INT32
      fetch_offset => INT64
      max_bytes => INT32