wolff_producer.erl

%% Copyright (c) 2018-2024 EMQ Technologies Co., Ltd. All Rights Reserved.
%%
%% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License.
%% You may obtain a copy of the License at
%%
%%     http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing, software
%% distributed under the License is distributed on an "AS IS" BASIS,
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and
%% limitations under the License.

-module(wolff_producer).

-include_lib("kafka_protocol/include/kpro.hrl").
-include("wolff.hrl").

-define(MIN_DISCARD_LOG_INTERVAL, 5000).

%% APIs
-export([start_link/5, stop/1, notify_stop/2, send/3, send/4, send_sync/3]).

%% gen_server callbacks
-export([code_change/3, handle_call/3, handle_cast/2, handle_continue/2, handle_info/2, init/1, terminate/2]).

%% replayq callbacks
-export([queue_item_sizer/1, queue_item_marshaller/1]).

%% for test
-export([batch_bytes/1, varint_bytes/1]).

-export_type([config_in/0, config_key/0]).

-type topic() :: kpro:topic().
-type partition() :: kpro:partition().
-type offset_reply() :: wolff:offset_reply().
-type config_key() :: group |
                      replayq_dir |
                      replayq_max_total_bytes |
                      replayq_seg_bytes |
                      replayq_offload_mode |
                      required_acks |
                      ack_timeout |
                      max_batch_bytes |
                      max_linger_ms |
                      max_linger_bytes |
                      max_send_ahead |
                      compression |
                      drop_if_highmem |
                      telemetry_meta_data |
                      max_partitions.

-type config_in() :: #{group => wolff_client:producer_group(),
                       replayq_dir => string() | binary(),
                       replayq_max_total_bytes => pos_integer(),
                       replayq_seg_bytes => pos_integer(),
                       replayq_offload_mode => boolean(),
                       required_acks => kpro:required_acks(),
                       ack_timeout => timeout(),
                       max_batch_bytes => pos_integer(),
                       max_linger_ms => non_neg_integer(),
                       max_linger_bytes => non_neg_integer(),
                       max_send_ahead => non_neg_integer(),
                       compression => kpro:compress_option(),
                       drop_if_highmem => boolean(),
                       telemetry_meta_data => map(),
                       max_partitions => pos_integer()
                      }.

%% Some keys are removed from `config_in()'.
-type config_state() :: #{group => wolff_client:producer_group(),
                          replayq_max_total_bytes => pos_integer(),
                          replayq_offload_mode => boolean(),
                          required_acks => kpro:required_acks(),
                          ack_timeout => timeout(),
                          max_batch_bytes => pos_integer(),
                          max_linger_ms => non_neg_integer(),
                          max_linger_bytes => non_neg_integer(),
                          max_send_ahead => non_neg_integer(),
                          compression => kpro:compress_option(),
                          drop_if_highmem => boolean(),
                          telemetry_meta_data => map(),
                          max_partitions => pos_integer()
                         }.

-define(no_timer, no_timer).
-define(reconnect, reconnect).
-define(linger_expire, linger_expire).
-define(linger_expire_timer, linger_expire_timer).
-define(DEFAULT_REPLAYQ_SEG_BYTES, 10 * 1024 * 1024).
-define(DEFAULT_REPLAYQ_LIMIT, 2000000000).
-define(Q_ITEM(CallId, Ts, Batch), {CallId, Ts, Batch}).
-define(SEND_REQ(From, Batch, AckFun), {send, From, Batch, AckFun}).
-define(queued, queued).
-define(no_queued_reply, no_queued_reply).
-define(ACK_CB(AckCb, Partition), {AckCb, Partition}).
-define(no_queue_ack, no_queue_ack).
-define(MAX_LINGER_BYTES, (10 bsl 20)).
-define(EMPTY, empty).
-define(SYNC_REF(Caller, Ref), {Caller, Ref}).
-define(IS_SYNC_REF(Caller, Ref), ((is_reference(Caller) orelse is_pid(Caller)) andalso is_reference(Ref))).
-define(DISCARD_OVERFLOW, "buffer_size_limit").
-define(DISCARD_HIGH_MEM, "high_system_RAM_usage").

-type ack_fun() :: wolff:ack_fun().
-type send_req() :: ?SEND_REQ({pid(), reference()}, [wolff:msg()], ack_fun()).
-type sent() :: #{req_ref := reference(),
                  q_items := [?Q_ITEM(_CallId, _Ts, _Batch)],
                  q_ack_ref := replayq:ack_ref(),
                  attempts := pos_integer()
                 }.
-type sync_refs() :: {pid() | reference(), reference()}.
-type calls() :: #{ts := pos_integer(),
                   bytes := pos_integer(),
                   batch_r := [send_req()]
                  }.
-type state() :: #{ client_id := wolff:client_id()
                  , config := config_state()
                  , conn := undefined | _
                  , ?linger_expire_timer := false | reference()
                  , partition := partition()
                  , pending_acks := wolff_pendack:acks()
                  , produce_api_vsn := undefined | _
                  , replayq := replayq:q()
                  , sent_reqs := queue:queue(sent())
                  , sent_reqs_count := non_neg_integer()
                  , inflight_calls := non_neg_integer()
                  , topic := topic()
                  , calls := ?EMPTY | calls()
                  , sender => pid()
                  }.

%% @doc Start a per-partition producer worker.
%% Configs:
%% * `replayq_dir': `replayq' config to queue messages on disk in order to survive restart.
%% * `replayq_seg_bytes': default 10MB, replayq segment size.
%% * `replayq_max_total_bytes': default 2GB, replayq max total size.
%% * `replayq_offload_mode': default false, replayq work in offload mode if set to true.
%% * `required_acks': `all_isr', `leader_only' or `none'.
%% * `ack_timeout': default 10_000, timeout leader wait for replicas before reply to producer
%% * `max_batch_bytes': Most number of bytes to collect into a produce request.
%%    NOTE: This is only a rough estimation of the batch size towards kafka, NOT the
%%    exact max allowed message size configured in kafka.
%% * `max_linger_ms': Age in milliseconds a baatch can stay in queue when the connection
%%    is idle (as in no pending acks). Default: 0 (as in send immediately)
%% * `max_linger_bytes': Number of bytes to collect before sending it to Kafka.
%%    If set to 0, `max_batch_bytes' is taken for mem-only mode, otherwise it's 10 times
%%    `max_batch_bytes' (but never exceeds 10MB) to optimize disk write.
%% * `max_send_ahead': Number of batches to be sent ahead without receiving ack for
%%    the last request. Must be 0 if messages must be delivered in strict order.
%% * `compression': `no_compression', `snappy' or `gzip'.
-spec start_link(wolff:client_id(), topic(), partition(), pid() | ?conn_down(any()), config_in()) ->
  {ok, pid()} | {error, any()}.
start_link(ClientId, Topic, Partition, MaybeConnPid, Config) ->
  St = #{client_id => ClientId,
         topic => Topic,
         partition => Partition,
         conn => MaybeConnPid,
         config => use_defaults(Config),
         ?linger_expire_timer => false
        },
  %% the garbage collection can be expensive if using the default 'on_heap' option.
  SpawnOpts = [{spawn_opt, [{message_queue_data, off_heap}]}],
  gen_server:start_link(?MODULE, St, SpawnOpts).

stop(Pid) ->
  gen_server:stop(Pid).

notify_stop(Pid, Reason) ->
  gen_server:cast(Pid, {stop, Reason}).

%% @equiv send(Pid, Batch, AckFun, wait_for_queued)
-spec send(pid(), [wolff:msg()], wolff:ack_fun()) -> ok.
send(Pid, Batch, AckFun) ->
  send(Pid, Batch, AckFun, wait_for_queued).

%% @doc Send a batch asynchronously.
%% The callback function is evaluated by producer process when ack is received from kafka.
%% In case `required_acks' is configured to `none', the callback is evaluated immediately after send.
%% NOTE: This API has no backpressure,
%%       high produce rate may cause excessive ram and disk usage.
%%       Even less backpressure when the 4th arg is `no_linger'.
%% NOTE: It's possible that two or more batches get included into one produce request.
%%       But a batch is never split into produce requests.
%%       Make sure it will not exceed the `max_batch_bytes' limit when sending a batch.
-spec send(pid(), [wolff:msg()], wolff:ack_fun() | sync_refs(), WaitForQueued::wait_for_queued | no_wait_for_queued) -> ok.
send(Pid, [_ | _] = Batch0, AckFun, wait_for_queued) ->
  Caller = self(),
  Mref = erlang:monitor(process, Pid),
  Batch = ensure_ts(Batch0),
  erlang:send(Pid, ?SEND_REQ({Caller, Mref}, Batch, AckFun)),
  receive
    {Mref, ?queued} ->
      erlang:demonitor(Mref, [flush]),
      ok;
    {'DOWN', Mref, _, _, Reason} ->
      erlang:error({producer_down, Reason})
  end;
send(Pid, [_ | _] = Batch0, AckFun, no_wait_for_queued) ->
  Batch = ensure_ts(Batch0),
  erlang:send(Pid, ?SEND_REQ(?no_queued_reply, Batch, AckFun)),
  ok.

%% @doc Send a batch synchronously.
%% Raise error exception in case produce pid is down or when timed out.
-spec send_sync(pid(), [wolff:msg()], timeout()) -> {partition(), offset_reply()}.
send_sync(Pid, Batch0, Timeout) ->
  Caller = caller(),
  Mref = erlang:monitor(process, Pid),
  Ack = ?SYNC_REF(Caller, Mref),
  ok = send(Pid, Batch0, Ack, no_wait_for_queued),
  receive
    {Mref, Partition, BaseOffset} ->
      %% sent from eval_ack_cb
      erlang:demonitor(Mref, [flush]),
      {Partition, BaseOffset};
    {'DOWN', Mref, _, _, Reason} ->
      erlang:error({producer_down, Reason})
  after
    Timeout ->
      deref_caller(Caller),
      erlang:demonitor(Mref, [flush]),
      receive
        {Mref, Partition, BaseOffset} ->
          {Partition, BaseOffset}
      after
        0 ->
          erlang:error(timeout)
      end
  end.

init(#{client_id := ClientId, topic := Topic, partition := Partition} = St) ->
  erlang:process_flag(trap_exit, true),
  ok = set_process_label(ClientId, Topic, Partition),
  {ok, Sender} = wolff_sender:start_link(self(), ClientId, Topic, Partition),
  {ok, St#{sender => Sender}, {continue, do_init}}.

do_init(#{client_id := ClientId,
          topic := Topic,
          partition := Partition,
          config := Config0
         } = St) ->
  PathSegment0 =
    case maps:find(group, Config0) of
      {ok, Group} when is_binary(Group) -> <<Group/binary, $_, Topic/binary>>;
      _ -> <<ClientId/binary, $_, Topic/binary>>
    end,
  PathSegment = escape(PathSegment0),
  QCfg = case maps:get(replayq_dir, Config0, false) of
           false ->
             #{mem_only => true};
           BaseDir ->
             Dir = filename:join([BaseDir, PathSegment, integer_to_list(Partition)]),
             SegBytes = maps:get(replayq_seg_bytes, Config0, ?DEFAULT_REPLAYQ_SEG_BYTES),
             Offload = maps:get(replayq_offload_mode, Config0, false),
             #{dir => Dir, seg_bytes => SegBytes, offload => Offload}
         end,
  MaxTotalBytes = maps:get(replayq_max_total_bytes, Config0, ?DEFAULT_REPLAYQ_LIMIT),
  Q = replayq:open(QCfg#{sizer => fun ?MODULE:queue_item_sizer/1,
                         marshaller => fun ?MODULE:queue_item_marshaller/1,
                         max_total_bytes => MaxTotalBytes
                        }),
  Config1 = maps:update_with(
              telemetry_meta_data,
              fun(Meta) -> Meta#{partition_id => Partition} end,
              #{partition_id => Partition},
              Config0),
  Config2 = resolve_max_linger_bytes(Config1, Q),
  %% replayq configs are kept in Q, there is no need to duplicate them
  Config = maps:without([replayq_dir, replayq_seg_bytes], Config2),
  wolff_metrics:queuing_set(Config, replayq:count(Q)),
  wolff_metrics:queuing_bytes_set(Config, replayq:bytes(Q)),
  wolff_metrics:inflight_set(Config, 0),
  %% The initial connect attempt is made by the caller (wolff_producers.erl)
  %% If succeeded, 'Conn' is a pid (but it may as well be dead by now),
  %% if failed, it's a `term()' to indicate the failure reason.
  %% When it's not a pid, retry timer will be started in mark_connection_down
  %% like when 'DOWN' message is received.
  handle_leader_connection(
    St#{replayq => Q,
        config => Config,
        pending_acks => wolff_pendack:new(),
        produce_api_vsn => undefined,
        sent_reqs => queue:new(), % {kpro:req(), replayq:ack_ref(), [{CallId, MsgCount}]}
        sent_reqs_count => 0,
        inflight_calls => 0,
        client_id => ClientId,
        calls => ?EMPTY
       }).

handle_call(stop, From, St) ->
  gen_server:reply(From, ok),
  {stop, normal, St};
handle_call(_Call, _From, St) ->
  {noreply, St}.

handle_continue(do_init, St0) ->
  {noreply, do_init(St0)}.

handle_info(?linger_expire, St0) ->
  St1 = enqueue_calls(St0#{?linger_expire_timer => false}, no_linger),
  St = maybe_send_to_kafka(St1),
  {noreply, St};
handle_info(?SEND_REQ(_, Batch, _) = Call, #{calls := Calls0, config := #{max_linger_bytes := Max}} = St0) ->
  Bytes = batch_bytes(Batch),
  Calls = collect_send_calls(Call, Bytes, Calls0, Max),
  St1 = enqueue_calls(St0#{calls => Calls}, maybe_linger),
  St = maybe_send_to_kafka(St1),
  {noreply, St};
handle_info({msg, Conn, Rsp}, #{conn := Conn} = St0) ->
  try handle_kafka_ack(Rsp, St0) of
    St1 ->
      St = maybe_send_to_kafka(St1),
      {noreply, St}
  catch
    throw : {kafka_error, Reason, St1} ->
      %% connection is not really down, but we need to
      %% stay down for a while and maybe restart sending on a new connection
      %% if Reason is not_leader_for_partition,
      %% wolff_client should expire old metadata while we are down
      %% and connect to the new leader when we request for a new connection
      St = mark_connection_down(St1, Reason),
      {noreply, St}
  end;
handle_info(?leader_connection(Conn), St) ->
  {noreply, handle_leader_connection(St#{conn := Conn})};
handle_info(?reconnect, St0) ->
  St = St0#{reconnect_timer => ?no_timer},
  {noreply, ensure_delayed_reconnect(St, normal_delay)};
handle_info({'DOWN', _, _, Conn, Reason}, #{conn := Conn} = St0) ->
  %% assert, connection can never down when pending on a reconnect
  #{reconnect_timer := ?no_timer} = St0,
  St = mark_connection_down(St0, Reason),
  {noreply, St};
handle_info({'EXIT', _, Reason}, St) ->
  %% linked process died
  {stop, Reason, St};
handle_info(_Info, St) ->
  {noreply, St}.

handle_cast({stop, Reason}, St) ->
  {stop, {shutdown, Reason}, St};
handle_cast(_Cast, St) ->
  {noreply, St}.

code_change(_OldVsn, St, _Extra) ->
  {ok, St}.

terminate(Reason, #{replayq := Q} = St) ->
  case Reason of
    {shutdown, ?partition_lost} ->
      ok = replayq:close_and_purge(Q),
      _ = reply_error_for_all_reqs(St, ?partition_lost);
    _ ->
      ok = replayq:close(Q)
  end,
  ok = clear_gauges(St, Q).

%% Share the same logic for initial connection and reconnect
handle_leader_connection(#{topic := Topic, partition := Partition, conn := Conn} = St0) when is_pid(Conn) ->
  Attempts = maps:get(reconnect_attempts, St0, 0),
  Attempts > 0 andalso
    log_info(Topic, Partition, "partition_leader_reconnected", #{conn_pid => Conn}),
  _ = erlang:monitor(process, Conn),
  St1 = St0#{reconnect_timer => ?no_timer,
             reconnect_attempts => 0, %% reset counter
             conn := Conn},
  St2 = get_produce_version(St1),
  St3 = resend_sent_reqs(St2, ?reconnect),
  maybe_send_to_kafka(St3);
handle_leader_connection(#{conn := Reason} = St) ->
  mark_connection_down(St#{reconnect_timer => ?no_timer}, Reason).

clear_gauges(#{config := Config}, Q) ->
  wolff_metrics:inflight_set(Config, 0),
  maybe_reset_queuing(Config, Q),
  ok.

maybe_reset_queuing(Config, Q) ->
  case {replayq:count(Q), is_replayq_durable(Config, Q)} of
    {0, _} ->
      wolff_metrics:queuing_set(Config, 0),
      wolff_metrics:queuing_bytes_set(Config, 0);
    {_, false} ->
      wolff_metrics:queuing_set(Config, 0),
      wolff_metrics:queuing_bytes_set(Config, 0);
    {_, _} ->
      ok
  end.

ensure_ts(Batch) ->
  lists:map(fun(#{ts := _} = Msg) -> Msg;
               (Msg) -> Msg#{ts => now_ts()}
            end, Batch).

resolve_max_linger_bytes(#{max_linger_bytes := 0,
                           max_batch_bytes := Mb
                          } = Config, Q) ->
  case replayq:is_mem_only(Q) of
    true ->
      Config#{max_linger_bytes => Mb};
    false ->
      %% when disk or offload mode, try to linger with more bytes
      %% to optimize disk write performance
      Config#{max_linger_bytes => min(?MAX_LINGER_BYTES, Mb * 10)}
  end;
resolve_max_linger_bytes(Config, _Q) ->
  Config.

use_defaults(Config) ->
  use_defaults(Config, [{required_acks, all_isr},
                        {ack_timeout, 10000},
                        {max_batch_bytes, ?WOLFF_KAFKA_DEFAULT_MAX_MESSAGE_BYTES},
                        {max_linger_ms, 0},
                        {max_linger_bytes, 0},
                        {max_send_ahead, 0},
                        {compression, no_compression},
                        {reconnect_delay_ms, 2000}
                       ]).

use_defaults(Config, []) -> Config;
use_defaults(Config, [{K, V} | Rest]) ->
  case maps:is_key(K, Config) of
    true -> use_defaults(Config, Rest);
    false -> use_defaults(Config#{K => V}, Rest)
  end.

resend_requests(#{q_items := Items, req_ref := {alias, OldRef}} = Sent, St, ?reconnect) ->
  erlang:unalias(OldRef),
  {ok, Ref} = send_request(Items, St),
  [Sent#{req_ref := Ref}];
resend_requests(#{q_items := Items, req_ref := {alias, OldRef}} = Sent, St, ?message_too_large) ->
  erlang:unalias(OldRef),
  Refs = lists:map(fun(Item) -> {ok, Ref} = send_request([Item], St), Ref end, Items),
  #{attempts := Attempts, q_ack_ref := Q_AckRef} = Sent,
  make_sent_items_list(Items, Refs, Attempts, Q_AckRef).

%% only ack replayq when the last message is accepted by Kafka
make_sent_items_list([LastItem], [Ref], Attempts, Q_AckRef) ->
  [#{req_ref => Ref,
     q_items => [LastItem],
     q_ack_ref => Q_AckRef,
     attempts => Attempts
    }];
make_sent_items_list([Item | Items], [Ref | Refs], Attempts, Q_AckRef) ->
  [#{req_ref => Ref,
     q_items => [Item],
     q_ack_ref => ?no_queue_ack,
     attempts => Attempts
    } | make_sent_items_list(Items, Refs, Attempts, Q_AckRef)].

send_request(QueueItems,
             #{config := #{required_acks := RequiredAcks,
                           ack_timeout := AckTimeout,
                           compression := Compression
                          },
               produce_api_vsn := Vsn,
               topic := Topic,
               partition := Partition,
               conn := Conn,
               sender := Sender
              }) ->
  Batch = get_batch_from_queue_items(QueueItems),
  Opts = #{ack_timeout => AckTimeout,
           required_acks => RequiredAcks,
           compression => Compression
           },
  Ref = {alias, erlang:alias([reply])},
  ok = wolff_sender:do(Sender, Conn, Ref, Vsn, Topic, Partition, Batch, Opts),
  {ok, Ref}.

resend_sent_reqs(#{sent_reqs := SentReqs,
                   config := Config
                  } = St, Reason) ->
  F = fun(Sent, Acc) ->
          wolff_metrics:retried_inc(Config, 1),
          NewSentList = resend_requests(Sent, St, Reason),
          lists:reverse(NewSentList, Acc)
      end,
  NewSentReqs = lists:foldl(F, [], queue:to_list(SentReqs)),
  St#{sent_reqs := queue:from_list(lists:reverse(NewSentReqs))}.

maybe_send_to_kafka(#{conn := Conn, replayq := Q} = St) ->
  case replayq:count(Q) =:= 0 of
    true ->
      %% nothing to send
      St;
    false when is_pid(Conn) ->
      %% has connection, try send
      maybe_send_to_kafka_has_pending(St);
    false ->
      %% no connection
      %% maybe the connection was closed after ideling
      %% re-connect is triggered immediately if this
      %% is the first attempt
      ensure_delayed_reconnect(St, no_delay_for_first_attempt)
  end.

maybe_send_to_kafka_has_pending(St) ->
  case is_send_ahead_allowed(St) of
    true -> send_to_kafka(St);
    false -> St %% reached max inflight limit
  end.

send_to_kafka(#{sent_reqs := SentReqs,
                sent_reqs_count := SentReqsCount,
                inflight_calls := InflightCalls,
                replayq := Q,
                config := #{required_acks := RequiredAcks,
                            max_batch_bytes := BytesLimit} = Config,
                pending_acks := PendingAcks
               } = St0) ->
  {NewQ, QAckRef, Items} =
    replayq:pop(Q, #{bytes_limit => BytesLimit, count_limit => 999999999}),
  IDs = lists:map(fun({ID, _}) -> ID end, get_calls_from_queue_items(Items)),
  NewPendingAcks = wolff_pendack:move_backlog_to_inflight(PendingAcks, IDs),
  wolff_metrics:queuing_set(Config, replayq:count(NewQ)),
  wolff_metrics:queuing_bytes_set(Config, replayq:bytes(NewQ)),
  NewSentReqsCount = SentReqsCount + 1,
  NrOfCalls = count_calls(Items),
  NewInflightCalls = InflightCalls + NrOfCalls,
  _ = wolff_metrics:inflight_set(Config, NewInflightCalls),
  NoAck = (RequiredAcks =:= none),
  {ok, Ref} = send_request(Items, St0),
  Sent = #{req_ref => Ref,
           q_items => Items,
           q_ack_ref => QAckRef,
           attempts => 1
          },
  St1 = St0#{replayq := NewQ,
             sent_reqs := queue:in(Sent, SentReqs),
             sent_reqs_count := NewSentReqsCount,
             inflight_calls := NewInflightCalls,
             pending_acks := NewPendingAcks
            },
  St2 = maybe_fake_kafka_ack(NoAck, St1),
  maybe_send_to_kafka(St2).

%% when require no acks do not add to sent_reqs and ack caller immediately
maybe_fake_kafka_ack(_NoAck = true, St) ->
  do_handle_kafka_ack(?no_error, ?UNKNOWN_OFFSET, St);
maybe_fake_kafka_ack(_NoAck, St) -> St.

is_send_ahead_allowed(#{config := #{max_send_ahead := Max},
                        sent_reqs_count := SentCount}) ->
  SentCount =< Max.

%% nothing sent and nothing pending to be sent
is_idle(#{replayq := Q, sent_reqs_count := SentReqsCount}) ->
  SentReqsCount =:= 0 andalso replayq:count(Q) =:= 0.

now_ts() ->
    erlang:system_time(millisecond).

make_queue_item(CallId, Batch) ->
  ?Q_ITEM(CallId, now_ts(), Batch).

queue_item_sizer(?Q_ITEM(_CallId, _Ts, Batch)) ->
  batch_bytes(Batch).

batch_bytes(Batch) ->
  lists:foldl(fun(M, Sum) -> oct(M) + Sum end, 0, Batch).

queue_item_marshaller(?Q_ITEM(_, _, _) = I) ->
  term_to_binary(I);
queue_item_marshaller(Bin) when is_binary(Bin) ->
  binary_to_term(Bin).

get_produce_version(#{conn := Conn} = St) when is_pid(Conn) ->
  Vsn = case kpro:get_api_vsn_range(Conn, produce) of
          {ok, {_Min, Max}} -> Max;
          {error, _} -> 3 %% minimum magic v2 produce API version
        end,
  St#{produce_api_vsn => Vsn}.

get_batch_from_queue_items(Items) ->
  get_batch_from_queue_items(Items, []).

get_batch_from_queue_items([], Acc) ->
  lists:reverse(Acc);
get_batch_from_queue_items([?Q_ITEM(_CallId, _Ts, Batch) | Items], Acc) ->
  get_batch_from_queue_items(Items, lists:reverse(Batch, Acc)).

count_calls(Items) -> length(Items).

count_msgs([]) ->
  0;
count_msgs([?Q_ITEM(_CallId, _Ts, Batch) | Rest]) ->
  length(Batch) + count_msgs(Rest).

get_calls_from_queue_items(Items) ->
  get_calls_from_queue_items(Items, []).

get_calls_from_queue_items([], Calls) ->
  lists:reverse(Calls);
get_calls_from_queue_items([QItem | Rest], Calls) ->
  ?Q_ITEM(CallId, _Ts, Batch) = QItem,
  get_calls_from_queue_items(Rest, [{CallId, length(Batch)} | Calls]).

-spec handle_kafka_ack(_, state()) -> state().
handle_kafka_ack(#kpro_rsp{api = produce,
                           ref = Ref,
                           msg = Rsp
                          },
                 #{sent_reqs := SentReqs} = St) ->
  [TopicRsp] = kpro:find(responses, Rsp),
  [PartitionRsp] = kpro:find(partition_responses, TopicRsp),
  ErrorCode = kpro:find(error_code, PartitionRsp),
  BaseOffset = kpro:find(base_offset, PartitionRsp),
  case queue:peek(SentReqs) of
    {value, #{req_ref := Ref1}} when Ref1 =:= Ref ->
      %% sent_reqs queue front matched the response reference
      do_handle_kafka_ack(ErrorCode, BaseOffset, St);
    _ ->
      %% stale response e.g. when inflight > 1, but we have to retry an older req
      St
  end.

note(Fmt, Args) ->
  lists:flatten(io_lib:format(Fmt, Args)).

-spec do_handle_kafka_ack(_, _, state()) -> state().
do_handle_kafka_ack(?no_error, BaseOffset, St) ->
  clear_sent_and_ack_callers(?no_error, BaseOffset, St);
do_handle_kafka_ack(EC, _BaseOffset, St) when EC =:= ?message_too_large orelse EC =:= ?record_list_too_large ->
  #{topic := Topic, partition := Partition, config := Config, sent_reqs := SentReqs} = St,
  {value, #{q_items := Items}} = queue:peek(SentReqs),
  Batch = get_batch_from_queue_items(Items),
  Bytes = batch_bytes(Batch),
  Hint = case EC of
    ?message_too_large ->
      "Consider increasing server side topic config 'max.message.bytes'!";
    ?record_list_too_large ->
      "Cnosider increasing server side topic config 'segment.bytes'!"
  end,
  %% Reason to drop request or split batch is message_too_large
  %% because record_list_too_large makes little sense for application.
  Reason = ?message_too_large,
  case length(Items) of
    1 ->
      %% This is a single message batch, but it's still too large
      Note = "A single-request batch is dropped because it's too large for this topic! " ++ Hint,
      log_error(Topic, Partition, EC, #{note => Note, estimated_batch_bytes => Bytes}),
      wolff_metrics:dropped_inc(Config, 1),
      clear_sent_and_ack_callers(EC, Reason, St);
    N ->
      %% This is a batch of more than one queue items (calls)
      %% Split the batch, and re-send
      #{max_batch_bytes := Max} = Config,
      NewMax = max(1, (Max + 1) div 2),
      St1 = St#{config := Config#{max_batch_bytes := NewMax}},
      Note = note("Config max_batch_bytes=~p is too large for this topic, "
                  "trying to split the current batch and retry. "
                  "Will use max_batch_bytes=~p to collect future batches. " ++ Hint,
                  [Max, NewMax]),
      log_warn(Topic, Partition, EC, #{note => Note, calls_count => N, encode_bytes => Bytes}),
      %% We do not incement the attempt counter for too large batch
      %% St2 = increment_attempt_for_first_sent_req(St),
      resend_sent_reqs(St1, Reason)
  end;
do_handle_kafka_ack(ErrorCode, _BaseOffset, St) ->
  %% Other errors, such as not_leader_for_partition
  #{sent_reqs := SentReqs} = St,
  {value, #{attempts := Attempts, q_items := Items}} = queue:peek(SentReqs),
  #{topic := Topic,
    partition := Partition,
    config := Config
   } = St,
  NrOfCalls = count_calls(Items),
  inc_sent_failed(Config, NrOfCalls, Attempts),
  log_warn(Topic, Partition, "error_in_produce_response",
           #{error_code => ErrorCode,
             batch_size => count_msgs(Items),
             attempts => Attempts}),
  St1 = increment_attempt(St),
  erlang:throw({kafka_error, ErrorCode, St1}).

%% Since we only handle Kafka ack for the first sent request in the sent queue
%% we only bump the first itme with attempt counter
increment_attempt(#{sent_reqs := SentReqs} = St) ->
  {{value, SentReq}, SentReqs1} = queue:out(SentReqs),
  #{attempts := Attempts} = SentReq,
  St#{sent_reqs := queue:in_r(SentReq#{attempts := Attempts + 1}, SentReqs1)}.

clear_sent_and_ack_callers(ErrorCode, BaseOffset, St) ->
  #{sent_reqs := SentReqs,
    sent_reqs_count := SentReqsCount,
    inflight_calls := InflightCalls,
    pending_acks := PendingAcks,
    replayq := Q,
    config := Config
  } = St,
  {{value, #{q_ack_ref := Q_AckRef,
             q_items := Items,
             attempts := Attempts}}, NewSentReqs} = queue:out(SentReqs),
  Calls = get_calls_from_queue_items(Items),
  NrOfCalls = count_calls(Items),
  case ErrorCode of
    ?no_error ->
      inc_sent_success(Config, NrOfCalls, Attempts);
    _ ->
      inc_sent_failed(Config, NrOfCalls, Attempts)
  end,
  NewSentReqsCount = SentReqsCount - 1,
  NewInflightCalls = InflightCalls - NrOfCalls,
  wolff_metrics:inflight_set(Config, NewInflightCalls),
  ok = replayq_ack(Q, Q_AckRef),
  NewPendingAcks = evaluate_pending_ack_funs(PendingAcks, Calls, BaseOffset),
  St#{sent_reqs := NewSentReqs,
      sent_reqs_count := NewSentReqsCount,
      inflight_calls := NewInflightCalls,
      pending_acks := NewPendingAcks
     }.

replayq_ack(_Q, ?no_queue_ack) ->
  ok;
replayq_ack(Q, Q_AckRef) ->
  replayq:ack(Q, Q_AckRef).

%% @private This function is called in below scenarios
%% * Failed to connect any of the brokers
%% * Failed to connect to partition leader
%% * Connection 'DOWN' due to Kafka initiated socket close
%% * Produce request failure
%%   - Socket error
%%   - Error code received from Kafka
mark_connection_down(#{topic := Topic,
                       partition := Partition,
                       conn := Old
                      } = St0, Reason) ->
  false = is_pid(Reason),
  St = St0#{conn := Reason},
  ok = log_connection_down(Topic, Partition, Old, Reason),
  case is_idle(St) of
    true ->
      St;
    false ->
      %% ensure delayed reconnect timer is started
      ensure_delayed_reconnect(St, normal_delay)
  end.

log_connection_down(_Topic, _Partition, _, to_be_discovered) ->
  %% this is the initial state of the connection
  ok;
log_connection_down(Topic, Partition, Conn, Reason) ->
  log_info(Topic, Partition,
           "connection_to_partition_leader_error",
           #{conn => Conn, reason => Reason}).

is_timer_on(?no_timer) ->
  false;
is_timer_on({_, Ref}) when is_reference(Ref) ->
  %% started by timer:apply_after
  is_timer_on(Ref);
is_timer_on(Ref) when is_reference(Ref) ->
  erlang:read_timer(Ref) =/= false.

ensure_delayed_reconnect(St, DelayStrategy) ->
  Tref = maps:get(reconnect_timer, St, ?no_timer),
  case is_timer_on(Tref) of
    true ->
      St;
    false ->
      do_ensure_delayed_reconnect(St, DelayStrategy)
  end.

do_ensure_delayed_reconnect(
  #{config := #{reconnect_delay_ms := Delay0} = Config,
    client_id := ClientId,
    topic := Topic,
    partition := Partition
   } = St, DelayStrategy) ->
  Attempts = maps:get(reconnect_attempts, St, 0),
  MaxPartitions = maps:get(max_partitions, Config, all_partitions),
  Attempts > 0 andalso Attempts rem 10 =:= 0 andalso
    log_error(Topic, Partition,
              "producer_is_still_disconnected_after_retry",
              #{attempts => Attempts}),
  Delay =
    case DelayStrategy of
      no_delay_for_first_attempt when Attempts =:= 0 ->
        %% this is the first attempt after disconnected, no delay
        0;
      _ ->
        %% add an extra random delay to avoid all partition workers
        %% try to reconnect around the same moment
        Delay0 + rand:uniform(1000)
    end,
  case wolff_client_sup:find_client(ClientId) of
    {ok, ClientPid} ->
      Group = maps:get(group, Config, ?NO_GROUP),
      Args = [ClientPid, Group, Topic, Partition, self(), MaxPartitions],
      {ok, Tref} = timer:apply_after(Delay, wolff_client, recv_leader_connection, Args),
      St#{reconnect_timer => Tref, reconnect_attempts => Attempts + 1};
    {error, Reason} ->
      log_warn(Topic, Partition,
               "failed_to_find_client_will_retry_after_delay",
               #{delay => Delay, reason => Reason}),
      %% call timer:apply_after for both cases, do not use send_after here
      {ok, Tref} = timer:apply_after(Delay, erlang, send, [self(), ?reconnect]),
      St#{reconnect_timer => Tref, reconnect_attempts => Attempts + 1}
  end.

evaluate_pending_ack_funs(PendingAcks, [], _BaseOffset) -> PendingAcks;
evaluate_pending_ack_funs(PendingAcks, [{CallId, BatchSize} | Rest], BaseOffset) ->
  NewPendingAcks =
    case wolff_pendack:take_inflight(PendingAcks, CallId) of
      {ok, AckCb, PendingAcks1} ->
        ok = eval_ack_cb(AckCb, BaseOffset),
        PendingAcks1;
      false ->
        PendingAcks
    end,
  evaluate_pending_ack_funs(NewPendingAcks, Rest, offset(BaseOffset, BatchSize)).

offset(BaseOffset, Delta) when is_integer(BaseOffset) andalso BaseOffset >= 0 -> BaseOffset + Delta;
offset(BaseOffset, _Delta) -> BaseOffset.

log_info(Topic, Partition, Msg, Args) ->
  log(info, Args#{topic => Topic, partition => Partition, msg => Msg}).

log_warn(Topic, Partition, Msg, Args) ->
  log(warning, Args#{topic => Topic, partition => Partition, msg => Msg}).

log_error(Topic, Partition, Msg, Args) ->
  log(error, Args#{topic => Topic, partition => Partition, msg => Msg}).

log(Level, Report) ->
    logger:log(Level, Report).

%% Estimation of size in bytes of one payload sent to Kafka.
%% According to Kafka protocol, a v2 record consists of below fields:
%%   Length => varint              # varint_bytes(SizeOfAllRestFields)
%%   Attributes => int8            # 1 -- always zero (the byte value)
%%   TimestampDelta => varint      # 4 -- a wild guess
%%   OffsetDelta => varint         # 2 -- never exceeds 3 bytes
%%   KeyLen => varint              # varint_bytes(size(Key))
%%   Key => data                   # size(Key)
%%   ValueLen => varint            # varint_bytes(size(Value))
%%   Value => data                 # size(Value)
%%   Headers => [Header]           # headers_bytes(Headers)
oct(#{key := K, value := V} = Msg) ->
  HeadersBytes = headers_bytes(maps:get(headers, Msg, [])),
  FieldsBytes = HeadersBytes + 7 + encoded_bin_bytes(K) + encoded_bin_bytes(V),
  FieldsBytes + varint_bytes(HeadersBytes + FieldsBytes).

headers_bytes(Headers) ->
  headers_bytes(Headers, 0, 0).

headers_bytes([], Bytes, Count) ->
  Bytes + varint_bytes(Count);
headers_bytes([{Name, Value} | T], Bytes, Count) ->
  NewBytes = Bytes + encoded_bin_bytes(Name) + encoded_bin_bytes(Value),
  headers_bytes(T, NewBytes, Count+1).

-compile({inline, encoded_bin_bytes/1}).
encoded_bin_bytes(B) ->
  varint_bytes(size(B)) + size(B).

%% varint encoded bytes of a unsigned integer
-compile({inline, vb/1}).
vb(N) when N < 128 -> 1;
vb(N) when N < 16384 -> 2;
vb(N) when N < 2097152 -> 3;
vb(N) when N < 268435456 -> 4;
vb(N) when N < 34359738368 -> 5;
vb(N) when N < 4398046511104 -> 6;
vb(N) when N < 562949953421312 -> 7;
vb(N) when N < 72057594037927936 -> 8;
vb(N) when N < 9223372036854775808 -> 9;
vb(_) -> 10.

%% varint encoded bytes of a signed integer
-compile({inline, varint_bytes/1}).
varint_bytes(N) -> vb(zz(N)).

%% zigzag encode a signed integer
-compile({inline, zz/1}).
zz(I) -> (I bsl 1) bxor (I bsr 63).

%% collect send calls which are already sent to mailbox,
%% the collection is size-limited by the max_linger_bytes config.
collect_send_calls(Call, Bytes, ?EMPTY, Max) ->
  Init = #{ts => now_ts(), bytes => 0, batch_r => []},
  collect_send_calls(Call, Bytes, Init, Max);
collect_send_calls(Call, Bytes, Calls, Max) ->
  #{bytes := Bytes0, batch_r := BatchR} = Calls,
  Sum = Bytes0 + Bytes,
  R = Calls#{bytes => Sum,
             batch_r => [Call | BatchR]
            },
  case Sum < Max of
    true ->
      collect_send_calls2(R, Max);
    false ->
      R
  end.

%% Collect all send requests which are already in process mailbox
collect_send_calls2(Calls, Max) ->
  receive
    ?SEND_REQ(_, Batch, _) = Call ->
      Bytes = batch_bytes(Batch),
      collect_send_calls(Call, Bytes, Calls, Max)
  after
    0 ->
      Calls
  end.

ensure_linger_expire_timer_start(#{?linger_expire_timer := false} = St, Timeout) ->
  %% delay enqueue, try to accumulate more into the batch
  Ref = erlang:send_after(Timeout, self(), ?linger_expire),
  St#{?linger_expire_timer := Ref};
ensure_linger_expire_timer_start(St, _Timeout) ->
  %% timer is already started
  St.

ensure_linger_expire_timer_cancel(#{?linger_expire_timer := LTimer} = St) ->
  _ = is_reference(LTimer) andalso erlang:cancel_timer(LTimer),
  St#{?linger_expire_timer => false}.

%% check if the call collection should continue to linger before enqueue
is_linger_continue(#{config := #{max_linger_ms := 0}}) ->
  false;
is_linger_continue(#{calls := Calls, config := Config, replayq := Q}) ->
  case replayq:is_writing_to_disk(Q) of
    true ->
      #{max_linger_ms := MaxLingerMs, max_linger_bytes := MaxLingerBytes} = Config,
      #{ts := Ts, bytes := Bytes} = Calls,
      case Bytes < MaxLingerBytes of
        true ->
          TimeLeft = MaxLingerMs - (now_ts() - Ts),
          (TimeLeft > 0) andalso {true, TimeLeft};
        false ->
          false
      end;
    false ->
      false
  end.

enqueue_calls(#{calls := ?EMPTY} = St, _) ->
  %% no call to enqueue
  St;
enqueue_calls(St, maybe_linger) ->
  case is_linger_continue(St) of
    {true, Timeout} ->
      ensure_linger_expire_timer_start(St, Timeout);
    false ->
      enqueue_calls(St, no_linger)
  end;
enqueue_calls(#{calls := #{batch_r := CallsR}} = St0, no_linger) ->
  Calls = lists:reverse(CallsR),
  St = ensure_linger_expire_timer_cancel(St0),
  enqueue_calls2(Calls, St#{calls => ?EMPTY}).

enqueue_calls2(Calls,
               #{replayq := Q,
                 pending_acks := PendingAcks0,
                 partition := Partition,
                 config := Config0
                } = St0) ->
  {QueueItems, PendingAcks, CallByteSize} =
    lists:foldl(
      fun(?SEND_REQ(_From, Batch, AckFun), {Items, PendingAcksIn, Size}) ->
          %% keep callback funs in memory, do not seralize it into queue because
          %% saving anonymous function to disk may easily lead to badfun exception
          %% in case of restart on newer version beam.
          {CallId, PendingAcksOut} = wolff_pendack:insert_backlog(PendingAcksIn, ?ACK_CB(AckFun, Partition)),
          NewItems = [make_queue_item(CallId, Batch) | Items],
          {NewItems, PendingAcksOut, Size + batch_bytes(Batch)}
      end, {[], PendingAcks0, 0}, Calls),
   NewQ = replayq:append(Q, lists:reverse(QueueItems)),
   wolff_metrics:queuing_set(Config0, replayq:count(NewQ)),
   wolff_metrics:queuing_bytes_set(Config0, replayq:bytes(NewQ)),
   lists:foreach(fun maybe_reply_queued/1, Calls),
   IsHighMemOverflow =
        maps:get(drop_if_highmem, Config0, false)
        andalso replayq:is_mem_only(NewQ)
        andalso load_ctl:is_high_mem(),
   Overflow = case IsHighMemOverflow of
                  true ->
                      max(replayq:overflow(NewQ), CallByteSize);
                  false ->
                      replayq:overflow(NewQ)
              end,
   handle_overflow(St0#{replayq := NewQ,
                        pending_acks := PendingAcks
                       },
                   IsHighMemOverflow,
                   Overflow).

maybe_reply_queued(?SEND_REQ(?no_queued_reply, _, _)) ->
    ok;
maybe_reply_queued(?SEND_REQ({Caller, Ref}, _, _)) ->
    erlang:send(Caller, {Ref, ?queued}).

eval_ack_cb(?ACK_CB(AckFun, Partition), BaseOffset) when is_function(AckFun, 2) ->
  ok = AckFun(Partition, BaseOffset);
eval_ack_cb(?ACK_CB({F, A}, Partition), BaseOffset) when is_function(F) ->
  true = is_function(F, length(A) + 2),
  ok = erlang:apply(F, [Partition, BaseOffset | A]);
eval_ack_cb(?ACK_CB({Caller, Ref}, Partition), BaseOffset) when ?IS_SYNC_REF(Caller, Ref) ->
  _ = erlang:send(Caller, {Ref, Partition, BaseOffset}),
  ok.

handle_overflow(St, _IsHighMemOverflow, Overflow) when Overflow =< 0 ->
    ok = maybe_log_discard(St, 0, ?DISCARD_OVERFLOW),
    St;
handle_overflow(#{replayq := Q,
                  pending_acks := PendingAcks,
                  config := Config
                 } = St,
                IsHighMemOverflow,
                Overflow) ->
  BytesMode =
    case IsHighMemOverflow of
      true -> at_least;
      false -> at_most
    end,
  OverflowReason =
    case IsHighMemOverflow of
      true ->
        ?DISCARD_HIGH_MEM;
      false ->
        ?DISCARD_OVERFLOW
    end,
  {NewQ, QAckRef, Items} =
    replayq:pop(Q, #{bytes_limit => {BytesMode, Overflow}, count_limit => 999999999}),
  ok = replayq:ack(NewQ, QAckRef),
  Calls = get_calls_from_queue_items(Items),
  CallIDs = lists:map(fun({CallId, _BatchSize}) -> CallId end, Calls),
  NrOfCalls = count_calls(Items),
  wolff_metrics:dropped_queue_full_inc(Config, NrOfCalls),
  wolff_metrics:dropped_inc(Config, NrOfCalls),
  wolff_metrics:queuing_set(Config, replayq:count(NewQ)),
  wolff_metrics:queuing_bytes_set(Config, replayq:bytes(NewQ)),
  ok = maybe_log_discard(St, NrOfCalls, OverflowReason),
  {CbList, NewPendingAcks} = wolff_pendack:drop_backlog(PendingAcks, CallIDs),
  lists:foreach(fun(Cb) -> eval_ack_cb(Cb, ?buffer_overflow_discarded) end, CbList),
  St#{replayq := NewQ, pending_acks := NewPendingAcks}.

reply_error_for_all_reqs(St, Reason) ->
  #{pending_acks := PendingAcks, config := Config, sent_reqs := SentReqs} = St,
  F = fun(Cb, Acc) ->
    eval_ack_cb(Cb, Reason),
    Acc + 1
  end,
  Count = wolff_pendack:fold(PendingAcks, F, 0),
  AlreadyInc = lists:foldl(
    fun(#{attempts := Attempts}, Acc) when Attempts > 1 ->
      %% 'failed' counter is already incremented for this sent request
      Acc + 1;
    (_, Acc) ->
      Acc
  end, 0, queue:to_list(SentReqs)),
  inc_sent_failed(Config, Count - AlreadyInc, 1).

%% use process dictionary for upgrade without restart
maybe_log_discard(St, Increment, Reason) ->
  Last = get_overflow_log_state(),
  #{last_cnt := LastCnt, acc_cnt := AccCnt} = Last,
  case LastCnt =:= AccCnt andalso Increment =:= 0 of
    true -> %% no change
      ok;
    false ->
      maybe_log_discard(St, Increment, Last, Reason)
  end.

maybe_log_discard(#{topic := Topic, partition := Partition},
                  Increment,
                  #{last_ts := LastTs, last_cnt := LastCnt, acc_cnt := AccCnt},
                 Reason) ->
  NowTs = now_ts(),
  NewAccCnt = AccCnt + Increment,
  DiffCnt = NewAccCnt - LastCnt,
  case NowTs - LastTs > ?MIN_DISCARD_LOG_INTERVAL of
    true ->
      log_warn(Topic, Partition,
               "dropped_produce_requests",
               #{count => DiffCnt, cause => Reason}),
      put_overflow_log_state(NowTs, NewAccCnt, NewAccCnt);
    false ->
      put_overflow_log_state(LastTs, LastCnt, NewAccCnt)
  end.

get_overflow_log_state() ->
  case get(?buffer_overflow_discarded) of
    undefined -> #{last_ts => 0, last_cnt => 0, acc_cnt => 0};
    V when is_map(V) -> V
  end.

put_overflow_log_state(Ts, Cnt, Acc) ->
  put(?buffer_overflow_discarded, #{last_ts => Ts, last_cnt => Cnt, acc_cnt => Acc}),
  ok.

inc_sent_failed(Config, NrOfCalls, Attempts) when Attempts > 1 ->
    wolff_metrics:retried_failed_inc(Config, NrOfCalls);
inc_sent_failed(Config, NrOfCalls, _Attempts) ->
    wolff_metrics:failed_inc(Config, NrOfCalls).

inc_sent_success(Config, NrOfCalls, Attempts) when Attempts > 1 ->
    wolff_metrics:retried_success_inc(Config, NrOfCalls);
inc_sent_success(Config, NrOfCalls, _Attempts) ->
    wolff_metrics:success_inc(Config, NrOfCalls).

is_replayq_durable(#{replayq_offload_mode := true}, _Q) ->
    false;
is_replayq_durable(_, Q) ->
    not replayq:is_mem_only(Q).

-spec escape(string() | binary()) -> binary().
escape(Str) ->
    NormalizedStr = unicode:characters_to_nfd_list(Str),
    iolist_to_binary(escape_uri(NormalizedStr)).

%% copied from `edoc_lib' because dialyzer cannot see this private
%% function there.
escape_uri([C | Cs]) when C >= $a, C =< $z ->
    [C | escape_uri(Cs)];
escape_uri([C | Cs]) when C >= $A, C =< $Z ->
    [C | escape_uri(Cs)];
escape_uri([C | Cs]) when C >= $0, C =< $9 ->
    [C | escape_uri(Cs)];
escape_uri([C = $. | Cs]) ->
    [C | escape_uri(Cs)];
escape_uri([C = $- | Cs]) ->
    [C | escape_uri(Cs)];
escape_uri([C = $_ | Cs]) ->
    [C | escape_uri(Cs)];
escape_uri([C | Cs]) when C > 16#7f ->
    %% This assumes that characters are at most 16 bits wide.
    escape_byte(((C band 16#c0) bsr 6) + 16#c0)
	++ escape_byte(C band 16#3f + 16#80)
	++ escape_uri(Cs);
escape_uri([C | Cs]) ->
    escape_byte(C) ++ escape_uri(Cs);
escape_uri([]) ->
    [].

%% copied from `edoc_lib' because dialyzer cannot see this private
%% function there.
%% has a small modification: it uses `=' in place of `%' so that it
%% won't generate invalid paths in windows.
escape_byte(C) when C >= 0, C =< 255 ->
    [$=, hex_digit(C bsr 4), hex_digit(C band 15)].

%% copied from `edoc_lib' because dialyzer cannot see this private
%% function there.
hex_digit(N) when N >= 0, N =< 9 ->
    N + $0;
hex_digit(N) when N > 9, N =< 15 ->
    N + $a - 10.

%% Alias in OTP 24/25 has memory leak.
%% https://github.com/erlang/otp/issues/6947
-if(?OTP_RELEASE >= 26).
caller() -> erlang:alias([reply]).
deref_caller(Alias) ->
    erlang:unalias(Alias).
-else.
caller() -> self().
deref_caller(_Pid) -> ok.
-endif.

-if(?OTP_RELEASE >= 27).
set_process_label(ClientId, Topic, Partition) ->
    proc_lib:set_label({?MODULE, ClientId, Topic, Partition}).
-else.
set_process_label(_ClientId, _Topic, _Partition) ->
    ok.
-endif.

-ifdef(TEST).
-include_lib("eunit/include/eunit.hrl").

maybe_log_discard_test_() ->
    [ {"no-increment", fun() -> maybe_log_discard(undefined, 0, ?DISCARD_OVERFLOW) end}
    , {"fake-last-old",
       fun() ->
         Ts0 = now_ts() - ?MIN_DISCARD_LOG_INTERVAL - 1,
         ok = put_overflow_log_state(Ts0, 2, 2),
         ok = maybe_log_discard(#{topic => <<"a">>, partition => 0}, 1, ?DISCARD_OVERFLOW),
         St = get_overflow_log_state(),
         ?assertMatch(#{last_cnt := 3, acc_cnt := 3}, St),
         ?assert(maps:get(last_ts, St) - Ts0 > ?MIN_DISCARD_LOG_INTERVAL)
       end}
    , {"fake-last-fresh",
       fun() ->
         Ts0 = now_ts(),
         ok = put_overflow_log_state(Ts0, 2, 2),
         ok = maybe_log_discard(#{topic => <<"a">>, partition => 0}, 2, ?DISCARD_OVERFLOW),
         St = get_overflow_log_state(),
         ?assertMatch(#{last_cnt := 2, acc_cnt := 4}, St),
         ?assert(maps:get(last_ts, St) - Ts0 < ?MIN_DISCARD_LOG_INTERVAL)
       end}
    ].
-endif.