Packages

brod

3.6.0
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

HexDocs

Current section

113 Versions

Jump to
Readme
113 Versions
2 Dependencies
Dependants
Files
Activity
Hide whitespace

Compare versions

30 files changed
+1855 additions
-2865 deletions
  @@ -1,42 +1,17 @@
1 + KAFKA_VERSION ?= 1.1
1 2 PROJECT = brod
2 3 PROJECT_DESCRIPTION = Kafka client library in Erlang
3 - PROJECT_VERSION = 3.5.2
4 + PROJECT_VERSION = 3.6.0
4 5
5 6 DEPS = supervisor3 kafka_protocol
6 - TEST_DEPS = docopt jsone meck proper
7 - REL_DEPS = docopt jsone
8 7
9 - ERLC_OPTS = -Werror +warn_unused_vars +warn_shadow_vars +warn_unused_import +warn_obsolete_guard +debug_info
10 - TEST_ERLC_OPTS = -Werror +warn_unused_vars +warn_shadow_vars +warn_unused_import +warn_obsolete_guard +debug_info
8 + ERLC_OPTS = -Werror +warn_unused_vars +warn_shadow_vars +warn_unused_import +warn_obsolete_guard +debug_info -Dbuild_brod_cli
11 9
12 - dep_supervisor3_commit = 1.1.5
13 - dep_kafka_protocol_commit = 1.1.2
14 - dep_docopt = git https://github.com/zmstone/docopt-erl.git 0.1.3
10 + dep_supervisor3_commit = 1.1.7
11 + dep_kafka_protocol_commit = 2.1.1
12 + dep_kafka_protocol = git https://github.com/klarna/kafka_protocol.git $(dep_kafka_protocol_commit)
15 13
16 - ERTS_VSN = $(shell erl -noshell -eval 'io:put_chars(erlang:system_info(version)), halt()')
17 - ESCRIPT_FILE = scripts/brod_cli
18 - ESCRIPT_EMU_ARGS = -sname brod_cli
19 -
20 - COVER = true
21 -
22 - EUNIT_OPTS = verbose
23 -
24 - CT_OPTS = -ct_use_short_names true
25 -
26 - ERL_LIBS := $(ERL_LIBS):$(CURDIR)
27 -
28 - ifeq ($(MAKECMDGOALS),)
29 - export BROD_CLI=true
30 - else ifneq ($(filter rel,$(MAKECMDGOALS)),)
31 - export BROD_CLI=true
32 - else ifneq ($(filter escript,$(MAKECMDGOALS)),)
33 - export BROD_CLI=true
34 - endif
35 -
36 - ifeq ($(BROD_CLI),true)
37 - ERLC_OPTS += -DBROD_CLI
38 - TEST_ERLC_OPTS += -DBROD_CLI
39 - endif
14 + EDOC_OPTS = preprocess, {macros, [{build_brod_cli, true}]}
40 15
41 16 ## Make app the default target
42 17 ## To avoid building a release when brod is used as a erlang.mk project's dependency
  @@ -44,20 +19,50 @@ app::
44 19
45 20 include erlang.mk
46 21
47 - rel:: escript
48 - @cp scripts/brod _rel/brod/bin/brod
49 - @cp $(ESCRIPT_FILE) _rel/brod/bin/brod_cli
50 - @ln -sf erts-$(ERTS_VSN) _rel/brod/erts
51 - @tar -pczf _rel/brod.tar.gz -C _rel brod
22 + compile:
23 + @rebar3 compile
52 24
53 25 test-env:
54 - ./scripts/setup-test-env.sh
26 + @./scripts/setup-test-env.sh
55 27
56 - t: eunit ct
57 - ./scripts/cover-summary.escript eunit.coverdata ct.coverdata
28 + ut:
29 + @rebar3 eunit -v --cover_export_name ut-$(KAFKA_VERSION)
30 +
31 + # version check, eunit and all common tests
32 + t: vsn-check ut
33 + @rebar3 ct -v --cover_export_name ct-$(KAFKA_VERSION)
58 34
59 35 vsn-check:
60 - $(verbose) ./scripts/vsn-check.sh $(PROJECT_VERSION)
36 + @./scripts/vsn-check.sh $(PROJECT_VERSION)
37 +
38 + distclean:: rebar-clean
39 +
40 + rebar-clean:
41 + @rebar3 clean
42 + @rm -rf _build
43 + @rm -rf doc
44 + @rm -f pipe.testdata
61 45
62 46 hex-publish: distclean
63 - $(verbose) rebar3 hex publish
47 + @rebar3 hex publish
48 +
49 + ## tests that require kafka running at localhost
50 + INTEGRATION_CTS = brod_cg_commits brod_client brod_compression brod_consumer brod_producer brod_group_subscriber brod_topic_subscriber
51 +
52 + ## integration tests
53 + it: ut $(INTEGRATION_CTS:%=it-%)
54 +
55 + $(INTEGRATION_CTS:%=it-%):
56 + @rebar3 ct -v --suite=test/$(subst it-,,$@)_SUITE --cover_export_name $@-$(KAFKA_VERSION)
57 +
58 + ## build escript and a releas, and copy escript to release bin dir
59 + brod-cli:
60 + @rebar3 as brod_cli escriptize
61 + @rebar3 as brod_cli release
62 + @cp _build/brod_cli/bin/brod_cli _build/brod_cli/rel/brod/bin/
63 +
64 + cover:
65 + @rebar3 cover -v
66 +
67 + coveralls:
68 + @rebar3 coveralls send
  @@ -1,8 +1,8 @@
1 - # Brod - Apache Kafka Erlang client library
1 + [![Build Status](https://travis-ci.org/klarna/brod.svg)](https://travis-ci.org/klarna/brod) [![Coverage Status](https://coveralls.io/repos/github/klarna/brod/badge.svg?branch=master)](https://coveralls.io/github/klarna/brod?branch=master)
2 2
3 - Brod is an erlang implementation of the Apache Kafka protocol, providing support for both producers and consumers.
3 + # Brod - Apache Kafka Client for Erlang/Elixir
4 4
5 - [![Build Status](https://travis-ci.org/klarna/brod.svg)](https://travis-ci.org/klarna/brod)
5 + Brod is an Erlang implementation of the Apache Kafka protocol, providing support for both producers and consumers.
6 6
7 7 Why "brod"? [http://en.wikipedia.org/wiki/Max_Brod](http://en.wikipedia.org/wiki/Max_Brod)
8 8
  @@ -11,23 +11,20 @@ Why "brod"? [http://en.wikipedia.org/wiki/Max_Brod](http://en.wikipedia.org/wiki
11 11 * Supports Apache Kafka v0.8+
12 12 * Robust producer implementation supporting in-flight requests and asynchronous acknowledgements
13 13 * Both consumer and producer handle leader re-election and other cluster disturbances internally
14 - * Opens max 1 tcp connection to a broker per "brod_client", one can create more clients if needed
14 + * Opens max 1 tcp connection to a broker per `brod_client`, one can create more clients if needed
15 15 * Producer: will start to batch automatically when number of unacknowledged (in flight) requests exceeds configurable maximum
16 16 * Producer: will try to re-send buffered messages on common errors like "Not a leader for partition", errors are resolved automatically by refreshing metadata
17 17 * 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"
18 18 * Group subscriber: Support for consumer groups with options to have Kafka as offset storage or a custom one
19 19 * Topic subscriber: Subscribe on messages from all or selected topic partitions without using consumer groups
20 20 * Pick latest supported version when sending requests to kafka.
21 -
22 - # Missing features
23 -
24 - * lz4 compression & decompression
25 - * new 0.10.1.0 create/delete topic api
21 + * Direct APIs for message send/fetch and cluster inspection/management without having to start clients/producers/consumers.
22 + * A escriptized command-line tool for message send/fetch and cluster inspection/management.
26 23
27 24 # Building and testing
28 25
29 - make
30 - make test-env t # requires docker-composer in place
26 + make compile
27 + make test-env t # requires docker-compose in place
31 28
32 29 # Working With Kafka 0.9.x or Earlier
33 30
  @@ -48,45 +45,46 @@ e.g. in sys.config:
48 45
49 46 # Quick Demo
50 47
51 - Assuming kafka is running at `localhost:9092`
52 - and there is a topic named `brod-test`.
48 + Assuming kafka is running at `localhost:9092` and there is a topic named `test-topic`.
53 49
54 - Below code snippet is copied from Erlang shell
55 - with some non-important printouts trimmed.
50 + Start Erlang shell by `make compile; erl -pa _build/default/lib/*/ebin`, then paste lines below into shell:
56 51
57 52 ```erlang
58 - > rr(brod).
59 - > {ok, _} = application:ensure_all_started(brod).
60 - > KafkaBootstrapEndpoints = [{"localhost", 9092}].
61 - > Topic = <<"brod-test">>.
62 - > Partition = 0.
63 - > ok = brod:start_client(KafkaBootstrapEndpoints, client1).
64 - > ok = brod:start_producer(client1, Topic, _ProducerConfig = []).
65 - > ok = brod:produce_sync(client1, Topic, Partition, <<"key1">>, <<"value1">>).
66 - > ok = brod:produce_sync(client1, Topic, Partition, <<"key2">>, <<"value2">>).
67 - > SubscriberCallbackFun =
68 - fun(Partition, Msg, ShellPid = CallbackState) ->
69 - ShellPid ! Msg,
70 - {ok, ack, CallbackState}
71 - end.
72 - > Receive = fun() -> receive Msg -> Msg after 1000 -> timeout end end.
73 - > brod_topic_subscriber:start_link(client1, Topic, Partitions=[Partition],
74 - _ConsumerConfig=[{begin_offset, earliest}],
75 - _CommittdOffsets=[], message, SubscriberCallbackFun,
76 - _CallbackState=self()).
77 - > Receive().
78 - #kafka_message{offset = 0,magic_byte = 0,attributes = 0,
79 - key = <<"key1">>,value = <<"value1">>,crc = 1978725405}
80 - > Receive().
81 - #kafka_message{offset = 1,magic_byte = 0,attributes = 0,
82 - key = <<"key2">>,value = <<"value2">>,crc = 1964753830}
83 - > {ok, CallRef} = brod:produce(client1, Topic, Partition, <<"key3">>, <<"value3">>).
84 - > #brod_produce_reply{ call_ref = CallRef,
85 - result = brod_produce_req_acked
86 - } = Receive().
87 - > Receive().
88 - #kafka_message{offset = 2,magic_byte = 0,attributes = 0,
89 - key = <<"key3">>,value = <<"value3">>,crc = -1013830416}
53 + rr(brod),
54 + {ok, _} = application:ensure_all_started(brod),
55 + KafkaBootstrapEndpoints = [{"localhost", 9092}],
56 + Topic = <<"test-topic">>,
57 + Partition = 0,
58 + ok = brod:start_client(KafkaBootstrapEndpoints, client1),
59 + ok = brod:start_producer(client1, Topic, _ProducerConfig = []),
60 + {ok, FirstOffset} = brod:produce_sync_offset(client1, Topic, Partition, <<"key1">>, <<"value1">>),
61 + ok = brod:produce_sync(client1, Topic, Partition, <<"key2">>, <<"value2">>),
62 + SubscriberCallbackFun = fun(Partition, Msg, ShellPid = CallbackState) -> ShellPid ! Msg, {ok, ack, CallbackState} end,
63 + Receive = fun() -> receive Msg -> Msg after 1000 -> timeout end end,
64 + brod_topic_subscriber:start_link(client1, Topic, Partitions=[Partition],
65 + _ConsumerConfig=[{begin_offset, FirstOffset}],
66 + _CommittdOffsets=[], message, SubscriberCallbackFun,
67 + _CallbackState=self()),
68 + AckCb = fun(Partition, BaseOffset) -> io:format(user, "\nProduced to partition ~p at base-offset ~p\n", [Partition, BaseOffset]) end,
69 + ok = brod:produce_cb(client1, Topic, Partition, <<>>, [{<<"key3">>, <<"value3">>}], AckCb).
70 + Receive().
71 + Receive().
72 + {ok, {_, [Msg]}} = brod:fetch(KafkaBootstrapEndpoints, Topic, Partition, FirstOffset + 2), Msg.
73 + ```
74 +
75 + Example outputs:
76 +
77 + ```erlang
78 + #kafka_message{offset = 0,key = <<"key1">>,
79 + value = <<"value1">>,ts_type = create,ts = 1531995555085,
80 + headers = []}
81 + #kafka_message{offset = 1,key = <<"key2">>,
82 + value = <<"value2">>,ts_type = create,ts = 1531995555107,
83 + headers = []}
84 + Produced to partition 0 at base-offset 406
85 + #kafka_message{offset = 2,key = <<"key3">>,
86 + value = <<"value3">>,ts_type = create,ts = 1531995555129,
87 + headers = []}
90 88 ```
91 89
92 90 # Overview
  @@ -154,7 +152,7 @@ Put below configs to client config in sys.config or app env:
154 152 {default_producer_config, []}
155 153 ```
156 154
157 - ## Start a producer on demand
155 + ## Start a Producer on Demand
158 156
159 157 ```erlang
160 158 brod:start_producer(_Client = brod_client_1,
  @@ -162,30 +160,41 @@ brod:start_producer(_Client = brod_client_1,
162 160 _ProducerConfig = []).
163 161 ```
164 162
165 - ## Produce to a known topic-partition:
163 + ## Supported Message Input Format
164 +
165 + Brod supports below produce APIs:
166 +
167 + - `brod:produce`: Async produce with ack message sent back to caller.
168 + - `brod:produce_cb`: Async produce with a callback evaluated when ack is received.
169 + - `brod:produce_sync`: Sync produce returns `ok`.
170 + - `brod:produce_sync_offset`: Sync produce returns `{ok, BaseOffset}`.
171 +
172 + The `Value` arg in these APIs can be:
173 +
174 + - `binary()`: One single message
175 + - `{brod:msg_ts(), binary()}`: One single message with its create-time timestamp
176 + - `#{ts => brod:msg_ts(), value => binary(), headers => [{_, _}]}`:
177 + One single message. If this map does not have a `key` field, the `Key` argument is used.
178 + - `[{K, V} | {T, K, V}]`: A batch, where `V` could be a nested list of such representation.
179 + - `[#{key => K, value => V, ts => T, headers => [{_, _}]}]`: A batch.
180 +
181 + When `Value` is a batch, the `Key` argument is only used as partitioner input.
182 + All messages are unified into a batch format of below spec:
183 + `[#{key => K, value => V, ts => T, headers => [{_, _}]}]`.
184 + `ts` field is dropped for kafka prior to version `0.10`
185 + `headers` field is dropped for kafka prior to version `0.11`
186 +
187 + ## Synchronized Produce API
166 188
167 189 ```erlang
168 - {ok, CallRef} =
169 - brod:produce(_Client = brod_client_1,
170 - _Topic = <<"brod-test-topic-1">>,
171 - _Partition = 0
172 - _Key = <<"some-key">>
173 - _Value = <<"some-value">>),
174 -
175 - %% just to illustrate what message to expect
176 - receive
177 - #brod_produce_reply{ call_ref = CallRef
178 - , result = brod_produce_req_acked
179 - } ->
180 - ok
181 - after 5000 ->
182 - erlang:exit(timeout)
183 - end.
190 + brod:produce_sync(_Client = brod_client_1,
191 + _Topic = <<"brod-test-topic-1">>,
192 + _Partition = 0
193 + _Key = <<"some-key">>
194 + _Value = <<"some-value">>).
184 195 ```
185 196
186 - ## Synchronized produce request
187 -
188 - Block calling process until Kafka confirmed the message:
197 + Or block calling process until Kafka confirmed the message:
189 198
190 199 ```erlang
191 200 {ok, CallRef} =
  @@ -197,17 +206,15 @@ Block calling process until Kafka confirmed the message:
197 206 brod:sync_produce_request(CallRef).
198 207 ```
199 208
200 - or the same in one call:
209 + ## Produce One Message and Receive Its Offset in Kafka
201 210
202 211 ```erlang
203 - brod:produce_sync(_Client = brod_client_1,
204 - _Topic = <<"brod-test-topic-1">>,
205 - _Partition = 0
206 - _Key = <<"some-key">>
207 - _Value = <<"some-value">>).
212 + Client = brod_client_1,
213 + Topic = <<"brod-test-topic-1">>,
214 + {ok, Offset} = brod:produce_sync_offset(Client, Topic, 0, <<>>, <<"value">>).
208 215 ```
209 216
210 - ## Produce with random partitioner
217 + ## Produce with Random Partitioner
211 218
212 219 ```erlang
213 220 Client = brod_client_1,
  @@ -215,60 +222,54 @@ Topic = <<"brod-test-topic-1">>,
215 222 PartitionFun = fun(_Topic, PartitionsCount, _Key, _Value) ->
216 223 {ok, crypto:rand_uniform(0, PartitionsCount)}
217 224 end,
218 - {ok, CallRef} = brod:produce(Client, Topic, PartitionFun, Key, Value).
225 + ok = brod:produce_sync(Client, Topic, PartitionFun, Key, Value).
219 226 ```
220 227
221 - ## Produce a batch of (maybe nested) Key-Value list
228 + ## Produce a Batch
222 229
223 230 ```erlang
224 - %% The top-level key is used for partitioning
225 - %% and nested keys are discarded.
226 - %% Nested messages are serialized into a message set to the same partition.
227 231 brod:produce(_Client = brod_client_1,
228 232 _Topic = <<"brod-test-topic-1">>,
229 233 _Partition = MyPartitionerFun
230 234 _Key = KeyUsedForPartitioning
231 - _Value = [ {<<"k1", <<"v1">>}
232 - , {<<"k2", <<"v2">>}
233 - , { _KeyDiscarded = <<>>
234 - , [ {<<"k3">>, <<"v3">>}
235 - , {<<"k4">>, <<"v4">>}
236 - ]}
235 + _Value = [ #{key => "k1" value => "v1", headers = [{"foo", "bar"}]}
236 + , #{key => "k2" value => "v2"}
237 237 ]).
238 238 ```
239 239
240 - ## Produce message create time.
240 + ## Handle Acks from Kafka as Messages
241 241
242 - ```erlang
243 - brod:produce(Client, Topic, Partition, _Key = <<>>,
244 - _Value = [{Ts1, Key1, Value1}, {Ts2, Key2, Value2}]).
245 - ```
246 -
247 - ## Handle acks from kafka
248 -
249 - Unless brod:produce_sync was called, callers of brod:produce should
250 - expect a message of below pattern for each produce call.
251 - Add `-include_lib("brod/include/brod.hrl").` to use the record.
242 + For async produce APIs `brod:produce/3` and `brod:produce/5`,
243 + the caller should expect a message of below pattern for each produce call.
252 244
253 245 ```erlang
254 246 #brod_produce_reply{ call_ref = CallRef %% returned from brod:produce
255 247 , result = brod_produce_req_acked
256 248 }
257 249 ```
250 + Add `-include_lib("brod/include/brod.hrl").` to use the record.
258 251
259 - NOTE: If required_acks is set to 0 in producer config,
260 - kafka will NOT ack the requests, and the reply message is sent back
261 - to caller immediately after the message has been sent to the socket process.
262 -
263 - In case the brod:produce caller is a process like gen_server which
252 + In case the `brod:produce` caller is a process like `gen_server` which
264 253 receives ALL messages, the callers should keep the call references in its
265 254 looping state and match the replies against them when received.
266 - Otherwise brod:sync_produce_request/1 can be used to block-wait for acks.
255 + Otherwise `brod:sync_produce_request/1` can be used to block-wait for acks.
256 +
257 + NOTE: If `required_acks` is set to `none` in producer config,
258 + kafka will NOT ack the requests, and the reply message is sent back
259 + to caller immediately after the message has been sent to the socket process.
267 260
268 261 NOTE: The replies are only strictly ordered per-partition.
269 262 i.e. if the caller is producing to two or more partitions,
270 263 it may receive replies ordered differently than in which order
271 - brod:produce API was called.
264 + `brod:produce` API was called.
265 +
266 + ## Handle Acks from Kafka in Callback Function
267 +
268 + Async APIs `brod:produce_cb/4` and `brod:produce_cb/6` allow callers to
269 + provided a callback function to handle acknowledgements from kafka.
270 + In this case, the caller may want to monitor the producer process because
271 + then they know that the callbacks will not be evaluated if the producer is 'DOWN',
272 + and there is perhaps a need for retry.
272 273
273 274 # Consumers
274 275
  @@ -371,9 +372,13 @@ start(ClientId) ->
371 372
372 373 # Authentication support
373 374
374 - brod supports SASL PLAIN authentication mechanism out of the box. To use it
375 - add `{sasl, {plain, Username, Password}}` to client config. Also, brod has authentication
376 - plugins support. Authentication callback module should implement `brod_auth_backend` behaviour.
375 + brod supports SASL `PLAIN`, `SCRAM-SHA-256` and `SCRAM-SHA-512` authentication mechanisms out of the box.
376 + To use it, add `{sasl, {Mechanism, Username, Password}}` or `{sasl, {Mechanism, File}}` to client config.
377 + Where `Mechanism` is `plain | scram_sha_256 | scram_sha_512`, and `File` is the path to a text file
378 + which contains two lines, first line for username and second line for password
379 +
380 + Also, brod has authentication plugins support with `{sasl, {callback, Module, Opts}}` in client config.
381 + Authentication callback module should implement `brod_auth_backend` behaviour.
377 382 Auth function spec:
378 383
379 384 ```erlang
  @@ -403,7 +408,6 @@ Partition = 0.
403 408 brod:get_metadata(Hosts).
404 409 brod:get_metadata(Hosts, [Topic]).
405 410 brod:resolve_offset(Hosts, Topic, Partition).
406 - brod:fetch(Hosts, Topic, Partition, 1).
407 411 ```
408 412
409 413 # brod-cli: A command line tool to interact with Kafka
  @@ -411,38 +415,32 @@ brod:fetch(Hosts, Topic, Partition, 1).
411 415 This will build a self-contained binary with brod application
412 416
413 417 ```bash
414 - make rel
415 - ./_rel/brod/bin/brod --help
418 + make brod-cli
419 + _build/brod_cli/rel/brod/bin/brod -h
416 420 ```
417 421
418 422 Disclaimer: This script is NOT designed for use cases where fault-tolerance is a hard requirement.
419 423 As it may crash when e.g. kafka cluster is temporarily unreachable,
420 424 or (for fetch command) when the parition leader migrates to another broker in the cluster.
421 425
422 - Start an Erlang shell with brod started
423 -
424 - ```bash
425 - ./_rel/brod/bin/brod-i
426 - ```
427 -
428 - ## brod-cli examples:
426 + ## brod-cli examples (with `alias brod=_build/brod_cli/rel/brod/bin/brod`):
429 427
430 428 ### Fetch and print metadata
431 429 ```
432 - ./scripts/brod meta -b localhost
430 + brod meta -b localhost
433 431 ```
434 432
435 433 ### Produce a Message
436 434
437 435 ```
438 - ./scripts/brod send -b localhost -t test-topic -p 0 -k "key" -v "value"
436 + brod send -b localhost -t test-topic -p 0 -k "key" -v "value"
439 437
440 438 ```
441 439
442 440 ### Fetch a Message
443 441
444 442 ```
445 - ./scripts/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])'
443 + 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])'
446 444 ```
447 445
448 446 Bound variables to be used in `--fmt` expression:
  @@ -450,7 +448,6 @@ Bound variables to be used in `--fmt` expression:
450 448 - `Offset`: Message offset
451 449 - `Key`: Kafka key
452 450 - `Value`: Kafka Value
453 - - `CRC`: Message CRC
454 451 - `TsType`: Timestamp type either `create` or `append`
455 452 - `Ts`: Timestamp, `-1` as no value
456 453
  @@ -458,33 +455,33 @@ Bound variables to be used in `--fmt` expression:
458 455
459 456 Send `README.md` to kafka one line per kafka message
460 457 ```
461 - ./scripts/brod pipe -b localhost:9092 -t test-topic -p 0 -s @./README.md
458 + brod pipe -b localhost:9092 -t test-topic -p 0 -s @./README.md
462 459 ```
463 460
464 461 ### Resolve Offset
465 462
466 463 ```
467 - ./scripts/brod offset -b localhost:9092 -t test-topic -p 0
464 + brod offset -b localhost:9092 -t test-topic -p 0
468 465 ```
469 466
470 467 ### List or Describe Groups
471 468
472 469 ```
473 470 # List all groups
474 - ./scripts/brod groups -b localhost:9092
471 + brod groups -b localhost:9092
475 472
476 473 # Describe groups
477 - ./scripts/brod groups -b localhost:9092 --ids group-1,group-2
474 + brod groups -b localhost:9092 --ids group-1,group-2
478 475 ```
479 476
480 477 ### Display Committed Offsets
481 478
482 479 ```
483 480 # all topics
484 - ./scripts/brod commits -b localhost:9092 --id the-group-id --describe
481 + brod commits -b localhost:9092 --id the-group-id --describe
485 482
486 483 # a specific topic
487 - ./scripts/brod commits -b localhost:9092 --id the-group-id --describe --topic topic-name
484 + brod commits -b localhost:9092 --id the-group-id --describe --topic topic-name
488 485 ```
489 486
490 487 ### Commit Offsets
  @@ -493,15 +490,23 @@ NOTE: This feature is designed for force overwriting commits, not for regular us
493 490
494 491 ```
495 492 # Commit 'latest' offsets of all partitions with 2 days retention
496 - ./scripts/brod commits -b localhost:9092 --id the-group-id --topic topic-name --offsets latest --retention 2d
493 + brod commits -b localhost:9092 --id the-group-id --topic topic-name --offsets latest --retention 2d
497 494
498 495 # Commit offset=100 for partition 0 and 200 for partition 1
499 - ./scripts/brod commits -b localhost:9092 --id the-group-id --topic topic-name --offsets "0:100,1:200"
496 + brod commits -b localhost:9092 --id the-group-id --topic topic-name --offsets "0:100,1:200"
500 497
501 498 # Use --retention 0 to delete commits (may linger in kafka before cleaner does its job)
502 - ./scripts/brod commits -b localhost:9092 --id the-group-id --topic topic-name --offsets latest --retention 0
499 + brod commits -b localhost:9092 --id the-group-id --topic topic-name --offsets latest --retention 0
503 500
504 501 # Try join an active consumer group using 'range' protocol and steal one partition assignment then commit offset=10000
505 - ./scripts/brod commits -b localhost:9092 -i the-group-id -t topic-name -o "0:10000" --protocol range
502 + brod commits -b localhost:9092 -i the-group-id -t topic-name -o "0:10000" --protocol range
506 503 ```
507 504
505 + ## TODOs
506 +
507 + * HTML tagged EDoc
508 + * Support scram-sasl in brod-cli
509 + * lz4 compression & decompression
510 + * Create/delete topic APIs
511 + * Transactional produce APIs
512 +
unknownerlang.mk
File is too large to be displayed (100 KB limit).
  @@ -1,12 +1,12 @@
1 1 {<<"name">>,<<"brod">>}.
2 - {<<"version">>,<<"3.5.2">>}.
2 + {<<"version">>,<<"3.6.0">>}.
3 3 {<<"requirements">>,
4 4 #{<<"kafka_protocol">> =>
5 5 #{<<"app">> => <<"kafka_protocol">>,<<"optional">> => false,
6 - <<"requirement">> => <<"1.1.2">>},
6 + <<"requirement">> => <<"2.1.1">>},
7 7 <<"supervisor3">> =>
8 8 #{<<"app">> => <<"supervisor3">>,<<"optional">> => false,
9 - <<"requirement">> => <<"1.1.5">>}}}.
9 + <<"requirement">> => <<"1.1.7">>}}}.
10 10 {<<"app">>,<<"brod">>}.
11 11 {<<"maintainers">>,[<<"Ivan Dyachkov">>,<<"Zaiming Shi">>]}.
12 12 {<<"precompiled">>,false}.
  @@ -15,15 +15,13 @@
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/brod.erl">>,<<"src/brod_auth_backend.erl">>,
19 - <<"src/brod_cg_commits.erl">>,<<"src/brod_cli.erl">>,
18 + <<"src/brod.erl">>,<<"src/brod_cg_commits.erl">>,<<"src/brod_cli.erl">>,
20 19 <<"src/brod_cli_pipe.erl">>,<<"src/brod_client.erl">>,
21 20 <<"src/brod_consumer.erl">>,<<"src/brod_consumers_sup.erl">>,
22 21 <<"src/brod_group_coordinator.erl">>,<<"src/brod_group_member.erl">>,
23 22 <<"src/brod_group_subscriber.erl">>,<<"src/brod_kafka_apis.erl">>,
24 - <<"src/brod_kafka_request.erl">>,<<"src/brod_kafka_requests.erl">>,
25 - <<"src/brod_producer.erl">>,<<"src/brod_producer_buffer.erl">>,
26 - <<"src/brod_producers_sup.erl">>,<<"src/brod_sock.erl">>,
23 + <<"src/brod_kafka_request.erl">>,<<"src/brod_producer.erl">>,
24 + <<"src/brod_producer_buffer.erl">>,<<"src/brod_producers_sup.erl">>,
27 25 <<"src/brod_sup.erl">>,<<"src/brod_topic_subscriber.erl">>,
28 26 <<"src/brod_utils.erl">>]}.
29 27 {<<"licenses">>,[<<"Apache License 2.0">>]}.
  @@ -1,5 +1,5 @@
1 1 %%%
2 - %%% Copyright (c) 2014-2017, Klarna AB
2 + %%% Copyright (c) 2014-2018, Klarna Bank AB (publ)
3 3 %%%
4 4 %%% Licensed under the Apache License, Version 2.0 (the "License");
5 5 %%% you may not use this file except in compliance with the License.
  @@ -28,26 +28,26 @@
28 28 , partition :: brod:partition()
29 29 , high_wm_offset :: integer() %% max offset of the partition
30 30 , messages :: [brod:message()] %% exposed to brod user
31 - | kpro:incomplete_message() %% this union member
32 - %% is internal only
31 + | kpro:incomplete_batch() %% this union member
32 + %% is internal only
33 33 }).
34 34
35 35 -record(kafka_fetch_error,
36 36 { topic :: brod:topic()
37 37 , partition :: brod:partition()
38 38 , error_code :: brod:error_code()
39 - , error_desc :: binary()
39 + , error_desc = ""
40 40 }).
41 41
42 - -record(brod_call_ref, { caller :: pid()
43 - , callee :: pid()
44 - , ref :: reference()
42 + -record(brod_call_ref, { caller :: undefined | pid()
43 + , callee :: undefined | pid()
44 + , ref :: undefined | reference()
45 45 }).
46 46
47 47 -define(BROD_PRODUCE_UNKNOWN_OFFSET, -1).
48 48
49 49 -record(brod_produce_reply, { call_ref :: brod:call_ref()
50 - , base_offset :: brod:offset()
50 + , base_offset :: undefined | brod:offset()
51 51 , result :: brod:produce_result()
52 52 }).
Loading more files…