diff --git a/changelog.md b/changelog.md index 9f19e92..9ae7642 100644 --- a/changelog.md +++ b/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. diff --git a/src/wolff_producer.erl b/src/wolff_producer.erl index 58820ac..9a8c98e 100644 --- a/src/wolff_producer.erl +++ b/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 diff --git a/test/wolff_tests.erl b/test/wolff_tests.erl index 6159ed0..6b3608a 100644 --- a/test/wolff_tests.erl +++ b/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() ->