Kafka Avro CLI
A CLI for inspecting the confluent schema-registry, produce and consume avro kafka events.
Usage
You can install it locally for your project
yarn add @ovotech/kafka-avro-cli
yarn kac --helpOr globally:
yarn global add @ovotech/kafka-avro-cli
kac --helpkac <command>
Commands:
kac schema [subject] Serach for a given topic
kac topic [name] Serach for a given topic
kac create-topic <topic> Create kafka topic
kac produce [file] Produce events in kafka topic from a file
kac consume <topic> Consume events from kafka topic. Display them in the
console or save them to a file
Options:
--version Show version number [boolean]
--config, -c Path to the configuration file [default: "kac.config.json"]
--help Show help [boolean]
copyright OVO Energy 2018
Commands
Command configuration
This cli tool communicates with schema registry and kafka which require credentials and addresses. In order to configure them you need to create a kac.config.json file or to provide the path for it wit the --config option.
The file should look like this:
{
"kafkaClient": {
"kafkaHost": "kafka.example.com:13581",
"sslOptions": {
"ca": "-----BEGIN CERTIFICATE-----\n...\n-----END CERTIFICATE-----",
"key": "-----BEGIN PRIVATE KEY-----\n...\n-----END PRIVATE KEY-----",
"cert": "-----BEGIN CERTIFICATE-----\n...\n-----END CERTIFICATE-----"
}
},
"schemaRegistry": "https://example-user:example-password@registry.example.com:13584"
}-
kafkaClient- options are passed directly to the KafkaClient from node-kafka. -
schemaRegistry- url to the schema registry, with credentials information.
kac --config my/config.json ...schema command
kac.ts schema [subject]
Used to search for, filter and get details of a particular schema in the schema
registry. [subject] is a partial name of a subject. If no subject if provided,
all subjects are returned.
Options:
--version Show version number [boolean]
--config, -c Path to the configuration file [default: "kac.config.json"]
--help Show help [boolean]
--output-dir, -o Save the results into a folder. One file per versionIf more than one subject is found, it will result in a list of partial matches.
$ kac schema migration
Searching for "migration" in localhost:8081
Found 4 matching "migration"
----------------------------------------
migration-scheduled-value
migration-completed-2-value
migration-completed-value
If only one is matched, it will display the full schemas of all the versions for this subject
$ kac schema migration-scheduled-value
Searching for "migration-scheduled-value" in localhost:8081
Subject migration-scheduled-value found [ 1 ] versions
Version: 1 ----------------------------------------
{ type: 'record',
name: 'TestSchema1',
fields: [ { name: 'accountId', type: 'string' } ] }
topic command
kac topic [name]
Used to search for, filter and get details of a particular topic in the kafka
server. [name] is a partial name of a topic. If no name if provided, all topics
are returned.
Options:
--version Show version number [boolean]
--config, -c Path to the configuration file [default: "kac.config.json"]
--help Show help
If more than one topic is found, it will result in a list of partial matches.
$ kac topic migration
Searching for "migration" in localhost:29092
Found 3 matching "migration"
----------------------------------------
migration-scheduled
migration-completed-2
migration-completed
If only one is matched, it will display the partitions and offsets of a topic
$ kac topic migration-completed-2
Searching for "migration-completed-2" in localhost:29092
Partitions and their offsets for migration-completed-2
----------------------------------------
{ migration-completed-2:
{ '0': [ 143024920 ],
'1': [ 142607286 ],
'2': [ 142860803 ],
'3': [ 143160301 ] } }
create-topic command
kac create-topic <topic>
Creates a topic in the kafka server with configurable partitions count and
replication factor.
Options:
--version Show version number [boolean]
--config, -c Path to the configuration file
[default: "kac.config.json"]
--help Show help [boolean]
--partitions Number of partitions for topic [number] [default: 1]
--replication-factor [number] [default: 1]
For example:
$ kac create-topic test-topic --partitions 2
Topic created test-topic partitions 2 replication factor 1
produce command
kac produce [file]
Produce events in kafka topic from a file
Options:
--version Show version number [boolean]
--config, -c Path to the configuration file [default: "kac.config.json"]
--help Show help [boolean]
The file needed to produce the topics should look something like this:
{
"0": {
"topic": "migration-completed-2",
"schema": {
"type": "record",
"name": "TestSchema1",
"fields": [{ "name": "accountId", "type": "string" }]
},
"partition": 0,
"key": "some key",
"messages": [{ "accountId": "67096" }, { "accountId": "57096" }]
},
"1": {
"topic": "migration-scheduled",
"schema": {
"type": "record",
"name": "TestSchema1",
"fields": [{ "name": "customerId", "type": "string" }]
},
"partition": 1,
"key": "some other key",
"messages": [{ "customerId": "12321" }, { "customerId": "32131" }]
}
}This should allow you to produce events in different partitions.
consume command
kac consume <topic>
Consume events from kafka topic. Display them in the console or save them to a
file
Options:
--version Show version number [boolean]
--config, -c Path to the configuration file [default: "kac.config.json"]
--help Show help [boolean]
--group Consumer Group Id [default: a prefixed random name]
--latest Start listening from the latest offsets
[boolean] [default: false]
--tail Keep listening after all messages consumed
[boolean] [default: false]
--output-file, -o Save the results into a file, that can later be used to
produce those events
To display the contents of a topic simply write the full topic name.
The consumer group will terminate once all the messages have been consumed, but you can keep listening with the --tail option.
$ kac consume migration-completed-2
Consume messages in migration-completed-2
----------------------------------------
Offset 0/1 partition 0 key null
BalanceMigrationCompletedEvent {
accountId: '6666666' }
----------------------------------------
Offset 1/1 partition 0 key null
BalanceMigrationCompletedEvent {
accountId: '5555555' }
----------------------------------------
Offset 0/1 partition 1 key null
BalanceMigrationCompletedEvent {
accountId: '4444444' }
----------------------------------------
Offset 1/1 partition 1 key null
BalanceMigrationCompletedEvent {
accountId: '3333333' }
----------------------------------------
[============================================================] 100% Elapsed: 0.0s
Consumed 4 messages
- Partition 0 : 2 messages.
- Partition 1 : 2 messages.
You can also write the output to a file with the --output-file option. The contents of that file would be in the same format, as the one used to produce the events with the produce command.
$ kac consume migration-completed-2 --output-file test2.json
Writing to file test2.json from topic migration-completed-2
----------------------------------------
[============================================================] 100% Elapsed: 0.0s
Consumed 8 messages
- Partition 0 : 4 messages.
- Partition 1 : 4 messages.
Running the tests
The tests require a running kafka server and schema-registry service. This is setup easily with a docker-compose from root project folder:
docker-compose upThen you can run the tests with:
yarn testCoding style (linting, etc) tests
Style is maintained with prettier and tslint
yarn lint
Deployment
Deployment is preferment by lerna automatically on merge / push to master, but you'll need to bump the package version numbers yourself. Only updated packages with newer versions will be pushed to the npm registry.
Contributing
Have a bug? File an issue with a simple example that reproduces this so we can take a look & confirm.
Want to make a change? Submit a PR, explain why it's useful, and make sure you've updated the docs (this file) and the tests (see test folder).
License
This project is licensed under Apache 2 - see the LICENSE file for details