Packages
brod
3.9.2
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
113 Versions
Jump to
Current section
113 Versions
Compare versions
23
files changed
+3
additions
-10065
deletions
| @@ -1,5 +1,5 @@ | |
| 1 1 | {<<"name">>,<<"brod">>}. |
| 2 | - {<<"version">>,<<"3.9.1">>}. |
| 2 | + {<<"version">>,<<"3.9.2">>}. |
| 3 3 | {<<"requirements">>, |
| 4 4 | #{<<"kafka_protocol">> => |
| 5 5 | #{<<"app">> => <<"kafka_protocol">>,<<"optional">> => false, |
| @@ -15,28 +15,7 @@ | |
| 15 15 | [<<"src/brod.app.src">>,<<"LICENSE">>,<<"Makefile">>,<<"NOTICE">>, |
| 16 16 | <<"README.md">>,<<"erlang.mk">>,<<"include/brod.hrl">>, |
| 17 17 | <<"include/brod_int.hrl">>,<<"rebar.config">>,<<"rebar.config.script">>, |
| 18 | - <<"src/_build/test/extras/.rebar3/erlcinfo">>, |
| 19 | - <<"src/_build/test/extras/brod.erl">>, |
| 20 | - <<"src/_build/test/extras/brod_cg_commits.erl">>, |
| 21 | - <<"src/_build/test/extras/brod_cli.erl">>, |
| 22 | - <<"src/_build/test/extras/brod_cli_pipe.erl">>, |
| 23 | - <<"src/_build/test/extras/brod_client.erl">>, |
| 24 | - <<"src/_build/test/extras/brod_consumer.erl">>, |
| 25 | - <<"src/_build/test/extras/brod_consumers_sup.erl">>, |
| 26 | - <<"src/_build/test/extras/brod_group_coordinator.erl">>, |
| 27 | - <<"src/_build/test/extras/brod_group_member.erl">>, |
| 28 | - <<"src/_build/test/extras/brod_group_subscriber.erl">>, |
| 29 | - <<"src/_build/test/extras/brod_group_subscriber_v2.erl">>, |
| 30 | - <<"src/_build/test/extras/brod_group_subscriber_worker.erl">>, |
| 31 | - <<"src/_build/test/extras/brod_kafka_apis.erl">>, |
| 32 | - <<"src/_build/test/extras/brod_kafka_request.erl">>, |
| 33 | - <<"src/_build/test/extras/brod_producer.erl">>, |
| 34 | - <<"src/_build/test/extras/brod_producer_buffer.erl">>, |
| 35 | - <<"src/_build/test/extras/brod_producers_sup.erl">>, |
| 36 | - <<"src/_build/test/extras/brod_sup.erl">>, |
| 37 | - <<"src/_build/test/extras/brod_topic_subscriber.erl">>, |
| 38 | - <<"src/_build/test/extras/brod_utils.erl">>,<<"src/brod.erl">>, |
| 39 | - <<"src/brod_cg_commits.erl">>,<<"src/brod_cli.erl">>, |
| 18 | + <<"src/brod.erl">>,<<"src/brod_cg_commits.erl">>,<<"src/brod_cli.erl">>, |
| 40 19 | <<"src/brod_cli_pipe.erl">>,<<"src/brod_client.erl">>, |
| 41 20 | <<"src/brod_consumer.erl">>,<<"src/brod_consumers_sup.erl">>, |
| 42 21 | <<"src/brod_group_coordinator.erl">>,<<"src/brod_group_member.erl">>, |
| @@ -1,905 +0,0 @@ | |
| 1 | - %%% |
| 2 | - %%% Copyright (c) 2014-2018, Klarna Bank AB (publ) |
| 3 | - %%% |
| 4 | - %%% Licensed under the Apache License, Version 2.0 (the "License"); |
| 5 | - %%% you may not use this file except in compliance with the License. |
| 6 | - %%% You may obtain a copy of the License at |
| 7 | - %%% |
| 8 | - %%% http://www.apache.org/licenses/LICENSE-2.0 |
| 9 | - %%% |
| 10 | - %%% Unless required by applicable law or agreed to in writing, software |
| 11 | - %%% distributed under the License is distributed on an "AS IS" BASIS, |
| 12 | - %%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 13 | - %%% See the License for the specific language governing permissions and |
| 14 | - %%% limitations under the License. |
| 15 | - %%% |
| 16 | - |
| 17 | - -module(brod). |
| 18 | - -behaviour(application). |
| 19 | - |
| 20 | - %% Application |
| 21 | - -export([ start/0 |
| 22 | - , start/2 |
| 23 | - , stop/0 |
| 24 | - , stop/1 |
| 25 | - ]). |
| 26 | - |
| 27 | - %% Client API |
| 28 | - -export([ get_partitions_count/2 |
| 29 | - , start_client/1 |
| 30 | - , start_client/2 |
| 31 | - , start_client/3 |
| 32 | - , start_consumer/3 |
| 33 | - , start_producer/3 |
| 34 | - , stop_client/1 |
| 35 | - ]). |
| 36 | - |
| 37 | - -export([ start_link_client/1 |
| 38 | - , start_link_client/2 |
| 39 | - , start_link_client/3 |
| 40 | - ]). |
| 41 | - |
| 42 | - %% Producer API |
| 43 | - -export([ get_producer/3 |
| 44 | - , produce/2 |
| 45 | - , produce/3 |
| 46 | - , produce/5 |
| 47 | - , produce_cb/4 |
| 48 | - , produce_cb/6 |
| 49 | - , produce_sync/2 |
| 50 | - , produce_sync/3 |
| 51 | - , produce_sync/5 |
| 52 | - , produce_sync_offset/5 |
| 53 | - , produce_no_ack/5 |
| 54 | - , sync_produce_request/1 |
| 55 | - , sync_produce_request/2 |
| 56 | - , sync_produce_request_offset/1 |
| 57 | - , sync_produce_request_offset/2 |
| 58 | - ]). |
| 59 | - |
| 60 | - %% Simple Consumer API |
| 61 | - -export([ consume_ack/2 |
| 62 | - , consume_ack/4 |
| 63 | - , get_consumer/3 |
| 64 | - , subscribe/3 |
| 65 | - , subscribe/5 |
| 66 | - , unsubscribe/1 |
| 67 | - , unsubscribe/2 |
| 68 | - , unsubscribe/3 |
| 69 | - , unsubscribe/4 |
| 70 | - ]). |
| 71 | - |
| 72 | - %% Subscriber API |
| 73 | - -export([ start_link_group_subscriber/7 |
| 74 | - , start_link_group_subscriber/8 |
| 75 | - , start_link_group_subscriber_v2/1 |
| 76 | - , start_link_topic_subscriber/5 |
| 77 | - , start_link_topic_subscriber/6 |
| 78 | - , start_link_topic_subscriber/7 |
| 79 | - ]). |
| 80 | - |
| 81 | - %% APIs for quick metadata or message inspection and brod_cli |
| 82 | - -export([ get_metadata/1 |
| 83 | - , get_metadata/2 |
| 84 | - , get_metadata/3 |
| 85 | - , resolve_offset/3 |
| 86 | - , resolve_offset/4 |
| 87 | - , resolve_offset/5 |
| 88 | - , fetch/4 |
| 89 | - , fetch/5 |
| 90 | - , fold/8 |
| 91 | - , connect_leader/4 |
| 92 | - , list_all_groups/2 |
| 93 | - , list_groups/2 |
| 94 | - , describe_groups/3 |
| 95 | - , connect_group_coordinator/3 |
| 96 | - , fetch_committed_offsets/2 |
| 97 | - , fetch_committed_offsets/3 |
| 98 | - ]). |
| 99 | - |
| 100 | - -deprecated([ {fetch, 7, next_version} |
| 101 | - , {fetch, 8, next_version} |
| 102 | - ]). |
| 103 | - |
| 104 | - -export([ fetch/7 |
| 105 | - , fetch/8 |
| 106 | - ]). |
| 107 | - |
| 108 | - -ifdef(build_brod_cli). |
| 109 | - -export([main/1]). |
| 110 | - -endif. |
| 111 | - |
| 112 | - -export_type([ batch_input/0 |
| 113 | - , bootstrap/0 |
| 114 | - , call_ref/0 |
| 115 | - , cg/0 |
| 116 | - , cg_protocol_type/0 |
| 117 | - , client/0 |
| 118 | - , client_config/0 |
| 119 | - , client_id/0 |
| 120 | - , compression/0 |
| 121 | - , connection/0 |
| 122 | - , conn_config/0 |
| 123 | - , consumer_config/0 |
| 124 | - , consumer_option/0 |
| 125 | - , consumer_options/0 |
| 126 | - , endpoint/0 |
| 127 | - , error_code/0 |
| 128 | - , fetch_opts/0 |
| 129 | - , fold_acc/0 |
| 130 | - , fold_fun/1 |
| 131 | - , fold_limits/0 |
| 132 | - , fold_stop_reason/0 |
| 133 | - , fold_result/0 |
| 134 | - , group_config/0 |
| 135 | - , group_generation_id/0 |
| 136 | - , group_id/0 |
| 137 | - , group_member/0 |
| 138 | - , group_member_id/0 |
| 139 | - , hostname/0 |
| 140 | - , key/0 |
| 141 | - , msg_input/0 |
| 142 | - , msg_ts/0 |
| 143 | - , message/0 |
| 144 | - , message_set/0 |
| 145 | - , offset/0 |
| 146 | - , offset_time/0 |
| 147 | - , partition/0 |
| 148 | - , partition_assignment/0 |
| 149 | - , partition_fun/0 |
| 150 | - , partitioner/0 |
| 151 | - , portnum/0 |
| 152 | - , produce_ack_cb/0 |
| 153 | - , producer_config/0 |
| 154 | - , produce_reply/0 |
| 155 | - , produce_result/0 |
| 156 | - , received_assignments/0 |
| 157 | - , topic/0 |
| 158 | - , topic_partition/0 |
| 159 | - , value/0 |
| 160 | - ]). |
| 161 | - |
| 162 | - -include("brod_int.hrl"). |
| 163 | - |
| 164 | - %%%_* Types ==================================================================== |
| 165 | - |
| 166 | - %% basics |
| 167 | - -type hostname() :: kpro:hostname(). |
| 168 | - -type portnum() :: pos_integer(). |
| 169 | - -type endpoint() :: {hostname(), portnum()}. |
| 170 | - -type topic() :: kpro:topic(). |
| 171 | - -type partition() :: kpro:partition(). |
| 172 | - -type topic_partition() :: {topic(), partition()}. |
| 173 | - -type offset() :: kpro:offset(). |
| 174 | - -type key() :: undefined %% no key, transformed to <<>> |
| 175 | - | binary(). |
| 176 | - -type value() :: undefined %% no value, transformed to <<>> |
| 177 | - | iodata() %% single value |
| 178 | - | {msg_ts(), binary()} %% one message with timestamp |
| 179 | - | [?KV(key(), value())] %% backward compatible |
| 180 | - | [?TKV(msg_ts(), key(), value())] %% backward compatible |
| 181 | - | kpro:msg_input() %% one magic v2 message |
| 182 | - | kpro:batch_input(). %% maybe nested batch |
| 183 | - |
| 184 | - -type msg_input() :: kpro:msg_input(). |
| 185 | - -type batch_input() :: [msg_input()]. |
| 186 | - |
| 187 | - -type msg_ts() :: kpro:msg_ts(). |
| 188 | - -type client_id() :: atom(). |
| 189 | - -type client() :: client_id() | pid(). |
| 190 | - -type client_config() :: brod_client:config(). |
| 191 | - -type bootstrap() :: [endpoint()] %% default client config |
| 192 | - | {[endpoint()], client_config()}. |
| 193 | - -type offset_time() :: integer() |
| 194 | - | ?OFFSET_EARLIEST |
| 195 | - | ?OFFSET_LATEST. |
| 196 | - -type message() :: kpro:message(). |
| 197 | - -type message_set() :: #kafka_message_set{}. |
| 198 | - -type error_code() :: kpro:error_code(). |
| 199 | - |
| 200 | - %% producers |
| 201 | - -type produce_reply() :: #brod_produce_reply{}. |
| 202 | - -type producer_config() :: brod_producer:config(). |
| 203 | - -type partition_fun() :: fun((topic(), pos_integer(), key(), value()) -> |
| 204 | - {ok, partition()}). |
| 205 | - -type partitioner() :: partition_fun() | random | hash. |
| 206 | - -type produce_ack_cb() :: fun((partition(), offset()) -> _). |
| 207 | - -type compression() :: no_compression | gzip | snappy. |
| 208 | - -type call_ref() :: #brod_call_ref{}. |
| 209 | - -type produce_result() :: brod_produce_req_buffered |
| 210 | - | brod_produce_req_acked. |
| 211 | - |
| 212 | - |
| 213 | - %% consumers |
| 214 | - -type consumer_option() :: begin_offset |
| 215 | - | min_bytes |
| 216 | - | max_bytes |
| 217 | - | max_wait_time |
| 218 | - | sleep_timeout |
| 219 | - | prefetch_count |
| 220 | - | prefetch_bytes |
| 221 | - | offset_reset_policy |
| 222 | - | size_stat_window. |
| 223 | - -type consumer_options() :: [{consumer_option(), integer()}]. |
| 224 | - -type consumer_config() :: brod_consumer:config(). |
| 225 | - -type connection() :: kpro:connection(). |
| 226 | - -type conn_config() :: [{atom(), term()}] | kpro:conn_config(). |
| 227 | - |
| 228 | - %% consumer groups |
| 229 | - -type group_id() :: kpro:group_id(). |
| 230 | - -type group_member_id() :: binary(). |
| 231 | - -type group_member() :: {group_member_id(), #kafka_group_member_metadata{}}. |
| 232 | - -type group_generation_id() :: non_neg_integer(). |
| 233 | - -type group_config() :: proplists:proplist(). |
| 234 | - -type partition_assignment() :: {topic() , [partition()]}. |
| 235 | - -type received_assignments() :: [#brod_received_assignment{}]. |
| 236 | - -type cg() :: #brod_cg{}. |
| 237 | - -type cg_protocol_type() :: binary(). |
| 238 | - -type fetch_opts() :: kpro:fetch_opts(). |
| 239 | - -type fold_acc() :: term(). |
| 240 | - -type fold_fun(Acc) :: fun((message(), Acc) -> {ok, Acc} | {error, any()}). |
| 241 | - %% fold always returns when reaches the high watermark offset |
| 242 | - %% fold also returns when any of the limits is hit |
| 243 | - -type fold_limits() :: #{ message_count => pos_integer() |
| 244 | - , reach_offset => offset() |
| 245 | - }. |
| 246 | - -type fold_stop_reason() :: reached_end_of_partition |
| 247 | - | reached_message_count_limit |
| 248 | - | reached_target_offset |
| 249 | - | {error, any()}. |
| 250 | - %% OffsetToContinue: begin offset for the next fold call |
| 251 | - -type fold_result() :: ?BROD_FOLD_RET(fold_acc(), |
| 252 | - OffsetToContinue :: offset(), |
| 253 | - fold_stop_reason()). |
| 254 | - |
| 255 | - %%%_* APIs ===================================================================== |
| 256 | - |
| 257 | - %% @doc Start brod application. |
| 258 | - -spec start() -> ok | no_return(). |
| 259 | - start() -> |
| 260 | - {ok, _Apps} = application:ensure_all_started(brod), |
| 261 | - ok. |
| 262 | - |
| 263 | - %% @doc Stop brod application. |
| 264 | - -spec stop() -> ok. |
| 265 | - stop() -> |
| 266 | - application:stop(brod). |
| 267 | - |
| 268 | - %% @doc Application behaviour callback |
| 269 | - start(_StartType, _StartArgs) -> brod_sup:start_link(). |
| 270 | - |
| 271 | - %% @doc Application behaviour callback |
| 272 | - stop(_State) -> ok. |
| 273 | - |
| 274 | - %% @equiv start_client(BootstrapEndpoints, brod_default_client) |
| 275 | - -spec start_client([endpoint()]) -> ok | {error, any()}. |
| 276 | - start_client(BootstrapEndpoints) -> |
| 277 | - start_client(BootstrapEndpoints, ?BROD_DEFAULT_CLIENT_ID). |
| 278 | - |
| 279 | - %% @equiv start_client(BootstrapEndpoints, ClientId, []) |
| 280 | - -spec start_client([endpoint()], client_id()) -> ok | {error, any()}. |
| 281 | - start_client(BootstrapEndpoints, ClientId) -> |
| 282 | - start_client(BootstrapEndpoints, ClientId, []). |
| 283 | - |
| 284 | - %% @doc Start a client. |
| 285 | - %% |
| 286 | - %% BootstrapEndpoints: |
| 287 | - %% Kafka cluster endpoints, can be any of the brokers in the cluster |
| 288 | - %% which does not necessarily have to be a leader of any partition, |
| 289 | - %% e.g. a load-balanced entrypoint to the remote kakfa cluster. |
| 290 | - %% |
| 291 | - %% ClientId: Atom to identify the client process. |
| 292 | - %% |
| 293 | - %% Config is a proplist, possible values: |
| 294 | - %% restart_delay_seconds (optional, default=10) |
| 295 | - %% How much time to wait between attempts to restart brod_client |
| 296 | - %% process when it crashes |
| 297 | - %% |
| 298 | - %% get_metadata_timeout_seconds (optional, default=5) |
| 299 | - %% Return `{error, timeout}' from `brod_client:get_xxx' calls if |
| 300 | - %% responses for APIs such as `metadata', `find_coordinator' |
| 301 | - %% is not received in time. |
| 302 | - %% |
| 303 | - %% reconnect_cool_down_seconds (optional, default=1) |
| 304 | - %% Delay this configured number of seconds before retrying to |
| 305 | - %% estabilish a new connection to the kafka partition leader. |
| 306 | - %% |
| 307 | - %% allow_topic_auto_creation (optional, default=true) |
| 308 | - %% By default, brod respects what is configured in broker about |
| 309 | - %% topic auto-creation. i.e. whatever `auto.create.topics.enable' |
| 310 | - %% is set in broker configuration. |
| 311 | - %% However if `allow_topic_auto_creation' is set to `false' in client |
| 312 | - %% config, brod will avoid sending metadata requests that may cause an |
| 313 | - %% auto-creation of the topic regardless of what broker config is. |
| 314 | - %% |
| 315 | - %% auto_start_producers (optional, default=false) |
| 316 | - %% If true, brod client will spawn a producer automatically when |
| 317 | - %% user is trying to call `produce' but did not call `brod:start_producer' |
| 318 | - %% explicitly. Can be useful for applications which don't know beforehand |
| 319 | - %% which topics they will be working with. |
| 320 | - %% |
| 321 | - %% default_producer_config (optional, default=[]) |
| 322 | - %% Producer configuration to use when auto_start_producers is true. |
| 323 | - %% @see brod_producer:start_link/4. for details about producer config |
| 324 | - %% |
| 325 | - %% Connection config entries can be added in the same proplist. |
| 326 | - %% see `kpro_connection.erl' in `kafka_protocol' for more details. |
| 327 | - %% |
| 328 | - %% ssl (optional, default=false) |
| 329 | - %% `true | false | ssl:ssl_option()' |
| 330 | - %% `true' is translated to `[]' as `ssl:ssl_option()' i.e. all default. |
| 331 | - %% |
| 332 | - %% sasl (optional, default=undefined) |
| 333 | - %% Credentials for SASL/Plain authentication. |
| 334 | - %% `{mechanism(), Filename}' or `{mechanism(), UserName, Password}' |
| 335 | - %% where mechanism can be atoms: plain (for "PLAIN"), scram_sha_256 |
| 336 | - %% (for "SCRAM-SHA-256") or scram_sha_512 (for SCRAM-SHA-512). |
| 337 | - %% `Filename' should be a file consisting two lines, first line |
| 338 | - %% is the username and second line is the password. |
| 339 | - %% `Username', `Password' should be `string() | binary()' |
| 340 | - %% |
| 341 | - %% connect_timeout (optional, default=5000) |
| 342 | - %% Timeout when trying to connect to an endpoint. |
| 343 | - %% |
| 344 | - %% request_timeout (optional, default=240000, constraint: >= 1000) |
| 345 | - %% Timeout when waiting for a response, connection restart when timed out. |
| 346 | - %% query_api_versions (optional, default=true) |
| 347 | - %% Must be set to false to work with kafka versions prior to 0.10, |
| 348 | - %% When set to `true', at connection start, brod will send a query request |
| 349 | - %% to get the broker supported API version ranges. |
| 350 | - %% When set to 'false', brod will alway use the lowest supported API version |
| 351 | - %% when sending requests to kafka. |
| 352 | - %% Supported API version ranges can be found in: |
| 353 | - %% `brod_kafka_apis:supported_versions/1' |
| 354 | - %% extra_sock_opts (optional, default=[]) |
| 355 | - %% Extra socket options to tune socket performance. |
| 356 | - %% e.g. [{sndbuf, 1 bsl 20}]. |
| 357 | - %% ref: http://erlang.org/doc/man/gen_tcp.html#type-option |
| 358 | - -spec start_client([endpoint()], client_id(), client_config()) -> |
| 359 | - ok | {error, any()}. |
| 360 | - start_client(BootstrapEndpoints, ClientId, Config) -> |
| 361 | - case brod_sup:start_client(BootstrapEndpoints, ClientId, Config) of |
| 362 | - ok -> ok; |
| 363 | - {error, {already_started, _Pid}} -> ok; |
| 364 | - {error, Reason} -> {error, Reason} |
| 365 | - end. |
| 366 | - |
| 367 | - %% @equiv start_link_client(BootstrapEndpoints, brod_default_client) |
| 368 | - -spec start_link_client([endpoint()]) -> {ok, pid()} | {error, any()}. |
| 369 | - start_link_client(BootstrapEndpoints) -> |
| 370 | - start_link_client(BootstrapEndpoints, ?BROD_DEFAULT_CLIENT_ID). |
| 371 | - |
| 372 | - %% @equiv start_link_client(BootstrapEndpoints, ClientId, []) |
| 373 | - -spec start_link_client([endpoint()], client_id()) -> |
| 374 | - {ok, pid()} | {error, any()}. |
| 375 | - start_link_client(BootstrapEndpoints, ClientId) -> |
| 376 | - start_link_client(BootstrapEndpoints, ClientId, []). |
| 377 | - |
| 378 | - -spec start_link_client([endpoint()], client_id(), client_config()) -> |
| 379 | - {ok, pid()} | {error, any()}. |
| 380 | - start_link_client(BootstrapEndpoints, ClientId, Config) -> |
| 381 | - brod_client:start_link(BootstrapEndpoints, ClientId, Config). |
| 382 | - |
| 383 | - %% @doc Stop a client. |
| 384 | - -spec stop_client(client()) -> ok. |
| 385 | - stop_client(Client) when is_atom(Client) -> |
| 386 | - case brod_sup:find_client(Client) of |
| 387 | - [_Pid] -> brod_sup:stop_client(Client); |
| 388 | - [] -> brod_client:stop(Client) |
| 389 | - end; |
| 390 | - stop_client(Client) when is_pid(Client) -> |
| 391 | - brod_client:stop(Client). |
| 392 | - |
| 393 | - %% @doc Dynamically start a per-topic producer. |
| 394 | - %% @see brod_producer:start_link/4. for details about producer config. |
| 395 | - -spec start_producer(client(), topic(), producer_config()) -> |
| 396 | - ok | {error, any()}. |
| 397 | - start_producer(Client, TopicName, ProducerConfig) -> |
| 398 | - brod_client:start_producer(Client, TopicName, ProducerConfig). |
| 399 | - |
| 400 | - %% @doc Dynamically start a topic consumer. |
| 401 | - %% @see brod_consumer:start_link/5. for details about consumer config. |
| 402 | - -spec start_consumer(client(), topic(), consumer_config()) -> |
| 403 | - ok | {error, any()}. |
| 404 | - start_consumer(Client, TopicName, ConsumerConfig) -> |
| 405 | - brod_client:start_consumer(Client, TopicName, ConsumerConfig). |
| 406 | - |
| 407 | - %% @doc Get number of partitions for a given topic. |
| 408 | - %% The higher level producers may need the partition numbers to |
| 409 | - %% find the partition producer pid --- if the number of partitions |
| 410 | - %% is not statically configured for them. |
| 411 | - %% It is up to the callers how they want to distribute their data |
| 412 | - %% (e.g. random, roundrobin or consistent-hashing) to the partitions. |
| 413 | - -spec get_partitions_count(client(), topic()) -> |
| 414 | - {ok, pos_integer()} | {error, any()}. |
| 415 | - get_partitions_count(Client, Topic) -> |
| 416 | - brod_client:get_partitions_count(Client, Topic). |
| 417 | - |
| 418 | - -spec get_consumer(client(), topic(), partition()) -> |
| 419 | - {ok, pid()} | {error, Reason} |
| 420 | - when Reason :: client_down |
| 421 | - | {consumer_down, noproc} |
| 422 | - | {consumer_not_found, topic()} |
| 423 | - | {consumer_not_found, topic(), partition()}. |
| 424 | - get_consumer(Client, Topic, Partition) -> |
| 425 | - brod_client:get_consumer(Client, Topic, Partition). |
| 426 | - |
| 427 | - %% @equiv brod_client:get_producer/3 |
| 428 | - -spec get_producer(client(), topic(), partition()) -> |
| 429 | - {ok, pid()} | {error, Reason} |
| 430 | - when Reason :: client_down |
| 431 | - | {producer_down, noproc} |
| 432 | - | {producer_not_found, topic()} |
| 433 | - | {producer_not_found, topic(), partition()}. |
| 434 | - get_producer(Client, Topic, Partition) -> |
| 435 | - brod_client:get_producer(Client, Topic, Partition). |
| 436 | - |
| 437 | - %% @equiv produce(Pid, 0, <<>>, Value) |
| 438 | - -spec produce(pid(), value()) -> {ok, call_ref()} | {error, any()}. |
| 439 | - produce(Pid, Value) -> |
| 440 | - produce(Pid, _Key = <<>>, Value). |
| 441 | - |
| 442 | - %% @doc Produce one message if Value is binary or iolist, |
| 443 | - %% Or send a batch if Value is a (nested) kv-list or a list of maps, |
| 444 | - %% in this case Key is discarded (only the keys in kv-list are sent to kafka). |
| 445 | - %% The pid should be a partition producer pid, NOT client pid. |
| 446 | - %% The return value is a call reference of type `call_ref()', |
| 447 | - %% so the caller can used it to expect (match) a |
| 448 | - %% `#brod_produce_reply{result = brod_produce_req_acked}' |
| 449 | - %% message after the produce request has been acked by kafka. |
| 450 | - -spec produce(pid(), key(), value()) -> |
| 451 | - {ok, call_ref()} | {error, any()}. |
| 452 | - produce(ProducerPid, Key, Value) -> |
| 453 | - brod_producer:produce(ProducerPid, Key, Value). |
| 454 | - |
| 455 | - %% @doc Produce one message if Value is binary or iolist, |
| 456 | - %% Or send a batch if Value is a (nested) kv-list or a list of maps, |
| 457 | - %% in this case Key is used only for partitioning, or discarded if the 3rd |
| 458 | - %% arg is a partition number instead of a partitioner callback. |
| 459 | - %% This function first lookup the producer pid, |
| 460 | - %% then call `produce/3' to do the real work. |
| 461 | - %% The return value is a call reference of type `call_ref()', |
| 462 | - %% so the caller can used it to expect (match) a |
| 463 | - %% `#brod_produce_reply{result = brod_produce_req_acked}' |
| 464 | - %% message after the produce request has been acked by kafka. |
| 465 | - -spec produce(client(), topic(), partition() | partitioner(), |
| 466 | - key(), value()) -> {ok, call_ref()} | {error, any()}. |
| 467 | - produce(Client, Topic, Partition, Key, Value) when is_integer(Partition) -> |
| 468 | - case get_producer(Client, Topic, Partition) of |
| 469 | - {ok, Pid} -> produce(Pid, Key, Value); |
| 470 | - {error, Reason} -> {error, Reason} |
| 471 | - end; |
| 472 | - produce(Client, Topic, Partitioner, Key, Value) -> |
| 473 | - PartFun = brod_utils:make_part_fun(Partitioner), |
| 474 | - case brod_client:get_partitions_count(Client, Topic) of |
| 475 | - {ok, PartitionsCnt} -> |
| 476 | - {ok, Partition} = PartFun(Topic, PartitionsCnt, Key, Value), |
| 477 | - produce(Client, Topic, Partition, Key, Value); |
| 478 | - {error, Reason} -> |
| 479 | - {error, Reason} |
| 480 | - end. |
| 481 | - |
| 482 | - %% @doc Same as `produce/3' only the ack is not delivered as a message, |
| 483 | - %% instead, the callback is evaluated by producer worker when ack is received |
| 484 | - %% from kafka. |
| 485 | - -spec produce_cb(pid(), key(), value(), produce_ack_cb()) -> |
| 486 | - ok | {error, any()}. |
| 487 | - produce_cb(ProducerPid, Key, Value, AckCb) -> |
| 488 | - brod_producer:produce_cb(ProducerPid, Key, Value, AckCb). |
| 489 | - |
| 490 | - %% @doc Same as `produce/5' only the ack is not delivered as a message, |
| 491 | - %% instead, the callback is evaluated by producer worker when ack is received |
| 492 | - %% from kafka. Return the partition to caller as `{ok, Partition}' for caller |
| 493 | - %% to correlate the callback when the 3rd arg is not a partition number. |
| 494 | - -spec produce_cb(client(), topic(), partition() | partitioner(), |
| 495 | - key(), value(), produce_ack_cb()) -> |
| 496 | - ok | {ok, partition()} | {error, any()}. |
| 497 | - produce_cb(Client, Topic, Part, Key, Value, AckCb) when is_integer(Part) -> |
| 498 | - case get_producer(Client, Topic, Part) of |
| 499 | - {ok, Pid} -> produce_cb(Pid, Key, Value, AckCb); |
| 500 | - {error, Reason} -> {error, Reason} |
| 501 | - end; |
| 502 | - produce_cb(Client, Topic, Partitioner, Key, Value, AckCb) -> |
| 503 | - PartFun = brod_utils:make_part_fun(Partitioner), |
| 504 | - case brod_client:get_partitions_count(Client, Topic) of |
| 505 | - {ok, PartitionsCnt} -> |
| 506 | - {ok, Partition} = PartFun(Topic, PartitionsCnt, Key, Value), |
| 507 | - case produce_cb(Client, Topic, Partition, Key, Value, AckCb) of |
| 508 | - ok -> {ok, Partition}; |
| 509 | - {error, Reason} -> {error, Reason} |
| 510 | - end; |
| 511 | - {error, Reason} -> |
| 512 | - {error, Reason} |
| 513 | - end. |
| 514 | - |
| 515 | - %% @doc Send the message to partition worker without any ack. |
| 516 | - %% NOTE: This call has no back-pressure to the caller, |
| 517 | - %% excessive usage may cause beam to run out of memory. |
| 518 | - -spec produce_no_ack(pid(), key(), value()) -> ok | {error, any()}. |
| 519 | - produce_no_ack(ProducerPid, Key, Value) -> |
| 520 | - brod_producer:produce_no_ack(ProducerPid, Key, Value). |
| 521 | - |
| 522 | - %% @doc Find the partition worker and send message without any ack. |
| 523 | - %% NOTE: This call has no back-pressure to the caller, |
| 524 | - %% excessive usage may cause beam to run out of memory. |
| 525 | - -spec produce_no_ack(client(), topic(), partition() | partitioner(), |
| 526 | - key(), value()) -> ok | {error, any()}. |
| 527 | - produce_no_ack(Client, Topic, Part, Key, Value) when is_integer(Part) -> |
| 528 | - case get_producer(Client, Topic, Part) of |
| 529 | - {ok, Pid} -> produce_no_ack(Pid, Key, Value); |
| 530 | - {error, Reason} -> {error, Reason} |
| 531 | - end; |
| 532 | - produce_no_ack(Client, Topic, Partitioner, Key, Value) -> |
| 533 | - PartFun = brod_utils:make_part_fun(Partitioner), |
| 534 | - case brod_client:get_partitions_count(Client, Topic) of |
| 535 | - {ok, PartitionsCnt} -> |
| 536 | - {ok, Partition} = PartFun(Topic, PartitionsCnt, Key, Value), |
| 537 | - produce_no_ack(Client, Topic, Partition, Key, Value); |
| 538 | - {error, _Reason} -> |
| 539 | - %% error ignored |
| 540 | - ok |
| 541 | - end. |
| 542 | - |
| 543 | - %% @doc Same as `produce/5' only the ack is not d |
| 544 | - %% @equiv produce_sync(Pid, 0, <<>>, Value) |
| 545 | - -spec produce_sync(pid(), value()) -> ok. |
| 546 | - produce_sync(Pid, Value) -> |
| 547 | - produce_sync(Pid, _Key = <<>>, Value). |
| 548 | - |
| 549 | - %% @doc Sync version of produce/3 |
| 550 | - %% This function will not return until a response is received from kafka, |
| 551 | - %% however if producer is started with required_acks set to 0, this function |
| 552 | - %% will return onece the messages is buffered in the producer process. |
| 553 | - -spec produce_sync(pid(), key(), value()) -> |
| 554 | - ok | {error, any()}. |
| 555 | - produce_sync(Pid, Key, Value) -> |
| 556 | - case produce(Pid, Key, Value) of |
| 557 | - {ok, CallRef} -> |
| 558 | - %% Wait until the request is acked by kafka |
| 559 | - sync_produce_request(CallRef); |
| 560 | - {error, Reason} -> |
| 561 | - {error, Reason} |
| 562 | - end. |
| 563 | - |
| 564 | - %% @doc Sync version of produce/5 |
| 565 | - %% This function will not return until a response is received from kafka, |
| 566 | - %% however if producer is started with required_acks set to 0, this function |
| 567 | - %% will return once the messages are buffered in the producer process. |
| 568 | - -spec produce_sync(client(), topic(), partition() | partitioner(), |
| 569 | - key(), value()) -> ok | {error, any()}. |
| 570 | - produce_sync(Client, Topic, Partition, Key, Value) -> |
| 571 | - case produce_sync_offset(Client, Topic, Partition, Key, Value) of |
| 572 | - {ok, _} -> ok; |
| 573 | - Else -> Else |
| 574 | - end. |
| 575 | - |
| 576 | - %% @doc Version of produce_sync/5 that returns the offset assigned by Kafka |
| 577 | - %% If producer is started with required_acks set to 0, the offset will be |
| 578 | - %% `?BROD_PRODUCE_UNKNOWN_OFFSET'. |
| 579 | - -spec produce_sync_offset(client(), topic(), partition() | partitioner(), |
| 580 | - key(), value()) -> {ok, offset()} | {error, any()}. |
| 581 | - produce_sync_offset(Client, Topic, Partition, Key, Value) -> |
| 582 | - case produce(Client, Topic, Partition, Key, Value) of |
| 583 | - {ok, CallRef} -> |
| 584 | - sync_produce_request_offset(CallRef); |
| 585 | - {error, Reason} -> |
| 586 | - {error, Reason} |
| 587 | - end. |
| 588 | - |
| 589 | - %% @doc Block wait for sent produced request to be acked by kafka. |
| 590 | - -spec sync_produce_request(call_ref()) -> |
| 591 | - ok | {error, Reason :: any()}. |
| 592 | - sync_produce_request(CallRef) -> |
| 593 | - sync_produce_request(CallRef, infinity). |
| 594 | - |
| 595 | - -spec sync_produce_request(call_ref(), timeout()) -> |
| 596 | - ok | {error, Reason :: any()}. |
| 597 | - sync_produce_request(CallRef, Timeout) -> |
| 598 | - case sync_produce_request_offset(CallRef, Timeout) of |
| 599 | - {ok, _} -> ok; |
| 600 | - Else -> Else |
| 601 | - end. |
| 602 | - |
| 603 | - %% @doc As sync_produce_request_offset/1, but also returning assigned offset |
| 604 | - %% See produce_sync_offset/5. |
| 605 | - -spec sync_produce_request_offset(call_ref()) -> |
| 606 | - {ok, offset()} | {error, Reason :: any()}. |
| 607 | - sync_produce_request_offset(CallRef) -> |
| 608 | - sync_produce_request_offset(CallRef, infinity). |
| 609 | - |
| 610 | - -spec sync_produce_request_offset(call_ref(), timeout()) -> |
| 611 | - {ok, offset()} | {error, Reason :: any()}. |
| 612 | - sync_produce_request_offset(CallRef, Timeout) -> |
| 613 | - brod_producer:sync_produce_request(CallRef, Timeout). |
| 614 | - |
| 615 | - %% @doc Subscribe data stream from the given topic-partition. |
| 616 | - %% If {error, Reason} is returned, the caller should perhaps retry later. |
| 617 | - %% {ok, ConsumerPid} is returned if success, the caller may want to monitor |
| 618 | - %% the consumer pid to trigger a re-subscribe in case it crashes. |
| 619 | - %% |
| 620 | - %% If subscribed successfully, the subscriber process should expect messages |
| 621 | - %% of pattern: |
| 622 | - %% {ConsumerPid, #kafka_message_set{}} and |
| 623 | - %% {ConsumerPid, #kafka_fetch_error{}}, |
| 624 | - %% -include_lib(brod/include/brod.hrl) to access the records. |
| 625 | - %% In case #kafka_fetch_error{} is received the subscriber should re-subscribe |
| 626 | - %% itself to resume the data stream. |
| 627 | - -spec subscribe(client(), pid(), topic(), partition(), |
| 628 | - consumer_options()) -> {ok, pid()} | {error, any()}. |
| 629 | - subscribe(Client, SubscriberPid, Topic, Partition, Options) -> |
| 630 | - case brod_client:get_consumer(Client, Topic, Partition) of |
| 631 | - {ok, ConsumerPid} -> |
| 632 | - case subscribe(ConsumerPid, SubscriberPid, Options) of |
| 633 | - ok -> {ok, ConsumerPid}; |
| 634 | - Error -> Error |
| 635 | - end; |
| 636 | - {error, Reason} -> |
| 637 | - {error, Reason} |
| 638 | - end. |
| 639 | - |
| 640 | - -spec subscribe(pid(), pid(), consumer_options()) -> ok | {error, any()}. |
| 641 | - subscribe(ConsumerPid, SubscriberPid, Options) -> |
| 642 | - brod_consumer:subscribe(ConsumerPid, SubscriberPid, Options). |
| 643 | - |
| 644 | - %% @doc Unsubscribe the current subscriber. Assuming the subscriber is self(). |
| 645 | - -spec unsubscribe(client(), topic(), partition()) -> ok | {error, any()}. |
| 646 | - unsubscribe(Client, Topic, Partition) -> |
| 647 | - unsubscribe(Client, Topic, Partition, self()). |
| 648 | - |
| 649 | - %% @doc Unsubscribe the current subscriber. |
| 650 | - -spec unsubscribe(client(), topic(), partition(), pid()) -> ok | {error, any()}. |
| 651 | - unsubscribe(Client, Topic, Partition, SubscriberPid) -> |
| 652 | - case brod_client:get_consumer(Client, Topic, Partition) of |
| 653 | - {ok, ConsumerPid} -> unsubscribe(ConsumerPid, SubscriberPid); |
| 654 | - Error -> Error |
| 655 | - end. |
| 656 | - |
| 657 | - %% @doc Unsubscribe the current subscriber. Assuming the subscriber is self(). |
| 658 | - -spec unsubscribe(pid()) -> ok | {error, any()}. |
| 659 | - unsubscribe(ConsumerPid) -> |
| 660 | - unsubscribe(ConsumerPid, self()). |
| 661 | - |
| 662 | - %% @doc Unsubscribe the current subscriber. |
| 663 | - -spec unsubscribe(pid(), pid()) -> ok | {error, any()}. |
| 664 | - unsubscribe(ConsumerPid, SubscriberPid) -> |
| 665 | - brod_consumer:unsubscribe(ConsumerPid, SubscriberPid). |
| 666 | - |
| 667 | - -spec consume_ack(client(), topic(), partition(), offset()) -> |
| 668 | - ok | {error, any()}. |
| 669 | - consume_ack(Client, Topic, Partition, Offset) -> |
| 670 | - case brod_client:get_consumer(Client, Topic, Partition) of |
| 671 | - {ok, ConsumerPid} -> consume_ack(ConsumerPid, Offset); |
| 672 | - {error, Reason} -> {error, Reason} |
| 673 | - end. |
| 674 | - |
| 675 | - -spec consume_ack(pid(), offset()) -> ok | {error, any()}. |
| 676 | - consume_ack(ConsumerPid, Offset) -> |
| 677 | - brod_consumer:ack(ConsumerPid, Offset). |
| 678 | - |
| 679 | - %% @equiv brod_group_subscriber:start_link/7 |
| 680 | - -spec start_link_group_subscriber( |
| 681 | - client(), group_id(), [topic()], |
| 682 | - group_config(), consumer_config(), module(), term()) -> |
| 683 | - {ok, pid()} | {error, any()}. |
| 684 | - start_link_group_subscriber(Client, GroupId, Topics, GroupConfig, |
| 685 | - ConsumerConfig, CbModule, CbInitArg) -> |
| 686 | - brod_group_subscriber:start_link(Client, GroupId, Topics, GroupConfig, |
| 687 | - ConsumerConfig, CbModule, CbInitArg). |
| 688 | - |
| 689 | - %% @doc Start group_subscriber_v2 |
| 690 | - -spec start_link_group_subscriber_v2( brod_group_subscriber_v2:subscriber_config() |
| 691 | - ) -> |
| 692 | - {ok, pid()} | {error, any()}. |
| 693 | - start_link_group_subscriber_v2(Config) -> |
| 694 | - brod_group_subscriber_v2:start_link(Config). |
| 695 | - |
| 696 | - %% @equiv brod_group_subscriber:start_link/8 |
| 697 | - -spec start_link_group_subscriber( |
| 698 | - client(), group_id(), [topic()], group_config(), |
| 699 | - consumer_config(), message | message_set, |
| 700 | - module(), term()) -> |
| 701 | - {ok, pid()} | {error, any()}. |
| 702 | - start_link_group_subscriber(Client, GroupId, Topics, GroupConfig, |
| 703 | - ConsumerConfig, MessageType, |
| 704 | - CbModule, CbInitArg) -> |
| 705 | - brod_group_subscriber:start_link(Client, GroupId, Topics, GroupConfig, |
| 706 | - ConsumerConfig, MessageType, |
| 707 | - CbModule, CbInitArg). |
| 708 | - |
| 709 | - %% @equiv start_link_topic_subscriber(Client, Topic, 'all', ConsumerConfig, |
| 710 | - %% CbModule, CbInitArg) |
| 711 | - -spec start_link_topic_subscriber( |
| 712 | - client(), topic(), consumer_config(), module(), term()) -> |
| 713 | - {ok, pid()} | {error, any()}. |
| 714 | - start_link_topic_subscriber(Client, Topic, ConsumerConfig, |
| 715 | - CbModule, CbInitArg) -> |
| 716 | - start_link_topic_subscriber(Client, Topic, all, ConsumerConfig, |
| 717 | - CbModule, CbInitArg). |
| 718 | - |
| 719 | - %% @equiv start_link_topic_subscriber(Client, Topic, Partitions, |
| 720 | - %% ConsumerConfig, message, |
| 721 | - %% CbModule, CbInitArg) |
| 722 | - -spec start_link_topic_subscriber( |
| 723 | - client(), topic(), all | [partition()], |
| 724 | - consumer_config(), module(), term()) -> |
| 725 | - {ok, pid()} | {error, any()}. |
| 726 | - start_link_topic_subscriber(Client, Topic, Partitions, |
| 727 | - ConsumerConfig, CbModule, CbInitArg) -> |
| 728 | - start_link_topic_subscriber(Client, Topic, Partitions, |
| 729 | - ConsumerConfig, message, CbModule, CbInitArg). |
| 730 | - |
| 731 | - %% @equiv brod_topic_subscriber:start_link/7 |
| 732 | - -spec start_link_topic_subscriber( |
| 733 | - client(), topic(), all | [partition()], |
| 734 | - consumer_config(), message | message_set, |
| 735 | - module(), term()) -> |
| 736 | - {ok, pid()} | {error, any()}. |
| 737 | - start_link_topic_subscriber(Client, Topic, Partitions, |
| 738 | - ConsumerConfig, MessageType, CbModule, CbInitArg) -> |
| 739 | - brod_topic_subscriber:start_link(Client, Topic, Partitions, |
| 740 | - ConsumerConfig, MessageType, |
| 741 | - CbModule, CbInitArg). |
| 742 | - |
| 743 | - %% @doc Fetch broker metadata |
| 744 | - %% Return the message body of `metadata' response. |
| 745 | - %% See `kpro_schema.erl' for details |
| 746 | - -spec get_metadata([endpoint()]) -> {ok, kpro:struct()} | {error, any()}. |
| 747 | - get_metadata(Hosts) -> |
| 748 | - brod_utils:get_metadata(Hosts). |
| 749 | - |
| 750 | - %% @doc Fetch broker/topic metadata |
| 751 | - %% Return the message body of `metadata' response. |
| 752 | - %% See `kpro_schema.erl' for struct details |
| 753 | - -spec get_metadata([endpoint()], all | [topic()]) -> |
| 754 | - {ok, kpro:struct()} | {error, any()}. |
| 755 | - get_metadata(Hosts, Topics) -> |
| 756 | - brod_utils:get_metadata(Hosts, Topics). |
| 757 | - |
| 758 | - %% @doc Fetch broker/topic metadata |
| 759 | - %% Return the message body of `metadata' response. |
| 760 | - %% See `kpro_schema.erl' for struct details |
| 761 | - -spec get_metadata([endpoint()], all | [topic()], conn_config()) -> |
| 762 | - {ok, kpro:struct()} | {error, any()}. |
| 763 | - get_metadata(Hosts, Topics, Options) -> |
| 764 | - brod_utils:get_metadata(Hosts, Topics, Options). |
| 765 | - |
| 766 | - %% @equiv resolve_offset(Hosts, Topic, Partition, latest, 1) |
| 767 | - -spec resolve_offset([endpoint()], topic(), partition()) -> |
| 768 | - {ok, offset()} | {error, any()}. |
| 769 | - resolve_offset(Hosts, Topic, Partition) -> |
| 770 | - resolve_offset(Hosts, Topic, Partition, ?OFFSET_LATEST). |
| 771 | - |
| 772 | - %% @doc Resolve semantic offset or timestamp to real offset. |
| 773 | - -spec resolve_offset([endpoint()], topic(), partition(), offset_time()) -> |
| 774 | - {ok, offset()} | {error, any()}. |
| 775 | - resolve_offset(Hosts, Topic, Partition, Time) -> |
| 776 | - resolve_offset(Hosts, Topic, Partition, Time, []). |
| 777 | - |
| 778 | - %% @doc Resolve semantic offset or timestamp to real offset. |
| 779 | - -spec resolve_offset([endpoint()], topic(), partition(), |
| 780 | - offset_time(), conn_config()) -> |
| 781 | - {ok, offset()} | {error, any()}. |
| 782 | - resolve_offset(Hosts, Topic, Partition, Time, ConnCfg) -> |
| 783 | - brod_utils:resolve_offset(Hosts, Topic, Partition, Time, ConnCfg). |
| 784 | - |
| 785 | - %% @doc Fetch a single message set from the given topic-partition. |
| 786 | - %% The first arg can either be an already established connection to leader, |
| 787 | - %% or `{Endpoints, ConnConfig}' so to establish a new connection before fetch. |
| 788 | - -spec fetch(connection() | client_id() | bootstrap(), |
| 789 | - topic(), partition(), integer()) -> |
| 790 | - {ok, {HwOffset :: offset(), [message()]}} | {error, any()}. |
| 791 | - fetch(ConnOrBootstrap, Topic, Partition, Offset) -> |
| 792 | - Opts = #{ max_wait_time => 1000 |
| 793 | - , min_bytes => 1 |
| 794 | - , max_bytes => 1 bsl 20 |
| 795 | - }, |
| 796 | - fetch(ConnOrBootstrap, Topic, Partition, Offset, Opts). |
| 797 | - |
| 798 | - %% @doc Fetch a single message set from the given topic-partition. |
| 799 | - %% The first arg can either be an already established connection to leader, |
| 800 | - %% or `{Endpoints, ConnConfig}' so to establish a new connection before fetch. |
| 801 | - -spec fetch(connection() | client_id() | bootstrap(), |
| 802 | - topic(), partition(), offset(), fetch_opts()) -> |
| 803 | - {ok, {HwOffset :: offset(), [message()]}} | {error, any()}. |
| 804 | - fetch(ConnOrBootstrap, Topic, Partition, Offset, Opts) -> |
| 805 | - brod_utils:fetch(ConnOrBootstrap, Topic, Partition, Offset, Opts). |
| 806 | - |
| 807 | - %% @doc Fold through messages in a partition. |
| 808 | - %% Works like `lists:foldl/2' but with below stop conditions: |
| 809 | - %% * Always return after reach high watermark offset |
| 810 | - %% * Return after the given message count limit is reached |
| 811 | - %% * Return after the given kafka offset is reached. |
| 812 | - %% * Return if the `FoldFun' returns an `{error, Reason}' tuple. |
| 813 | - %% NOTE: Exceptions from evaluating `FoldFun' are not caught. |
| 814 | - -spec fold(connection() | client_id() | bootstrap(), |
| 815 | - topic(), partition(), offset(), fetch_opts(), |
| 816 | - Acc, fold_fun(Acc), fold_limits()) -> |
| 817 | - fold_result() when Acc :: fold_acc(). |
| 818 | - fold(Bootstrap, Topic, Partition, Offset, Opts, Acc, Fun, Limits) -> |
| 819 | - brod_utils:fold(Bootstrap, Topic, Partition, Offset, Opts, Acc, Fun, Limits). |
| 820 | - |
| 821 | - %% @deprecated |
| 822 | - %% fetch(Hosts, Topic, Partition, Offset, Wait, MinBytes, MaxBytes, []) |
| 823 | - -spec fetch([endpoint()], topic(), partition(), offset(), |
| 824 | - non_neg_integer(), non_neg_integer(), pos_integer()) -> |
| 825 | - {ok, [message()]} | {error, any()}. |
| 826 | - fetch(Hosts, Topic, Partition, Offset, MaxWaitTime, MinBytes, MaxBytes) -> |
| 827 | - fetch(Hosts, Topic, Partition, Offset, MaxWaitTime, MinBytes, MaxBytes, []). |
| 828 | - |
| 829 | - %% @deprecated Fetch a single message set from the given topic-partition. |
| 830 | - -spec fetch([endpoint()], topic(), partition(), offset(), |
| 831 | - non_neg_integer(), non_neg_integer(), pos_integer(), |
| 832 | - conn_config()) -> {ok, [message()]} | {error, any()}. |
| 833 | - fetch(Hosts, Topic, Partition, Offset, |
| 834 | - MaxWaitTime, MinBytes, MaxBytes, ConnConfig) -> |
| 835 | - FetchOpts = #{ max_wait_time => MaxWaitTime |
| 836 | - , min_bytes => MinBytes |
| 837 | - , max_bytes => MaxBytes |
| 838 | - }, |
| 839 | - case fetch({Hosts, ConnConfig}, Topic, Partition, Offset, FetchOpts) of |
| 840 | - {ok, {_HwOffset, Batch}} -> {ok, Batch}; %% backward compatible |
| 841 | - {error, Reason} -> {error, Reason} |
| 842 | - end. |
| 843 | - |
| 844 | - %% @doc Connect partition leader. |
| 845 | - -spec connect_leader([endpoint()], topic(), partition(), |
| 846 | - conn_config()) -> {ok, pid()}. |
| 847 | - connect_leader(Hosts, Topic, Partition, ConnConfig) -> |
| 848 | - kpro:connect_partition_leader(Hosts, ConnConfig, Topic, Partition). |
| 849 | - |
| 850 | - %% @doc List ALL consumer groups in the given kafka cluster. |
| 851 | - %% NOTE: Exception if failed to connect any of the coordinator brokers. |
| 852 | - -spec list_all_groups([endpoint()], conn_config()) -> |
| 853 | - [{endpoint(), [cg()] | {error, any()}}]. |
| 854 | - list_all_groups(Endpoints, ConnCfg) -> |
| 855 | - brod_utils:list_all_groups(Endpoints, ConnCfg). |
| 856 | - |
| 857 | - %% @doc List consumer groups in the given group coordinator broker. |
| 858 | - -spec list_groups(endpoint(), conn_config()) -> {ok, [cg()]} | {error, any()}. |
| 859 | - list_groups(CoordinatorEndpoint, ConnCfg) -> |
| 860 | - brod_utils:list_groups(CoordinatorEndpoint, ConnCfg). |
| 861 | - |
| 862 | - %% @doc Describe consumer groups. The given consumer group IDs should be all |
| 863 | - %% managed by the coordinator-broker running at the given endpoint. |
| 864 | - %% Otherwise error codes will be returned in the result structs. |
| 865 | - %% Return `describe_groups' response body field named `groups'. |
| 866 | - %% See `kpro_schema.erl' for struct details |
| 867 | - -spec describe_groups(endpoint(), conn_config(), [group_id()]) -> |
| 868 | - {ok, [kpro:struct()]} | {error, any()}. |
| 869 | - describe_groups(CoordinatorEndpoint, ConnCfg, IDs) -> |
| 870 | - brod_utils:describe_groups(CoordinatorEndpoint, ConnCfg, IDs). |
| 871 | - |
| 872 | - %% @doc Connect to consumer group coordinator broker. |
| 873 | - %% Done in steps: 1) connect to any of the given bootstrap ednpoints; |
| 874 | - %% 2) send group_coordinator_request to resolve group coordinator endpoint;; |
| 875 | - %% 3) connect to the resolved endpoint and return the connection pid |
| 876 | - -spec connect_group_coordinator([endpoint()], conn_config(), group_id()) -> |
| 877 | - {ok, pid()} | {error, any()}. |
| 878 | - connect_group_coordinator(BootstrapEndpoints, ConnCfg, GroupId) -> |
| 879 | - Args = #{type => group, id => GroupId}, |
| 880 | - kpro:connect_coordinator(BootstrapEndpoints, ConnCfg, Args). |
| 881 | - |
| 882 | - %% @doc Fetch committed offsets for ALL topics in the given consumer group. |
| 883 | - %% Return the `responses' field of the `offset_fetch' response. |
| 884 | - %% See `kpro_schema.erl' for struct details. |
| 885 | - -spec fetch_committed_offsets([endpoint()], conn_config(), group_id()) -> |
| 886 | - {ok, [kpro:struct()]} | {error, any()}. |
| 887 | - fetch_committed_offsets(BootstrapEndpoints, ConnCfg, GroupId) -> |
| 888 | - brod_utils:fetch_committed_offsets(BootstrapEndpoints, ConnCfg, GroupId, []). |
| 889 | - |
| 890 | - %% @doc Same as `fetch_committed_offsets/3', |
| 891 | - %% but works with a started `brod_client' |
| 892 | - -spec fetch_committed_offsets(client(), group_id()) -> |
| 893 | - {ok, [kpro:struct()]} | {error, any()}. |
| 894 | - fetch_committed_offsets(Client, GroupId) -> |
| 895 | - brod_utils:fetch_committed_offsets(Client, GroupId, []). |
| 896 | - |
| 897 | - -ifdef(build_brod_cli). |
| 898 | - main(X) -> brod_cli:main(X). |
| 899 | - -endif. |
| 900 | - |
| 901 | - %%%_* Emacs ==================================================================== |
| 902 | - %%% Local Variables: |
| 903 | - %%% allout-layout: t |
| 904 | - %%% erlang-indent-level: 2 |
| 905 | - %%% End: |
| @@ -1,305 +0,0 @@ | |
| 1 | - %%% |
| 2 | - %%% Copyright (c) 2017-2018 Klarna Bank AB (publ) |
| 3 | - %%% |
| 4 | - %%% Licensed under the Apache License, Version 2.0 (the "License"); |
| 5 | - %%% you may not use this file except in compliance with the License. |
| 6 | - %%% You may obtain a copy of the License at |
| 7 | - %%% |
| 8 | - %%% http://www.apache.org/licenses/LICENSE-2.0 |
| 9 | - %%% |
| 10 | - %%% Unless required by applicable law or agreed to in writing, software |
| 11 | - %%% distributed under the License is distributed on an "AS IS" BASIS, |
| 12 | - %%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 13 | - %%% See the License for the specific language governing permissions and |
| 14 | - %%% limitations under the License. |
| 15 | - %%% |
| 16 | - |
| 17 | - %%%============================================================================= |
| 18 | - %%% @doc This is a utility module to help force commit offsets to kafka. |
| 19 | - %%%============================================================================= |
| 20 | - |
| 21 | - -module(brod_cg_commits). |
| 22 | - |
| 23 | - -behaviour(gen_server). |
| 24 | - -behaviour(brod_group_member). |
| 25 | - |
| 26 | - -export([ run/2 |
| 27 | - ]). |
| 28 | - |
| 29 | - -export([ start_link/2 |
| 30 | - , stop/1 |
| 31 | - , sync/1 |
| 32 | - ]). |
| 33 | - |
| 34 | - %% callbacks for brod_group_coordinator |
| 35 | - -export([ get_committed_offsets/2 |
| 36 | - , assignments_received/4 |
| 37 | - , assignments_revoked/1 |
| 38 | - , assign_partitions/3 |
| 39 | - ]). |
| 40 | - |
| 41 | - -export([ code_change/3 |
| 42 | - , handle_call/3 |
| 43 | - , handle_cast/2 |
| 44 | - , handle_info/2 |
| 45 | - , init/1 |
| 46 | - , terminate/2 |
| 47 | - ]). |
| 48 | - |
| 49 | - -include("brod_int.hrl"). |
| 50 | - |
| 51 | - -type topic() :: brod:topic(). |
| 52 | - -type partition() :: brod:partition(). |
| 53 | - -type offset() :: brod:offset(). |
| 54 | - -type group_id() :: brod:group_id(). |
| 55 | - -type member_id() :: brod:group_member_id(). |
| 56 | - -type retention() :: integer(). %% -1 to use whatever configured in kafka |
| 57 | - -type offsets() :: latest | earliest | [{partition(), offset()}]. |
| 58 | - -type prop_key() :: id | topic | retention | protocol | offsets. |
| 59 | - -type prop_val() :: group_id() | topic() | retention() | offsets() |
| 60 | - | brod_group_coordinator:protocol_name(). |
| 61 | - -type group_input() :: [{prop_key(), prop_val()}]. |
| 62 | - |
| 63 | - -record(state, |
| 64 | - { client :: brod:client() |
| 65 | - , groupId :: brod:group_id() |
| 66 | - , memberId :: ?undef | member_id() |
| 67 | - , generationId :: ?undef | brod:group_generation_id() |
| 68 | - , coordinator :: pid() |
| 69 | - , topic :: ?undef | topic() |
| 70 | - , offsets :: ?undef | offsets() |
| 71 | - , is_elected = false :: boolean() |
| 72 | - , pending_sync :: ?undef | {pid(), reference()} |
| 73 | - , is_done = false :: boolean() |
| 74 | - }). |
| 75 | - |
| 76 | - %%%_* APIs ===================================================================== |
| 77 | - |
| 78 | - %% @doc Force commit offsets. |
| 79 | - run(ClientId, GroupInput) -> |
| 80 | - {ok, Pid} = start_link(ClientId, GroupInput), |
| 81 | - ok = sync(Pid), |
| 82 | - ok = stop(Pid). |
| 83 | - |
| 84 | - %% @doc Start (link) a group member. |
| 85 | - %% The member will try to join the consumer group and |
| 86 | - %% get assignments for the given topic-partitions, |
| 87 | - %% then commit given offsets to kafka. |
| 88 | - %% In case not all given partitions are assigned to it, |
| 89 | - %% it will terminate with an exit exception |
| 90 | - -spec start_link(brod:client(), group_input()) -> {ok, pid()} | {error, any()}. |
| 91 | - start_link(Client, GroupInput) -> |
| 92 | - gen_server:start_link(?MODULE, {Client, GroupInput}, []). |
| 93 | - |
| 94 | - %% @doc Stop the process. |
| 95 | - -spec stop(pid()) -> ok. |
| 96 | - stop(Pid) -> |
| 97 | - Mref = erlang:monitor(process, Pid), |
| 98 | - ok = gen_server:cast(Pid, stop), |
| 99 | - receive |
| 100 | - {'DOWN', Mref, process, Pid, _Reason} -> |
| 101 | - ok |
| 102 | - end. |
| 103 | - |
| 104 | - %% @doc Make a call to the resetter process, the call will be blocked |
| 105 | - %% until offsets are committed. |
| 106 | - -spec sync(pid()) -> ok. |
| 107 | - sync(Pid) -> |
| 108 | - ok = gen_server:call(Pid, sync, infinity). |
| 109 | - |
| 110 | - %%%_* APIs for group coordinator =============================================== |
| 111 | - |
| 112 | - %% @doc Called by group coordinator when there is new assignemnt received. |
| 113 | - -spec assignments_received(pid(), member_id(), integer(), |
| 114 | - brod:received_assignments()) -> ok. |
| 115 | - assignments_received(Pid, MemberId, GenerationId, TopicAssignments) -> |
| 116 | - gen_server:cast(Pid, {new_assignments, MemberId, |
| 117 | - GenerationId, TopicAssignments}). |
| 118 | - |
| 119 | - %% @doc Called by group coordinator before re-joinning the consumer group. |
| 120 | - -spec assignments_revoked(pid()) -> ok. |
| 121 | - assignments_revoked(Pid) -> |
| 122 | - gen_server:call(Pid, unsubscribe_all_partitions, infinity). |
| 123 | - |
| 124 | - %% @doc This function is called only when `partition_assignment_strategy' |
| 125 | - %% is set for `callback_implemented' in group config. |
| 126 | - -spec assign_partitions(pid(), [brod:group_member()], |
| 127 | - [{brod:topic(), brod:partition()}]) -> |
| 128 | - [{member_id(), [brod:partition_assignment()]}]. |
| 129 | - assign_partitions(Pid, Members, TopicPartitionList) -> |
| 130 | - Call = {assign_partitions, Members, TopicPartitionList}, |
| 131 | - gen_server:call(Pid, Call, infinity). |
| 132 | - |
| 133 | - %% @doc Called by group coordinator when initializing the assignments |
| 134 | - %% for subscriber. |
| 135 | - %% NOTE: this function is called only when it is DISABLED to commit offsets |
| 136 | - %% to kafka. i.e. offset_commit_policy is set to consumer_managed |
| 137 | - -spec get_committed_offsets(pid(), [{brod:topic(), brod:partition()}]) -> |
| 138 | - {ok, [{{brod:topic(), brod:partition()}, brod:offset()}]}. |
| 139 | - get_committed_offsets(_Pid, _TopicPartitions) -> {ok, []}. |
| 140 | - |
| 141 | - %%%_* gen_server callbacks ===================================================== |
| 142 | - |
| 143 | - init({Client, GroupInput}) -> |
| 144 | - ok = brod_utils:assert_client(Client), |
| 145 | - GroupId = proplists:get_value(id, GroupInput), |
| 146 | - ok = brod_utils:assert_group_id(GroupId), |
| 147 | - Topic = proplists:get_value(topic, GroupInput), |
| 148 | - ProtocolName = proplists:get_value(protocol, GroupInput), |
| 149 | - Retention = proplists:get_value(retention, GroupInput), |
| 150 | - Offsets = proplists:get_value(offsets, GroupInput), |
| 151 | - %% use callback_implemented strategy so I know I am elected leader |
| 152 | - %% when `assign_partitions' callback is called. |
| 153 | - Config = [ {partition_assignment_strategy, callback_implemented} |
| 154 | - , {offset_retention_seconds, Retention} |
| 155 | - , {protocol_name, ProtocolName} |
| 156 | - , {rejoin_delay_seconds, 2} |
| 157 | - ], |
| 158 | - {ok, Pid} = brod_group_coordinator:start_link(Client, GroupId, [Topic], |
| 159 | - Config, ?MODULE, self()), |
| 160 | - State = #state{ client = Client |
| 161 | - , groupId = GroupId |
| 162 | - , coordinator = Pid |
| 163 | - , topic = Topic |
| 164 | - , offsets = Offsets |
| 165 | - }, |
| 166 | - {ok, State}. |
| 167 | - |
| 168 | - handle_info(Info, State) -> |
| 169 | - log(State, info, "Info discarded:~p", [Info]), |
| 170 | - {noreply, State}. |
| 171 | - |
| 172 | - handle_call(sync, From, State0) -> |
| 173 | - State1 = State0#state{pending_sync = From}, |
| 174 | - State = maybe_reply_sync(State1), |
| 175 | - {noreply, State}; |
| 176 | - handle_call({assign_partitions, Members, TopicPartitions}, _From, |
| 177 | - #state{topic = MyTopic, |
| 178 | - offsets = Offsets} = State) -> |
| 179 | - log(State, info, "Assigning all topic partitions to self", []), |
| 180 | - MyTP = [{MyTopic, P} || {P, _} <- Offsets], |
| 181 | - %% Assert that my topic partitions are included in |
| 182 | - %% subscriptions collected from ALL members |
| 183 | - Pred = fun(TP) -> not lists:member(TP, TopicPartitions) end, |
| 184 | - case lists:filter(Pred, MyTP) of |
| 185 | - [] -> |
| 186 | - %% all of the give partitions are valid |
| 187 | - ok; |
| 188 | - BadPartitions -> |
| 189 | - PartitionNumbers = [P || {_T, P} <- BadPartitions], |
| 190 | - log(State, error, "Nonexisting partitions in input: ~p", |
| 191 | - [PartitionNumbers]), |
| 192 | - erlang:exit({non_existing_partitions, PartitionNumbers}) |
| 193 | - end, |
| 194 | - %% To keep it simple, assign all my topic-partitions to self |
| 195 | - %% but discard all other topic-partitions. |
| 196 | - %% After all, I will leave group as soon as offsets are committed |
| 197 | - Result = assign_all_to_self(Members, MyTP), |
| 198 | - {reply, Result, State#state{is_elected = true}}; |
| 199 | - handle_call(unsubscribe_all_partitions, _From, #state{} = State) -> |
| 200 | - %% nothing to do, because I do not subscribe to any partition |
| 201 | - {reply, ok, State}; |
| 202 | - handle_call(Call, _From, State) -> |
| 203 | - {reply, {error, {unknown_call, Call}}, State}. |
| 204 | - |
| 205 | - handle_cast({new_assignments, _MemberId, GenerationId, Assignments}, |
| 206 | - #state{ is_elected = IsLeader |
| 207 | - , offsets = OffsetsToCommit |
| 208 | - , coordinator = Pid |
| 209 | - , topic = MyTopic |
| 210 | - } = State) -> |
| 211 | - %% Write a log if I am not a leader, |
| 212 | - %% hope the desired partitions are all assigned to me |
| 213 | - IsLeader orelse log(State, info, "Not elected", []), |
| 214 | - Groupped0 = |
| 215 | - brod_utils:group_per_key( |
| 216 | - fun(#brod_received_assignment{ topic = Topic |
| 217 | - , partition = Partition |
| 218 | - , begin_offset = Offset |
| 219 | - }) -> |
| 220 | - {Topic, {Partition, Offset}} |
| 221 | - end, Assignments), |
| 222 | - %% Discard other topics if for whatever reason the group leader assigns |
| 223 | - %% irrelevant topic-partitions to me |
| 224 | - Groupped = lists:filter(fun({Topic, _}) -> Topic =:= MyTopic end, Groupped0), |
| 225 | - log(State, info, "current offsets:\n~p", [Groupped]), |
| 226 | - %% Assert all desired partitions are in assignment |
| 227 | - case Groupped of |
| 228 | - [] -> |
| 229 | - log(State, error, "Topic ~s is not received in assignment", [MyTopic]), |
| 230 | - erlang:exit({bad_topic_assignment, Groupped0}); |
| 231 | - [{MyTopic, PartitionOffsetList}] -> |
| 232 | - MyPartitions = [P || {P, _O} <- OffsetsToCommit], |
| 233 | - ReceivedPartitions = [P || {P, _O} <- PartitionOffsetList], |
| 234 | - case MyPartitions -- ReceivedPartitions of |
| 235 | - [] -> |
| 236 | - ok; |
| 237 | - Left -> |
| 238 | - log(State, error, |
| 239 | - "Partitions ~p are not received in assignment, " |
| 240 | - "There is probably another active group member subscribing " |
| 241 | - "to topic ~s, stop it and retry\n", [MyTopic, Left]), |
| 242 | - erlang:exit({unexpected_assignments, Left}) |
| 243 | - end |
| 244 | - end, |
| 245 | - %% Stage all offsets in coordinator process |
| 246 | - lists:foreach( |
| 247 | - fun({Partition, Offset}) -> |
| 248 | - %% -1 here, because brod_group_coordinator +1 to commit |
| 249 | - OffsetToCommit = Offset - 1, |
| 250 | - brod_group_coordinator:ack(Pid, GenerationId, MyTopic, |
| 251 | - Partition, OffsetToCommit) |
| 252 | - end, OffsetsToCommit), |
| 253 | - %% Now force it to commit |
| 254 | - case brod_group_coordinator:commit_offsets(Pid) of |
| 255 | - ok -> ok; |
| 256 | - {error, Reason} -> |
| 257 | - log(State, error, "Failed to commit, reason:\n~p", [Reason]), |
| 258 | - erlang:exit(commit_failed) |
| 259 | - end, |
| 260 | - {noreply, set_done(State)}; |
| 261 | - handle_cast(stop, State) -> |
| 262 | - {stop, normal, State}; |
| 263 | - handle_cast(_Cast, State) -> |
| 264 | - {noreply, State}. |
| 265 | - |
| 266 | - code_change(_OldVsn, State, _Extra) -> |
| 267 | - {ok, State}. |
| 268 | - |
| 269 | - terminate(_Reason, #state{}) -> |
| 270 | - ok. |
| 271 | - |
| 272 | - %%%_* Internal Functions ======================================================= |
| 273 | - |
| 274 | - set_done(State) -> |
| 275 | - maybe_reply_sync(State#state{is_done = true}). |
| 276 | - |
| 277 | - maybe_reply_sync(#state{is_done = false} = State) -> |
| 278 | - State; |
| 279 | - maybe_reply_sync(#state{pending_sync = ?undef} = State) -> |
| 280 | - State; |
| 281 | - maybe_reply_sync(#state{pending_sync = From} = State) -> |
| 282 | - gen_server:reply(From, ok), |
| 283 | - log(State, info, "done\n", []), |
| 284 | - State#state{pending_sync = ?undef}. |
| 285 | - |
| 286 | - %% I am the current leader because I am assigning partitions. |
| 287 | - %% My member ID should be positioned at the head of the member list. |
| 288 | - -spec assign_all_to_self([brod:group_member()], [{topic(), partition()}]) -> |
| 289 | - [{member_id(), [brod:partition_assignment()]}]. |
| 290 | - assign_all_to_self([{MyMemberId, _} | Members], TopicPartitions) -> |
| 291 | - Groupped = brod_utils:group_per_key(TopicPartitions), |
| 292 | - [ {MyMemberId, Groupped} |
| 293 | - | [{Id, []} || {Id, _MemberMeta} <- Members] |
| 294 | - ]. |
| 295 | - |
| 296 | - log(#state{groupId = GroupId}, Level, Fmt, Args) -> |
| 297 | - brod_utils:log(Level, |
| 298 | - "Group member (~s,coor=~p):\n" ++ Fmt, |
| 299 | - [GroupId, self() | Args]). |
| 300 | - |
| 301 | - %%%_* Emacs ==================================================================== |
| 302 | - %%% Local Variables: |
| 303 | - %%% allout-layout: t |
| 304 | - %%% erlang-indent-level: 2 |
| 305 | - %%% End: |
| @@ -1,1280 +0,0 @@ | |
| 1 | - %%% |
| 2 | - %%% Copyright (c) 2017-2018 Klarna Bank AB (publ) |
| 3 | - %%% |
| 4 | - %%% Licensed under the Apache License, Version 2.0 (the "License"); |
| 5 | - %%% you may not use this file except in compliance with the License. |
| 6 | - %%% You may obtain a copy of the License at |
| 7 | - %%% |
| 8 | - %%% http://www.apache.org/licenses/LICENSE-2.0 |
| 9 | - %%% |
| 10 | - %%% Unless required by applicable law or agreed to in writing, software |
| 11 | - %%% distributed under the License is distributed on an "AS IS" BASIS, |
| 12 | - %%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 13 | - %%% See the License for the specific language governing permissions and |
| 14 | - %%% limitations under the License. |
| 15 | - %%% |
| 16 | - |
| 17 | - -module(brod_cli). |
| 18 | - |
| 19 | - -ifdef(build_brod_cli). |
| 20 | - |
| 21 | - -export([main/1, main/2]). |
| 22 | - |
| 23 | - -include("brod_int.hrl"). |
| 24 | - |
| 25 | - -define(CLIENT, brod_cli_client). |
| 26 | - |
| 27 | - %% 'halt' is for escript, stop the vm immediately |
| 28 | - %% 'exit' is for testing, we want eunit or ct to be able to capture |
| 29 | - -define(STOP(How), |
| 30 | - begin |
| 31 | - try |
| 32 | - brod:stop_client(?CLIENT) |
| 33 | - catch |
| 34 | - exit : {noproc, _} -> |
| 35 | - ok |
| 36 | - end, |
| 37 | - _ = brod:stop(), |
| 38 | - case How of |
| 39 | - 'halt' -> erlang:halt(?LINE); |
| 40 | - 'exit' -> erlang:exit(?LINE) |
| 41 | - end |
| 42 | - end). |
| 43 | - |
| 44 | - -define(MAIN_DOC, "usage: |
| 45 | - brod -h|--help |
| 46 | - brod -v|--version |
| 47 | - brod <command> [options] [-h|--help] [--verbose|--debug] |
| 48 | - |
| 49 | - commands: |
| 50 | - meta: Inspect topic metadata |
| 51 | - offset: Inspect offsets |
| 52 | - fetch: Fetch messages |
| 53 | - send: Produce messages |
| 54 | - pipe: Pipe file or stdin as messages to kafka |
| 55 | - groups: List/describe consumer group |
| 56 | - commits: List/descibe committed offsets |
| 57 | - or force overwrite existing commits |
| 58 | - "). |
| 59 | - |
| 60 | - %% NOTE: bad indentation at the first line is intended |
| 61 | - -define(COMMAND_COMMON_OPTIONS, |
| 62 | - " --ssl Use TLS, validate server using trusted CAs |
| 63 | - --cacertfile=<cacert> Use TLS, validate server using the given certificate |
| 64 | - --certfile=<certfile> Client certificate in case client authentication |
| 65 | - is enabled in brokers |
| 66 | - --keyfile=<keyfile> Client private key in case client authentication |
| 67 | - is enabled in brokers |
| 68 | - --sasl-plain=<file> Tell brod to use username/password stored in the |
| 69 | - given file, the file should have username and |
| 70 | - password in two lines. |
| 71 | - --scram256=<file> Like sasl-plain option, but to use scram-sha-256 |
| 72 | - --scram512=<file> Like sasl-plain option, but to use scram-sha-512 |
| 73 | - --ebin-paths=<dirs> Comma separated directory names for extra beams, |
| 74 | - This is to support user compiled message formatters |
| 75 | - --no-api-vsn-query Do not query api version (for kafka 0.9 or earlier) |
| 76 | - Or set KAFKA_VERSION environment variable to 0.9 for |
| 77 | - the same effect |
| 78 | - " |
| 79 | - ). |
| 80 | - |
| 81 | - -define(META_CMD, "meta"). |
| 82 | - -define(META_DOC, "usage: |
| 83 | - brod meta [options] |
| 84 | - |
| 85 | - options: |
| 86 | - -b,--brokers=<brokers> Comma separated host:port pairs |
| 87 | - [default: localhost:9092] |
| 88 | - -t,--topic=<topic> Topic name [default: *] |
| 89 | - -T,--text Print metadata as aligned texts (default) |
| 90 | - -J,--json Print metadata as JSON object |
| 91 | - -L,--list List topics, no partition details, |
| 92 | - Applicable only for --text option |
| 93 | - -U,--under-replicated Display only under-replicated partitions |
| 94 | - " |
| 95 | - ?COMMAND_COMMON_OPTIONS |
| 96 | - "Text output schema (out of sync replicas are marked with *): |
| 97 | - brokers <count>: |
| 98 | - <broker-id>: <endpoint> |
| 99 | - topics <count>: |
| 100 | - <name> <count>: [[ERROR] [<reason>]] |
| 101 | - <partition>: <leader-broker-id> (replicas[*]...) [<error-reason>] |
| 102 | - " |
| 103 | - ). |
| 104 | - |
| 105 | - -define(OFFSET_CMD, "offset"). |
| 106 | - -define(OFFSET_DOC, "usage: |
| 107 | - brod offset [options] |
| 108 | - |
| 109 | - options: |
| 110 | - -b,--brokers=<brokers> Comma separated host:port pairs |
| 111 | - [default: localhost:9092] |
| 112 | - -t,--topic=<topic> Topic name |
| 113 | - -p,--partition=<parti> Partition number |
| 114 | - [default: all] |
| 115 | - -T,--time=<time> Unix epoch (in milliseconds) of the correlated offset |
| 116 | - to fetch. Special values: |
| 117 | - 'latest' or -1 for latest offset |
| 118 | - 'earliest' or -2 for earliest offset |
| 119 | - [default: latest] |
| 120 | - --one-line If it is to resolve 'all' offsets, |
| 121 | - Print results in one line. |
| 122 | - e.g. 0:111,1:222 |
| 123 | - " |
| 124 | - ?COMMAND_COMMON_OPTIONS |
| 125 | - ). |
| 126 | - |
| 127 | - -define(FETCH_CMD, "fetch"). |
| 128 | - -define(FETCH_DOC, "usage: |
| 129 | - brod fetch [options] |
| 130 | - |
| 131 | - options: |
| 132 | - -b,--brokers=<brokers> Comma separated host:port pairs |
| 133 | - [default: localhost:9092] |
| 134 | - -t,--topic=<topic> Topic name |
| 135 | - -p,--partition=<parti> Partition number |
| 136 | - -o,--offset=<offset> Offset to start fetching from |
| 137 | - latest: From the latest offset (not last) |
| 138 | - earliest: From earliest offset (first) |
| 139 | - last: From offset (latest - 1) |
| 140 | - <integer>: From a specific offset |
| 141 | - [default: last] |
| 142 | - -c,--count=<count> Number of messages to fetch (-1 as infinity) |
| 143 | - [default: 1] |
| 144 | - -w,--wait=<seconds> Time in seconds to wait for one message set |
| 145 | - [default: 5s] |
| 146 | - --kv-deli=<deli> Delimiter for offset, key and value output [default: :] |
| 147 | - --msg-deli=<deli> Delimiter between messages. [default: \\n] |
| 148 | - --max-bytes=<bytes> Max number of bytes kafka should try to accumulate |
| 149 | - within the --wait time |
| 150 | - [default: 1K] |
| 151 | - --fmt=<fmt> Output format. Assume keys and values are utf8 strings |
| 152 | - v: Print 'V <msg-deli>' |
| 153 | - kv: Print 'K <kv-deli> V <msg-deli>' |
| 154 | - okv: Print 'O <kv-deli> K <kv-deli> V <msg-deli>' |
| 155 | - eterm: Pretty print tuple '{Offse, Key, Value}.' |
| 156 | - to a consultable Erlang term format. |
| 157 | - Expr: An Erlang expression to be evaluated for each |
| 158 | - message. Bound variable to be used in the |
| 159 | - expression: Offset, Key, Value, TsType, Ts. |
| 160 | - Print nothing if the evaluation result in 'ok', |
| 161 | - otherwise print the evaluated io-list. |
| 162 | - [default: v] |
| 163 | - " |
| 164 | - ?COMMAND_COMMON_OPTIONS |
| 165 | - "NOTE: Reaching either --count or --wait limit will cause script to exit |
| 166 | - " |
| 167 | - ). |
| 168 | - |
| 169 | - -define(SEND_CMD, "send"). |
| 170 | - -define(SEND_DOC, "usage: |
| 171 | - brod send [options] |
| 172 | - |
| 173 | - options: |
| 174 | - -b,--brokers=<brokers> Comma separated host:port pairs |
| 175 | - [default: localhost:9092] |
| 176 | - -t,--topic=<topic> Topic name |
| 177 | - -p,--partition=<parti> Partition number [default: 0] |
| 178 | - Special values: |
| 179 | - random: randomly pick a partition |
| 180 | - hash: hash key to a partition |
| 181 | - -k,--key=<key> Key to produce [default: null] |
| 182 | - -v,--value=<value> Value to produce. Special values: |
| 183 | - null: No payload |
| 184 | - @/path/to/file: Send a whole file as payload |
| 185 | - --acks=<acks> Required acks. Supported values: |
| 186 | - all or -1: Require acks from all in-sync replica |
| 187 | - 1: Require acks from only partition leader |
| 188 | - 0: Require no acks |
| 189 | - [default: all] |
| 190 | - --ack-timeout=<time> How long the partition leader should wait for replicas |
| 191 | - to ack before sending response to producer |
| 192 | - The value can be an integer to indicate number of |
| 193 | - milliseconds or followed by s/m to indicate seconds |
| 194 | - or minutes [default: 10s] |
| 195 | - --compression=<compre> Supported values: none / gzip / snappy |
| 196 | - [default: none] |
| 197 | - " |
| 198 | - ?COMMAND_COMMON_OPTIONS |
| 199 | - ). |
| 200 | - |
| 201 | - -define(PIPE_CMD, "pipe"). |
| 202 | - -define(PIPE_DOC, "usage: |
| 203 | - brod pipe [options] |
| 204 | - |
| 205 | - options: |
| 206 | - -b,--brokers=<brokers> Comma separated host:port pairs |
| 207 | - [default: localhost:9092] |
| 208 | - -t,--topic=<topic> Topic name |
| 209 | - -p,--partition=<parti> Partition number [default: 0] |
| 210 | - Special values: |
| 211 | - random: randomly pick a partition |
| 212 | - hash: hash key to a partition |
| 213 | - -s,--source=<source> Data source. Special value: |
| 214 | - stdin: Reads messages from standard-input |
| 215 | - @path/to/file: Reads from file |
| 216 | - [default: stdin] |
| 217 | - --prompt Applicable when --source is stdin, enable input prompt |
| 218 | - --no-eof-exit Do not exit when reaching EOF |
| 219 | - --tail Applicable when --source is a file |
| 220 | - brod will start from EOF and keep tailing for new bytes |
| 221 | - --msg-deli=<msg-deli> Message delimiter |
| 222 | - NOTE: A message is always delimited when reaching EOF |
| 223 | - [default: \\n] |
| 224 | - --kv-deli=<kv-deli> Key-Value delimiter. |
| 225 | - when not provided, messages are produced with |
| 226 | - only value, key is set to null. [default: none] |
| 227 | - --blk-size=<size> Block size (bytes) when reading bytes from a file. |
| 228 | - Applicable when --source is file and --msg-deli is |
| 229 | - not \\n. [default: 1M] |
| 230 | - --acks=<acks> Required acks. [default: all] |
| 231 | - Supported values: |
| 232 | - all or -1: Require acks from all in-sync replica |
| 233 | - 1: Require acks from only partition leader |
| 234 | - 0: Require no acks |
| 235 | - --ack-timeout=<time> How long the partition leader should wait for replicas |
| 236 | - to ack before sending response to producer |
| 237 | - not applicable when --acks is not 'all' |
| 238 | - [default: 10s] |
| 239 | - --max-linger-ms=<ms> Max ms for messages to linger in buffer [default: 200] |
| 240 | - --max-linger-cnt=<N> Max messages to linger in buffer [default: 100] |
| 241 | - --max-batch=<bytes> Max size for one message-set (before compression) |
| 242 | - The value can be either an integer to indicate bytes |
| 243 | - or followed by K/M to indicate KBytes or MBytes |
| 244 | - [default: 1M] |
| 245 | - --compression=<compr> Supported values: none/gzip/snappy [default: none] |
| 246 | - " |
| 247 | - ?COMMAND_COMMON_OPTIONS |
| 248 | - "NOTE: When --source is path/to/file, it by default reads from BOF |
| 249 | - unless --tail is given. |
| 250 | - " |
| 251 | - ). |
| 252 | - |
| 253 | - -define(GROUPS_CMD, "groups"). |
| 254 | - -define(GROUPS_DOC, "usage: |
| 255 | - brod groups [options] |
| 256 | - |
| 257 | - options: |
| 258 | - -b,--brokers=<brokers> Comma separated host:port pairs |
| 259 | - [default: localhost:9092] |
| 260 | - --ids=<group-id> Comma separated group IDs to describe |
| 261 | - [default: all] |
| 262 | - " |
| 263 | - ?COMMAND_COMMON_OPTIONS |
| 264 | - ). |
| 265 | - |
| 266 | - |
| 267 | - -define(COMMITS_CMD, "commits"). |
| 268 | - -define(COMMITS_DOC, "usage: |
| 269 | - brod commits [options] |
| 270 | - |
| 271 | - options: |
| 272 | - -b,--brokers=<brokers> Comma separated host:port pairs |
| 273 | - [default: localhost:9092] |
| 274 | - -d,--describe Describe committed offsets, |
| 275 | - otherwise reset commit history |
| 276 | - -i,--id=<group-id> Group ID to describe, or to commit offsets. |
| 277 | - -t,--topic=<topic> Topic name to commit offsets |
| 278 | - -o,--offsets=<offsets> latest: commit latest offset for all partitions, |
| 279 | - earliest: commit earliest offset for all partitions, |
| 280 | - Comma separated 'partition:offset' pairs. |
| 281 | - -r,--retention=<time> An integer to indicate the retention for the commit, |
| 282 | - default time unit is seconds. Accepts one char suffix |
| 283 | - as time unit, s=second m=mintue h=hour d=day. |
| 284 | - Default value -1 is to respect kafka config. |
| 285 | - [default: -1] |
| 286 | - --protocol=<protocol> Protocol name to be used when trying to join group. |
| 287 | - [default: roundrobin] |
| 288 | - " |
| 289 | - ?COMMAND_COMMON_OPTIONS |
| 290 | - ). |
| 291 | - |
| 292 | - -define(DOCS, |
| 293 | - [ {?META_CMD, ?META_DOC} |
| 294 | - , {?OFFSET_CMD, ?OFFSET_DOC} |
| 295 | - , {?SEND_CMD, ?SEND_DOC} |
| 296 | - , {?FETCH_CMD, ?FETCH_DOC} |
| 297 | - , {?PIPE_CMD, ?PIPE_DOC} |
| 298 | - , {?GROUPS_CMD, ?GROUPS_DOC} |
| 299 | - , {?COMMITS_CMD, ?COMMITS_DOC} |
| 300 | - ]). |
| 301 | - |
| 302 | - -define(LOG_LEVEL_QUIET, 0). |
| 303 | - -define(LOG_LEVEL_VERBOSE, 1). |
| 304 | - -define(LOG_LEVEL_DEBUG, 2). |
| 305 | - |
| 306 | - -type log_level() :: non_neg_integer(). |
| 307 | - |
| 308 | - -type command() :: string(). |
| 309 | - |
| 310 | - main(Args) -> |
| 311 | - _ = main(Args, halt). |
| 312 | - |
| 313 | - -spec main([string()], halt | exit) -> no_return(). |
| 314 | - main(["-h" | _], _Stop) -> |
| 315 | - print(?MAIN_DOC); |
| 316 | - main(["--help" | _], _Stop) -> |
| 317 | - print(?MAIN_DOC); |
| 318 | - main(["-v" | _], _Stop) -> |
| 319 | - print_version(); |
| 320 | - main(["--version" | _], _Stop) -> |
| 321 | - print_version(); |
| 322 | - main(["-" ++ _ = Arg | _], Stop) -> |
| 323 | - logerr("Unknown option: ~s\n", [Arg]), |
| 324 | - print(?MAIN_DOC), |
| 325 | - ?STOP(Stop); |
| 326 | - main([Command | _] = Args, Stop) -> |
| 327 | - case lists:keyfind(Command, 1, ?DOCS) of |
| 328 | - {_, Doc} -> |
| 329 | - main(Command, Doc, Args, Stop); |
| 330 | - false -> |
| 331 | - logerr("Unknown command: ~s\n", [Command]), |
| 332 | - print(?MAIN_DOC), |
| 333 | - ?STOP(Stop) |
| 334 | - end; |
| 335 | - main(_, Stop) -> |
| 336 | - print(?MAIN_DOC), |
| 337 | - ?STOP(Stop). |
| 338 | - |
| 339 | - -spec main(command(), string(), [string()], halt | exit) -> _ | no_return(). |
| 340 | - main(Command, Doc, Args0, Stop) -> |
| 341 | - IsHelp = lists:member("--help", Args0) orelse lists:member("-h", Args0), |
| 342 | - IsVerbose = lists:member("--verbose", Args0), |
| 343 | - IsDebug = lists:member("--debug", Args0), |
| 344 | - Args = Args0 -- ["--verbose", "--debug"], |
| 345 | - LogLevels = [{IsDebug, ?LOG_LEVEL_DEBUG}, |
| 346 | - {IsVerbose, ?LOG_LEVEL_VERBOSE}, |
| 347 | - {true, ?LOG_LEVEL_QUIET}], |
| 348 | - {true, LogLevel} = lists:keyfind(true, 1, LogLevels), |
| 349 | - erlang:put(brod_cli_log_level, LogLevel), |
| 350 | - case IsHelp of |
| 351 | - true -> |
| 352 | - print(Doc); |
| 353 | - false -> |
| 354 | - main(Command, Doc, Args, Stop, LogLevel) |
| 355 | - end. |
| 356 | - |
| 357 | - -spec main(command(), string(), [string()], halt | exit, |
| 358 | - log_level()) -> _ | no_return(). |
| 359 | - main(Command, Doc, Args, Stop, LogLevel) -> |
| 360 | - ParsedArgs = |
| 361 | - try |
| 362 | - docopt:docopt(Doc, Args, [debug || LogLevel =:= ?LOG_LEVEL_DEBUG]) |
| 363 | - catch |
| 364 | - C1 : E1 ?BIND_STACKTRACE(Stack1) -> |
| 365 | - ?GET_STACKTRACE(Stack1), |
| 366 | - verbose("~p:~p\n~p\n", [C1, E1, Stack1]), |
| 367 | - ?STOP(Stop) |
| 368 | - end, |
| 369 | - case LogLevel =:= ?LOG_LEVEL_QUIET of |
| 370 | - true -> |
| 371 | - _ = error_logger:logfile({open, 'brod.log'}), |
| 372 | - _ = error_logger:tty(false); |
| 373 | - false -> |
| 374 | - ok |
| 375 | - end, |
| 376 | - %% So the linked processes won't take me with them |
| 377 | - %% This is to allow error_logger to write more logs |
| 378 | - %% before stopping init process immediately |
| 379 | - process_flag(trap_exit, true), |
| 380 | - ok = brod:start(), |
| 381 | - try |
| 382 | - Brokers = parse(ParsedArgs, "--brokers", fun parse_brokers/1), |
| 383 | - ConnConfig0 = parse_connection_config(ParsedArgs), |
| 384 | - Paths = parse(ParsedArgs, "--ebin-paths", fun parse_paths/1), |
| 385 | - NoApiQuery = parse(ParsedArgs, "--no-api-vsn-query", fun parse_boolean/1) |
| 386 | - orelse ({0, 9} =:= get_kafka_version()), |
| 387 | - ok = code:add_pathsa(Paths), |
| 388 | - SockOpts = [{query_api_versions, not NoApiQuery} | ConnConfig0], |
| 389 | - verbose("connection config: ~p\n", [SockOpts]), |
| 390 | - run(Command, Brokers, SockOpts, ParsedArgs) |
| 391 | - catch |
| 392 | - throw : Reason when is_binary(Reason) -> |
| 393 | - %% invalid options etc. |
| 394 | - logerr([Reason, "\n"]), |
| 395 | - ?STOP(Stop); |
| 396 | - C2 : E2 ?BIND_STACKTRACE(Stack2) -> |
| 397 | - ?GET_STACKTRACE(Stack2), |
| 398 | - logerr("~p:~p\n~p\n", [C2, E2, Stack2]), |
| 399 | - ?STOP(Stop) |
| 400 | - end. |
| 401 | - |
| 402 | - run(?META_CMD, Brokers, Topic, SockOpts, Args) -> |
| 403 | - Topics = case Topic of |
| 404 | - <<"*">> -> []; %% fetch all topics |
| 405 | - _ -> [Topic] |
| 406 | - end, |
| 407 | - IsJSON = parse(Args, "--json", fun parse_boolean/1), |
| 408 | - IsText = parse(Args, "--text", fun parse_boolean/1), |
| 409 | - Format = kf(true, [ {IsJSON, json} |
| 410 | - , {IsText, text} |
| 411 | - , {true, text} |
| 412 | - ]), |
| 413 | - IsList = parse(Args, "--list", fun parse_boolean/1), |
| 414 | - IsUrp = parse(Args, "--under-replicated", fun parse_boolean/1), |
| 415 | - {ok, Metadata} = brod:get_metadata(Brokers, Topics, SockOpts), |
| 416 | - format_metadata(Metadata, Format, IsList, IsUrp); |
| 417 | - run(?OFFSET_CMD, Brokers, Topic, SockOpts, Args) -> |
| 418 | - Partition = parse(Args, "--partition", fun("all") -> all; |
| 419 | - (Num) -> int(Num) |
| 420 | - end), |
| 421 | - IsOneLine = parse(Args, "--one-line", fun parse_boolean/1), |
| 422 | - Time = parse(Args, "--time", fun parse_offset_time/1), |
| 423 | - ok = start_client(Brokers, SockOpts), |
| 424 | - try |
| 425 | - resolve_offsets_print(Topic, Partition, Time, IsOneLine) |
| 426 | - after |
| 427 | - brod_client:stop(?CLIENT) |
| 428 | - end; |
| 429 | - run(?FETCH_CMD, Brokers, Topic, ConnOpts, Args) -> |
| 430 | - Partition = parse(Args, "--partition", fun int/1), %% not parse_partition/1 |
| 431 | - Count = parse(Args, "--count", fun int/1), |
| 432 | - Offset0 = parse(Args, "--offset", fun parse_offset_time/1), |
| 433 | - Wait = parse(Args, "--wait", fun parse_timeout/1), |
| 434 | - KvDeli = parse(Args, "--kv-deli", fun parse_delimiter/1), |
| 435 | - MsgDeli = parse(Args, "--msg-deli", fun parse_delimiter/1), |
| 436 | - FmtFun = parse(Args, "--fmt", |
| 437 | - fun(FmtOption) -> |
| 438 | - parse_fmt(FmtOption, KvDeli, MsgDeli) |
| 439 | - end), |
| 440 | - MaxBytes = parse(Args, "--max-bytes", fun parse_size/1), |
| 441 | - {ok, Conn} = brod:connect_leader(Brokers, Topic, Partition, ConnOpts), |
| 442 | - Offset = resolve_begin_offset(Conn, Topic, Partition, Offset0), |
| 443 | - FetchOpts = #{max_wait_time => Wait, max_bytes => MaxBytes}, |
| 444 | - FoldLimits = case Count < 0 of |
| 445 | - true -> #{}; |
| 446 | - false -> #{message_count => Count} |
| 447 | - end, |
| 448 | - FoldFun = |
| 449 | - fun(M, Acc) -> |
| 450 | - #kafka_message{offset = O, key = K, value = V} = M, |
| 451 | - R = case is_function(FmtFun, 3) of |
| 452 | - true -> FmtFun(O, K, V); |
| 453 | - false -> FmtFun(M) |
| 454 | - end, |
| 455 | - case R of |
| 456 | - ok -> ok; |
| 457 | - IoData -> print(IoData) |
| 458 | - end, |
| 459 | - {ok, Acc + 1} |
| 460 | - end, |
| 461 | - {FetchedCount, NextOffset, Reason} = |
| 462 | - brod:fold(Conn, Topic, Partition, Offset, FetchOpts, |
| 463 | - 0, FoldFun, FoldLimits), |
| 464 | - logerr("Fetch loop stopped after ~p messages~n", [FetchedCount]), |
| 465 | - logerr("Continue at Offset: ~p~n", [NextOffset]), |
| 466 | - logerr("Reason: ~p~n", [Reason]); |
| 467 | - run(?SEND_CMD, Brokers, Topic, SockOpts, Args) -> |
| 468 | - Partition = parse(Args, "--partition", fun parse_partition/1), |
| 469 | - Acks = parse(Args, "--acks", fun parse_acks/1), |
| 470 | - AckTimeout = parse(Args, "--ack-timeout", fun parse_timeout/1), |
| 471 | - Compression = parse(Args, "--compression", fun parse_compression/1), |
| 472 | - Key = parse(Args, "--key", fun("null") -> <<"">>; |
| 473 | - (K) -> bin(K) |
| 474 | - end), |
| 475 | - Value0 = parse(Args, "--value", fun("null") -> <<"">>; |
| 476 | - ("@" ++ F) -> {file, F}; |
| 477 | - (V) -> bin(V) |
| 478 | - end), |
| 479 | - Value = |
| 480 | - case Value0 of |
| 481 | - {file, File} -> |
| 482 | - {ok, Bin} = file:read_file(File), |
| 483 | - Bin; |
| 484 | - <<_/binary>> -> |
| 485 | - Value0 |
| 486 | - end, |
| 487 | - ProducerConfig = |
| 488 | - [ {required_acks, Acks} |
| 489 | - , {ack_timeout, AckTimeout} |
| 490 | - , {compression, Compression} |
| 491 | - , {min_compression_batch_size, 0} |
| 492 | - , {max_linger_ms, 0} |
| 493 | - ], |
| 494 | - ClientConfig = |
| 495 | - [ {auto_start_producers, true} |
| 496 | - , {default_producer_config, ProducerConfig} |
| 497 | - ] ++ SockOpts, |
| 498 | - ok = start_client(Brokers, ClientConfig), |
| 499 | - Msgs = [{brod_utils:epoch_ms(), Key, Value}], |
| 500 | - ok = brod:produce_sync(?CLIENT, Topic, Partition, <<>>, Msgs); |
| 501 | - run(?PIPE_CMD, Brokers, Topic, SockOpts, Args) -> |
| 502 | - Partition = parse(Args, "--partition", fun parse_partition/1), |
| 503 | - Acks = parse(Args, "--acks", fun parse_acks/1), |
| 504 | - AckTimeout = parse(Args, "--ack-timeout", fun parse_timeout/1), |
| 505 | - Compression = parse(Args, "--compression", fun parse_compression/1), |
| 506 | - KvDeli = parse(Args, "--kv-deli", fun parse_delimiter/1), |
| 507 | - MsgDeli = parse(Args, "--msg-deli", fun parse_delimiter/1), |
| 508 | - MaxLingerMs = parse(Args, "--max-linger-ms", fun int/1), |
| 509 | - MaxLingerCnt = parse(Args, "--max-linger-cnt", fun int/1), |
| 510 | - MaxBatch = parse(Args, "--max-batch", fun parse_size/1), |
| 511 | - Source = parse(Args, "--source", fun parse_source/1), |
| 512 | - IsPrompt = parse(Args, "--prompt", fun parse_boolean/1), |
| 513 | - IsTail = parse(Args, "--tail", fun parse_boolean/1), |
| 514 | - IsNoExit = parse(Args, "--no-eof-exit", fun parse_boolean/1), |
| 515 | - BlkSize = parse(Args, "--blk-size", fun parse_size/1), |
| 516 | - ProducerConfig = |
| 517 | - [ {required_acks, Acks} |
| 518 | - , {ack_timeout, AckTimeout} |
| 519 | - , {compression, Compression} |
| 520 | - , {min_compression_batch_size, 0} |
| 521 | - , {max_linger_ms, MaxLingerMs} |
| 522 | - , {max_linger_count, MaxLingerCnt} |
| 523 | - , {max_batch_size, MaxBatch} |
| 524 | - ], |
| 525 | - ClientConfig = |
| 526 | - [ {auto_start_producers, true} |
| 527 | - , {default_producer_config, ProducerConfig} |
| 528 | - ] ++ SockOpts, |
| 529 | - ok = start_client(Brokers, ClientConfig), |
| 530 | - SendFun = |
| 531 | - fun(?TKV(Ts, Key, Value), PendingAcks) -> |
| 532 | - {ok, CallRef} = |
| 533 | - brod:produce(?CLIENT, Topic, Partition, <<>>, [{Ts, Key, Value}]), |
| 534 | - debug("sent: ~w\n", [CallRef]), |
| 535 | - debug("value: ~P\n", [Value, 9]), |
| 536 | - queue:in(CallRef, PendingAcks) |
| 537 | - end, |
| 538 | - KvDeliForReader = case none =:= KvDeli of |
| 539 | - true -> none; |
| 540 | - false -> bin(KvDeli) |
| 541 | - end, |
| 542 | - ReaderArgs = [ {source, Source} |
| 543 | - , {kv_deli, KvDeliForReader} |
| 544 | - , {msg_deli, bin(MsgDeli)} |
| 545 | - , {prompt, IsPrompt} |
| 546 | - , {tail, IsTail} |
| 547 | - , {no_exit, IsNoExit} |
| 548 | - , {blk_size, BlkSize} |
| 549 | - , {retry_delay, 100} |
| 550 | - ], |
| 551 | - {ok, ReaderPid} = |
| 552 | - brod_cli_pipe:start_link(ReaderArgs), |
| 553 | - _ = erlang:monitor(process, ReaderPid), |
| 554 | - pipe(ReaderPid, SendFun, queue:new()). |
| 555 | - |
| 556 | - run(?GROUPS_CMD, Brokers, SockOpts, Args) -> |
| 557 | - IDs = parse(Args, "--ids", fun parse_cg_ids/1), |
| 558 | - cg(Brokers, SockOpts, IDs); |
| 559 | - run(?COMMITS_CMD, Brokers, SockOpts, Args) -> |
| 560 | - IsDesc = parse(Args, "--describe", fun parse_boolean/1), |
| 561 | - ID = parse(Args, "--id", fun bin/1), |
| 562 | - Topic = parse(Args, "--topic", |
| 563 | - fun(?undef) -> ?undef; |
| 564 | - (Name) -> bin(Name) |
| 565 | - end), |
| 566 | - ok = start_client(Brokers, SockOpts), |
| 567 | - case IsDesc of |
| 568 | - true -> show_commits(ID, Topic); |
| 569 | - false -> reset_commits(ID, Topic, Args) |
| 570 | - end; |
| 571 | - run(Cmd, Brokers, SockOpts, Args) -> |
| 572 | - %% Clause for all per-topic commands |
| 573 | - Topic = parse(Args, "--topic", fun bin/1), |
| 574 | - run(Cmd, Brokers, Topic, SockOpts, Args). |
| 575 | - |
| 576 | - resolve_offsets_print(Topic, all, Time, IsOneLine) -> |
| 577 | - Offsets = resolve_offsets(Topic, Time), |
| 578 | - Outputs = |
| 579 | - lists:map( |
| 580 | - fun({Partition, Offset}) -> |
| 581 | - io_lib:format("~p:~p", [Partition, Offset]) |
| 582 | - end, Offsets), |
| 583 | - Delimiter = case IsOneLine of |
| 584 | - true -> ","; |
| 585 | - false -> "\n" |
| 586 | - end, |
| 587 | - print(infix(Outputs, Delimiter)); |
| 588 | - resolve_offsets_print(Topic, Partition, Time, _) when is_integer(Partition) -> |
| 589 | - {ok, Offset} = resolve_offset(Topic, Partition, Time), |
| 590 | - print(integer_to_list(Offset)). |
| 591 | - |
| 592 | - resolve_offsets(Topic, Time) -> |
| 593 | - {ok, Count} = brod_client:get_partitions_count(?CLIENT, Topic), |
| 594 | - Partitions = lists:seq(0, Count - 1), |
| 595 | - lists:map( |
| 596 | - fun(P) -> |
| 597 | - {ok, Offset} = resolve_offset(Topic, P, Time), |
| 598 | - {P, Offset} |
| 599 | - end, Partitions). |
| 600 | - |
| 601 | - resolve_offset(Topic, Partition, Time) -> |
| 602 | - {ok, SockPid} = brod_client:get_leader_connection(?CLIENT, Topic, Partition), |
| 603 | - brod_utils:resolve_offset(SockPid, Topic, Partition, Time). |
| 604 | - |
| 605 | - show_commits(GroupId, Topic) -> |
| 606 | - case brod:fetch_committed_offsets(?CLIENT, GroupId) of |
| 607 | - {ok, PerTopicStructs0} -> |
| 608 | - Pred = fun(S) -> Topic =:= ?undef orelse Topic =:= kf(topic, S) end, |
| 609 | - PerTopicStructs = lists:filter(Pred, PerTopicStructs0), |
| 610 | - lists:foreach(fun print_commits/1, PerTopicStructs); |
| 611 | - {error, Reason} -> |
| 612 | - throw_bin("Failed to fetch commited offsets ~p\n", [Reason]) |
| 613 | - end. |
| 614 | - |
| 615 | - reset_commits(ID, Topic, Args) -> |
| 616 | - Retention = parse(Args, "--retention", fun parse_retention/1), |
| 617 | - ProtocolName = parse(Args, "--protocol", fun(X) -> X end), |
| 618 | - Offsets0 = parse(Args, "--offsets", fun parse_commit_offsets_input/1), |
| 619 | - Offsets = |
| 620 | - case is_atom(Offsets0) of |
| 621 | - true -> resolve_offsets(Topic, Offsets0); |
| 622 | - false -> Offsets0 |
| 623 | - end, |
| 624 | - Group = [ {id, ID} |
| 625 | - , {topic, Topic} |
| 626 | - , {retention, Retention} |
| 627 | - , {protocol, ProtocolName} |
| 628 | - , {offsets, Offsets} |
| 629 | - ], |
| 630 | - brod_cg_commits:run(?CLIENT, Group). |
| 631 | - |
| 632 | - parse_commit_offsets_input("latest") -> latest; |
| 633 | - parse_commit_offsets_input("earliest") -> earliest; |
| 634 | - parse_commit_offsets_input(PartitionOffsets) -> |
| 635 | - Pairs = string:tokens(PartitionOffsets, ","), |
| 636 | - F = fun(Pair) -> |
| 637 | - [Partition, Offset] = string:tokens(Pair, ":"), |
| 638 | - {int(Partition), parse_offset_time(Offset)} |
| 639 | - end, |
| 640 | - lists:map(F, Pairs). |
| 641 | - |
| 642 | - parse_retention("-1") -> -1; |
| 643 | - parse_retention([_|_] = R) -> |
| 644 | - case lists:last(R) of |
| 645 | - X when X >= $0 andalso X =< $9 -> |
| 646 | - int(R); |
| 647 | - Unit -> |
| 648 | - int(lists:reverse(tl(lists:reverse(R)))) * |
| 649 | - case Unit of |
| 650 | - $s -> 1; |
| 651 | - $S -> 1; |
| 652 | - $m -> 60; |
| 653 | - $M -> 60; |
| 654 | - $h -> 60 * 60; |
| 655 | - $H -> 60 * 60; |
| 656 | - $d -> 60 * 60 * 24; |
| 657 | - $D -> 60 * 60 * 24 |
| 658 | - end |
| 659 | - end. |
| 660 | - |
| 661 | - print_commits(Struct) -> |
| 662 | - Topic = kf(topic, Struct), |
| 663 | - PartRsps = kf(partition_responses, Struct), |
| 664 | - print([Topic, ":\n"]), |
| 665 | - print([pp_fmt_struct(1, P) || P <- PartRsps]). |
| 666 | - |
| 667 | - cg(BootstrapEndpoints, SockOpts, all) -> |
| 668 | - %% List all groups |
| 669 | - All = list_groups(BootstrapEndpoints, SockOpts), |
| 670 | - lists:foreach(fun print_cg_cluster/1, All); |
| 671 | - cg(BootstrapEndpoints, SockOpts, IDs) -> |
| 672 | - CgClusters = list_groups(BootstrapEndpoints, SockOpts), |
| 673 | - describe_cgs(CgClusters, SockOpts, lists:usort(IDs)). |
| 674 | - |
| 675 | - describe_cgs(_, _SockOpts, []) -> ok; |
| 676 | - describe_cgs([], _SockOpts, IDs) -> |
| 677 | - logerr("Unknown group IDs: ~s", [infix(IDs, ", ")]); |
| 678 | - describe_cgs([{Coordinator, CgList} | Rest], SockOpts, IDs) -> |
| 679 | - %% Get all IDs managed by current coordinator. |
| 680 | - ThisIDs = [ID || #brod_cg{id = ID} <- CgList, lists:member(ID, IDs)], |
| 681 | - ok = do_describe_cgs(Coordinator, SockOpts, ThisIDs), |
| 682 | - IDsRest = IDs -- ThisIDs, |
| 683 | - describe_cgs(Rest, SockOpts, IDsRest). |
| 684 | - |
| 685 | - do_describe_cgs(_Coordinator, _SockOpts, []) -> ok; |
| 686 | - do_describe_cgs(Coordinator, SockOpts, IDs) -> |
| 687 | - case brod:describe_groups(Coordinator, SockOpts, IDs) of |
| 688 | - {ok, DescArray} -> |
| 689 | - ok = print("~s\n", [fmt_endpoint(Coordinator)]), |
| 690 | - lists:foreach(fun print_cg_desc/1, DescArray); |
| 691 | - {error, Reason} -> |
| 692 | - logerr("Failed to describe IDs [~s] at broker ~s\nreason:~p\n", |
| 693 | - [infix(IDs, ","), fmt_endpoint(Coordinator), Reason]) |
| 694 | - end. |
| 695 | - |
| 696 | - print_cg_desc(Desc) -> |
| 697 | - EC = kf(error_code, Desc), |
| 698 | - GroupId = kf(group_id, Desc), |
| 699 | - case ?IS_ERROR(EC) of |
| 700 | - true -> |
| 701 | - logerr("Failed to describe group id=~s\nreason:~p\n", [GroupId, EC]); |
| 702 | - false -> |
| 703 | - D1 = lists:keydelete(error_code, 1, ensure_list(Desc)), |
| 704 | - D = lists:keydelete(group_id, 1, ensure_list(D1)), |
| 705 | - print(" ~s\n~s", [GroupId, pp_fmt_struct(_Indent = 2, D)]) |
| 706 | - end. |
| 707 | - |
| 708 | - ensure_list(Struct) when is_map(Struct) -> maps:to_list(Struct); |
| 709 | - ensure_list(List) when is_list(List) -> List. |
| 710 | - |
| 711 | - pp_fmt_struct(Indent, Map) when is_map(Map) -> |
| 712 | - pp_fmt_struct(Indent, maps:to_list(Map)); |
| 713 | - pp_fmt_struct(Indent, Fields0) when is_list(Fields0) -> |
| 714 | - Fields = case Fields0 of |
| 715 | - [_] -> Fields0; |
| 716 | - _ -> lists:keydelete(no_error, 2, Fields0) |
| 717 | - end, |
| 718 | - F = fun(IsFirst, {N, V}) when is_map(V) -> |
| 719 | - indent_fmt(IsFirst, Indent, |
| 720 | - "~p:\n~s", [N, pp_fmt_struct(Indent + 1, V)]); |
| 721 | - (IsFirst, {N, V}) -> |
| 722 | - indent_fmt(IsFirst, Indent, |
| 723 | - "~p: ~s", [N, pp_fmt_struct_value(Indent, V)]) |
| 724 | - end, |
| 725 | - [ F(true, hd(Fields)) |
| 726 | - | lists:map(fun(Fi) -> F(false, Fi) end, tl(Fields)) |
| 727 | - ]. |
| 728 | - |
| 729 | - pp_fmt_struct_value(_Indent, X) when is_integer(X) orelse |
| 730 | - is_atom(X) orelse |
| 731 | - is_binary(X) orelse |
| 732 | - X =:= [] -> |
| 733 | - [pp_fmt_prim(X), "\n"]; |
| 734 | - pp_fmt_struct_value(Indent, [H | _] = Array) when is_list(Array) -> |
| 735 | - case is_tuple(H) orelse is_map(H) of |
| 736 | - true -> |
| 737 | - %% array of sub struct |
| 738 | - ["\n", |
| 739 | - lists:map(fun(Item) -> |
| 740 | - pp_fmt_struct(Indent + 1, Item) |
| 741 | - end, Array) |
| 742 | - ]; |
| 743 | - false -> |
| 744 | - %% array of primitive values |
| 745 | - [[pp_fmt_prim(V) || V <- Array], "\n"] |
| 746 | - end. |
| 747 | - |
| 748 | - pp_fmt_prim([]) -> "[]"; |
| 749 | - pp_fmt_prim(N) when is_integer(N) -> integer_to_list(N); |
| 750 | - pp_fmt_prim(A) when is_atom(A) -> atom_to_list(A); |
| 751 | - pp_fmt_prim(S) when is_binary(S) -> S. |
| 752 | - |
| 753 | - indent_fmt(true, Indent, Fmt, Args) -> |
| 754 | - io_lib:format(lists:duplicate((Indent - 1) * 2, $\s) ++ "- " ++ Fmt, Args); |
| 755 | - indent_fmt(false, Indent, Fmt, Args) -> |
| 756 | - io_lib:format(lists:duplicate(Indent * 2, $\s) ++ Fmt, Args). |
| 757 | - |
| 758 | - print_cg_cluster({Endpoint, Cgs}) -> |
| 759 | - ok = print([fmt_endpoint(Endpoint), "\n"]), |
| 760 | - IoData = [ io_lib:format(" ~s (~s)\n", [Id, Type]) |
| 761 | - || #brod_cg{id = Id, protocol_type = Type} <- Cgs |
| 762 | - ], |
| 763 | - print(IoData). |
| 764 | - |
| 765 | - fmt_endpoint({Host, Port}) -> |
| 766 | - bin(io_lib:format("~s:~B", [Host, Port])). |
| 767 | - |
| 768 | - %% Return consumer groups clustered by group coordinator |
| 769 | - %% {CoordinatorEndpoint, [group_id()]}. |
| 770 | - list_groups(Brokers, SockOpts) -> |
| 771 | - Cgs = brod:list_all_groups(Brokers, SockOpts), |
| 772 | - lists:keysort(1, lists:foldl(fun do_list_groups/2, [], Cgs)). |
| 773 | - |
| 774 | - do_list_groups({_Endpoint, []}, Acc) -> Acc; |
| 775 | - do_list_groups({Endpoint, {error, Reason}}, Acc) -> |
| 776 | - logerr("Failed to list groups at kafka ~s\nreason~p", |
| 777 | - [fmt_endpoint(Endpoint), Reason]), |
| 778 | - Acc; |
| 779 | - do_list_groups({Endpoint, Cgs}, Acc) -> |
| 780 | - [{Endpoint, Cgs} | Acc]. |
| 781 | - |
| 782 | - pipe(ReaderPid, SendFun, PendingAcks0) -> |
| 783 | - PendingAcks1 = flush_pending_acks(PendingAcks0, _Timeout = 0), |
| 784 | - receive |
| 785 | - {pipe, ReaderPid, Messages} -> |
| 786 | - PendingAcks = lists:foldl(SendFun, PendingAcks1, Messages), |
| 787 | - pipe(ReaderPid, SendFun, PendingAcks); |
| 788 | - {'DOWN', _Ref, process, ReaderPid, Reason} -> |
| 789 | - %% Reader is down, flush pending acks |
| 790 | - debug("reader down, reason: ~p\n", [Reason]), |
| 791 | - _ = flush_pending_acks(PendingAcks1, infinity); |
| 792 | - #brod_produce_reply{ call_ref = CallRef |
| 793 | - , result = brod_produce_req_acked |
| 794 | - } -> |
| 795 | - {{value, CallRef}, PendingAcks} = queue:out(PendingAcks1), |
| 796 | - pipe(ReaderPid, SendFun, PendingAcks) |
| 797 | - end. |
| 798 | - |
| 799 | - flush_pending_acks(Queue, Timeout) -> |
| 800 | - case queue:peek(Queue) of |
| 801 | - empty -> |
| 802 | - Queue; |
| 803 | - {value, CallRef} -> |
| 804 | - case brod:sync_produce_request(CallRef, Timeout) of |
| 805 | - ok -> |
| 806 | - debug("acked: ~w\n", [CallRef]), |
| 807 | - {_, Rest} = queue:out(Queue), |
| 808 | - flush_pending_acks(Rest, Timeout); |
| 809 | - {error, timeout} -> |
| 810 | - Queue |
| 811 | - end |
| 812 | - end. |
| 813 | - |
| 814 | - resolve_begin_offset(_Sock, _T, _P, Offset) when is_integer(Offset) -> |
| 815 | - Offset; |
| 816 | - resolve_begin_offset(Sock, Topic, Partition, last) -> |
| 817 | - Earliest = resolve_begin_offset(Sock, Topic, Partition, earliest), |
| 818 | - Latest = resolve_begin_offset(Sock, Topic, Partition, latest), |
| 819 | - case Latest =:= Earliest of |
| 820 | - true -> erlang:throw(bin("partition is empty")); |
| 821 | - false -> Latest - 1 |
| 822 | - end; |
| 823 | - resolve_begin_offset(Sock, Topic, Partition, Time) -> |
| 824 | - {ok, Offset} = brod_utils:resolve_offset(Sock, Topic, Partition, Time), |
| 825 | - Offset. |
| 826 | - |
| 827 | - parse_source("stdin") -> |
| 828 | - standard_io; |
| 829 | - parse_source("@" ++ Path) -> |
| 830 | - parse_source(Path); |
| 831 | - parse_source(Path) -> |
| 832 | - case filelib:is_regular(Path) of |
| 833 | - true -> {file, Path}; |
| 834 | - false -> erlang:throw(bin(["bad file ", Path])) |
| 835 | - end. |
| 836 | - |
| 837 | - parse_size(Size) -> |
| 838 | - case lists:reverse(Size) of |
| 839 | - "K" ++ N -> int(lists:reverse(N)) * (1 bsl 10); |
| 840 | - "M" ++ N -> int(lists:reverse(N)) * (1 bsl 20); |
| 841 | - N -> int(lists:reverse(N)) |
| 842 | - end. |
| 843 | - |
| 844 | - format_metadata(Metadata, Format, IsList, IsToListUrp) -> |
| 845 | - Brokers = kf(brokers, Metadata), |
| 846 | - Topics0 = kf(topic_metadata, Metadata), |
| 847 | - Cluster = kf(cluster_id, Metadata, ?undef), |
| 848 | - Controller = kf(controller_id, Metadata, ?undef), |
| 849 | - Topics1 = case IsToListUrp of |
| 850 | - true -> lists:filter(fun is_ur_topic/1, Topics0); |
| 851 | - false -> Topics0 |
| 852 | - end, |
| 853 | - Topics = format_topics(Topics1), |
| 854 | - case Format of |
| 855 | - json -> |
| 856 | - JSON = jsone:encode([ {brokers, Brokers} |
| 857 | - , {topics, Topics} |
| 858 | - , {cluster_id, Cluster} |
| 859 | - , {controller_id, Controller} |
| 860 | - ]), |
| 861 | - print([JSON, "\n"]); |
| 862 | - text -> |
| 863 | - CL = case Cluster of |
| 864 | - ?undef -> ""; |
| 865 | - _ -> io_lib:format("cluster_id: ~s\n", [Cluster]) |
| 866 | - end, |
| 867 | - |
| 868 | - CT = case Controller of |
| 869 | - ?undef -> ""; |
| 870 | - _ -> io_lib:format("controller: ~p\n", [Controller]) |
| 871 | - end, |
| 872 | - BL = format_broker_lines(Brokers), |
| 873 | - TL = format_topics_lines(Topics, IsList), |
| 874 | - case IsList of |
| 875 | - true -> print(TL); |
| 876 | - false -> print([CL, CT, BL, TL]) |
| 877 | - end |
| 878 | - end. |
| 879 | - |
| 880 | - format_broker_lines(Brokers) -> |
| 881 | - Header = io_lib:format("brokers [~p]:\n", [length(Brokers)]), |
| 882 | - F = fun(Broker) -> |
| 883 | - Id = kf(node_id, Broker), |
| 884 | - Host = kf(host, Broker), |
| 885 | - Port = kf(port, Broker), |
| 886 | - Rack = kf(rack, Broker, <<>>), |
| 887 | - HostStr = fmt_endpoint({Host, Port}), |
| 888 | - format_broker_line(Id, Rack, HostStr) |
| 889 | - end, |
| 890 | - [Header, lists:map(F, Brokers)]. |
| 891 | - |
| 892 | - format_broker_line(Id, Rack, Endpoint) |
| 893 | - when Rack =:= ?kpro_null orelse Rack =:= <<>> -> |
| 894 | - io_lib:format(" ~p: ~s\n", [Id, Endpoint]); |
| 895 | - format_broker_line(Id, Rack, Endpoint) -> |
| 896 | - io_lib:format(" ~p(~s): ~s\n", [Id, Rack, Endpoint]). |
| 897 | - |
| 898 | - format_topics_lines(Topics, true) -> |
| 899 | - Header = io_lib:format("topics [~p]:\n", [length(Topics)]), |
| 900 | - [Header, lists:map(fun format_topic_list_line/1, Topics)]; |
| 901 | - format_topics_lines(Topics, false) -> |
| 902 | - Header = io_lib:format("topics [~p]:\n", [length(Topics)]), |
| 903 | - [Header, lists:map(fun format_topic_lines/1, Topics)]. |
| 904 | - |
| 905 | - format_topic_list_line({Name, Partitions}) when is_list(Partitions) -> |
| 906 | - io_lib:format(" ~s\n", [Name]); |
| 907 | - format_topic_list_line({Name, ErrorCode}) -> |
| 908 | - ErrorStr = format_error_code(ErrorCode), |
| 909 | - io_lib:format(" ~s: [ERROR] ~s\n", [Name, ErrorStr]). |
| 910 | - |
| 911 | - format_topic_lines({Name, Partitions}) when is_list(Partitions) -> |
| 912 | - Header = io_lib:format(" ~s [~p]:\n", [Name, length(Partitions)]), |
| 913 | - PartitionsText = format_partitions_lines(Partitions), |
| 914 | - [Header, PartitionsText]; |
| 915 | - format_topic_lines({Name, ErrorCode}) -> |
| 916 | - ErrorStr = format_error_code(ErrorCode), |
| 917 | - io_lib:format(" ~s: [ERROR] ~s\n", [Name, ErrorStr]). |
| 918 | - |
| 919 | - format_error_code(E) when is_atom(E) -> atom_to_list(E); |
| 920 | - format_error_code(E) when is_integer(E) -> integer_to_list(E). |
| 921 | - |
| 922 | - format_partitions_lines(Partitions0) -> |
| 923 | - Partitions1 = |
| 924 | - lists:map(fun({Pnr, Info}) -> |
| 925 | - {binary_to_integer(Pnr), Info} |
| 926 | - end, Partitions0), |
| 927 | - Partitions = lists:keysort(1, Partitions1), |
| 928 | - lists:map(fun format_partition_lines/1, Partitions). |
| 929 | - |
| 930 | - format_partition_lines({Partition, Info}) -> |
| 931 | - LeaderNodeId = kf(leader, Info), |
| 932 | - Status = kf(status, Info), |
| 933 | - Isr = kf(isr, Info), |
| 934 | - Osr = kf(osr, Info), |
| 935 | - MaybeWarning = case ?IS_ERROR(Status) of |
| 936 | - true -> [" [", atom_to_list(Status), "]"]; |
| 937 | - false -> "" |
| 938 | - end, |
| 939 | - ReplicaList = |
| 940 | - case Osr of |
| 941 | - [] -> format_list(Isr, ""); |
| 942 | - _ -> [format_list(Isr, ""), ",", format_list(Osr, "*")] |
| 943 | - end, |
| 944 | - io_lib:format("~7s: ~2s (~s)~s\n", |
| 945 | - [integer_to_list(Partition), |
| 946 | - integer_to_list(LeaderNodeId), |
| 947 | - ReplicaList, MaybeWarning]). |
| 948 | - |
| 949 | - format_list(List, Mark) -> |
| 950 | - infix(lists:map(fun(I) -> [integer_to_list(I), Mark] end, List), ","). |
| 951 | - |
| 952 | - infix([], _Sep) -> []; |
| 953 | - infix([_] = L, _Sep) -> L; |
| 954 | - infix([H | T], Sep) -> [H, Sep, infix(T, Sep)]. |
| 955 | - |
| 956 | - format_topics(Topics) -> |
| 957 | - TL = lists:map(fun format_topic/1, Topics), |
| 958 | - lists:keysort(1, TL). |
| 959 | - |
| 960 | - format_topic(Topic) -> |
| 961 | - TopicName = kf(topic, Topic), |
| 962 | - PL = kf(partition_metadata, Topic), |
| 963 | - {TopicName, format_partitions(PL)}. |
| 964 | - |
| 965 | - format_partitions(Partitions) -> |
| 966 | - PL = lists:map(fun format_partition/1, Partitions), |
| 967 | - lists:keysort(1, PL). |
| 968 | - |
| 969 | - format_partition(P) -> |
| 970 | - ErrorCode = kf(error_code, P), |
| 971 | - PartitionNr = kf(partition, P), |
| 972 | - LeaderNodeId = kf(leader, P), |
| 973 | - Replicas = kf(replicas, P), |
| 974 | - Isr = kf(isr, P), |
| 975 | - Data = [ {leader, LeaderNodeId} |
| 976 | - , {status, ErrorCode} |
| 977 | - , {isr, Isr} |
| 978 | - , {osr, Replicas -- Isr} |
| 979 | - ], |
| 980 | - {integer_to_binary(PartitionNr), Data}. |
| 981 | - |
| 982 | - %% Return true if a topics is under-replicated |
| 983 | - is_ur_topic(Topic) -> |
| 984 | - ErrorCode = kf(error_code, Topic), |
| 985 | - Partitions = kf(partition_metadata, Topic), |
| 986 | - %% when there is an error, we do not know if |
| 987 | - %% it is under-replicated or not |
| 988 | - %% retrun true to alert user |
| 989 | - ?IS_ERROR(ErrorCode) orelse lists:any(fun is_ur_partition/1, Partitions). |
| 990 | - |
| 991 | - %% Return true if a partition is under-replicated |
| 992 | - is_ur_partition(Partition) -> |
| 993 | - ErrorCode = kf(error_code, Partition), |
| 994 | - Replicas = kf(replicas, Partition), |
| 995 | - Isr = kf(isr, Partition), |
| 996 | - ?IS_ERROR(ErrorCode) orelse lists:sort(Isr) =/= lists:sort(Replicas). |
| 997 | - |
| 998 | - parse_delimiter("none") -> none; |
| 999 | - parse_delimiter(EscappedStr) -> eval_str(EscappedStr). |
| 1000 | - |
| 1001 | - eval_str([]) -> []; |
| 1002 | - eval_str([$\\, $n | Rest]) -> |
| 1003 | - [$\n | eval_str(Rest)]; |
| 1004 | - eval_str([$\\, $t | Rest]) -> |
| 1005 | - [$\t | eval_str(Rest)]; |
| 1006 | - eval_str([$\\, $s | Rest]) -> |
| 1007 | - [$\s | eval_str(Rest)]; |
| 1008 | - eval_str([C | Rest]) -> |
| 1009 | - [C | eval_str(Rest)]. |
| 1010 | - |
| 1011 | - parse_fmt("v", _KvDel, MsgDeli) -> |
| 1012 | - fun(_Offset, _Key, Value) -> [Value, MsgDeli] end; |
| 1013 | - parse_fmt("kv", KvDeli, MsgDeli) -> |
| 1014 | - fun(_Offset, Key, Value) -> [Key, KvDeli, Value, MsgDeli] end; |
| 1015 | - parse_fmt("okv", KvDeli, MsgDeli) -> |
| 1016 | - fun(Offset, Key, Value) -> |
| 1017 | - [integer_to_list(Offset), KvDeli, |
| 1018 | - Key, KvDeli, Value, MsgDeli] |
| 1019 | - end; |
| 1020 | - parse_fmt("eterm", _KvDeli, _MsgDeli) -> |
| 1021 | - fun(Offset, Key, Value) -> |
| 1022 | - io_lib:format("~p.\n", [{Offset, Key, Value}]) |
| 1023 | - end; |
| 1024 | - parse_fmt(FunLiteral0, _KvDeli, _MsgDeli) -> |
| 1025 | - FunLiteral = ensure_end_with_dot(FunLiteral0), |
| 1026 | - {ok, Tokens, _Line} = erl_scan:string(FunLiteral), |
| 1027 | - {ok, [Expr]} = erl_parse:parse_exprs(Tokens), |
| 1028 | - fun(#kafka_message{offset = Offset, |
| 1029 | - key = Key, |
| 1030 | - value = Value, |
| 1031 | - ts_type = TsType, |
| 1032 | - ts = Ts, |
| 1033 | - headers = Headers |
| 1034 | - }) -> |
| 1035 | - Bindings = |
| 1036 | - lists:foldl( |
| 1037 | - fun({VarName, VarValue}, Acc) -> |
| 1038 | - erl_eval:add_binding(VarName, VarValue, Acc) |
| 1039 | - end, erl_eval:new_bindings(), |
| 1040 | - [ {'Offset', Offset} |
| 1041 | - , {'Key', Key} |
| 1042 | - , {'Value', Value} |
| 1043 | - , {'TsType', TsType} |
| 1044 | - , {'Ts', Ts} |
| 1045 | - , {'Headers', Headers} |
| 1046 | - ]), |
| 1047 | - {value, Val, _NewBindings} = erl_eval:expr(Expr, Bindings), |
| 1048 | - case Val of |
| 1049 | - F when is_function(F, 3) -> |
| 1050 | - %% for backward compatibility |
| 1051 | - F(Offset, Key, Value); |
| 1052 | - V -> |
| 1053 | - V |
| 1054 | - end |
| 1055 | - end. |
| 1056 | - |
| 1057 | - %% Append a dot to the function literal. |
| 1058 | - ensure_end_with_dot(Str0) -> |
| 1059 | - Str = rstrip(Str0, [$\n, $\t, $\s, $.]), |
| 1060 | - Str ++ ".". |
| 1061 | - |
| 1062 | - rstrip(Str, CharSet) -> |
| 1063 | - lists:reverse(lstrip(lists:reverse(Str), CharSet)). |
| 1064 | - |
| 1065 | - lstrip([], _) -> []; |
| 1066 | - lstrip([C | Rest] = Str, CharSet) -> |
| 1067 | - case lists:member(C, CharSet) of |
| 1068 | - true -> lstrip(Rest, CharSet); |
| 1069 | - false -> Str |
| 1070 | - end. |
| 1071 | - |
| 1072 | - parse_partition("random") -> |
| 1073 | - fun(_Topic, PartitionsCount, _Key, _Value) -> |
| 1074 | - {_, _, Micro} = os:timestamp(), |
| 1075 | - {ok, Micro rem PartitionsCount} |
| 1076 | - end; |
| 1077 | - parse_partition("hash") -> |
| 1078 | - fun(_Topic, PartitionsCount, Key, _Value) -> |
| 1079 | - Hash = erlang:phash2(Key), |
| 1080 | - {ok, Hash rem PartitionsCount} |
| 1081 | - end; |
| 1082 | - parse_partition(I) -> |
| 1083 | - try |
| 1084 | - list_to_integer(I) |
| 1085 | - catch |
| 1086 | - _ : _ -> |
| 1087 | - erlang:throw(bin(["Bad partition: ", I])) |
| 1088 | - end. |
| 1089 | - |
| 1090 | - parse_acks("all") -> -1; |
| 1091 | - parse_acks("-1") -> -1; |
| 1092 | - parse_acks("0") -> 0; |
| 1093 | - parse_acks("1") -> 1; |
| 1094 | - parse_acks(X) -> erlang:throw(bin(["Bad --acks value: ", X])). |
| 1095 | - |
| 1096 | - parse_timeout(Str) -> |
| 1097 | - case lists:reverse(Str) of |
| 1098 | - "s" ++ R -> int(lists:reverse(R)) * 1000; |
| 1099 | - "m" ++ R -> int(lists:reverse(R)) * 60 * 1000; |
| 1100 | - _ -> int(Str) |
| 1101 | - end. |
| 1102 | - |
| 1103 | - parse_compression("none") -> no_compression; |
| 1104 | - parse_compression("gzip") -> gzip; |
| 1105 | - parse_compression("snappy") -> snappy; |
| 1106 | - parse_compression(X) -> erlang:throw(bin(["Unknown --compresion value: ", X])). |
| 1107 | - |
| 1108 | - parse_offset_time("earliest") -> earliest; |
| 1109 | - parse_offset_time("latest") -> latest; |
| 1110 | - parse_offset_time("last") -> last; |
| 1111 | - parse_offset_time(T) -> int(T). |
| 1112 | - |
| 1113 | - parse_connection_config(Args) -> |
| 1114 | - SslBool = parse(Args, "--ssl", fun parse_boolean/1), |
| 1115 | - CaCertFile = parse(Args, "--cacertfile", fun parse_file/1), |
| 1116 | - CertFile = parse(Args, "--certfile", fun parse_file/1), |
| 1117 | - KeyFile = parse(Args, "--keyfile", fun parse_file/1), |
| 1118 | - FilterPred = fun({_, V}) -> V =/= ?undef end, |
| 1119 | - SslOpt = |
| 1120 | - case CaCertFile of |
| 1121 | - ?undef -> |
| 1122 | - SslBool; |
| 1123 | - _ -> |
| 1124 | - Files = |
| 1125 | - [{cacertfile, CaCertFile}, |
| 1126 | - {certfile, CertFile}, |
| 1127 | - {keyfile, KeyFile}], |
| 1128 | - lists:filter(FilterPred, Files) |
| 1129 | - end, |
| 1130 | - SaslPlain = parse(Args, "--sasl-plain", fun parse_file/1), |
| 1131 | - SaslScram256 = parse(Args, "--scram256", fun parse_file/1), |
| 1132 | - SaslScram512 = parse(Args, "--scram512", fun parse_file/1), |
| 1133 | - SaslOpts0 = [ {scram_sha_512, SaslScram512} |
| 1134 | - , {scram_sha_256, SaslScram256} |
| 1135 | - , {plain, SaslPlain} |
| 1136 | - ], |
| 1137 | - SaslOpts = case lists:filter(FilterPred, SaslOpts0) of |
| 1138 | - [] -> []; |
| 1139 | - [H | _] -> [{sasl, H}] |
| 1140 | - end, |
| 1141 | - lists:filter(FilterPred, [{ssl, SslOpt} | SaslOpts]). |
| 1142 | - |
| 1143 | - parse_boolean(true) -> true; |
| 1144 | - parse_boolean(false) -> false; |
| 1145 | - parse_boolean("true") -> true; |
| 1146 | - parse_boolean("false") -> false; |
| 1147 | - parse_boolean(?undef) -> ?undef. |
| 1148 | - |
| 1149 | - parse_cg_ids("") -> []; |
| 1150 | - parse_cg_ids("all") -> all; |
| 1151 | - parse_cg_ids(Str) -> [bin(I) || I <- string:tokens(Str, ",")]. |
| 1152 | - |
| 1153 | - parse_file(?undef) -> |
| 1154 | - ?undef; |
| 1155 | - parse_file(Path) -> |
| 1156 | - case filelib:is_regular(Path) of |
| 1157 | - true -> Path; |
| 1158 | - false -> erlang:throw(bin(["bad file ", Path])) |
| 1159 | - end. |
| 1160 | - |
| 1161 | - parse(Args, OptName, ParseFun) -> |
| 1162 | - case lists:keyfind(OptName, 1, Args) of |
| 1163 | - {_, Arg} -> |
| 1164 | - try |
| 1165 | - ParseFun(Arg) |
| 1166 | - catch |
| 1167 | - C : E ?BIND_STACKTRACE(Stack) -> |
| 1168 | - ?GET_STACKTRACE(Stack), |
| 1169 | - verbose("~p:~p\n~p\n", [C, E, Stack]), |
| 1170 | - Reason = |
| 1171 | - case Arg of |
| 1172 | - ?undef -> ["Missing option ", OptName]; |
| 1173 | - _ -> ["Failed to parse ", OptName, ": ", Arg] |
| 1174 | - end, |
| 1175 | - erlang:throw(bin(Reason)) |
| 1176 | - end; |
| 1177 | - false -> |
| 1178 | - Reason = [OptName, " is missing"], |
| 1179 | - erlang:throw(bin(Reason)) |
| 1180 | - end. |
| 1181 | - |
| 1182 | - print_version() -> |
| 1183 | - _ = application:load(brod), |
| 1184 | - {_, _, V} = lists:keyfind(brod, 1, application:loaded_applications()), |
| 1185 | - print([V, "\n"]). |
| 1186 | - |
| 1187 | - print(IoData) -> io:put_chars(stdio(), IoData). |
| 1188 | - |
| 1189 | - print(Fmt, Args) -> io:put_chars(stdio(), io_lib:format(Fmt, Args)). |
| 1190 | - |
| 1191 | - logerr(IoData) -> io:put_chars(stderr(), ["*** ", IoData]). |
| 1192 | - |
| 1193 | - logerr(Fmt, Args) -> |
| 1194 | - io:put_chars(stderr(), io_lib:format("*** " ++ Fmt, Args)). |
| 1195 | - |
| 1196 | - verbose(Fmt, Args) -> |
| 1197 | - case erlang:get(brod_cli_log_level) >= ?LOG_LEVEL_VERBOSE of |
| 1198 | - true -> logerr("[verbo]: " ++ Fmt, Args); |
| 1199 | - false -> ok |
| 1200 | - end. |
| 1201 | - |
| 1202 | - debug(Fmt, Args) -> |
| 1203 | - case erlang:get(brod_cli_log_level) >= ?LOG_LEVEL_DEBUG of |
| 1204 | - true -> logerr("[debug]: " ++ Fmt, Args); |
| 1205 | - false -> ok |
| 1206 | - end. |
| 1207 | - |
| 1208 | - stdio() -> |
| 1209 | - case get(redirect_stdio) of |
| 1210 | - undefined -> user; |
| 1211 | - Other -> Other |
| 1212 | - end. |
| 1213 | - |
| 1214 | - stderr() -> |
| 1215 | - case get(redirect_stderr) of |
| 1216 | - undefined -> standard_error; |
| 1217 | - Other -> Other |
| 1218 | - end. |
| 1219 | - |
| 1220 | - int(Str) -> list_to_integer(trim(Str)). |
| 1221 | - |
| 1222 | - trim_h([$\s | T]) -> trim_h(T); |
| 1223 | - trim_h(X) -> X. |
| 1224 | - |
| 1225 | - trim(Str) -> trim_h(lists:reverse(trim_h(lists:reverse(Str)))). |
| 1226 | - |
| 1227 | - bin(IoData) -> iolist_to_binary(IoData). |
| 1228 | - |
| 1229 | - parse_brokers(HostsStr) -> |
| 1230 | - F = fun(HostPortStr) -> |
| 1231 | - Pair = string:tokens(HostPortStr, ":"), |
| 1232 | - case Pair of |
| 1233 | - [Host, PortStr] -> {Host, list_to_integer(PortStr)}; |
| 1234 | - [Host] -> {Host, 9092} |
| 1235 | - end |
| 1236 | - end, |
| 1237 | - shuffle(lists:map(F, string:tokens(HostsStr, ","))). |
| 1238 | - |
| 1239 | - %% Parse code paths. |
| 1240 | - parse_paths(?undef) -> []; |
| 1241 | - parse_paths(Str) -> string:tokens(Str, ","). |
| 1242 | - |
| 1243 | - %% Randomize the order. |
| 1244 | - shuffle(L) -> |
| 1245 | - RandList = lists:map(fun(_) -> element(3, os:timestamp()) end, L), |
| 1246 | - {_, SortedL} = lists:unzip(lists:keysort(1, lists:zip(RandList, L))), |
| 1247 | - SortedL. |
| 1248 | - |
| 1249 | - -spec kf(kpro:field_name(), kpro:struct()) -> kpro:field_value(). |
| 1250 | - kf(FieldName, Struct) -> kpro:find(FieldName, Struct). |
| 1251 | - |
| 1252 | - -spec kf(kpro:field_name(), kpro:struct(), kpro:field_value()) -> |
| 1253 | - kpro:field_value(). |
| 1254 | - kf(FieldName, Struct, Default) -> |
| 1255 | - kpro:find(FieldName, Struct, Default). |
| 1256 | - |
| 1257 | - start_client(BootstrapEndpoints, ClientConfig) -> |
| 1258 | - {ok, _} = brod_client:start_link(BootstrapEndpoints, ?CLIENT, ClientConfig), |
| 1259 | - ok. |
| 1260 | - |
| 1261 | - -spec throw_bin(string(), [term()]) -> no_return(). |
| 1262 | - throw_bin(Fmt, Args) -> |
| 1263 | - erlang:throw(bin(io_lib:format(Fmt, Args))). |
| 1264 | - |
| 1265 | - get_kafka_version() -> |
| 1266 | - case os:getenv("KAFKA_VERSION") of |
| 1267 | - false -> |
| 1268 | - ?LATEST_KAFKA_VERSION; |
| 1269 | - Vsn -> |
| 1270 | - [Major, Minor | _] = string:tokens(Vsn, "."), |
| 1271 | - {list_to_integer(Major), list_to_integer(Minor)} |
| 1272 | - end. |
| 1273 | - |
| 1274 | - -endif. |
| 1275 | - |
| 1276 | - %%%_* Emacs ==================================================================== |
| 1277 | - %%% Local Variables: |
| 1278 | - %%% allout-layout: t |
| 1279 | - %%% erlang-indent-level: 2 |
| 1280 | - %%% End: |
Loading more files…