Packages
brod
4.4.7
4.5.7
4.5.6
4.5.5
4.5.4
4.5.3
4.5.2
4.5.1
4.5.0
4.4.7
4.4.6
4.4.5
4.4.4
4.4.3
4.4.2
4.4.1
4.4.0
4.3.3
4.3.2
4.3.1
4.3.0
4.2.0
4.1.1
4.1.0
4.0.0
3.19.1
3.19.0
3.18.0
3.17.1
3.17.0
3.16.5
3.16.4
3.16.3
3.16.2
3.16.1
3.16.0
3.15.6
3.15.5
3.15.4
3.15.3
3.15.1
3.15.0
3.14.0
3.13.0
3.12.0
3.11.0
3.10.0
3.9.5
3.9.3
3.9.2
3.9.1
3.9.0
3.8.1
3.8.0
3.7.11
3.7.10
3.7.9
3.7.8
3.7.7
3.7.6
3.7.5
3.7.4
3.7.3
3.7.2
3.7.1
3.7.0
3.6.2
3.6.1
3.6.0
3.5.2
3.5.1
3.5.0
3.4.0
3.3.5
3.3.4
3.3.3
3.3.2
3.3.1
3.3.0
3.2.0
3.0.0
2.5.0
2.4.1
2.4.0
2.3.7
2.3.6
2.3.5
2.3.4
2.3.3
2.3.1
2.2.16
2.2.15
2.2.14
2.2.12
2.2.11
2.2.10
2.2.9
2.2.8
2.2.7
2.2.6
2.2.5
2.2.4
2.2.3
2.2.2
2.2.1
2.2.0
2.1.12
2.1.11
2.1.10
2.1.8
2.1.7
2.1.4
2.1.2
2.0.0
Apache Kafka Erlang client library
Current section
Files
Jump to
Current section
Files
src/brod_kafka_request.erl
%%%
%%% Copyright (c) 2017-2021 Klarna Bank AB (publ)
%%%
%%% 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.
%%%
%% @doc Helper functions for building request messages.
-module(brod_kafka_request).
-export([ create_topics/3
, delete_topics/3
, fetch/8
, list_groups/1
, list_offsets/4
, join_group/2
, metadata/2
, offset_commit/2
, offset_fetch/3
, produce/7
, sync_group/2
]).
-include("brod_int.hrl").
-type api() :: brod_kafka_apis:api().
-type vsn() :: brod_kafka_apis:vsn().
-type topic() :: brod:topic().
-type topic_config() :: kpro:struct().
-type partition() :: brod:partition().
-type offset() :: brod:offset().
-type conn() :: kpro:connection().
%% @doc Make a produce request, If the first arg is a connection pid, call
%% `brod_kafka_apis:pick_version/2' to resolve version.
-spec produce(conn() | vsn(), topic(), partition(),
kpro:batch_input(), integer(), integer(),
brod:compression()) -> kpro:req().
produce(MaybePid, Topic, Partition, BatchInput,
RequiredAcks, AckTimeout, Compression) ->
Vsn = pick_version(produce, MaybePid),
kpro_req_lib:produce(Vsn, Topic, Partition, BatchInput,
#{ required_acks => RequiredAcks
, ack_timeout => AckTimeout
, compression => Compression
}).
%% @doc Make a create_topics request.
-spec create_topics(vsn() | conn(), [topic_config()], #{timeout => kpro:int32(),
validate_only => boolean()}) -> kpro:req().
create_topics(Connection, TopicConfigs, RequestConfigs)
when is_pid(Connection) ->
Vsn = brod_kafka_apis:pick_version(Connection, create_topics),
create_topics(Vsn, TopicConfigs, RequestConfigs);
create_topics(Vsn, TopicConfigs, RequestConfigs) ->
kpro_req_lib:create_topics(Vsn, TopicConfigs, RequestConfigs).
%% @doc Make a delete_topics request.
-spec delete_topics(vsn() | conn(), [topic()], pos_integer()) -> kpro:req().
delete_topics(Connection, Topics, Timeout) when is_pid(Connection) ->
Vsn = brod_kafka_apis:pick_version(Connection, delete_topics),
delete_topics(Vsn, Topics, Timeout);
delete_topics(Vsn, Topics, Timeout) ->
kpro_req_lib:delete_topics(Vsn, Topics, #{timeout => Timeout}).
%% @doc Make a fetch request, If the first arg is a connection pid, call
%% `brod_kafka_apis:pick_version/2' to resolve version.
-spec fetch(conn(), topic(), partition(), offset(),
kpro:wait(), kpro:count(), kpro:count(),
kpro:isolation_level()) -> kpro:req().
fetch(Pid, Topic, Partition, Offset,
WaitTime, MinBytes, MaxBytes, IsolationLevel) ->
Vsn = pick_version(fetch, Pid),
kpro_req_lib:fetch(Vsn, Topic, Partition, Offset,
#{ max_wait_time => WaitTime
, min_bytes => MinBytes
, max_bytes => MaxBytes
, isolation_level => IsolationLevel
}).
%% @doc Make a `list_offsets' request message for offset resolution.
%% In kafka protocol, -2 and -1 are semantic 'time' to request for
%% 'earliest' and 'latest' offsets.
%% In brod implementation, -2, -1, 'earliest' and 'latest'
%% are semantic 'offset', this is why often a variable named
%% Offset is used as the Time argument.
-spec list_offsets(conn(), topic(), partition(), brod:offset_time()) ->
kpro:req().
list_offsets(Connection, Topic, Partition, TimeOrSemanticOffset) ->
Time = ensure_integer_offset_time(TimeOrSemanticOffset),
Vsn = pick_version(list_offsets, Connection),
kpro_req_lib:list_offsets(Vsn, Topic, Partition, Time).
%% @doc Make a metadata request.
-spec metadata(vsn() | conn(), all | [topic()]) -> kpro:req().
metadata(Connection, Topics) when is_pid(Connection) ->
Vsn = brod_kafka_apis:pick_version(Connection, metadata),
metadata(Vsn, Topics);
metadata(Vsn, Topics) ->
kpro_req_lib:metadata(Vsn, Topics).
%% @doc Make a offset fetch request.
%% NOTE: empty topics list only works for kafka 0.10.2.0 or later
-spec offset_fetch(conn(), brod:group_id(), Topics) -> kpro:req()
when Topics :: [{topic(), [partition()]}].
offset_fetch(Connection, GroupId, Topics0) ->
Topics =
lists:map(
fun({Topic, Partitions}) ->
[ {name, Topic}
, {partition_indexes, Partitions}
]
end, Topics0),
Body = [ {group_id, GroupId}
, {topics, case Topics of
[] -> ?kpro_null;
_ -> Topics
end}
],
Vsn = pick_version(offset_fetch, Connection),
kpro:make_request(offset_fetch, Vsn, Body).
%% @doc Make a `list_groups' request.
-spec list_groups(conn()) -> kpro:req().
list_groups(Connection) ->
Vsn = pick_version(list_groups, Connection),
case Vsn >= 3 of
true ->
kpro:make_request(list_groups, Vsn, #{tagged_fields => []});
false ->
kpro:make_request(list_groups, Vsn, [])
end.
%% @doc Make a `join_group' request.
-spec join_group(conn(), kpro:struct()) -> kpro:req().
join_group(Conn, Fields) ->
make_req(join_group, Conn, Fields).
%% @doc Make a `sync_group' request.
-spec sync_group(conn(), kpro:struct()) -> kpro:req().
sync_group(Conn, Fields) ->
make_req(sync_group, Conn, Fields).
%% @doc Make a `offset_commit' request.
-spec offset_commit(conn(), kpro:struct()) -> kpro:req().
offset_commit(Conn, Fields) ->
make_req(offset_commit, Conn, Fields).
%%%_* Internal Functions =======================================================
make_req(API, Conn, Fields) when is_pid(Conn) ->
Vsn = pick_version(API, Conn),
make_req(API, Vsn, Fields);
make_req(API, Vsn, Fields) ->
kpro:make_request(API, Vsn, Fields).
-spec pick_version(api(), pid()) -> vsn().
pick_version(_API, Vsn) when is_integer(Vsn) -> Vsn;
pick_version(API, Connection) when is_pid(Connection) ->
brod_kafka_apis:pick_version(Connection, API);
pick_version(API, _) ->
brod_kafka_apis:default_version(API).
-spec ensure_integer_offset_time(brod:offset_time()) -> integer().
ensure_integer_offset_time(?OFFSET_EARLIEST) -> -2;
ensure_integer_offset_time(?OFFSET_LATEST) -> -1;
ensure_integer_offset_time(T) when is_integer(T) -> T.
%%%_* Emacs ====================================================================
%%% Local Variables:
%%% allout-layout: t
%%% erlang-indent-level: 2
%%% End: