anchor.diff:        decode_response(Rest, [Response | Acc])
     end.
   
  -%% @doc help function to encode kpro_XxxRequest into kafka wire format.
  --spec encode_request(client_id(), corr_id(), kpro_RequestMessage()) -> iodata().
  +%% @doc help function to encode kpro_*_request_v* into kafka wire format.
  +-spec encode_request(client_id(), corr_id(), kpro_request_message()) -> iodata().
   encode_request(ClientId, CorrId, Request) ->
  -  R = #kpro_Request{ correlationId  = CorrId
  -                   , clientId       = ClientId
  -                   , requestMessage = Request
  +  %% FIXME huston, we have a problem
  +  H = #kpro_request_header{ correlation_id  = CorrId
  +                          , client_id       = ClientId
  +                          },
  +  R = #kpro_request{ header = H
  +                   , message = Request
                      },
     encode_request(R).
   
  -%% @doc Encode #kpro_Request{} records into kafka wire format.
  --spec encode_request(kpro_Request()) -> iodata().
  -encode_request(#kpro_Request{ apiVersion     = ApiVersion0
  -                            , correlationId  = CorrId0
  -                            , clientId       = ClientId
  -                            , requestMessage = RequestMessage
  -                            }) ->
  +%% @doc Encode #kpro_request{} records into kafka wire format.
  +-spec encode_request(kpro_request()) -> iodata().
  +encode_request(#kpro_request{ message = RequestMessage,
  +                              header = #kpro_request_header{
  +                                  api_version     = ApiVersion0,
  +                                  correlation_id  = CorrId0,
  +                                  client_id       = ClientId }}) ->
     true = (CorrId0 =< ?MAX_CORR_ID), %% assert
     ApiKey = req_to_api_key(RequestMessage),
     CorrId = (ApiKey bsl ?CORR_ID_BITS) bor CorrId0,
  
positive.code: next_corr_id(CorrId)       -> CorrId + 1.
  
  %% @doc Parse binary stream received from kafka broker.
  %%      Return a list of kpro_Response() and the remaining bytes.
  %% @end
  -spec decode_response(binary()) -> {[kpro_Response()], binary()}.
  decode_response(Bin) ->
    decode_response(Bin, []).
  
  
negative.code: 
  -define(INT, signed-integer).
  
  %% @doc Help function to contruct a #kpro_offsets_request_v0{} for requests
  %% against one single topic-partition.
  %% @end
  -spec offsets_request(topic(), partition(), integer(), non_neg_integer()) ->
          kpro_offsets_request_v0().
  offsets_request(Topic, Partition, Time, MaxNoOffsets) ->
    PartitionReq =
      #kpro_offsets_request_v0_partition{ partition       = Partition
                                        , timestamp       = Time
                                        , max_num_offsets = MaxNoOffsets
                                        },
    TopicReq =
      #kpro_offsets_request_v0_topic{ topic      = Topic
                                    , partitions = [PartitionReq]
                                    },
    #kpro_offsets_request_v0{ replica_id = ?KPRO_REPLICA_ID
                            , topics     = [TopicReq]
                            }.
  
  %% @doc Help function to construct a #kpro_fetch_request_v0{} against one single
  %% topic-partition.
  %% @end
  -spec fetch_request(topic(), partition(), offset(),
                      non_neg_integer(), non_neg_integer(), pos_integer()) ->
                        kpro_fetch_request_v0().
  fetch_request(Topic, Partition, Offset, MaxWaitTime, MinBytes, MaxBytes) ->
    PerPartition =
      #kpro_fetch_request_v0_partition{ partition    = Partition
                                      , fetch_offset = Offset
                                      , max_bytes    = MaxBytes
                                      },
    PerTopic =
      #kpro_fetch_request_v0_topic{ topic      = Topic
                                  , partitions = [PerPartition]
                                  },
    #kpro_fetch_request_v0{ replica_id    = ?KPRO_REPLICA_ID
                          , max_wait_time = MaxWaitTime
                          , min_bytes     = MinBytes
                          , topics        = [PerTopic]
                          }.
  
  
  %% @equiv produce_request(Topic, Partition, KvList, RequiredAcks,
  