Packages
brod
3.16.5
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
13
files changed
+781
additions
-261
deletions
| @@ -1,4 +1,3 @@ | |
| 1 | -  |
| 2 1 | # NOTICE |
| 3 2 | |
| 4 3 | This product includes software developed by |
| @@ -6,45 +5,47 @@ This product includes software developed by | |
| 6 5 | |
| 7 6 | # Brod - Apache Kafka Client for Erlang/Elixir |
| 8 7 | |
| 8 | +  |
| 9 | + |
| 9 10 | Brod is an Erlang implementation of the Apache Kafka protocol, providing support for both producers and consumers. |
| 10 11 | |
| 11 12 | Why "brod"? [http://en.wikipedia.org/wiki/Max_Brod](http://en.wikipedia.org/wiki/Max_Brod) |
| 12 13 | |
| 13 | - # Features |
| 14 | + ## Features |
| 14 15 | |
| 15 | - * Supports Apache Kafka v0.8+ |
| 16 | - * Robust producer implementation supporting in-flight requests and asynchronous acknowledgements |
| 17 | - * Both consumer and producer handle leader re-election and other cluster disturbances internally |
| 18 | - * Opens max 1 tcp connection to a broker per `brod_client`, one can create more clients if needed |
| 19 | - * Producer: will start to batch automatically when number of unacknowledged (in flight) requests exceeds configurable maximum |
| 20 | - * Producer: will try to re-send buffered messages on common errors like "Not a leader for partition", errors are resolved automatically by refreshing metadata |
| 21 | - * Simple consumer: The poller, has a configurable "prefetch count" - it will continue sending fetch requests as long as total number of unprocessed messages (not message-sets) is less than "prefetch count" |
| 22 | - * Group subscriber: Support for consumer groups with options to have Kafka as offset storage or a custom one |
| 23 | - * Topic subscriber: Subscribe on messages from all or selected topic partitions without using consumer groups |
| 24 | - * Pick latest supported version when sending requests to kafka. |
| 25 | - * Direct APIs for message send/fetch and cluster inspection/management without having to start clients/producers/consumers. |
| 26 | - * A escriptized command-line tool for message send/fetch and cluster inspection/management. |
| 27 | - * Configurable compression library. `snappy` compression is supported by default. |
| 16 | + - Supports Apache Kafka v0.8+ |
| 17 | + - Robust producer implementation supporting in-flight requests and asynchronous acknowledgements |
| 18 | + - Both consumer and producer handle leader re-election and other cluster disturbances internally |
| 19 | + - Opens max 1 tcp connection to a broker per `brod_client`, one can create more clients if needed |
| 20 | + - Producer: will start to batch automatically when number of unacknowledged (in flight) requests exceeds configurable maximum |
| 21 | + - Producer: will try to re-send buffered messages on common errors like "Not a leader for partition", errors are resolved automatically by refreshing metadata |
| 22 | + - Simple consumer: The poller, has a configurable "prefetch count" - it will continue sending fetch requests as long as total number of unprocessed messages (not message-sets) is less than "prefetch count" |
| 23 | + - Group subscriber: Support for consumer groups with options to have Kafka as offset storage or a custom one |
| 24 | + - Topic subscriber: Subscribe on messages from all or selected topic partitions without using consumer groups |
| 25 | + - Pick latest supported version when sending requests to kafka. |
| 26 | + - Direct APIs for message send/fetch and cluster inspection/management without having to start clients/producers/consumers. |
| 27 | + - A escriptized command-line tool for message send/fetch and cluster inspection/management. |
| 28 | + - Configurable compression library. `snappy` compression is supported by default. |
| 28 29 | For more compression options, see [kafka_protocol/README](https://github.com/kafka4beam/kafka_protocol/blob/master/README.md#compression-support) |
| 29 30 | |
| 30 | - # Building and testing |
| 31 | + ## Building and testing |
| 31 32 | |
| 32 33 | NOTE: Min Erlang/OTP version 22 |
| 33 34 | |
| 34 | - ``` |
| 35 | + ```sh |
| 35 36 | make compile |
| 36 37 | make test-env t # requires docker-compose in place |
| 37 38 | ``` |
| 38 39 | |
| 39 | - # Working With Kafka 0.9.x or Earlier |
| 40 | + ## Working With Kafka 0.9.x or Earlier |
| 40 41 | |
| 41 42 | Make sure `{query_api_versions, false}` exists in client config. |
| 42 43 | This is because `ApiVersionRequest` was introduced in kafka 0.10, |
| 43 44 | sending such request to older version brokers will cause connection failure. |
| 44 45 | |
| 45 | - e.g. in sys.config: |
| 46 | + e.g. in `sys.config`: |
| 46 47 | |
| 47 | - ``` |
| 48 | + ```erlang |
| 48 49 | [{brod, |
| 49 50 | [ { clients |
| 50 51 | , [ { brod_client_1 %% registered name |
| @@ -53,7 +54,7 @@ e.g. in sys.config: | |
| 53 54 | ]}]}]}] |
| 54 55 | ``` |
| 55 56 | |
| 56 | - # Quick Demo |
| 57 | + ## Quick Demo |
| 57 58 | |
| 58 59 | Assuming kafka is running at `localhost:9092` and there is a topic named `test-topic`. |
| 59 60 | |
| @@ -97,25 +98,28 @@ Produced to partition 0 at base-offset 406 | |
| 97 98 | headers = []} |
| 98 99 | ``` |
| 99 100 | |
| 100 | - # Overview |
| 101 | + ## Overview |
| 101 102 | |
| 102 103 | Brod supervision (and process link) tree. |
| 103 104 | |
| 104 | -  |
| 105 | +  |
| 105 106 | |
| 106 | - # Clients |
| 107 | + ## Clients |
| 107 108 | |
| 108 109 | A `brod_client` in brod is a `gen_server` responsible for establishing and |
| 109 110 | maintaining tcp sockets connecting to kafka brokers. |
| 110 111 | It also manages per-topic-partition producer and consumer processes under |
| 111 112 | two-level supervision trees. |
| 112 113 | |
| 113 | - ## Start clients by default |
| 114 | + To use producers or consumers, you have to start at least one cient that |
| 115 | + will manage them. |
| 114 116 | |
| 115 | - You may include client configs in sys.config have them started by default |
| 117 | + ### Start clients by default |
| 118 | + |
| 119 | + You may include client configs in `sys.config` have them started by default |
| 116 120 | (by application controller) |
| 117 121 | |
| 118 | - Example of configuration (for sys.config): |
| 122 | + Example of configuration (for `sys.config`): |
| 119 123 | |
| 120 124 | ```erlang |
| 121 125 | [{brod, |
| @@ -133,7 +137,20 @@ Example of configuration (for sys.config): | |
| 133 137 | }] |
| 134 138 | ``` |
| 135 139 | |
| 136 | - ## Start brod client on demand |
| 140 | + Example of configuration in Elixir (for `config/dev.exs` or `config/prod.exs`, etc.): |
| 141 | + |
| 142 | + ```elixir |
| 143 | + config :brod, |
| 144 | + clients: [ |
| 145 | + # :brod_client_1 is the registered name of the client |
| 146 | + brod_client_1: [ |
| 147 | + endpoints: [{"localhost", 9092}], |
| 148 | + reconnect_cool_down_seconds: 10 |
| 149 | + ] |
| 150 | + ] |
| 151 | + ``` |
| 152 | + |
| 153 | + ### Start brod client on demand |
| 137 154 | |
| 138 155 | You may also call `brod:start_client/1,2,3` to start a client on demand, |
| 139 156 | which will be added to brod supervision tree. |
| @@ -151,18 +168,27 @@ ExtraSockOpts = [{sndbuf, 1024*1024}], | |
| 151 168 | ok = brod:start_client([{"localhost", 9092}], brod_client_1, [{extra_sock_opts, ExtraSockOpts}]). |
| 152 169 | ``` |
| 153 170 | |
| 154 | - # Producers |
| 171 | + ## Producers |
| 155 172 | |
| 156 | - ## Auto start producer with default producer config |
| 173 | + A `brod_producer` is a `gen_server` that is responsible for producing messages to a given |
| 174 | + partition of a given topic. |
| 157 175 | |
| 158 | - Put below configs to client config in sys.config or app env: |
| 176 | + Producers may be started either manually or automatically in the moment you call `brod:produce` |
| 177 | + but did not call `brod:start_producer` beforehand. |
| 178 | + |
| 179 | + ### Auto start producer with default producer config |
| 180 | + |
| 181 | + Put below configs to client config in `sys.config` or app env if you start client statically: |
| 159 182 | |
| 160 183 | ```erlang |
| 161 184 | {auto_start_producers, true} |
| 162 185 | {default_producer_config, []} |
| 163 186 | ``` |
| 164 187 | |
| 165 | - ## Start a Producer on Demand |
| 188 | + Or pass the `{auto_start_producers, true}` option to `brod:start_client` if you start the client |
| 189 | + dynamically. |
| 190 | + |
| 191 | + ### Start a Producer on Demand |
| 166 192 | |
| 167 193 | ```erlang |
| 168 194 | brod:start_producer(_Client = brod_client_1, |
| @@ -170,14 +196,15 @@ brod:start_producer(_Client = brod_client_1, | |
| 170 196 | _ProducerConfig = []). |
| 171 197 | ``` |
| 172 198 | |
| 173 | - ## Supported Message Input Format |
| 199 | + ### Supported Message Input Format |
| 174 200 | |
| 175 201 | Brod supports below produce APIs: |
| 176 202 | |
| 177 | - - `brod:produce`: Async produce with ack message sent back to caller. |
| 178 | - - `brod:produce_cb`: Async produce with a callback evaluated when ack is received. |
| 179 | - - `brod:produce_sync`: Sync produce returns `ok`. |
| 180 | - - `brod:produce_sync_offset`: Sync produce returns `{ok, BaseOffset}`. |
| 203 | + - [`brod:produce`](https://hexdocs.pm/brod/brod.html#produce/5): Async produce with ack message sent back to caller. |
| 204 | + - [`brod:produce_cb`](https://hexdocs.pm/brod/brod.html#produce_cb/6): Async produce with a callback evaluated when ack is received. |
| 205 | + - [`brod:produce_sync`](https://hexdocs.pm/brod/brod.html#produce_sync/5): Sync produce that returns `ok`. |
| 206 | + - [`brod:produce_sync_offset`](https://hexdocs.pm/brod/brod.html#produce_sync_offset/5): Sync produce that returns `{ok, BaseOffset}`. |
| 207 | + - [`brod:produce_no_ack`](https://hexdocs.pm/brod/brod.html#produce_no_ack/5): Async produce without backpressure (use with care!). |
| 181 208 | |
| 182 209 | The `Value` arg in these APIs can be: |
| 183 210 | |
| @@ -191,10 +218,10 @@ The `Value` arg in these APIs can be: | |
| 191 218 | When `Value` is a batch, the `Key` argument is only used as partitioner input and all messages are written on the same partition. |
| 192 219 | All messages are unified into a batch format of below spec: |
| 193 220 | `[#{key => K, value => V, ts => T, headers => [{_, _}]}]`. |
| 194 | - `ts` field is dropped for kafka prior to version `0.10` (produce API version 0, magic version 0) |
| 195 | - `headers` field is dropped for kafka prior to version `0.11` (produce API version 0-2, magic version 0-1) |
| 221 | + `ts` field is dropped for kafka prior to version `0.10` (produce API version 0, magic version 0). |
| 222 | + `headers` field is dropped for kafka prior to version `0.11` (produce API version 0-2, magic version 0-1). |
| 196 223 | |
| 197 | - ## Synchronized Produce API |
| 224 | + ### Synchronized Produce API |
| 198 225 | |
| 199 226 | ```erlang |
| 200 227 | brod:produce_sync(_Client = brod_client_1, |
| @@ -216,7 +243,7 @@ Or block calling process until Kafka confirmed the message: | |
| 216 243 | brod:sync_produce_request(CallRef). |
| 217 244 | ``` |
| 218 245 | |
| 219 | - ## Produce One Message and Receive Its Offset in Kafka |
| 246 | + ### Produce One Message and Receive Its Offset in Kafka |
| 220 247 | |
| 221 248 | ```erlang |
| 222 249 | Client = brod_client_1, |
| @@ -224,7 +251,7 @@ Topic = <<"brod-test-topic-1">>, | |
| 224 251 | {ok, Offset} = brod:produce_sync_offset(Client, Topic, 0, <<>>, <<"value">>). |
| 225 252 | ``` |
| 226 253 | |
| 227 | - ## Produce with Random Partitioner |
| 254 | + ### Produce with Random Partitioner |
| 228 255 | |
| 229 256 | ```erlang |
| 230 257 | Client = brod_client_1, |
| @@ -232,7 +259,7 @@ Topic = <<"brod-test-topic-1">>, | |
| 232 259 | ok = brod:produce_sync(Client, Topic, random, Key, Value). |
| 233 260 | ``` |
| 234 261 | |
| 235 | - ## Produce a Batch |
| 262 | + ### Produce a Batch |
| 236 263 | |
| 237 264 | ```erlang |
| 238 265 | brod:produce(_Client = brod_client_1, |
| @@ -244,7 +271,7 @@ brod:produce(_Client = brod_client_1, | |
| 244 271 | ]). |
| 245 272 | ``` |
| 246 273 | |
| 247 | - ## Handle Acks from Kafka as Messages |
| 274 | + ### Handle Acks from Kafka as Messages |
| 248 275 | |
| 249 276 | For async produce APIs `brod:produce/3` and `brod:produce/5`, |
| 250 277 | the caller should expect a message of below pattern for each produce call. |
| @@ -254,6 +281,7 @@ the caller should expect a message of below pattern for each produce call. | |
| 254 281 | , result = brod_produce_req_acked |
| 255 282 | } |
| 256 283 | ``` |
| 284 | + |
| 257 285 | Add `-include_lib("brod/include/brod.hrl").` to use the record. |
| 258 286 | |
| 259 287 | In case the `brod:produce` caller is a process like `gen_server` which |
| @@ -270,7 +298,7 @@ i.e. if the caller is producing to two or more partitions, | |
| 270 298 | it may receive replies ordered differently than in which order |
| 271 299 | `brod:produce` API was called. |
| 272 300 | |
| 273 | - ## Handle Acks from Kafka in Callback Function |
| 301 | + ### Handle Acks from Kafka in Callback Function |
| 274 302 | |
| 275 303 | Async APIs `brod:produce_cb/4` and `brod:produce_cb/6` allow callers to |
| 276 304 | provided a callback function to handle acknowledgements from kafka. |
| @@ -278,7 +306,7 @@ In this case, the caller may want to monitor the producer process because | |
| 278 306 | then they know that the callbacks will not be evaluated if the producer is 'DOWN', |
| 279 307 | and there is perhaps a need for retry. |
| 280 308 | |
| 281 | - # Consumers |
| 309 | + ## Consumers |
| 282 310 | |
| 283 311 | Kafka consumers work in poll mode. In brod, `brod_consumer` is the poller, |
| 284 312 | which is constantly asking for more data from the kafka node which is a leader |
| @@ -297,35 +325,48 @@ per-partition subscriber. | |
| 297 325 | Below diagrams illustrate 3 examples of how subscriber processes may work |
| 298 326 | with `brod_consumer`. |
| 299 327 | |
| 300 | - ## Partition subscriber |
| 301 | -  |
| 328 | + ### Partition subscriber |
| 329 | + |
| 330 | +  |
| 302 331 | |
| 303 332 | This gives the best flexibility as the per-partition subscribers work |
| 304 | - directly with per-partition pollers. |
| 333 | + directly with per-partition pollers (`brod_consumer`s). |
| 305 334 | |
| 306 335 | The messages are delivered to subscribers in message sets (batches), |
| 307 336 | not individual messages, (however the subscribers are allowed to |
| 308 337 | ack individual offsets). |
| 309 338 | |
| 310 | - ## Topic subscriber (`brod_topic_subscriber`) |
| 311 | -  |
| 339 | + Example: |
| 340 | + ```erlang |
| 341 | + ok = brod:start_client([{"localhost", 9092}], my_client). % one client per application is enough |
| 342 | + ok = brod:start_consumer(my_client, <<"my_topic">>, []). |
| 343 | + |
| 344 | + % Now in a separate process for each partition of my_topic call: |
| 345 | + {ok, ConsumerPid} = brod:subscribe(my_client, self(), <<"my_topic">>, Partition, []). |
| 346 | + % The process should now receive messages sets as regular messages |
| 347 | + ``` |
| 348 | + |
| 349 | + ### Topic subscriber (`brod_topic_subscriber`) |
| 350 | + |
| 351 | +  |
| 312 352 | |
| 313 353 | A topic subscriber provides the easiest way to receive and process messages from |
| 314 354 | ALL partitions of a given topic. See |
| 315 | - [brod_demo_cg_collector](test/brod_demo_cg_collector.erl) and |
| 316 | - [brod_demo_topic_subscriber](test/brod_demo_topic_subscriber.erl) for example. |
| 355 | + [brod_demo_cg_collector](https://github.com/kafka4beam/brod/blob/master/test/brod_demo_cg_collector.erl) and |
| 356 | + [brod_demo_topic_subscriber](https://github.com/kafka4beam/brod/blob/master/test/brod_demo_topic_subscriber.erl) for example. |
| 317 357 | |
| 318 358 | Users may choose to implement the `brod_topic_subscriber` behaviour callbacks |
| 319 359 | in a module, or simply provide an anonymous callback function to have the |
| 320 360 | individual messages processed. |
| 321 361 | |
| 322 | - ## Group subscriber (`brod_group_subscriber`) |
| 323 | -  |
| 362 | + ### Group subscriber (`brod_group_subscriber`) |
| 363 | + |
| 364 | +  |
| 324 365 | |
| 325 366 | Similar to topic subscriber, the `brod_group_subscriber` behaviour callbacks are |
| 326 367 | to be implemented to process individual messages. See |
| 327 | - [brod_demo_group_subscriber_koc](test/brod_demo_group_subscriber_koc.erl) and |
| 328 | - [brod_demo_group_subscriber_loc](test/brod_demo_group_subscriber_loc.erl) for |
| 368 | + [brod_demo_group_subscriber_koc](https://github.com/kafka4beam/brod/blob/master/test/brod_demo_group_subscriber_koc.erl) and |
| 369 | + [brod_demo_group_subscriber_loc](https://github.com/kafka4beam/brod/blob/master/test/brod_demo_group_subscriber_loc.erl) for |
| 329 370 | example. |
| 330 371 | |
| 331 372 | A group subscriber is started by giving a set of topics, some |
| @@ -338,7 +379,7 @@ for `brod_group_coordinator`) for a different group subscriber (e.g. spawn | |
| 338 379 | one subscriber per partition), see [brucke](https://github.com/klarna/brucke) |
| 339 380 | for example. |
| 340 381 | |
| 341 | - ### Example of group consumer which commits offsets to Kafka |
| 382 | + #### Example of group consumer which commits offsets to Kafka |
| 342 383 | |
| 343 384 | ```erlang |
| 344 385 | -module(my_subscriber). |
| @@ -379,7 +420,7 @@ start(ClientId) -> | |
| 379 420 | _CallbackInitArg = []). |
| 380 421 | ``` |
| 381 422 | |
| 382 | - # Authentication support |
| 423 | + ## Authentication support |
| 383 424 | |
| 384 425 | brod supports SASL `PLAIN`, `SCRAM-SHA-256` and `SCRAM-SHA-512` authentication mechanisms out of the box. |
| 385 426 | To use it, add `{sasl, {Mechanism, Username, Password}}` or `{sasl, {Mechanism, File}}` to client config. |
| @@ -398,14 +439,14 @@ auth(Host :: string(), Sock :: gen_tcp:socket() | ssl:sslsocket(), | |
| 398 439 | ``` |
| 399 440 | |
| 400 441 | If authentication is successful - callback function should return an atom `ok`, otherwise - error tuple with reason description. |
| 401 | - For example, you can use `brod_gssapi` plugin (https://github.com/kafka4beam/brod_gssapi) for SASL GSSAPI authentication. |
| 442 | + For example, you can use [`brod_gssapi` plugin](https://github.com/kafka4beam/brod_gssapi) for SASL GSSAPI authentication. |
| 402 443 | To use it - add it as dependency to your top level project that uses brod. |
| 403 444 | Then add `{sasl, {callback, brod_gssapi, {gssapi, Keytab, Principal}}}` to client config. |
| 404 445 | Keytab should be the keytab file path, and Principal should be a byte-list or binary string. |
| 405 446 | |
| 406 | - See also: https://github.com/klarna/brod/wiki/SASL-gssapi-(kerberos)-authentication |
| 447 | + See also: <https://github.com/klarna/brod/wiki/SASL-gssapi-(kerberos)-authentication> |
| 407 448 | |
| 408 | - # Other API to play with/inspect kafka |
| 449 | + ## Other API to play with/inspect kafka |
| 409 450 | |
| 410 451 | These functions open a connection to kafka cluster, send a request, |
| 411 452 | await response and then close the connection. |
| @@ -433,7 +474,7 @@ brod:delete_topics(Hosts, [Topic], Timeout). | |
| 433 474 | |
| 434 475 | Caution the above delete_topics can fail if you do not have `delete.topic.enable` set to true in your kafka config |
| 435 476 | |
| 436 | - # brod-cli: A command line tool to interact with Kafka |
| 477 | + ## brod-cli: A command line tool to interact with Kafka |
| 437 478 | |
| 438 479 | This will build a self-contained binary with brod application |
| 439 480 | |
| @@ -446,23 +487,24 @@ Disclaimer: This script is NOT designed for use cases where fault-tolerance is a | |
| 446 487 | As it may crash when e.g. kafka cluster is temporarily unreachable, |
| 447 488 | or (for fetch command) when the partition leader migrates to another broker in the cluster. |
| 448 489 | |
| 449 | - ## brod-cli examples (with `alias brod=_build/brod_cli/rel/brod/bin/brod`): |
| 490 | + ### brod-cli examples (with `alias brod=_build/brod_cli/rel/brod/bin/brod`): |
| 450 491 | |
| 451 | - ### Fetch and print metadata |
| 452 | - ``` |
| 492 | + #### Fetch and print metadata |
| 493 | + |
| 494 | + ```sh |
| 453 495 | brod meta -b localhost |
| 454 496 | ``` |
| 455 497 | |
| 456 | - ### Produce a Message |
| 498 | + #### Produce a Message |
| 457 499 | |
| 458 | - ``` |
| 500 | + ```sh |
| 459 501 | brod send -b localhost -t test-topic -p 0 -k "key" -v "value" |
| 460 502 | |
| 461 503 | ``` |
| 462 504 | |
| 463 | - ### Fetch a Message |
| 505 | + #### Fetch a Message |
| 464 506 | |
| 465 | - ``` |
| 507 | + ```sh |
| 466 508 | brod fetch -b localhost -t test-topic -p 0 --fmt 'io:format("offset=~p, ts=~p, key=~s, value=~s\n", [Offset, Ts, Key, Value])' |
| 467 509 | ``` |
| 468 510 | |
| @@ -474,22 +516,23 @@ Bound variables to be used in `--fmt` expression: | |
| 474 516 | - `TsType`: Timestamp type either `create` or `append` |
| 475 517 | - `Ts`: Timestamp, `-1` as no value |
| 476 518 | |
| 477 | - ### Stream Messages to Kafka |
| 519 | + #### Stream Messages to Kafka |
| 478 520 | |
| 479 521 | Send `README.md` to kafka one line per kafka message |
| 480 | - ``` |
| 522 | + |
| 523 | + ```sh |
| 481 524 | brod pipe -b localhost:9092 -t test-topic -p 0 -s @./README.md |
| 482 525 | ``` |
| 483 526 | |
| 484 | - ### Resolve Offset |
| 527 | + #### Resolve Offset |
| 485 528 | |
| 486 | - ``` |
| 529 | + ```sh |
| 487 530 | brod offset -b localhost:9092 -t test-topic -p 0 |
| 488 531 | ``` |
| 489 532 | |
| 490 | - ### List or Describe Groups |
| 533 | + #### List or Describe Groups |
| 491 534 | |
| 492 | - ``` |
| 535 | + ```sh |
| 493 536 | # List all groups |
| 494 537 | brod groups -b localhost:9092 |
| 495 538 | |
| @@ -497,9 +540,9 @@ brod groups -b localhost:9092 | |
| 497 540 | brod groups -b localhost:9092 --ids group-1,group-2 |
| 498 541 | ``` |
| 499 542 | |
| 500 | - ### Display Committed Offsets |
| 543 | + #### Display Committed Offsets |
| 501 544 | |
| 502 | - ``` |
| 545 | + ```sh |
| 503 546 | # all topics |
| 504 547 | brod commits -b localhost:9092 --id the-group-id --describe |
| 505 548 | |
| @@ -507,11 +550,11 @@ brod commits -b localhost:9092 --id the-group-id --describe | |
| 507 550 | brod commits -b localhost:9092 --id the-group-id --describe --topic topic-name |
| 508 551 | ``` |
| 509 552 | |
| 510 | - ### Commit Offsets |
| 553 | + #### Commit Offsets |
| 511 554 | |
| 512 555 | NOTE: This feature is designed for force overwriting commits, not for regular use of offset commit. |
| 513 556 | |
| 514 | - ``` |
| 557 | + ```sh |
| 515 558 | # Commit 'latest' offsets of all partitions with 2 days retention |
| 516 559 | brod commits -b localhost:9092 --id the-group-id --topic topic-name --offsets latest --retention 2d |
| 517 560 | |
| @@ -525,7 +568,7 @@ brod commits -b localhost:9092 --id the-group-id --topic topic-name --offsets la | |
| 525 568 | brod commits -b localhost:9092 -i the-group-id -t topic-name -o "0:10000" --protocol range |
| 526 569 | ``` |
| 527 570 | |
| 528 | - ## TODOs |
| 571 | + ### TODOs |
| 529 572 | |
| 530 | - * Support scram-sasl in brod-cli |
| 531 | - * Transactional produce APIs |
| 573 | + - Support scram-sasl in brod-cli |
| 574 | + - Transactional produce APIs |
| @@ -31,4 +31,4 @@ | |
| 31 31 | [{<<"app">>,<<"supervisor3">>}, |
| 32 32 | {<<"optional">>,false}, |
| 33 33 | {<<"requirement">>,<<"1.1.11">>}]}]}. |
| 34 | - {<<"version">>,<<"3.16.4">>}. |
| 34 | + {<<"version">>,<<"3.16.5">>}. |
| @@ -36,7 +36,7 @@ | |
| 36 36 | ]}. |
| 37 37 | {ex_doc, |
| 38 38 | [ {extras, |
| 39 | - [ {"changelog.md", #{title => "Changelog"}} |
| 39 | + [ {"CHANGELOG.md", #{title => "Changelog"}} |
| 40 40 | , {"README.md", #{title => "Overview"}} |
| 41 41 | , {"LICENSE", #{title => "License"}} |
| 42 42 | , "guides/examples/elixir/Publisher.md" |
| @@ -50,6 +50,7 @@ | |
| 50 50 | , {main, "README.md"} |
| 51 51 | , {homepage_url, "https://hexdocs.pm/brod"} |
| 52 52 | , {source_url, "https://github.com/kafka4beam/brod"} |
| 53 | + , {source_ref, "master"} |
| 53 54 | , {api_reference, false} |
| 54 55 | ]}. |
| 55 56 | {hex, [{doc, ex_doc}]}. |
| @@ -1,6 +1,6 @@ | |
| 1 1 | {application,brod, |
| 2 2 | [{description,"Apache Kafka Erlang client library"}, |
| 3 | - {vsn,"3.16.4"}, |
| 3 | + {vsn,"3.16.5"}, |
| 4 4 | {registered,[]}, |
| 5 5 | {applications,[kernel,stdlib,kafka_protocol,supervisor3, |
| 6 6 | snappyer]}, |
| @@ -26,6 +26,7 @@ | |
| 26 26 | |
| 27 27 | %% Client API |
| 28 28 | -export([ get_partitions_count/2 |
| 29 | + , get_partitions_count_safe/2 |
| 29 30 | , start_client/1 |
| 30 31 | , start_client/2 |
| 31 32 | , start_client/3 |
| @@ -133,8 +134,6 @@ | |
| 133 134 | , connection/0 |
| 134 135 | , conn_config/0 |
| 135 136 | , consumer_config/0 |
| 136 | - , consumer_option/0 |
| 137 | - , consumer_options/0 |
| 138 137 | , endpoint/0 |
| 139 138 | , error_code/0 |
| 140 139 | , fetch_opts/0 |
| @@ -183,7 +182,7 @@ | |
| 183 182 | -type topic_config() :: kpro:struct(). |
| 184 183 | -type partition() :: kpro:partition(). |
| 185 184 | -type topic_partition() :: {topic(), partition()}. |
| 186 | - -type offset() :: kpro:offset(). |
| 185 | + -type offset() :: kpro:offset(). %% Physical offset (an integer) |
| 187 186 | -type key() :: undefined %% no key, transformed to <<>> |
| 188 187 | | binary(). |
| 189 188 | -type value() :: undefined %% no value, transformed to <<>> |
| @@ -197,46 +196,60 @@ | |
| 197 196 | -type msg_input() :: kpro:msg_input(). |
| 198 197 | -type batch_input() :: [msg_input()]. |
| 199 198 | |
| 200 | - -type msg_ts() :: kpro:msg_ts(). |
| 199 | + -type msg_ts() :: kpro:msg_ts(). %% Unix time in milliseconds |
| 201 200 | -type client_id() :: atom(). |
| 202 201 | -type client() :: client_id() | pid(). |
| 203 202 | -type client_config() :: brod_client:config(). |
| 204 203 | -type bootstrap() :: [endpoint()] %% default client config |
| 205 204 | | {[endpoint()], client_config()}. |
| 206 | - -type offset_time() :: offset() |
| 205 | + -type offset_time() :: msg_ts() |
| 207 206 | | ?OFFSET_EARLIEST |
| 208 207 | | ?OFFSET_LATEST. |
| 209 | - -type message() :: kpro:message(). |
| 208 | + -type message() :: kpro:message(). %% A record with offset, key, value, ts_type, ts, and headers. |
| 210 209 | -type message_set() :: #kafka_message_set{}. |
| 210 | + %% A record with topic, partition, high_wm_offset (max offset of the partition), and messages. |
| 211 | + %% |
| 212 | + %% See <a href="https://github.com/kafka4beam/brod/blob/master/include/brod.hrl#L26"> |
| 213 | + %% the definition</a> for more information. |
| 211 214 | -type error_code() :: kpro:error_code(). |
| 212 215 | |
| 213 216 | %% producers |
| 214 217 | -type produce_reply() :: #brod_produce_reply{}. |
| 218 | + %% A record with call_ref, base_offset, and result. |
| 219 | + %% |
| 220 | + %% See the <a href="https://github.com/kafka4beam/brod/blob/master/include/brod.hrl#L49"> |
| 221 | + %% the definition</a> for more information. |
| 215 222 | -type producer_config() :: brod_producer:config(). |
| 216 223 | -type partition_fun() :: fun((topic(), pos_integer(), key(), value()) -> |
| 217 224 | {ok, partition()}). |
| 218 225 | -type partitioner() :: partition_fun() | random | hash. |
| 219 226 | -type produce_ack_cb() :: fun((partition(), offset()) -> _). |
| 220 227 | -type compression() :: no_compression | gzip | snappy. |
| 221 | - -type call_ref() :: #brod_call_ref{}. |
| 228 | + -type call_ref() :: #brod_call_ref{}. %% A record with caller, callee, and ref. |
| 222 229 | -type produce_result() :: brod_produce_req_buffered |
| 223 230 | | brod_produce_req_acked. |
| 224 231 | |
| 225 232 | |
| 226 233 | %% consumers |
| 227 | - -type consumer_option() :: begin_offset |
| 228 | - | min_bytes |
| 229 | - | max_bytes |
| 230 | - | max_wait_time |
| 231 | - | sleep_timeout |
| 232 | - | prefetch_count |
| 233 | - | prefetch_bytes |
| 234 | - | offset_reset_policy |
| 235 | - | size_stat_window. |
| 236 | - -type consumer_options() :: [{consumer_option(), integer()}]. |
| 237 | - -type consumer_config() :: brod_consumer:config(). |
| 234 | + -type consumer_config() :: [ {begin_offset, offset_time()} |
| 235 | + | {min_bytes, non_neg_integer()} |
| 236 | + | {max_bytes, non_neg_integer()} |
| 237 | + | {max_wait_time, integer()} |
| 238 | + | {sleep_timeout, integer()} |
| 239 | + | {prefetch_count, integer()} |
| 240 | + | {prefetch_bytes, non_neg_integer()} |
| 241 | + | {offset_reset_policy, brod_consumer:offset_reset_policy()} |
| 242 | + | {size_stat_window, non_neg_integer()} |
| 243 | + | {isolation_level, brod_consumer:isolation_level()} |
| 244 | + ]. |
| 245 | + %% Consumer configuration. |
| 246 | + %% |
| 247 | + %% The meaning of the options is documented at {@link brod_consumer:start_link/5}. |
| 238 248 | -type connection() :: kpro:connection(). |
| 239 249 | -type conn_config() :: [{atom(), term()}] | kpro:conn_config(). |
| 250 | + %% Connection configuration that will be passed to `kpro' calls. |
| 251 | + %% |
| 252 | + %% For more info, see the {@link kpro_connection:config()} type. |
| 240 253 | |
| 241 254 | %% consumer groups |
| 242 255 | -type group_id() :: kpro:group_id(). |
| @@ -294,7 +307,7 @@ start_client(BootstrapEndpoints) -> | |
| 294 307 | start_client(BootstrapEndpoints, ClientId) -> |
| 295 308 | start_client(BootstrapEndpoints, ClientId, []). |
| 296 309 | |
| 297 | - %% @doc Start a client. |
| 310 | + %% @doc Start a client ({@link brod_client}). |
| 298 311 | %% |
| 299 312 | %% `BootstrapEndpoints': |
| 300 313 | %% Kafka cluster endpoints, can be any of the brokers in the cluster, |
| @@ -341,7 +354,7 @@ start_client(BootstrapEndpoints, ClientId) -> | |
| 341 354 | %% <li>`default_producer_config' (optional, default=[]) |
| 342 355 | %% |
| 343 356 | %% Producer configuration to use when auto_start_producers is true. |
| 344 | - %% See `brod_producer:start_link/4' for details about producer config</li> |
| 357 | + %% See {@link brod_producer:start_link/4} for details about producer config</li> |
| 345 358 | %% |
| 346 359 | %% </ul> |
| 347 360 | %% |
| @@ -392,6 +405,9 @@ start_client(BootstrapEndpoints, ClientId) -> | |
| 392 405 | %% </a> |
| 393 406 | %% </li> |
| 394 407 | %% </ul> |
| 408 | + %% |
| 409 | + %% You can read more about clients in the |
| 410 | + %% <a href="https://hexdocs.pm/brod/readme.html#clients">overview</a>. |
| 395 411 | -spec start_client([endpoint()], client_id(), client_config()) -> |
| 396 412 | ok | {error, any()}. |
| 397 413 | start_client(BootstrapEndpoints, ClientId, Config) -> |
| @@ -427,31 +443,69 @@ stop_client(Client) when is_atom(Client) -> | |
| 427 443 | stop_client(Client) when is_pid(Client) -> |
| 428 444 | brod_client:stop(Client). |
| 429 445 | |
| 430 | - %% @doc Dynamically start a per-topic producer. |
| 431 | - %% @see brod_producer:start_link/4 |
| 446 | + %% @doc Dynamically start a per-topic producer and register it in the client. |
| 447 | + %% |
| 448 | + %% You have to start a producer for each topic you want to produce messages |
| 449 | + %% into, unless you have specified `auto_start_producers = true' when starting |
| 450 | + %% the client (in that case you don't have to call this function at all). |
| 451 | + %% |
| 452 | + %% After starting the producer, you can call {@link produce/5} and friends |
| 453 | + %% for producing messages. |
| 454 | + %% |
| 455 | + %% You can read more about producers in the |
| 456 | + %% <a href="https://hexdocs.pm/brod/readme.html#producers">overview</a>. |
| 457 | + %% |
| 458 | + %% A client has to be already started before making this call (e.g. by calling |
| 459 | + %% {@link start_client/3}). |
| 460 | + %% |
| 461 | + %% See {@link brod_producer:start_link/4} for a list of available configuration |
| 462 | + %% options. |
| 463 | + %% |
| 464 | + %% Example: |
| 465 | + %% ``` |
| 466 | + %% > brod:start_producer(my_client, <<"my_topic">>, [{max_retries, 5}]). |
| 467 | + %% ok |
| 468 | + %% ''' |
| 432 469 | -spec start_producer(client(), topic(), producer_config()) -> |
| 433 470 | ok | {error, any()}. |
| 434 471 | start_producer(Client, TopicName, ProducerConfig) -> |
| 435 472 | brod_client:start_producer(Client, TopicName, ProducerConfig). |
| 436 473 | |
| 437 | - %% @doc Dynamically start a topic consumer. |
| 438 | - %% @see brod_consumer:start_link/5. for details about consumer config. |
| 474 | + %% @doc Dynamically start topic consumer(s) and register it in the client. |
| 475 | + %% |
| 476 | + %% A {@link brod_consumer} is started for each partition of the given topic. |
| 477 | + %% Note that you can have only one consumer per client-topic. |
| 478 | + %% |
| 479 | + %% See {@link brod_consumer:start_link/5} for details about consumer config. |
| 480 | + %% |
| 481 | + %% You can read more about consumers in the |
| 482 | + %% <a href="https://hexdocs.pm/brod/readme.html#consumers">overview</a>. |
| 439 483 | -spec start_consumer(client(), topic(), consumer_config()) -> |
| 440 484 | ok | {error, any()}. |
| 441 485 | start_consumer(Client, TopicName, ConsumerConfig) -> |
| 442 486 | brod_client:start_consumer(Client, TopicName, ConsumerConfig). |
| 443 487 | |
| 444 488 | %% @doc Get number of partitions for a given topic. |
| 489 | + %% |
| 445 490 | %% The higher level producers may need the partition numbers to |
| 446 | - %% find the partition producer pid --- if the number of partitions |
| 491 | + %% find the partition producer pid – if the number of partitions |
| 447 492 | %% is not statically configured for them. |
| 448 493 | %% It is up to the callers how they want to distribute their data |
| 449 494 | %% (e.g. random, roundrobin or consistent-hashing) to the partitions. |
| 495 | + %% NOTE: The partitions count is cached for 120 seconds. |
| 450 496 | -spec get_partitions_count(client(), topic()) -> |
| 451 497 | {ok, pos_integer()} | {error, any()}. |
| 452 498 | get_partitions_count(Client, Topic) -> |
| 453 499 | brod_client:get_partitions_count(Client, Topic). |
| 454 500 | |
| 501 | + %% @doc The same as `get_partitions_count(Client, Topic)' |
| 502 | + %% but ensured not to auto-create topics in Kafka even |
| 503 | + %% when Kafka has topic auto-creation configured. |
| 504 | + -spec get_partitions_count_safe(client(), topic()) -> |
| 505 | + {ok, pos_integer()} | {error, any()}. |
| 506 | + get_partitions_count_safe(Client, Topic) -> |
| 507 | + brod_client:get_partitions_count_safe(Client, Topic). |
| 508 | + |
| 455 509 | -spec get_consumer(client(), topic(), partition()) -> |
| 456 510 | {ok, pid()} | {error, Reason} |
| 457 511 | when Reason :: client_down |
| @@ -462,7 +516,7 @@ get_partitions_count(Client, Topic) -> | |
| 462 516 | get_consumer(Client, Topic, Partition) -> |
| 463 517 | brod_client:get_consumer(Client, Topic, Partition). |
| 464 518 | |
| 465 | - %% @equiv brod_client:get_producer/3 |
| 519 | + %% @equiv brod_client:get_producer(Client, Topic, Partition) |
| 466 520 | -spec get_producer(client(), topic(), partition()) -> |
| 467 521 | {ok, pid()} | {error, Reason} |
| 468 522 | when Reason :: client_down |
| @@ -478,29 +532,73 @@ get_producer(Client, Topic, Partition) -> | |
| 478 532 | produce(Pid, Value) -> |
| 479 533 | produce(Pid, _Key = <<>>, Value). |
| 480 534 | |
| 481 | - %% @doc Produce one message if `Value' is a binary or an |
| 482 | - %% iolist. Otherwise send a batch, if `Value' is a (nested) key-value |
| 483 | - %% list, or a list of maps. In this case `Key' is discarded (only the |
| 484 | - %% keys in the key-value list are sent to Kafka). The pid should be a |
| 485 | - %% partition producer pid, NOT client pid. The return value is a call |
| 486 | - %% reference of type `call_ref()', so the caller can use it to expect |
| 487 | - %% (match) a `#brod_produce_reply{result = brod_produce_req_acked}' |
| 535 | + %% @doc Produce one or more messages. |
| 536 | + %% |
| 537 | + %% See {@link produce/5} for information about possible shapes |
| 538 | + %% of `Value'. |
| 539 | + %% |
| 540 | + %% The pid should be a partition producer pid, NOT client pid. |
| 541 | + %% |
| 542 | + %% The return value is a call reference of type `call_ref()', |
| 543 | + %% so the caller can use it to expect (match) |
| 544 | + %% a `#brod_produce_reply{result = brod_produce_req_acked}' |
| 488 545 | %% message after the produce request has been acked by Kafka. |
| 489 546 | -spec produce(pid(), key(), value()) -> |
| 490 547 | {ok, call_ref()} | {error, any()}. |
| 491 548 | produce(ProducerPid, Key, Value) -> |
| 492 549 | brod_producer:produce(ProducerPid, Key, Value). |
| 493 550 | |
| 494 | - %% @doc Produce one message if `Value' is a binary or an iolist. |
| 495 | - %% Otherwise send a batch if `Value' is a (nested) key-value list, or |
| 496 | - %% a list of maps. In this case `Key' is used only for partitioning, |
| 497 | - %% or discarded if the 3rd argument is a partition number instead of a |
| 498 | - %% partitioner callback. This function first looks up the producer |
| 499 | - %% pid, then calls `produce/3' to do the real work. The return value |
| 500 | - %% is a call reference of type `call_ref()', so the caller can used it |
| 501 | - %% to expect (match) a `#brod_produce_reply{result = |
| 502 | - %% brod_produce_req_acked}' message after the produce request has been |
| 503 | - %% acked by Kafka. |
| 551 | + %% @doc Produce one or more messages. |
| 552 | + %% |
| 553 | + %% `Value' can have many different forms: |
| 554 | + %% <ul> |
| 555 | + %% <li>`binary()': Single message with key from the `Key' argument</li> |
| 556 | + %% <li>`{brod:msg_ts(), binary()}': Single message with |
| 557 | + %% its create-time timestamp and key from `Key'</li> |
| 558 | + %% <li>`#{ts => brod:msg_ts(), value => binary(), headers => [{_, _}]}': |
| 559 | + %% Single message; if this map does not have a `key' |
| 560 | + %% field, `Key' is used instead</li> |
| 561 | + %% <li>`[{K, V} | {T, K, V}]': A batch, where `V' could be |
| 562 | + %% a nested list of such representation</li> |
| 563 | + %% <li>`[#{key => K, value => V, ts => T, headers => [{_, _}]}]': |
| 564 | + %% A batch</li> |
| 565 | + %% </ul> |
| 566 | + %% |
| 567 | + %% When `Value' is a batch, the `Key' argument is only used |
| 568 | + %% as partitioner input and all messages are written on the |
| 569 | + %% same partition. |
| 570 | + %% |
| 571 | + %% `ts' field is dropped for kafka prior to version `0.10' |
| 572 | + %% (produce API version 0, magic version 0). `headers' field |
| 573 | + %% is dropped for kafka prior to version `0.11' (produce API |
| 574 | + %% version 0-2, magic version 0-1). |
| 575 | + %% |
| 576 | + %% `Partition' may be either a concrete partition (an integer) |
| 577 | + %% or a partitioner (see {@link partitioner()} for more info). |
| 578 | + %% |
| 579 | + %% A producer for the particular topic has to be already started |
| 580 | + %% (by calling {@link start_producer/3}), unless you have specified |
| 581 | + %% `auto_start_producers = true' when starting the client. |
| 582 | + %% |
| 583 | + %% This function first looks up the producer pid, then calls {@link produce/3} |
| 584 | + %% to do the real work. |
| 585 | + %% |
| 586 | + %% The return value is a call reference of type {@link call_ref()}, so the caller |
| 587 | + %% can used it to expect (match) |
| 588 | + %% a `#brod_produce_reply{result = brod_produce_req_acked}' |
| 589 | + %% (see the {@link produce_reply()} type) message after the |
| 590 | + %% produce request has been acked by Kafka. |
| 591 | + %% |
| 592 | + %% Example: |
| 593 | + %% ``` |
| 594 | + %% > brod:produce(my_client, <<"my_topic">>, 0, "key", <<"Hello from erlang!">>). |
| 595 | + %% {ok,{brod_call_ref,<0.83.0>,<0.133.0>,#Ref<0.3024768151.2556690436.92841>}} |
| 596 | + %% > flush(). |
| 597 | + %% Shell got {brod_produce_reply, |
| 598 | + %% {brod_call_ref,<0.83.0>,<0.133.0>, |
| 599 | + %% #Ref<0.3024768151.2556690436.92841>}, |
| 600 | + %% 12,brod_produce_req_acked} |
| 601 | + %% ''' |
| 504 602 | -spec produce(client(), topic(), partition() | partitioner(), |
| 505 603 | key(), value()) -> {ok, call_ref()} | {error, any()}. |
| 506 604 | produce(Client, Topic, Partition, Key, Value) when is_integer(Partition) -> |
| @@ -518,17 +616,19 @@ produce(Client, Topic, Partitioner, Key, Value) -> | |
| 518 616 | {error, Reason} |
| 519 617 | end. |
| 520 618 | |
| 521 | - %% @doc Same as `produce/3', only the ack is not delivered as a message, |
| 619 | + %% @doc Same as {@link produce/3}, only the ack is not delivered as a message, |
| 522 620 | %% instead, the callback is evaluated by producer worker when ack is received |
| 523 | - %% from kafka. |
| 621 | + %% from kafka (see the {@link produce_ack_cb()} type). |
| 524 622 | -spec produce_cb(pid(), key(), value(), produce_ack_cb()) -> |
| 525 623 | ok | {error, any()}. |
| 526 624 | produce_cb(ProducerPid, Key, Value, AckCb) -> |
| 527 625 | brod_producer:produce_cb(ProducerPid, Key, Value, AckCb). |
| 528 626 | |
| 529 | - %% @doc Same as `produce/5' only the ack is not delivered as a message, |
| 627 | + %% @doc Same as {@link produce/5} only the ack is not delivered as a message, |
| 530 628 | %% instead, the callback is evaluated by producer worker when ack is received |
| 531 | - %% from kafka. Return the partition to caller as `{ok, Partition}' for caller |
| 629 | + %% from kafka (see the {@link produce_ack_cb()} type). |
| 630 | + %% |
| 631 | + %% Return the partition to caller as `{ok, Partition}' for caller |
| 532 632 | %% to correlate the callback when the 3rd arg is not a partition number. |
| 533 633 | -spec produce_cb(client(), topic(), partition() | partitioner(), |
| 534 634 | key(), value(), produce_ack_cb()) -> |
| @@ -552,6 +652,7 @@ produce_cb(Client, Topic, Partitioner, Key, Value, AckCb) -> | |
| 552 652 | end. |
| 553 653 | |
| 554 654 | %% @doc Send the message to partition worker without any ack. |
| 655 | + %% |
| 555 656 | %% NOTE: This call has no back-pressure to the caller, |
| 556 657 | %% excessive usage may cause BEAM to run out of memory. |
| 557 658 | -spec produce_no_ack(pid(), key(), value()) -> ok | {error, any()}. |
| @@ -559,6 +660,7 @@ produce_no_ack(ProducerPid, Key, Value) -> | |
| 559 660 | brod_producer:produce_no_ack(ProducerPid, Key, Value). |
| 560 661 | |
| 561 662 | %% @doc Find the partition worker and send message without any ack. |
| 663 | + %% |
| 562 664 | %% NOTE: This call has no back-pressure to the caller, |
| 563 665 | %% excessive usage may cause BEAM to run out of memory. |
| 564 666 | -spec produce_no_ack(client(), topic(), partition() | partitioner(), |
| @@ -579,13 +681,12 @@ produce_no_ack(Client, Topic, Partitioner, Key, Value) -> | |
| 579 681 | ok |
| 580 682 | end. |
| 581 683 | |
| 582 | - %% @doc Same as `produce/5' only the ack is not d |
| 583 684 | %% @equiv produce_sync(Pid, <<>>, Value) |
| 584 | - -spec produce_sync(pid(), value()) -> ok. |
| 685 | + -spec produce_sync(pid(), value()) -> ok | {error, any()}. |
| 585 686 | produce_sync(Pid, Value) -> |
| 586 687 | produce_sync(Pid, _Key = <<>>, Value). |
| 587 688 | |
| 588 | - %% @doc Sync version of produce/3 |
| 689 | + %% @doc Sync version of {@link produce/3}. |
| 589 690 | %% |
| 590 691 | %% This function will not return until the response is received from |
| 591 692 | %% Kafka. But when producer is started with `required_acks' set to 0, |
| @@ -602,9 +703,10 @@ produce_sync(Pid, Key, Value) -> | |
| 602 703 | {error, Reason} |
| 603 704 | end. |
| 604 705 | |
| 605 | - %% @doc Sync version of produce/5 |
| 706 | + %% @doc Sync version of {@link produce/5}. |
| 707 | + %% |
| 606 708 | %% This function will not return until a response is received from kafka, |
| 607 | - %% however if producer is started with required_acks set to 0, this function |
| 709 | + %% however if producer is started with `required_acks' set to 0, this function |
| 608 710 | %% will return once the messages are buffered in the producer process. |
| 609 711 | -spec produce_sync(client(), topic(), partition() | partitioner(), |
| 610 712 | key(), value()) -> ok | {error, any()}. |
| @@ -614,8 +716,9 @@ produce_sync(Client, Topic, Partition, Key, Value) -> | |
| 614 716 | Else -> Else |
| 615 717 | end. |
| 616 718 | |
| 617 | - %% @doc Version of produce_sync/5 that returns the offset assigned by Kafka |
| 618 | - %% If producer is started with required_acks set to 0, the offset will be |
| 719 | + %% @doc Version of {@link produce_sync/5} that returns the offset assigned by Kafka. |
| 720 | + %% |
| 721 | + %% If producer is started with `required_acks' set to 0, the offset will be |
| 619 722 | %% `?BROD_PRODUCE_UNKNOWN_OFFSET'. |
| 620 723 | -spec produce_sync_offset(client(), topic(), partition() | partitioner(), |
| 621 724 | key(), value()) -> {ok, offset()} | {error, any()}. |
| @@ -627,12 +730,25 @@ produce_sync_offset(Client, Topic, Partition, Key, Value) -> | |
| 627 730 | {error, Reason} |
| 628 731 | end. |
| 629 732 | |
| 630 | - %% @doc Block wait for sent produced request to be acked by kafka. |
| 733 | + %% @equiv sync_produce_request(CallRef, infinity) |
| 631 734 | -spec sync_produce_request(call_ref()) -> |
| 632 735 | ok | {error, Reason :: any()}. |
| 633 736 | sync_produce_request(CallRef) -> |
| 634 737 | sync_produce_request(CallRef, infinity). |
| 635 738 | |
| 739 | + %% @doc Block wait for sent produced request to be acked by kafka. |
| 740 | + %% |
| 741 | + %% This way, you can turn asynchronous requests, made by {@link produce/5} |
| 742 | + %% and friends, into synchronous ones. |
| 743 | + %% |
| 744 | + %% Example: |
| 745 | + %% ``` |
| 746 | + %% {ok, CallRef} = brod:produce( |
| 747 | + %% brod_client_1, <<"my_topic">>, 0, <<"some-key">>, <<"some-value">>) |
| 748 | + %% ). % returns immediately |
| 749 | + %% % the following call waits and returns after the ack is received or timed out |
| 750 | + %% brod:sync_produce_request(CallRef, 5_000). |
| 751 | + %% ''' |
| 636 752 | -spec sync_produce_request(call_ref(), timeout()) -> |
| 637 753 | ok | {error, Reason :: any()}. |
| 638 754 | sync_produce_request(CallRef, Timeout) -> |
| @@ -641,13 +757,15 @@ sync_produce_request(CallRef, Timeout) -> | |
| 641 757 | Else -> Else |
| 642 758 | end. |
| 643 759 | |
| 644 | - %% @doc As sync_produce_request_offset/1, but also returning assigned offset |
| 645 | - %% See produce_sync_offset/5. |
| 760 | + %% @equiv sync_produce_request_offset(CallRef, infinity) |
| 646 761 | -spec sync_produce_request_offset(call_ref()) -> |
| 647 762 | {ok, offset()} | {error, Reason :: any()}. |
| 648 763 | sync_produce_request_offset(CallRef) -> |
| 649 764 | sync_produce_request_offset(CallRef, infinity). |
| 650 765 | |
| 766 | + %% @doc As {@link sync_produce_request/2}, but also returning assigned offset. |
| 767 | + %% |
| 768 | + %% See @{link produce_sync_offset/5}. |
| 651 769 | -spec sync_produce_request_offset(call_ref(), timeout()) -> |
| 652 770 | {ok, offset()} | {error, Reason :: any()}. |
| 653 771 | sync_produce_request_offset(CallRef, Timeout) -> |
| @@ -655,6 +773,14 @@ sync_produce_request_offset(CallRef, Timeout) -> | |
| 655 773 | |
| 656 774 | %% @doc Subscribe to a data stream from the given topic-partition. |
| 657 775 | %% |
| 776 | + %% A client has to be already started (by calling {@link start_client/3}, |
| 777 | + %% one client per multiple topics is enough) and a corresponding consumer |
| 778 | + %% for the topic and partition as well (by calling {@link start_consumer/3}), |
| 779 | + %% before calling this function. |
| 780 | + %% |
| 781 | + %% Caller may specify a set of options extending consumer config. |
| 782 | + %% See {@link brod_consumer:subscribe/3} for more info on that. |
| 783 | + %% |
| 658 784 | %% If `{error, Reason}' is returned, the caller should perhaps retry later. |
| 659 785 | %% |
| 660 786 | %% `{ok, ConsumerPid}' is returned on success. The caller may want to |
| @@ -669,8 +795,18 @@ sync_produce_request_offset(CallRef, Timeout) -> | |
| 669 795 | %% |
| 670 796 | %% In case `#kafka_fetch_error{}' is received the subscriber should |
| 671 797 | %% re-subscribe itself to resume the data stream. |
| 798 | + %% |
| 799 | + %% To provide a mechanism to handle backpressure, brod requires all messages |
| 800 | + %% sent to a subscriber to be acked by calling {@link consume_ack/4} after |
| 801 | + %% they are processed. If there are too many not-acked messages received by |
| 802 | + %% the subscriber, the consumer will stop to fetch new ones so the subscriber |
| 803 | + %% won't get overwhelmed. |
| 804 | + %% |
| 805 | + %% Only one process can be subscribed to a consumer. This means that if |
| 806 | + %% you want to read at different places (or at different paces), you have |
| 807 | + %% to create separate consumers (and thus also separate clients). |
| 672 808 | -spec subscribe(client(), pid(), topic(), partition(), |
| 673 | - consumer_options()) -> {ok, pid()} | {error, any()}. |
| 809 | + consumer_config()) -> {ok, pid()} | {error, any()}. |
| 674 810 | subscribe(Client, SubscriberPid, Topic, Partition, Options) -> |
| 675 811 | case brod_client:get_consumer(Client, Topic, Partition) of |
| 676 812 | {ok, ConsumerPid} -> |
| @@ -682,12 +818,16 @@ subscribe(Client, SubscriberPid, Topic, Partition, Options) -> | |
| 682 818 | {error, Reason} |
| 683 819 | end. |
| 684 820 | |
| 685 | - -spec subscribe(pid(), pid(), consumer_options()) -> ok | {error, any()}. |
| 821 | + %% @doc Subscribe to a data stream from the given consumer. |
| 822 | + %% |
| 823 | + %% See {@link subscribe/5} for more information. |
| 824 | + -spec subscribe(pid(), pid(), consumer_config()) -> ok | {error, any()}. |
| 686 825 | subscribe(ConsumerPid, SubscriberPid, Options) -> |
| 687 826 | brod_consumer:subscribe(ConsumerPid, SubscriberPid, Options). |
| 688 827 | |
| 689 | - %% @doc Unsubscribe the current subscriber. Assuming the subscriber is |
| 690 | - %% `self()'. |
| 828 | + %% @doc Unsubscribe the current subscriber. |
| 829 | + %% |
| 830 | + %% Assuming the subscriber is %% `self()'. |
| 691 831 | -spec unsubscribe(client(), topic(), partition()) -> ok | {error, any()}. |
| 692 832 | unsubscribe(Client, Topic, Partition) -> |
| 693 833 | unsubscribe(Client, Topic, Partition, self()). |
| @@ -700,8 +840,9 @@ unsubscribe(Client, Topic, Partition, SubscriberPid) -> | |
| 700 840 | Error -> Error |
| 701 841 | end. |
| 702 842 | |
| 703 | - %% @doc Unsubscribe the current subscriber. Assuming the subscriber is |
| 704 | - %% `self()'. |
| 843 | + %% @doc Unsubscribe the current subscriber. |
| 844 | + %% |
| 845 | + %% Assuming the subscriber is %% `self()'. |
| 705 846 | -spec unsubscribe(pid()) -> ok | {error, any()}. |
| 706 847 | unsubscribe(ConsumerPid) -> |
| 707 848 | unsubscribe(ConsumerPid, self()). |
| @@ -711,6 +852,31 @@ unsubscribe(ConsumerPid) -> | |
| 711 852 | unsubscribe(ConsumerPid, SubscriberPid) -> |
| 712 853 | brod_consumer:unsubscribe(ConsumerPid, SubscriberPid). |
| 713 854 | |
| 855 | + %% @doc Acknowledge that one or more messages have been processed. |
| 856 | + %% |
| 857 | + %% {@link brod_consumer} sends message-sets to the subscriber process, and keep |
| 858 | + %% the messages in a 'pending' queue. |
| 859 | + %% The subscriber may choose to ack any received offset. |
| 860 | + %% Acknowledging a greater offset will automatically acknowledge |
| 861 | + %% the messages before this offset. |
| 862 | + %% For example, if message `[1, 2, 3, 4]' have been sent to (as one or more message-sets) |
| 863 | + %% to the subscriber, the subscriber may acknowledge with offset `3' to indicate that |
| 864 | + %% the first three messages are successfully processed, leaving behind only message `4' |
| 865 | + %% pending. |
| 866 | + %% |
| 867 | + %% |
| 868 | + %% The 'pending' queue has a size limit (see `prefetch_count' consumer config) |
| 869 | + %% which is to provide a mechanism to handle back-pressure. |
| 870 | + %% If there are too many messages pending on ack, the consumer will stop |
| 871 | + %% fetching new ones so the subscriber won't get overwhelmed. |
| 872 | + %% |
| 873 | + %% Note, there is no range check done for the acknowledging offset, meaning if offset `[M, N]' |
| 874 | + %% are pending to be acknowledged, acknowledging with `Offset > N' will cause all offsets to be |
| 875 | + %% removed from the pending queue, and acknowledging with `Offset < M' has no effect. |
| 876 | + %% |
| 877 | + %% Use this function only with plain partition subscribers (i.e., when you |
| 878 | + %% manually call {@link subscribe/5}). Behaviours like |
| 879 | + %% {@link brod_topic_subscriber} have their own way how to ack messages. |
| 714 880 | -spec consume_ack(client(), topic(), partition(), offset()) -> |
| 715 881 | ok | {error, any()}. |
| 716 882 | consume_ack(Client, Topic, Partition, Offset) -> |
| @@ -719,6 +885,8 @@ consume_ack(Client, Topic, Partition, Offset) -> | |
| 719 885 | {error, Reason} -> {error, Reason} |
| 720 886 | end. |
| 721 887 | |
| 888 | + %% @equiv brod_consumer:ack(ConsumerPid, Offset) |
| 889 | + %% @doc See {@link consume_ack/4} for more information. |
| 722 890 | -spec consume_ack(pid(), offset()) -> ok | {error, any()}. |
| 723 891 | consume_ack(ConsumerPid, Offset) -> |
| 724 892 | brod_consumer:ack(ConsumerPid, Offset). |
| @@ -733,7 +901,7 @@ start_link_group_subscriber(Client, GroupId, Topics, GroupConfig, | |
| 733 901 | brod_group_subscriber:start_link(Client, GroupId, Topics, GroupConfig, |
| 734 902 | ConsumerConfig, CbModule, CbInitArg). |
| 735 903 | |
| 736 | - %% @doc Start group_subscriber_v2 |
| 904 | + %% @doc Start group_subscriber_v2. |
| 737 905 | -spec start_link_group_subscriber_v2( |
| 738 906 | brod_group_subscriber_v2:subscriber_config() |
| 739 907 | ) -> {ok, pid()} | {error, any()}. |
| @@ -798,18 +966,64 @@ start_link_topic_subscriber(Config) -> | |
| 798 966 | brod_topic_subscriber:start_link(Config). |
| 799 967 | |
| 800 968 | %% @equiv create_topics(Hosts, TopicsConfigs, RequestConfigs, []) |
| 801 | - -spec create_topics([endpoint()], [topic_config()], #{timeout => kpro:int32(), |
| 802 | - validate_only => boolean()}) -> |
| 803 | - ok | {ok, kpro:struct()} | {error, any()}. |
| 969 | + -spec create_topics([endpoint()], [topic_config()], #{timeout => kpro:int32()}) -> |
| 970 | + ok | {error, any()}. |
| 804 971 | create_topics(Hosts, TopicConfigs, RequestConfigs) -> |
| 805 972 | brod_utils:create_topics(Hosts, TopicConfigs, RequestConfigs). |
| 806 973 | |
| 807 | - %% @doc Create topic(s) in kafka |
| 808 | - %% Return the message body of `create_topics', response. |
| 809 | - %% See `kpro_schema.erl' for struct details |
| 810 | - -spec create_topics([endpoint()], [topic_config()], #{timeout => kpro:int32(), |
| 811 | - validate_only => boolean()}, conn_config()) -> |
| 812 | - ok | {ok, kpro:struct()} | {error, any()}. |
| 974 | + %% @doc Create topic(s) in kafka. |
| 975 | + %% |
| 976 | + %% `TopicConfigs' is a list of topic configurations. |
| 977 | + %% A topic configuration is a map (or tuple list for backward compatibility) |
| 978 | + %% with the following keys (all of them are reuired): |
| 979 | + %% <ul> |
| 980 | + %% <li>`name' |
| 981 | + %% |
| 982 | + %% The topic name.</li> |
| 983 | + %% |
| 984 | + %% <li>`num_partitions' |
| 985 | + %% |
| 986 | + %% The number of partitions to create in the topic, or -1 if we are |
| 987 | + %% either specifying a manual partition assignment or using the default |
| 988 | + %% partitions.</li> |
| 989 | + %% |
| 990 | + %% <li>`replication_factor' |
| 991 | + %% |
| 992 | + %% The number of replicas to create for each partition in the topic, |
| 993 | + %% or -1 if we are either specifying a manual partition assignment |
| 994 | + %% or using the default replication factor.</li> |
| 995 | + %% |
| 996 | + %% <li>`assignments' |
| 997 | + %% |
| 998 | + %% The manual partition assignment, or the empty list if we let Kafka |
| 999 | + %% automatically assign them. It is a list of maps (or tuple lists) with the |
| 1000 | + %% following keys: `partition_index' and `broker_ids' (a list of of brokers to |
| 1001 | + %% place the partition on).</li> |
| 1002 | + %% |
| 1003 | + %% <li>`configs' |
| 1004 | + %% |
| 1005 | + %% The custom topic configurations to set. It is a list of of maps (or |
| 1006 | + %% tuple lists) with keys `name' and `value'. You can find possible |
| 1007 | + %% options in the Kafka documentation.</li> |
| 1008 | + %% </ul> |
| 1009 | + %% |
| 1010 | + %% Example: |
| 1011 | + %% ``` |
| 1012 | + %% > TopicConfigs = [ |
| 1013 | + %% #{ |
| 1014 | + %% name => <<"my_topic">>, |
| 1015 | + %% num_partitions => 1, |
| 1016 | + %% replication_factor => 1, |
| 1017 | + %% assignments => [], |
| 1018 | + %% configs => [ #{name => <<"cleanup.policy">>, value => "compact"}] |
| 1019 | + %% } |
| 1020 | + %% ]. |
| 1021 | + %% > brod:create_topics([{"localhost", 9092}], TopicConfigs, #{timeout => 1000}, []). |
| 1022 | + %% ok |
| 1023 | + %% ''' |
| 1024 | + -spec create_topics([endpoint()], [topic_config()], #{timeout => kpro:int32()}, |
| 1025 | + conn_config()) -> |
| 1026 | + ok | {error, any()}. |
| 813 1027 | create_topics(Hosts, TopicConfigs, RequestConfigs, Options) -> |
| 814 1028 | brod_utils:create_topics(Hosts, TopicConfigs, RequestConfigs, Options). |
| 815 1029 | |
| @@ -819,50 +1033,89 @@ create_topics(Hosts, TopicConfigs, RequestConfigs, Options) -> | |
| 819 1033 | delete_topics(Hosts, Topics, Timeout) -> |
| 820 1034 | brod_utils:delete_topics(Hosts, Topics, Timeout). |
| 821 1035 | |
| 822 | - %% @doc Delete topic(s) from kafka |
| 823 | - %% Return the message body of `delete_topics', response. |
| 824 | - %% See `kpro_schema.erl' for struct details |
| 1036 | + %% @doc Delete topic(s) from kafka. |
| 1037 | + %% |
| 1038 | + %% Example: |
| 1039 | + %% ``` |
| 1040 | + %% > brod:delete_topics([{"localhost", 9092}], ["my_topic"], 5000, []). |
| 1041 | + %% ok |
| 1042 | + %% ''' |
| 825 1043 | -spec delete_topics([endpoint()], [topic()], pos_integer(), conn_config()) -> |
| 826 1044 | ok | {error, any()}. |
| 827 1045 | delete_topics(Hosts, Topics, Timeout, Options) -> |
| 828 1046 | brod_utils:delete_topics(Hosts, Topics, Timeout, Options). |
| 829 1047 | |
| 830 | - %% @doc Fetch broker metadata |
| 831 | - %% Return the message body of `metadata' response. |
| 832 | - %% See `kpro_schema.erl' for details |
| 1048 | + %% @doc Fetch broker metadata for all topics. |
| 1049 | + %% |
| 1050 | + %% See {@link get_metadata/3} for more information. |
| 833 1051 | -spec get_metadata([endpoint()]) -> {ok, kpro:struct()} | {error, any()}. |
| 834 1052 | get_metadata(Hosts) -> |
| 835 1053 | brod_utils:get_metadata(Hosts). |
| 836 1054 | |
| 837 | - %% @doc Fetch broker/topic metadata |
| 838 | - %% Return the message body of `metadata' response. |
| 839 | - %% See `kpro_schema.erl' for struct details |
| 1055 | + %% @doc Fetch broker metadata for the given topics. |
| 1056 | + %% |
| 1057 | + %% See {@link get_metadata/3} for more information. |
| 840 1058 | -spec get_metadata([endpoint()], all | [topic()]) -> |
| 841 1059 | {ok, kpro:struct()} | {error, any()}. |
| 842 1060 | get_metadata(Hosts, Topics) -> |
| 843 1061 | brod_utils:get_metadata(Hosts, Topics). |
| 844 1062 | |
| 845 | - %% @doc Fetch broker/topic metadata |
| 846 | - %% Return the message body of `metadata' response. |
| 847 | - %% See `kpro_schema.erl' for struct details |
| 1063 | + %% @doc Fetch broker metadata for the given topics using the given connection options. |
| 1064 | + %% |
| 1065 | + %% The response differs in each version of the `Metadata' API call. |
| 1066 | + %% The last supported `Metadata' API version is 2, so this will be |
| 1067 | + %% probably used (if your Kafka supports it too). See |
| 1068 | + %% <a href="https://github.com/kafka4beam/kafka_protocol/blob/master/priv/kafka.bnf">kafka.bnf</a> |
| 1069 | + %% (search for `MetadataResponseV2') for response schema with comments. |
| 1070 | + %% |
| 1071 | + %% Beware that when `auto.create.topics.enable' is set to true in |
| 1072 | + %% the broker configuration, fetching metadata with a concrete |
| 1073 | + %% topic specified (in the `Topics' parameter) may cause creation of |
| 1074 | + %% the topic when it does not exist. If you want a safe `get_metadata' |
| 1075 | + %% call, always pass `all' as `Topics' and then filter them. |
| 1076 | + %% |
| 1077 | + %% |
| 1078 | + %% ``` |
| 1079 | + %% > brod:get_metadata([{"localhost", 9092}], [<<"my_topic">>], []). |
| 1080 | + %% {ok,#{brokers => |
| 1081 | + %% [#{host => <<"localhost">>,node_id => 1,port => 9092, |
| 1082 | + %% rack => <<>>}], |
| 1083 | + %% cluster_id => <<"jTb2faMLRf6p21yD1y3v-A">>, |
| 1084 | + %% controller_id => 1, |
| 1085 | + %% topics => |
| 1086 | + %% [#{error_code => no_error,is_internal => false, |
| 1087 | + %% name => <<"my_topic">>, |
| 1088 | + %% partitions => |
| 1089 | + %% [#{error_code => no_error, |
| 1090 | + %% isr_nodes => [1], |
| 1091 | + %% leader_id => 1,partition_index => 1, |
| 1092 | + %% replica_nodes => [1]}, |
| 1093 | + %% #{error_code => no_error, |
| 1094 | + %% isr_nodes => [1], |
| 1095 | + %% leader_id => 1,partition_index => 0, |
| 1096 | + %% replica_nodes => [1]}]}]}} |
| 1097 | + %% ''' |
| 848 1098 | -spec get_metadata([endpoint()], all | [topic()], conn_config()) -> |
| 849 1099 | {ok, kpro:struct()} | {error, any()}. |
| 850 1100 | get_metadata(Hosts, Topics, Options) -> |
| 851 1101 | brod_utils:get_metadata(Hosts, Topics, Options). |
| 852 1102 | |
| 853 | - %% @equiv resolve_offset(Hosts, Topic, Partition, latest, 1) |
| 1103 | + %% @equiv resolve_offset(Hosts, Topic, Partition, latest, []) |
| 854 1104 | -spec resolve_offset([endpoint()], topic(), partition()) -> |
| 855 1105 | {ok, offset()} | {error, any()}. |
| 856 1106 | resolve_offset(Hosts, Topic, Partition) -> |
| 857 1107 | resolve_offset(Hosts, Topic, Partition, ?OFFSET_LATEST). |
| 858 1108 | |
| 859 | - %% @doc Resolve semantic offset or timestamp to real offset. |
| 1109 | + %% @equiv resolve_offset(Hosts, Topic, Partition, Time, []) |
| 860 1110 | -spec resolve_offset([endpoint()], topic(), partition(), offset_time()) -> |
| 861 1111 | {ok, offset()} | {error, any()}. |
| 862 1112 | resolve_offset(Hosts, Topic, Partition, Time) -> |
| 863 1113 | resolve_offset(Hosts, Topic, Partition, Time, []). |
| 864 1114 | |
| 865 1115 | %% @doc Resolve semantic offset or timestamp to real offset. |
| 1116 | + %% |
| 1117 | + %% The same as {@link resolve_offset/6} but the timeout is |
| 1118 | + %% extracted from connection config. |
| 866 1119 | -spec resolve_offset([endpoint()], topic(), partition(), |
| 867 1120 | offset_time(), conn_config()) -> |
| 868 1121 | {ok, offset()} | {error, any()}. |
| @@ -870,6 +1123,38 @@ resolve_offset(Hosts, Topic, Partition, Time, ConnCfg) -> | |
| 870 1123 | brod_utils:resolve_offset(Hosts, Topic, Partition, Time, ConnCfg). |
| 871 1124 | |
| 872 1125 | %% @doc Resolve semantic offset or timestamp to real offset. |
| 1126 | + %% |
| 1127 | + %% The function returns the offset of the first message |
| 1128 | + %% with the given timestamp, or of the first message after |
| 1129 | + %% the given timestamp (in case no message matches the |
| 1130 | + %% timestamp exactly), or -1 if the timestamp is newer |
| 1131 | + %% than (>) all messages in the topic. |
| 1132 | + %% |
| 1133 | + %% You can also use two semantic offsets instead of |
| 1134 | + %% a timestamp: `earliest' gives you the offset of the |
| 1135 | + %% first message in the topic and `latest' gives you |
| 1136 | + %% the offset of the last message incremented by 1. |
| 1137 | + %% |
| 1138 | + %% If the topic is empty, both `earliest' and `latest' |
| 1139 | + %% return the same value (which is 0 unless some messages |
| 1140 | + %% were deleted from the topic), and any timestamp returns |
| 1141 | + %% -1. |
| 1142 | + %% |
| 1143 | + %% An example for illustration: |
| 1144 | + %% ``` |
| 1145 | + %% Messages: |
| 1146 | + %% offset 0 1 2 3 |
| 1147 | + %% timestamp 10 20 20 30 |
| 1148 | + %% |
| 1149 | + %% Calls: |
| 1150 | + %% resolve_offset(Endpoints, Topic, Partition, 5) → 0 |
| 1151 | + %% resolve_offset(Endpoints, Topic, Partition, 10) → 0 |
| 1152 | + %% resolve_offset(Endpoints, Topic, Partition, 13) → 1 |
| 1153 | + %% resolve_offset(Endpoints, Topic, Partition, 20) → 1 |
| 1154 | + %% resolve_offset(Endpoints, Topic, Partition, 31) → -1 |
| 1155 | + %% resolve_offset(Endpoints, Topic, Partition, earliest) → 0 |
| 1156 | + %% resolve_offset(Endpoints, Topic, Partition, latest) → 4 |
| 1157 | + %% ''' |
| 873 1158 | -spec resolve_offset([endpoint()], topic(), partition(), |
| 874 1159 | offset_time(), conn_config(), |
| 875 1160 | #{timeout => kpro:int32()}) -> |
| @@ -878,8 +1163,11 @@ resolve_offset(Hosts, Topic, Partition, Time, ConnCfg, Opts) -> | |
| 878 1163 | brod_utils:resolve_offset(Hosts, Topic, Partition, Time, ConnCfg, Opts). |
| 879 1164 | |
| 880 1165 | %% @doc Fetch a single message set from the given topic-partition. |
| 881 | - %% The first arg can either be an already established connection to leader, |
| 882 | - %% or `{Endpoints, ConnConfig}' so to establish a new connection before fetch. |
| 1166 | + %% |
| 1167 | + %% Calls {@link fetch/5} with the default options: `max_wait_time' = 1 second, |
| 1168 | + %% `min_bytes' = 1 B, and `max_bytes' = 2^20 B (1 MB). |
| 1169 | + %% |
| 1170 | + %% See {@link fetch/5} for more information. |
| 883 1171 | -spec fetch(connection() | client_id() | bootstrap(), |
| 884 1172 | topic(), partition(), integer()) -> |
| 885 1173 | {ok, {HwOffset :: offset(), [message()]}} | {error, any()}. |
| @@ -891,8 +1179,57 @@ fetch(ConnOrBootstrap, Topic, Partition, Offset) -> | |
| 891 1179 | fetch(ConnOrBootstrap, Topic, Partition, Offset, Opts). |
| 892 1180 | |
| 893 1181 | %% @doc Fetch a single message set from the given topic-partition. |
| 1182 | + %% |
| 894 1183 | %% The first arg can either be an already established connection to leader, |
| 895 | - %% or `{Endpoints, ConnConfig}' so to establish a new connection before fetch. |
| 1184 | + %% or `{Endpoints, ConnConfig}' (or just `Endpoints') so to establish a new |
| 1185 | + %% connection before fetch. |
| 1186 | + %% |
| 1187 | + %% The fourth argument is the start offset of the query. Messages with offset |
| 1188 | + %% greater or equal will be fetched. |
| 1189 | + %% |
| 1190 | + %% You can also pass options for the fetch query. |
| 1191 | + %% See the {@link kpro_req_lib:fetch_opts()} type for their documentation. |
| 1192 | + %% Only `max_wait_time', `min_bytes', `max_bytes', and `isolation_level' |
| 1193 | + %% options are currently supported. The defaults are the same as documented |
| 1194 | + %% in the linked type, except for `min_bytes' which defaults to 1 in `brod'. |
| 1195 | + %% Note that `max_bytes' will be rounded up so that full messages are |
| 1196 | + %% retrieved. For example, if you specify `max_bytes = 42' and there |
| 1197 | + %% are three messages of size 40 bytes, two of them will be fetched. |
| 1198 | + %% |
| 1199 | + %% On success, the function returns the messages along with the <i>last stable |
| 1200 | + %% offset</i> (when using `read_committed' mode, the last committed offset) or the |
| 1201 | + %% <i>high watermark offset</i> (offset of the last message that was successfully |
| 1202 | + %% copied to all replicas, incremented by 1), whichever is lower. In essence, this |
| 1203 | + %% is the offset up to which it was possible to read the messages at the time of |
| 1204 | + %% fetching. This is similar to what {@link resolve_offset/6} with `latest' |
| 1205 | + %% returns. You can use this information to determine how far from the end of the |
| 1206 | + %% topic you currently are. Note that when you use this offset as the start offset |
| 1207 | + %% for a subseuqent call, an empty list of messages will be returned (assuming the |
| 1208 | + %% topic hasn't changed, e.g. no new message arrived). Only when you use an offset |
| 1209 | + %% greater than this one, `{error, offset_out_of_range}' will be returned. |
| 1210 | + %% |
| 1211 | + %% Note also that Kafka batches messages in a message set only up to the end of |
| 1212 | + %% a topic segment in which the first retrieved message is, so there may actually |
| 1213 | + %% be more messages behind the last fetched offset even if the fetched size is |
| 1214 | + %% significantly less than `max_bytes' provided in `fetch_opts()'. |
| 1215 | + %% See <a href="https://github.com/kafka4beam/brod/issues/251">this issue</a> |
| 1216 | + %% for more details. |
| 1217 | + %% |
| 1218 | + %% Example (the topic has only two messages): |
| 1219 | + %% ``` |
| 1220 | + %% > brod:fetch([{"localhost", 9092}], <<"my_topic">>, 0, 0, #{max_bytes => 1024}). |
| 1221 | + %% {ok,{2, |
| 1222 | + %% [{kafka_message,0,<<"some_key">>,<<"Hello world!">>, |
| 1223 | + %% create,1663940976473,[]}, |
| 1224 | + %% {kafka_message,1,<<"another_key">>,<<"This is a message with offset 1.">>, |
| 1225 | + %% create,1663940996335,[]}]}} |
| 1226 | + %% |
| 1227 | + %% > brod:fetch([{"localhost", 9092}], <<"my_topic">>, 0, 2, #{max_bytes => 1024}). |
| 1228 | + %% {ok,{2,[]}} |
| 1229 | + %% |
| 1230 | + %% > brod:fetch([{"localhost", 9092}], <<"my_topic">>, 0, 3, #{max_bytes => 1024}). |
| 1231 | + %% {error,offset_out_of_range} |
| 1232 | + %% ''' |
| 896 1233 | -spec fetch(connection() | client_id() | bootstrap(), |
| 897 1234 | topic(), partition(), offset(), fetch_opts()) -> |
| 898 1235 | {ok, {HwOffset :: offset(), [message()]}} | {error, any()}. |
| @@ -900,11 +1237,15 @@ fetch(ConnOrBootstrap, Topic, Partition, Offset, Opts) -> | |
| 900 1237 | brod_utils:fetch(ConnOrBootstrap, Topic, Partition, Offset, Opts). |
| 901 1238 | |
| 902 1239 | %% @doc Fold through messages in a partition. |
| 1240 | + %% |
| 903 1241 | %% Works like `lists:foldl/2' but with below stop conditions: |
| 904 | - %% * Always return after reach high watermark offset |
| 905 | - %% * Return after the given message count limit is reached |
| 906 | - %% * Return after the given kafka offset is reached. |
| 907 | - %% * Return if the `FoldFun' returns an `{error, Reason}' tuple. |
| 1242 | + %% <ul> |
| 1243 | + %% <li> Always return after reach high watermark offset </li> |
| 1244 | + %% <li> Return after the given message count limit is reached </li> |
| 1245 | + %% <li> Return after the given kafka offset is reached </li> |
| 1246 | + %% <li> Return if the `FoldFun' returns an `{error, Reason}' tuple </li> |
| 1247 | + %% </ul> |
| 1248 | + %% |
| 908 1249 | %% NOTE: Exceptions from evaluating `FoldFun' are not caught. |
| 909 1250 | -spec fold(connection() | client_id() | bootstrap(), |
| 910 1251 | topic(), partition(), offset(), fetch_opts(), |
| @@ -913,15 +1254,16 @@ fetch(ConnOrBootstrap, Topic, Partition, Offset, Opts) -> | |
| 913 1254 | fold(Bootstrap, Topic, Partition, Offset, Opts, Acc, Fun, Limits) -> |
| 914 1255 | brod_utils:fold(Bootstrap, Topic, Partition, Offset, Opts, Acc, Fun, Limits). |
| 915 1256 | |
| 916 | - %% @deprecated |
| 917 | - %% fetch(Hosts, Topic, Partition, Offset, Wait, MinBytes, MaxBytes, []) |
| 1257 | + %% @equiv fetch(Hosts, Topic, Partition, Offset, Wait, MinBytes, MaxBytes, []) |
| 1258 | + %% @deprecated Please use {@link fetch/5} instead |
| 918 1259 | -spec fetch([endpoint()], topic(), partition(), offset(), |
| 919 1260 | non_neg_integer(), non_neg_integer(), pos_integer()) -> |
| 920 1261 | {ok, [message()]} | {error, any()}. |
| 921 1262 | fetch(Hosts, Topic, Partition, Offset, MaxWaitTime, MinBytes, MaxBytes) -> |
| 922 1263 | fetch(Hosts, Topic, Partition, Offset, MaxWaitTime, MinBytes, MaxBytes, []). |
| 923 1264 | |
| 924 | - %% @deprecated Fetch a single message set from the given topic-partition. |
| 1265 | + %% @doc Fetch a single message set from the given topic-partition. |
| 1266 | + %% @deprecated Please use {@link fetch/5} instead |
| 925 1267 | -spec fetch([endpoint()], topic(), partition(), offset(), |
| 926 1268 | non_neg_integer(), non_neg_integer(), pos_integer(), |
| 927 1269 | conn_config()) -> {ok, [message()]} | {error, any()}. |
| @@ -944,6 +1286,7 @@ connect_leader(Hosts, Topic, Partition, ConnConfig) -> | |
| 944 1286 | kpro:connect_partition_leader(Hosts, ConnConfig, Topic, Partition, KproOptions). |
| 945 1287 | |
| 946 1288 | %% @doc List ALL consumer groups in the given kafka cluster. |
| 1289 | + %% |
| 947 1290 | %% NOTE: Exception if failed to connect any of the coordinator brokers. |
| 948 1291 | -spec list_all_groups([endpoint()], conn_config()) -> |
| 949 1292 | [{endpoint(), [cg()] | {error, any()}}]. |
| @@ -955,11 +1298,13 @@ list_all_groups(Endpoints, ConnCfg) -> | |
| 955 1298 | list_groups(CoordinatorEndpoint, ConnCfg) -> |
| 956 1299 | brod_utils:list_groups(CoordinatorEndpoint, ConnCfg). |
| 957 1300 | |
| 958 | - %% @doc Describe consumer groups. The given consumer group IDs should be all |
| 1301 | + %% @doc Describe consumer groups. |
| 1302 | + %% |
| 1303 | + %% The given consumer group IDs should be all |
| 959 1304 | %% managed by the coordinator-broker running at the given endpoint. |
| 960 1305 | %% Otherwise error codes will be returned in the result structs. |
| 961 1306 | %% Return `describe_groups' response body field named `groups'. |
| 962 | - %% See `kpro_schema.erl' for struct details |
| 1307 | + %% See `kpro_schema.erl' for struct details. |
| 963 1308 | -spec describe_groups(endpoint(), conn_config(), [group_id()]) -> |
| 964 1309 | {ok, [kpro:struct()]} | {error, any()}. |
| 965 1310 | describe_groups(CoordinatorEndpoint, ConnCfg, IDs) -> |
| @@ -983,6 +1328,7 @@ connect_group_coordinator(BootstrapEndpoints, ConnCfg, GroupId) -> | |
| 983 1328 | kpro:connect_coordinator(BootstrapEndpoints, ConnCfg, Args). |
| 984 1329 | |
| 985 1330 | %% @doc Fetch committed offsets for ALL topics in the given consumer group. |
| 1331 | + %% |
| 986 1332 | %% Return the `responses' field of the `offset_fetch' response. |
| 987 1333 | %% See `kpro_schema.erl' for struct details. |
| 988 1334 | -spec fetch_committed_offsets([endpoint()], conn_config(), group_id()) -> |
| @@ -990,7 +1336,7 @@ connect_group_coordinator(BootstrapEndpoints, ConnCfg, GroupId) -> | |
| 990 1336 | fetch_committed_offsets(BootstrapEndpoints, ConnCfg, GroupId) -> |
| 991 1337 | brod_utils:fetch_committed_offsets(BootstrapEndpoints, ConnCfg, GroupId, []). |
| 992 1338 | |
| 993 | - %% @doc Same as `fetch_committed_offsets/3', |
| 1339 | + %% @doc Same as @{link fetch_committed_offsets/3}, |
| 994 1340 | %% but works with a started `brod_client' |
| 995 1341 | -spec fetch_committed_offsets(client(), group_id()) -> |
| 996 1342 | {ok, [kpro:struct()]} | {error, any()}. |
Loading more files…