diff --git a/changelog.md b/changelog.md index 7ca2f7d..316e645 100644 --- a/changelog.md +++ b/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. diff --git a/src/wolff_producer.erl b/src/wolff_producer.erl index aae7bb9..c48eca7 100644 --- a/src/wolff_producer.erl +++ b/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); diff --git a/test/wolff_tests.erl b/test/wolff_tests.erl index e015d2c..69a15b8 100644 --- a/test/wolff_tests.erl +++ b/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) ->