fix: handle record_list_too_large error in produce response

Logic is similar to message_too_large: split the batch,
if a single message is still too large, drop it.

Author: zmstone

changelog.md

@@ -1,3 +1,6 @@
+* 1.5.19
+  - Handle `record_list_too_large` error returned from Kafka.
+    Similar to `message_too_large`  error, the batch is split, then dropped if single call is still too large.
 * 1.5.18
   - Partition metadata handling.
     - Fixed an issue introduced in 1.5.15 where temporarily missing partitions in the metadata response could leave a `wolff_producer` process permanently disconnected.

src/wolff_producer.erl

@@ -492,6 +492,8 @@ handle_kafka_ack(#kpro_rsp{api = produce,
       do_handle_kafka_ack(Ref, BaseOffset, St, normal);
     ?message_too_large ->
       do_handle_kafka_ack(Ref, BaseOffset, St, ?message_too_large);
+    ?record_list_too_large ->
+      do_handle_kafka_ack(Ref, BaseOffset, St, ?record_list_too_large);
     _ ->
       #{topic := Topic, partition := Partition} = St,
       log_warn(Topic, Partition, "Produce response error-code = ~0p", [ErrorCode]),
@@ -503,16 +505,20 @@ do_handle_kafka_ack(Ref, BaseOffset, #{sent_reqs := SentReqs } = St, Reason) ->
     {value, ?sent_req(#kpro_req{ref = Ref}, Q_AckRef, Calls)} ->
       %% this clause is kept only to be hot-upgrade safe
       clear_sent_and_ack_callers(Q_AckRef, Calls, BaseOffset, St);
-    {value, ?sent_items(Ref, Items, Q_AckRef)} when Reason =:= ?message_too_large ->
+    {value, ?sent_items(Ref, Items, Q_AckRef)} when Reason =:= ?message_too_large orelse Reason =:= ?record_list_too_large ->
       case Items of
         [?Q_ITEM(_CallId, _Ts, Batch)] ->
           %% This is one call, but it's still too large
           #{topic := Topic, partition := Partition} = St,
           #kpro_req{msg = IoData} = make_request(Items, St),
           EncodedBytes = iolist_size(IoData),
-          log_error(Topic, Partition, "One request is dropped due to message_too_large! "
-                    "This single-request includes ~p message(s), and encoded to ~p bytes. "
-                    "Please consider increasing max.message.bytes on the server side!",
+          Hint = case Reason 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,
+          log_error(Topic, Partition, "Produce request dropped (with ~w messages encoded to ~w bytes). " ++ Hint,
                     [length(Batch), EncodedBytes]),
           Calls = get_calls_from_queue_items(Items),
           clear_sent_and_ack_callers(Q_AckRef, Calls, ?message_too_large, St);

test/wolff_tests.erl

@@ -445,7 +445,10 @@ test_leader_restart() ->
   end).
 
 with_topic(Topic, Partitions, ReplicationFactor, MaxMessageBytes, TestFunc) ->
-  ok = create_topic(Topic, Partitions, ReplicationFactor, MaxMessageBytes),
+    with_topic(Topic, Partitions, ReplicationFactor, MaxMessageBytes, default_sgement_bytes, TestFunc).
+
+with_topic(Topic, Partitions, ReplicationFactor, MaxMessageBytes, SegmentBytes, TestFunc) ->
+  ok = create_topic(Topic, Partitions, ReplicationFactor, MaxMessageBytes, SegmentBytes),
   try
     _ = application:stop(wolff), %% ensure stopped
     {ok, _} = application:ensure_all_started(wolff),
@@ -454,23 +457,13 @@ with_topic(Topic, Partitions, ReplicationFactor, MaxMessageBytes, TestFunc) ->
     ok = delete_topic(Topic)
   end.
 
-
-message_too_large_test_() ->
-  {timeout, 60,
-   fun() -> test_message_too_large() end}.
-
-test_message_too_large() ->
-  Topic = "message-too-large-" ++ integer_to_list(abs(erlang:monotonic_time())),
-  Partitions = 1,
-  ReplicationFactor = 1,
-  MaxMessageBytes = 100,
-  with_topic(Topic, Partitions, ReplicationFactor, MaxMessageBytes, fun() ->
+test_batch_split_then_drop(Topic, MaxMessageBytes) ->
     ClientCfg = client_config(),
     ClientId = iolist_to_binary("client-" ++ Topic),
     {ok, Client} = start_client(ClientId, ?HOSTS, ClientCfg#{connection_strategy => per_partition}),
     TopicBin = iolist_to_binary(Topic),
     %% try to batch more messages than Kafka's limit,
-    %% the producer will get message_too_large error back
+    %% the producer will get message_too_large or record_list_too_large error back
     %% then it should retry sending one message at a time
     ProducerCfg = #{partitioner => fun(_, _) -> 0 end,
                     max_batch_bytes => MaxMessageBytes * 3,
@@ -504,8 +497,38 @@ test_message_too_large() ->
     ?assertEqual(message_too_large, (SendFunc([Msg(<<"0123456789">>)]))()),
     ok = wolff:stop_producers(Producers),
     ok = stop_client(Client),
-    ok = application:stop(wolff)
-  end).
+    ok = application:stop(wolff).
+
+%% Max message size is smaller than segment bytes to tigger record_list_too_large error.
+%% This is usually a bad server/topic configuration, but we need to cover it anyways.
+record_list_too_large_test_() ->
+  {timeout, 60,
+   fun() -> test_record_list_too_large() end}.
+
+test_record_list_too_large() ->
+  Topic = "record-list-too-large-" ++ integer_to_list(abs(erlang:monotonic_time())),
+  Partitions = 1,
+  ReplicationFactor = 1,
+  MaxMessageBytes = 1000,
+  SegmentBytes = 100,
+  with_topic(Topic, Partitions, ReplicationFactor, MaxMessageBytes, SegmentBytes,
+    fun() ->
+      test_batch_split_then_drop(Topic, SegmentBytes)
+    end).
+
+message_too_large_test_() ->
+  {timeout, 60,
+   fun() -> test_message_too_large() end}.
+
+test_message_too_large() ->
+  Topic = "message-too-large-" ++ integer_to_list(abs(erlang:monotonic_time())),
+  Partitions = 1,
+  ReplicationFactor = 1,
+  MaxMessageBytes = 100,
+  with_topic(Topic, Partitions, ReplicationFactor, MaxMessageBytes,
+    fun() ->
+      test_batch_split_then_drop(Topic, MaxMessageBytes)
+    end).
 
 one_byte_limit_test() ->
   Topic = "one-byte-limit-" ++ integer_to_list(abs(erlang:monotonic_time())),
@@ -616,8 +639,8 @@ encoded_bytes(Batch) ->
   Encoded = kpro_batch:encode(2, Batch, no_compression),
   iolist_size(Encoded).
 
-create_topic(Topic, Partitions, ReplicationFactor, MaxMessageBytes) ->
-  Cmd = create_topic_cmd(Topic, Partitions, ReplicationFactor, MaxMessageBytes),
+create_topic(Topic, Partitions, ReplicationFactor, MaxMessageBytes, SegmentBytes) ->
+  Cmd = create_topic_cmd(Topic, Partitions, ReplicationFactor, MaxMessageBytes, SegmentBytes),
   Result = os:cmd(Cmd),
   Expected = "Created topic " ++ Topic ++ ".\n",
   ?assertEqual(Expected, Result),
@@ -631,7 +654,7 @@ delete_topic(Topic) ->
     _ -> throw(Result)
   end.
 
-create_topic_cmd(Topic, Partitions, ReplicationFactor, MaxMessageBytes) ->
+create_topic_cmd(Topic, Partitions, ReplicationFactor, MaxMessageBytes, SegmentBytes) ->
   "docker exec wolff-kafka-1 /opt/kafka/bin/kafka-topics.sh" ++
   " --zookeeper zookeeper:2181" ++
   " --create" ++
@@ -643,6 +666,12 @@ create_topic_cmd(Topic, Partitions, ReplicationFactor, MaxMessageBytes) ->
       " --config max.message.bytes=" ++ integer_to_list(MaxMessageBytes);
     false ->
       ""
+  end ++
+  case is_integer(SegmentBytes) of
+    true ->
+      " --config segment.bytes=" ++ integer_to_list(SegmentBytes);
+    false ->
+      ""
   end.
 
 delete_topic_cmd(Topic) ->