anchor.diff:  
   -module(kpro).
   
  -%% help apis to build some of most common kpro_Xxx structures.
  +%% help apis to build some of most common kpro_xxx structures.
   -export([ fetch_request/6
  -        , offset_request/4
  +        , offsets_request/4
           , produce_request/5
           , produce_request/6
           ]).
  
positive.code: 
  -define(INT, signed-integer).
  
  %% @doc Help function to contruct a #kpro_OffsetRequest{} for requests
  %% against one single topic-partition.
  %% @end
  -spec offset_request(topic(), partition(), integer(), non_neg_integer()) ->
          kpro_OffsetRequest().
  offset_request(Topic, Partition, Time, MaxNoOffsets) ->
    PartitionReq =
      #kpro_OffsetRequestPartition{ partition          = Partition
                                  , time               = Time
                                  , maxNumberOfOffsets = MaxNoOffsets
                                  },
    TopicReq =
      #kpro_OffsetRequestTopic{ topicName                = Topic
                              , offsetRequestPartition_L = [PartitionReq]
                              },
    #kpro_OffsetRequest{ replicaId            = ?KPRO_REPLICA_ID
                       , offsetRequestTopic_L = [TopicReq]
                       }.
  
  %% @doc Help function to construct a #kpro_FetchRequest{} against one signle
  %% topic-partition.
  %% @end
  -spec fetch_request(topic(), partition(), offset(),
                      non_neg_integer(), non_neg_integer(), pos_integer()) ->
                        kpro_FetchRequest().
  fetch_request(Topic, Partition, Offset, MaxWaitTime, MinBytes, MaxBytes) ->
    PerPartition =
      #kpro_FetchRequestPartition{ partition   = Partition
                                 , fetchOffset = Offset
                                 , maxBytes    = MaxBytes
                                 },
    PerTopic =
      #kpro_FetchRequestTopic{ topicName               = Topic
                             , fetchRequestPartition_L = [PerPartition]
                             },
    #kpro_FetchRequest{ replicaId           = ?KPRO_REPLICA_ID
                      , maxWaitTime         = MaxWaitTime
                      , minBytes            = MinBytes
                      , fetchRequestTopic_L = [PerTopic]
                      }.
  
  
  %% @equiv produce_request(Topic, Partition, KvList, RequiredAcks,
  
negative.code:       decode_response(Rest, [Response | Acc])
    end.
  
  %% @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) ->
    %% 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{ 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,
  