Kafka
kafka create_topickafka create_topickafka producekafka producekafka produce_batchkafka produce_batchkafka consumekafka consumekafka topicskafka topic_infokafka lagkafka lagkafka delete_topic
kafka create_topic <config> <topic>
Description
Create a topic with one partition and the replication of the broker. Result: topic, created, partitions, replication. An existing topic gives an error
Create a topic with one partition and the replication of the broker. Result: topic, created, partitions, replication. An existing topic gives an error
Parameters
config: The connection, as JSON: bootstrap_servers, timeout_ms (10000), security_protocol, sasl_mechanism, username with password or password_env, properties - String - required
topic: The topic name (1 to 249 letters, digits, . _ -) - String - required
config: The connection, as JSON: bootstrap_servers, timeout_ms (10000), security_protocol, sasl_mechanism, username with password or password_env, properties - String - required
topic: The topic name (1 to 249 letters, digits, . _ -) - String - required
admin
kafka create_topic "{\"bootstrap_servers\": \"127.0.0.1:29092\"}" "doc_events"
mentdb
{"topic": "doc_events", "created": true, "partitions": 1, "replication": 1}
kafka create_topic <config> <topic> <json_params>
Description
Create a topic with parameters: partitions (1), replication (default of the broker), configs (topic settings, JSON object)
Create a topic with parameters: partitions (1), replication (default of the broker), configs (topic settings, JSON object)
Parameters
config: The connection, as JSON: bootstrap_servers, timeout_ms (10000), security_protocol, sasl_mechanism, username with password or password_env, properties - String - required
topic: The topic name (1 to 249 letters, digits, . _ -) - String - required
json_params: The parameters - String - required
config: The connection, as JSON: bootstrap_servers, timeout_ms (10000), security_protocol, sasl_mechanism, username with password or password_env, properties - String - required
topic: The topic name (1 to 249 letters, digits, . _ -) - String - required
json_params: The parameters - String - required
admin
kafka create_topic "{\"bootstrap_servers\": \"127.0.0.1:29092\"}" "doc_orders" "{\"partitions\": 3, \"replication\": 1, \"configs\": {\"retention.ms\": \"604800000\"}}"
mentdb
{"topic": "doc_orders", "created": true, "partitions": 3, "replication": 1}
kafka produce <config> <topic> <value>
Description
Send one message and wait for the acknowledgement of the broker. Result: topic, partition, offset, timestamp (milliseconds)
Send one message and wait for the acknowledgement of the broker. Result: topic, partition, offset, timestamp (milliseconds)
Parameters
config: The connection, as JSON: bootstrap_servers, timeout_ms (10000), security_protocol, sasl_mechanism, username with password or password_env, properties - String - required
topic: The topic name (1 to 249 letters, digits, . _ -) - String - required
value: The message, UTF-8 text - String - required
config: The connection, as JSON: bootstrap_servers, timeout_ms (10000), security_protocol, sasl_mechanism, username with password or password_env, properties - String - required
topic: The topic name (1 to 249 letters, digits, . _ -) - String - required
value: The message, UTF-8 text - String - required
admin
kafka produce "{\"bootstrap_servers\": \"127.0.0.1:29092\"}" "doc_events" "{\"event\": \"login\", \"user\": \"jim\"}"
mentdb
{"topic": "doc_events", "partition": 0, "offset": 0, "timestamp": 1793766367798}
with SCRAM authentication, the password in an environment variable (fixed result: real local Apache Kafka 3.9.1 KRaft test broker)
admin
kafka produce "{\"bootstrap_servers\": \"127.0.0.1:29094\", \"security_protocol\": \"SASL_PLAINTEXT\", \"sasl_mechanism\": \"SCRAM-SHA-256\", \"username\": \"bob\", \"password_env\": \"KAFKA_PASSWORD\", \"timeout_ms\": 5000}" "doc_events" "hello"
mentdb
{"topic": "doc_events", "partition": 0, "offset": 1, "timestamp": 1793766367883}
kafka produce <config> <topic> <value> <json_params>
Description
Send one message with parameters: key, headers (JSON object or array of {key, value}), partition (else chosen by the key), acks (all | 1 | 0: with 0 the offset is -1), compression (none | gzip | snappy | lz4 | zstd)
Send one message with parameters: key, headers (JSON object or array of {key, value}), partition (else chosen by the key), acks (all | 1 | 0: with 0 the offset is -1), compression (none | gzip | snappy | lz4 | zstd)
Parameters
config: The connection, as JSON: bootstrap_servers, timeout_ms (10000), security_protocol, sasl_mechanism, username with password or password_env, properties - String - required
topic: The topic name (1 to 249 letters, digits, . _ -) - String - required
value: The message - String - required
json_params: The parameters - String - required
config: The connection, as JSON: bootstrap_servers, timeout_ms (10000), security_protocol, sasl_mechanism, username with password or password_env, properties - String - required
topic: The topic name (1 to 249 letters, digits, . _ -) - String - required
value: The message - String - required
json_params: The parameters - String - required
admin
kafka produce "{\"bootstrap_servers\": \"127.0.0.1:29092\"}" "doc_orders" "order 1" "{\"key\": \"customer-42\", \"headers\": {\"source\": \"mentdb\", \"version\": 1}, \"partition\": 1, \"acks\": \"all\", \"compression\": \"zstd\"}"
mentdb
{"topic": "doc_orders", "partition": 1, "offset": 0, "timestamp": 1793766367901}
kafka produce_batch <config> <topic> <json_rows>
Description
Send a JSON array of rows, one message per row (the row as JSON). Result: topic, sent, failed, errors (row and message, first 20). After a time-out or an authentication error the next rows are not tried: failed counts them too
Send a JSON array of rows, one message per row (the row as JSON). Result: topic, sent, failed, errors (row and message, first 20). After a time-out or an authentication error the next rows are not tried: failed counts them too
Parameters
config: The connection, as JSON: bootstrap_servers, timeout_ms (10000), security_protocol, sasl_mechanism, username with password or password_env, properties - String - required
topic: The topic name (1 to 249 letters, digits, . _ -) - String - required
json_rows: The rows: JSON array of objects - String - required
config: The connection, as JSON: bootstrap_servers, timeout_ms (10000), security_protocol, sasl_mechanism, username with password or password_env, properties - String - required
topic: The topic name (1 to 249 letters, digits, . _ -) - String - required
json_rows: The rows: JSON array of objects - String - required
admin
kafka produce_batch "{\"bootstrap_servers\": \"127.0.0.1:29092\"}" "doc_events" "[{\"id\": 1, \"shop\": \"Saint-Denis\", \"amount\": 120.5}, {\"id\": 2, \"shop\": \"Le Tampon\", \"amount\": 87.25}, {\"id\": 3, \"shop\": \"Saint-Pierre\", \"amount\": 12}]"
mentdb
{"topic": "doc_events", "sent": 3, "failed": 0, "errors": []}
kafka produce_batch <config> <topic> <json_rows> <json_params>
Description
Send rows with parameters: key_field (the field used as key), value_field (the field used as value, a null value is a tombstone; else the row as JSON), max_rows (100000, 1000000 at most: more rows give an error before sending), partition, acks, compression
Send rows with parameters: key_field (the field used as key), value_field (the field used as value, a null value is a tombstone; else the row as JSON), max_rows (100000, 1000000 at most: more rows give an error before sending), partition, acks, compression
Parameters
config: The connection, as JSON: bootstrap_servers, timeout_ms (10000), security_protocol, sasl_mechanism, username with password or password_env, properties - String - required
topic: The topic name (1 to 249 letters, digits, . _ -) - String - required
json_rows: The rows - String - required
json_params: The parameters - String - required
config: The connection, as JSON: bootstrap_servers, timeout_ms (10000), security_protocol, sasl_mechanism, username with password or password_env, properties - String - required
topic: The topic name (1 to 249 letters, digits, . _ -) - String - required
json_rows: The rows - String - required
json_params: The parameters - String - required
admin
kafka produce_batch "{\"bootstrap_servers\": \"127.0.0.1:29092\"}" "doc_orders" "[{\"id\": 1, \"shop\": \"Saint-Denis\", \"amount\": 120.5}, {\"id\": 2, \"shop\": \"Le Tampon\", \"amount\": 87.25}, {\"id\": 3, \"shop\": \"Saint-Pierre\", \"amount\": 12}]" "{\"key_field\": \"id\", \"compression\": \"lz4\"}"
mentdb
{"topic": "doc_orders", "sent": 3, "failed": 0, "errors": []}
kafka consume <config> <topic>
Description
Read up to 100 messages of all the partitions from the beginning, without a group, nothing committed. Result: rows with topic, partition, offset, timestamp, key, value, headers
Read up to 100 messages of all the partitions from the beginning, without a group, nothing committed. Result: rows with topic, partition, offset, timestamp, key, value, headers
Parameters
config: The connection, as JSON: bootstrap_servers, timeout_ms (10000), security_protocol, sasl_mechanism, username with password or password_env, properties - String - required
topic: The topic name (1 to 249 letters, digits, . _ -) - String - required
config: The connection, as JSON: bootstrap_servers, timeout_ms (10000), security_protocol, sasl_mechanism, username with password or password_env, properties - String - required
topic: The topic name (1 to 249 letters, digits, . _ -) - String - required
admin
kafka consume "{\"bootstrap_servers\": \"127.0.0.1:29092\"}" "doc_events"
mentdb
[{"topic": "doc_events", "partition": 0, "offset": 0, "timestamp": 1793766367798, "key": null, "value": "{\"event\": \"login\", \"user\": \"jim\"}", "headers": []}, {"topic": "doc_events", "partition": 0, "offset": 1, "timestamp": 1793766367883, "key": null, "value": "hello", "headers": []}, {"topic": "doc_events", "partition": 0, "offset": 2, "timestamp": 1793766368050, "key": null, "value": "{\"id\": 1, \"shop\": \"Saint-Denis\", \"amount\": 120.5}", "headers": []}, {"topic": "doc_events", "partition": 0, "offset": 3, "timestamp": 1793766368050, "key": null, "value": "{\"id\": 2, \"shop\": \"Le Tampon\", \"amount\": 87.25}", "headers": []}, {"topic": "doc_events", "partition": 0, "offset": 4, "timestamp": 1793766368050, "key": null, "value": "{\"id\": 3, \"shop\": \"Saint-Pierre\", \"amount\": 12}", "headers": []}]
kafka consume <config> <topic> <json_params>
Description
Read with parameters: group_id (join a consumer group), from (earliest | latest | committed; committed is the default with a group, at the committed offsets or the beginning if none, and earliest without), max_messages (100, 100000 at most), max_bytes (64 MB, 1 GB at most, keys + values + headers; a single larger message gives an error), timeout_ms (5000: time to wait for messages, 0 = one short poll), commit (true, with a group), partition and offset (without a group: a partition, from an offset that must exist)
Read with parameters: group_id (join a consumer group), from (earliest | latest | committed; committed is the default with a group, at the committed offsets or the beginning if none, and earliest without), max_messages (100, 100000 at most), max_bytes (64 MB, 1 GB at most, keys + values + headers; a single larger message gives an error), timeout_ms (5000: time to wait for messages, 0 = one short poll), commit (true, with a group), partition and offset (without a group: a partition, from an offset that must exist)
Parameters
config: The connection, as JSON: bootstrap_servers, timeout_ms (10000), security_protocol, sasl_mechanism, username with password or password_env, properties - String - required
topic: The topic name (1 to 249 letters, digits, . _ -) - String - required
json_params: The parameters - String - required
config: The connection, as JSON: bootstrap_servers, timeout_ms (10000), security_protocol, sasl_mechanism, username with password or password_env, properties - String - required
topic: The topic name (1 to 249 letters, digits, . _ -) - String - required
json_params: The parameters - String - required
admin
kafka consume "{\"bootstrap_servers\": \"127.0.0.1:29092\"}" "doc_orders" "{\"group_id\": \"doc_billing\", \"max_messages\": 2}"
mentdb
[{"topic": "doc_orders", "partition": 2, "offset": 0, "timestamp": 1793766368189, "key": "2", "value": "{\"id\": 2, \"shop\": \"Le Tampon\", \"amount\": 87.25}", "headers": []}, {"topic": "doc_orders", "partition": 2, "offset": 1, "timestamp": 1793766368189, "key": "3", "value": "{\"id\": 3, \"shop\": \"Saint-Pierre\", \"amount\": 12}", "headers": []}]
one partition from an offset (fixed result: real local Apache Kafka 3.9.1 KRaft test broker)
admin
kafka consume "{\"bootstrap_servers\": \"127.0.0.1:29092\"}" "doc_orders" "{\"partition\": 1, \"offset\": 0, \"max_messages\": 1}"
mentdb
[{"topic": "doc_orders", "partition": 1, "offset": 0, "timestamp": 1793766367901, "key": "customer-42", "value": "order 1", "headers": [{"key": "source", "value": "mentdb"}, {"key": "version", "value": "1"}]}]
new messages only: nothing arrived during the second of waiting (fixed result: real local Apache Kafka 3.9.1 KRaft test broker)
admin
kafka consume "{\"bootstrap_servers\": \"127.0.0.1:29092\"}" "doc_orders" "{\"from\": \"latest\", \"timeout_ms\": 1000}"
mentdb
[]
kafka topics <config>
Description
The topics of the cluster, sorted: name, internal
The topics of the cluster, sorted: name, internal
Parameters
config: The connection, as JSON: bootstrap_servers, timeout_ms (10000), security_protocol, sasl_mechanism, username with password or password_env, properties - String - required
config: The connection, as JSON: bootstrap_servers, timeout_ms (10000), security_protocol, sasl_mechanism, username with password or password_env, properties - String - required
admin
kafka topics "{\"bootstrap_servers\": \"127.0.0.1:29092\"}"
mentdb
[{"name": "__consumer_offsets", "internal": true}, {"name": "doc_events", "internal": false}, {"name": "doc_orders", "internal": false}]
kafka topic_info <config> <topic>
Description
A topic: partition_count, messages (sum of end - begin offsets), and for each partition: leader, replicas, isr, begin_offset, end_offset
A topic: partition_count, messages (sum of end - begin offsets), and for each partition: leader, replicas, isr, begin_offset, end_offset
Parameters
config: The connection, as JSON: bootstrap_servers, timeout_ms (10000), security_protocol, sasl_mechanism, username with password or password_env, properties - String - required
topic: The topic name (1 to 249 letters, digits, . _ -) - String - required
config: The connection, as JSON: bootstrap_servers, timeout_ms (10000), security_protocol, sasl_mechanism, username with password or password_env, properties - String - required
topic: The topic name (1 to 249 letters, digits, . _ -) - String - required
admin
kafka topic_info "{\"bootstrap_servers\": \"127.0.0.1:29092\"}" "doc_orders"
mentdb
{"topic": "doc_orders", "internal": false, "partition_count": 3, "messages": 4, "partitions": [{"partition": 0, "leader": 1, "replicas": [1], "isr": [1], "begin_offset": 0, "end_offset": 1}, {"partition": 1, "leader": 1, "replicas": [1], "isr": [1], "begin_offset": 0, "end_offset": 1}, {"partition": 2, "leader": 1, "replicas": [1], "isr": [1], "begin_offset": 0, "end_offset": 2}]}
kafka lag <config> <group_id>
Description
The lag of a consumer group: group_id, topic, partition, committed, end, lag (partitions with a committed offset; an unknown group gives [])
The lag of a consumer group: group_id, topic, partition, committed, end, lag (partitions with a committed offset; an unknown group gives [])
Parameters
config: The connection, as JSON: bootstrap_servers, timeout_ms (10000), security_protocol, sasl_mechanism, username with password or password_env, properties - String - required
group_id: The consumer group - String - required
config: The connection, as JSON: bootstrap_servers, timeout_ms (10000), security_protocol, sasl_mechanism, username with password or password_env, properties - String - required
group_id: The consumer group - String - required
admin
kafka lag "{\"bootstrap_servers\": \"127.0.0.1:29092\"}" "doc_billing"
mentdb
[{"group_id": "doc_billing", "topic": "doc_orders", "partition": 2, "committed": 2, "end": 2, "lag": 0}]
kafka lag <config> <group_id> <topic>
Description
The lag of a consumer group on one topic
The lag of a consumer group on one topic
Parameters
config: The connection, as JSON: bootstrap_servers, timeout_ms (10000), security_protocol, sasl_mechanism, username with password or password_env, properties - String - required
group_id: The consumer group - String - required
topic: The topic name (1 to 249 letters, digits, . _ -) - String - required
config: The connection, as JSON: bootstrap_servers, timeout_ms (10000), security_protocol, sasl_mechanism, username with password or password_env, properties - String - required
group_id: The consumer group - String - required
topic: The topic name (1 to 249 letters, digits, . _ -) - String - required
admin
kafka lag "{\"bootstrap_servers\": \"127.0.0.1:29092\"}" "doc_billing" "doc_orders"
mentdb
[{"group_id": "doc_billing", "topic": "doc_orders", "partition": 2, "committed": 2, "end": 2, "lag": 0}]
kafka delete_topic <config> <topic>
Description
Delete a topic and its messages. Result: topic, deleted
Delete a topic and its messages. Result: topic, deleted
Parameters
config: The connection, as JSON: bootstrap_servers, timeout_ms (10000), security_protocol, sasl_mechanism, username with password or password_env, properties - String - required
topic: The topic name (1 to 249 letters, digits, . _ -) - String - required
config: The connection, as JSON: bootstrap_servers, timeout_ms (10000), security_protocol, sasl_mechanism, username with password or password_env, properties - String - required
topic: The topic name (1 to 249 letters, digits, . _ -) - String - required
admin
kafka delete_topic "{\"bootstrap_servers\": \"127.0.0.1:29092\"}" "doc_events"
mentdb
{"topic": "doc_events", "deleted": true}