4.x: handle record_list_too_large error

changelog.md

@@ -1,3 +1,7 @@
+* 4.0.13 (merge 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.
+
 * 4.0.12 (merge 1.5.18)
   - Partition metadata handling.
     - Fixed an issue introduced in 4.0.7 (1.5.15) where temporarily missing partitions in the metadata response could leave a `wolff_producer` process permanently disconnected.

src/wolff_producer.erl

@@ -614,19 +614,27 @@ note(Fmt, Args) ->
 -spec do_handle_kafka_ack(_, _, sent(), state()) -> state().
 do_handle_kafka_ack(?no_error, BaseOffset, _Sent, St) ->
   clear_sent_and_ack_callers(?no_error, BaseOffset, St);
-do_handle_kafka_ack(?message_too_large = EC, _BaseOffset, Sent, St) ->
+do_handle_kafka_ack(EC, _BaseOffset, Sent, St) when EC =:= ?message_too_large orelse EC =:= ?record_list_too_large ->
   #{topic := Topic, partition := Partition, config := Config} = St,
   #{q_items := Items} = Sent,
   #kpro_req{msg = IoData} = make_request(Items, St),
   Bytes = iolist_size(IoData),
+  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! "
-             "Consider increasing 'max.message.bytes' config on the server side!",
+      Note = "A single-request batch is dropped because it's too large for this topic! " ++ Hint,
       log_error(Topic, Partition, EC, #{note => Note, encode_bytes => Bytes}),
       wolff_metrics:dropped_inc(Config, 1),
-      clear_sent_and_ack_callers(EC, EC, St);
+      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
@@ -635,10 +643,10 @@ do_handle_kafka_ack(?message_too_large = EC, _BaseOffset, Sent, St) ->
       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.",
+                  "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}),
-      resend_sent_reqs(St1, EC)
+      resend_sent_reqs(St1, Reason)
   end;
 do_handle_kafka_ack(ErrorCode, _BaseOffset, Sent, St) ->
   %% Other errors, such as not_leader_for_partition

test/wolff_tests.erl

@@ -833,7 +833,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),
@@ -842,23 +845,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,
@@ -894,8 +887,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 greater than segment bytes to trigger 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_() ->
   {timeout, 10, fun one_byte_limit/0}.
@@ -1010,8 +1033,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),
   Pattern = "Created topic ",
   ?assert(string:str(Result, Pattern) > 0, Result),
@@ -1020,7 +1043,7 @@ create_topic(Topic, Partitions, ReplicationFactor, MaxMessageBytes) ->
 delete_topic(Topic) ->
   wolff_test_utils:delete_topic(Topic).
 
-create_topic_cmd(Topic, Partitions, ReplicationFactor, MaxMessageBytes) ->
+create_topic_cmd(Topic, Partitions, ReplicationFactor, MaxMessageBytes, SegmentBytes) ->
   wolff_test_utils:topics_cmd_base(Topic) ++
   " --create" ++
   " --partitions " ++ integer_to_list(Partitions) ++
@@ -1030,6 +1053,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.
 
 stop_kafka_2() ->