anchor.diff:    State1 = State#state
       { begin_offset        = NewBeginOffset
       , min_bytes           = F(min_bytes, State#state.min_bytes)
  -    , max_bytes           = F(max_bytes, State#state.max_bytes)
  +    , max_bytes_orig      = F(max_bytes, State#state.max_bytes_orig)
       , max_wait_time       = F(max_wait_time, State#state.max_wait_time)
       , sleep_timeout       = F(sleep_timeout, State#state.sleep_timeout)
       , prefetch_count      = F(prefetch_count, State#state.prefetch_count)
  
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:   State = maybe_delay_fetch_request(State0),
    {noreply, State};
  handle_message_set(#kafka_message_set{messages = [?incomplete_message]},
                     #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?">>}),
    State1 = State0#state{max_bytes = NewMaxBytes},
    State = maybe_send_fetch_request(State1),
    {noreply, State};
  