anchor.diff:    State = maybe_delay_fetch_request(State0),
     {noreply, State};
   handle_message_set(#kafka_message_set{messages = [?incomplete_message]},
  -                   #state{max_bytes = MaxBytes} = State0) ->
  +                   #state{ max_bytes = MaxBytes} = State0) ->
  +  %% max_bytes is too small to fetch ONE complete message, double it.
     NewMaxBytes = MaxBytes * 2,
     NewMaxBytes >= (1 bsl 31) andalso
         erlang:error({max_bytes_overflow,
                      <<"perhaps message corruption in broker?">>}),
  -  error_logger:warning_msg("~p ~p max_bytes ~p is too small, trying with ~p",
  -                           [?MODULE, self(), MaxBytes, NewMaxBytes]),
     State1 = State0#state{max_bytes = NewMaxBytes},
     State = maybe_send_fetch_request(State1),
     {noreply, State};
  
positive.code:              , begin_offset        = BeginOffset
               , max_wait_time       = MaxWaitTime
               , min_bytes           = MinBytes
               , max_bytes           = MaxBytes
               , sleep_timeout       = SleepTimeout
               , prefetch_count      = PrefetchCount
               , socket_pid          = ?undef
               , is_suspended        = false
               , offset_reset_policy = OffsetResetPolicy
               }}.
  
  handle_info(?INIT_SOCKET, #state{subscriber = Subscriber} = State0) ->
  
negative.code:                              | reset_to_earliest
                               | reset_to_latest.
  
  -type bytes() :: non_neg_integer().
  
  -record(state, { client_pid          :: pid()
                 , socket_pid          :: pid()
                 , topic               :: binary()
                 , partition           :: integer()
                 , begin_offset        :: offset_time()
                 , max_wait_time       :: integer()
                 , min_bytes           :: bytes()
                 , max_bytes_orig      :: bytes()
                 , sleep_timeout       :: integer()
                 , prefetch_count      :: integer()
                 , last_corr_id        :: corr_id()
  