From 246361f33d88e5f17bfe4274b62bc6baa164ffc7 Mon Sep 17 00:00:00 2001
From: zmstone <zmstone@gmail.com>
Date: Sat, 5 Oct 2024 16:07:02 +0200
Subject: [PATCH] refactor: delete brod_cli code

moved to https://github.com/kafka4beam/brod-cli
---
 .github/workflows/build.yml  |    2 -
 CHANGELOG.md                 |    3 +
 Makefile                     |    7 -
 README.md                    |   99 ---
 elvis.config                 |    2 -
 rebar.config                 |   17 +-
 scripts/brod                 |   32 -
 scripts/brod.escript         |   31 -
 src/brod.erl                 |   10 +-
 src/brod_cli.erl             | 1320 ----------------------------------
 src/brod_cli_pipe.erl        |  390 ----------
 test/brod_cli_pipe_tests.erl |  162 -----
 test/brod_cli_tests.erl      |  253 -------
 13 files changed, 5 insertions(+), 2323 deletions(-)
 delete mode 100755 scripts/brod
 delete mode 100755 scripts/brod.escript
 delete mode 100644 src/brod_cli.erl
 delete mode 100644 src/brod_cli_pipe.erl
 delete mode 100644 test/brod_cli_pipe_tests.erl
 delete mode 100644 test/brod_cli_tests.erl

diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml
index 11954fbc..b4097e95 100644
--- a/.github/workflows/build.yml
+++ b/.github/workflows/build.yml
@@ -63,8 +63,6 @@ jobs:
           rebar3-version: ${{ env.REBAR_VERSION }}
       - name: Compile
         run: rebar3 do compile
-      - name: Make brod_cli script
-        run: rebar3 as brod_cli escriptize
       - name: Run tests
         run: |
           export KAFKA_VERSION=${{ matrix.kafka }}
diff --git a/CHANGELOG.md b/CHANGELOG.md
index efe60210..e28a5215 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,8 @@
 # Changelog
 
+- 4.3.0
+  - Split brod-cli out to a separate project [kafka4beam/brod-cli](https://github.com/kafka4beam/brod-cli)
+
 - 4.2.0
   - Optimize consumer fetch latency.
     Introduced the `share_leader_conn` consumer configuration option (default: `false`).
diff --git a/Makefile b/Makefile
index 28cfca52..91e02240 100644
--- a/Makefile
+++ b/Makefile
@@ -35,13 +35,6 @@ hex-publish: clean
 ## tests that require kafka running at localhost
 INTEGRATION_CTS = brod_cg_commits brod_client brod_compression brod_consumer brod_producer brod_group_subscriber brod_topic_subscriber brod
 
-## build escript and a release, and copy escript to release bin dir
-brod-cli:
-	@rebar3 as brod_cli do compile,escriptize,release
-	@cp _build/brod_cli/bin/brod_cli _build/brod_cli/rel/brod/bin/
-	@cp scripts/brod _build/brod_cli/rel/brod/bin/
-	@cp scripts/brod.escript _build/brod_cli/rel/brod/bin/
-
 cover:
 	@rebar3 cover -v
 
diff --git a/README.md b/README.md
index 8a1f439e..824bc360 100644
--- a/README.md
+++ b/README.md
@@ -490,102 +490,3 @@ brod:delete_topics(Hosts, [Topic], Timeout).
 ```
 
 Caution the above delete_topics can fail if you do not have `delete.topic.enable` set to true in your kafka config
-
-## brod-cli: A command line tool to interact with Kafka
-
-This will build a self-contained binary with brod application
-
-```bash
-make brod-cli
-_build/brod_cli/rel/brod/bin/brod -h
-```
-
-Disclaimer: This script is NOT designed for use cases where fault-tolerance is a hard requirement.
-As it may crash when e.g. kafka cluster is temporarily unreachable,
-or (for fetch command) when the partition leader migrates to another broker in the cluster.
-
-### brod-cli examples (with `alias brod=_build/brod_cli/rel/brod/bin/brod`):
-
-#### Fetch and print metadata
-
-```sh
-brod meta -b localhost
-```
-
-#### Produce a Message
-
-```sh
-brod send -b localhost -t test-topic -p 0 -k "key" -v "value"
-
-```
-
-#### Fetch a Message
-
-```sh
-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])'
-```
-
-Bound variables to be used in `--fmt` expression:
-
-- `Offset`: Message offset
-- `Key`: Kafka key
-- `Value`: Kafka Value
-- `TsType`: Timestamp type either `create` or `append`
-- `Ts`: Timestamp, `-1` as no value
-
-#### Stream Messages to Kafka
-
-Send `README.md` to kafka one line per kafka message
-
-```sh
-brod pipe -b localhost:9092 -t test-topic -p 0 -s @./README.md
-```
-
-#### Resolve Offset
-
-```sh
-brod offset -b localhost:9092 -t test-topic -p 0
-```
-
-#### List or Describe Groups
-
-```sh
-# List all groups
-brod groups -b localhost:9092
-
-# Describe groups
-brod groups -b localhost:9092 --ids group-1,group-2
-```
-
-#### Display Committed Offsets
-
-```sh
-# all topics
-brod commits -b localhost:9092 --id the-group-id --describe
-
-# a specific topic
-brod commits -b localhost:9092 --id the-group-id --describe --topic topic-name
-```
-
-#### Commit Offsets
-
-NOTE: This feature is designed for force overwriting commits, not for regular use of offset commit.
-
-```sh
-# Commit 'latest' offsets of all partitions with 2 days retention
-brod commits -b localhost:9092 --id the-group-id --topic topic-name --offsets latest --retention 2d
-
-# Commit offset=100 for partition 0 and 200 for partition 1
-brod commits -b localhost:9092 --id the-group-id --topic topic-name --offsets "0:100,1:200"
-
-# Use --retention 0 to delete commits (may linger in kafka before cleaner does its job)
-brod commits -b localhost:9092 --id the-group-id --topic topic-name --offsets latest --retention 0
-
-# Try join an active consumer group using 'range' protocol and steal one partition assignment then commit offset=10000
-brod commits -b localhost:9092 -i the-group-id -t topic-name -o "0:10000" --protocol range
-```
-
-### TODOs
-
-- Support scram-sasl in brod-cli
-- Transactional produce APIs
diff --git a/elvis.config b/elvis.config
index c15b9788..03f535dd 100644
--- a/elvis.config
+++ b/elvis.config
@@ -30,8 +30,6 @@
                       #{ level => 3,
                          ignore => [ brod_group_coordinator
                                    , brod_utils
-                                   , brod_cli
-                                   , brod_cli_pipe
                                    ]
                        }}
                    , {elvis_style, god_modules,
diff --git a/rebar.config b/rebar.config
index 9b828b7d..fb86e6fb 100644
--- a/rebar.config
+++ b/rebar.config
@@ -1,26 +1,11 @@
 {deps, [{kafka_protocol, "4.1.9"}]}.
 {project_plugins, [{rebar3_lint, "~> 3.2.5"}]}.
-{edoc_opts, [{preprocess, true}, {macros, [{build_brod_cli, true}]}]}.
+{edoc_opts, [{preprocess, true}]}.
 {erl_opts, [warnings_as_errors, warn_unused_vars,warn_shadow_vars,warn_obsolete_guard,debug_info]}.
 {xref_checks, [undefined_function_calls, undefined_functions,
                locals_not_used, deprecated_function_calls,
                deprecated_functions]}.
 {profiles, [
-  {brod_cli, [
-    {deps, [ {docopt, {git, "https://github.com/zmstone/docopt-erl.git", {tag, "0.1.3"}}}
-           , {jsone, "1.7.0"}
-           , {snappyer, "1.2.9"}
-    ]},
-    {erl_opts, [warnings_as_errors, {d, build_brod_cli}]},
-    {escript_name, brod_cli},
-    {relx, [{release, {brod, "i"}, % release the interactive shell as brod-i
-      [brod, jsone, docopt]},
-      {include_erts, true},
-      {overlay, [{copy, "scripts/brod", "bin/"},
-        {copy, "{{lib_dirs}}/crc32cer/priv/crc32cer*.so", "bin/"},
-        {copy, "{{lib_dirs}}/snappyer/priv/snappyer.so", "bin/"}
-      ]}
-    ]}]},
   {test, [
     {deps, [ {docopt, {git, "https://github.com/zmstone/docopt-erl.git", {tag, "0.1.3"}}}
            , {hut, "1.3.0"}
diff --git a/scripts/brod b/scripts/brod
deleted file mode 100755
index 56361e54..00000000
--- a/scripts/brod
+++ /dev/null
@@ -1,32 +0,0 @@
-#!/bin/sh
-
-set -eu
-
-THIS="$0"
-CUR_DIR="$(pwd)"
-
-# resolve the dir (maybe sym-linked) in which THIS script resides
-cd "$(dirname "$THIS")"
-TARGET_FILE="$(basename "$THIS")"
-# use a while loop because readlink -f does not always work in all systems
-while [ -L "$TARGET_FILE" ]; do
-    TARGET_FILE="$(readlink "$TARGET_FILE")"
-    cd "$(dirname "$TARGET_FILE")"
-    TARGET_FILE="$(basename "$TARGET_FILE")"
-done
-BINDIR="$(pwd -P)"
-
-# go back to the original PWD
-cd "$CUR_DIR"
-
-if [ -x  "$BINDIR/brod-i" ]; then
-  # this is a release, try to use erts in release
-  # brod-i find erts bin dir for us
-  exec "$BINDIR/brod-i" escript bin/brod.escript "$@"
-else
-  # this is not a release, so we assume:
-  # 1. the user has erts in PATH
-  # 2. nif binary files are located in the bin dir
-  export NIF_BIN_DIR=$BINDIR
-  exec "$BINDIR/brod_cli" "$@"
-fi
diff --git a/scripts/brod.escript b/scripts/brod.escript
deleted file mode 100755
index b5075037..00000000
--- a/scripts/brod.escript
+++ /dev/null
@@ -1,31 +0,0 @@
-#!/usr/bin/env escript
-
-%% this script forwards the call to brod_cli module
-%% we avoid using the escriptised file due to the need of loading nif binaries
-
-main(Args) ->
-    ok = add_libs_dir(),
-    brod_cli:main(Args).
-
-add_libs_dir() ->
-    %% RELEASE_ROOT_DIR is set by bord-i boot script
-    [_ | _] = RootDir = os:getenv("RELEASE_ROOT_DIR"),
-    RelFile = filename:join([RootDir, "releases", "RELEASES"]),
-    case file:consult(RelFile) of
-        {ok, [Releases]} ->
-            Release = lists:keyfind("i", 3, Releases),
-            {release, _Name, _AppVsn, _ErtsVsn, Libs, _State} = Release,
-            lists:foreach(
-              fun({Name, Vsn, _}) ->
-                      add_lib_dir(RootDir, Name, Vsn)
-              end, Libs);
-        {error, Reason} ->
-            error({failed_to_read_RELEASES_file, RelFile, Reason})
-    end.
-
-add_lib_dir(RootDir, Name, Vsn) ->
-    LibDir = filename:join([RootDir, lib, atom_to_list(Name) ++ "-" ++ Vsn, ebin]),
-    case code:add_patha(LibDir) of
-        true -> ok;
-        {error, _} -> error(LibDir)
-    end.
diff --git a/src/brod.erl b/src/brod.erl
index d7dbedfb..859a2657 100644
--- a/src/brod.erl
+++ b/src/brod.erl
@@ -100,7 +100,7 @@
         , delete_topics/4
         ]).
 
-%% APIs for quick metadata or message inspection and brod_cli
+%% APIs for quick metadata or message inspection
 -export([ get_metadata/1
         , get_metadata/2
         , get_metadata/3
@@ -128,10 +128,6 @@
         , fetch/8
         ]).
 
--ifdef(build_brod_cli).
--export([main/1]).
--endif.
-
 -export_type([ batch_input/0
              , bootstrap/0
              , call_ref/0
@@ -1365,10 +1361,6 @@ fetch_committed_offsets(BootstrapEndpoints, ConnCfg, GroupId) ->
 fetch_committed_offsets(Client, GroupId) ->
   brod_utils:fetch_committed_offsets(Client, GroupId, []).
 
--ifdef(build_brod_cli).
-main(X) -> brod_cli:main(X).
--endif.
-
 %% @doc Start a new transaction, `TxId' will be the id of the transaction
 %% @equiv brod_transaction:start_link/3
 -spec transaction(client(), transactional_id(), transaction_config()) -> {ok, transaction()}.
diff --git a/src/brod_cli.erl b/src/brod_cli.erl
deleted file mode 100644
index e3f3cef8..00000000
--- a/src/brod_cli.erl
+++ /dev/null
@@ -1,1320 +0,0 @@
-%%%
-%%%   Copyright (c) 2017-2021 Klarna Bank AB (publ)
-%%%
-%%%   Licensed under the Apache License, Version 2.0 (the "License");
-%%%   you may not use this file except in compliance with the License.
-%%%   You may obtain a copy of the License at
-%%%
-%%%       http://www.apache.org/licenses/LICENSE-2.0
-%%%
-%%%   Unless required by applicable law or agreed to in writing, software
-%%%   distributed under the License is distributed on an "AS IS" BASIS,
-%%%   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-%%%   See the License for the specific language governing permissions and
-%%%   limitations under the License.
-%%%
-
-%% @private
--module(brod_cli).
-
--ifdef(build_brod_cli).
-
--export([main/1, main/2]).
-
--include("brod_int.hrl").
-
--define(CLIENT, brod_cli_client).
-
-%% 'halt' is for escript, stop the vm immediately
-%% 'exit' is for testing, we want eunit or ct to be able to capture
--define(STOP(How),
-        begin
-          try
-            brod:stop_client(?CLIENT)
-          catch
-            exit : {noproc, _} ->
-              ok
-          end,
-          _ = brod:stop(),
-          case How of
-            'halt' -> erlang:halt(?LINE);
-            'exit' -> erlang:exit(?LINE)
-          end
-        end).
-
--define(MAIN_DOC, "usage:
-  brod -h|--help
-  brod -v|--version
-  brod <command> [options] [-h|--help] [--verbose|--debug]
-
-commands:
-  meta:    Inspect topic metadata
-  offset:  Inspect offsets
-  fetch:   Fetch messages
-  send:    Produce messages
-  pipe:    Pipe file or stdin as messages to kafka
-  groups:  List/describe consumer group
-  commits: List/describe committed offsets
-           or force overwrite existing commits
-").
-
-%% NOTE: bad indentation at the first line is intended
--define(COMMAND_COMMON_OPTIONS,
-"  --ssl                  Use TLS, validate server using trusted CAs
-  --ssl-versions=<vsns>  Specify SSL versions. Comma separated versions,
-                         e.g. 1.3,1.2
-  --cacertfile=<cacert>  Use TLS, validate server using the given certificate
-  --certfile=<certfile>  Client certificate in case client authentication
-                         is enabled in brokers
-  --keyfile=<keyfile>    Client private key in case client authentication
-                         is enabled in brokers
-  --sasl-plain=<file>    Tell brod to use username/password stored in the
-                         given file, the file should have username and
-                         password in two lines.
-  --scram256=<file>      Like sasl-plain option, but to use scram-sha-256
-  --scram512=<file>      Like sasl-plain option, but to use scram-sha-512
-  --ebin-paths=<dirs>    Comma separated directory names for extra beams,
-                         This is to support user compiled message formatters
-  --no-api-vsn-query     Do not query api version (for kafka 0.9 or earlier)
-                         Or set KAFKA_VERSION environment variable to 0.9 for
-                         the same effect
-"
-).
-
--define(META_CMD, "meta").
--define(META_DOC, "usage:
-  brod meta [options]
-
-options:
-  -b,--brokers=<brokers> Comma separated host:port pairs
-                         [default: localhost:9092]
-  -t,--topic=<topic>     Topic name [default: *]
-  -T,--text              Print metadata as aligned texts (default)
-  -J,--json              Print metadata as JSON object
-  -L,--list              List topics, no partition details,
-                         Applicable only for --text option
-  -U,--under-replicated  Display only under-replicated partitions
-"
-?COMMAND_COMMON_OPTIONS
-"Text output schema (out of sync replicas are marked with *):
-brokers <count>:
-  <broker-id>: <endpoint>
-topics <count>:
-  <name> <count>: [[ERROR] [<reason>]]
-    <partition>: <leader-broker-id> (replicas[*]...) [<error-reason>]
-"
-).
-
--define(OFFSET_CMD, "offset").
--define(OFFSET_DOC, "usage:
-  brod offset [options]
-
-options:
-  -b,--brokers=<brokers> Comma separated host:port pairs
-                         [default: localhost:9092]
-  -t,--topic=<topic>     Topic name
-  -p,--partition=<parti> Partition number
-                         [default: all]
-  -T,--time=<time>       Unix epoch (in milliseconds) of the correlated offset
-                         to fetch. Special values:
-                           'latest' or -1 for latest offset
-                           'earliest' or -2 for earliest offset
-                         [default: latest]
-  --one-line             If it is to resolve 'all' offsets,
-                         Print results in one line.
-                         e.g. 0:111,1:222
-"
-?COMMAND_COMMON_OPTIONS
-).
-
--define(FETCH_CMD, "fetch").
--define(FETCH_DOC, "usage:
-  brod fetch [options]
-
-options:
-  -b,--brokers=<brokers> Comma separated host:port pairs
-                         [default: localhost:9092]
-  -t,--topic=<topic>     Topic name
-  -p,--partition=<parti> Partition number
-  -o,--offset=<offset>   Offset to start fetching from
-                           latest: From the latest offset (not last)
-                           earliest: From earliest offset (first)
-                           last: From offset (latest - 1)
-                           <integer>: From a specific offset
-                         [default: last]
-  -c,--count=<count>     Number of messages to fetch (-1 as infinity)
-                         [default: 1]
-  -w,--wait=<seconds>    Time in seconds to wait for one message set
-                         [default: 5s]
-  --kv-deli=<deli>       Delimiter for offset, key and value output [default: :]
-  --msg-deli=<deli>      Delimiter between messages. [default: \\n]
-  --max-bytes=<bytes>    Max number of bytes kafka should try to accumulate
-                         within the --wait time
-                         [default: 1K]
-  --fmt=<fmt>            Output format. Assume keys and values are utf8 strings
-                         v:     Print 'V <msg-deli>'
-                         kv:    Print 'K <kv-deli> V <msg-deli>'
-                         okv:   Print 'O <kv-deli> K <kv-deli> V <msg-deli>'
-                         eterm: Pretty print tuple '{Offse, Key, Value}.'
-                                to a consultable Erlang term format.
-                         Expr:  An Erlang expression to be evaluated for each
-                                message. Bound variable to be used in the
-                                expression: Offset, Key, Value, TsType, Ts.
-                                Print nothing if the evaluation result in 'ok',
-                                otherwise print the evaluated io-list.
-                         [default: v]
-"
-?COMMAND_COMMON_OPTIONS
-"NOTE: Reaching either --count or --wait limit will cause script to exit
-"
-).
-
--define(SEND_CMD, "send").
--define(SEND_DOC, "usage:
-  brod send [options]
-
-options:
-  -b,--brokers=<brokers> Comma separated host:port pairs
-                         [default: localhost:9092]
-  -t,--topic=<topic>     Topic name
-  -p,--partition=<parti> Partition number [default: 0]
-                         Special values:
-                           random: randomly pick a partition
-                           hash: hash key to a partition
-  -k,--key=<key>         Key to produce [default: null]
-  -v,--value=<value>     Value to produce. Special values:
-                           null: No payload
-                           @/path/to/file: Send a whole file as payload
-  --acks=<acks>          Required acks. Supported values:
-                           all or -1: Require acks from all in-sync replica
-                           1: Require acks from only partition leader
-                           0: Require no acks
-                         [default: all]
-  --ack-timeout=<time>   How long the partition leader should wait for replicas
-                         to ack before sending response to producer
-                         The value can be an integer to indicate number of
-                         milliseconds or followed by s/m to indicate seconds
-                         or minutes [default: 10s]
-  --compression=<compre> Supported values: none / gzip / snappy
-                         [default: none]
-"
-?COMMAND_COMMON_OPTIONS
-).
-
--define(PIPE_CMD, "pipe").
--define(PIPE_DOC, "usage:
-  brod pipe [options]
-
-options:
-  -b,--brokers=<brokers> Comma separated host:port pairs
-                         [default: localhost:9092]
-  -t,--topic=<topic>     Topic name
-  -p,--partition=<parti> Partition number [default: 0]
-                         Special values:
-                           random: randomly pick a partition
-                           hash: hash key to a partition
-  -s,--source=<source>   Data source. Special value:
-                           stdin: Reads messages from standard-input
-                           @path/to/file: Reads from file
-                         [default: stdin]
-  --prompt               Applicable when --source is stdin, enable input prompt
-  --no-eof-exit          Do not exit when reaching EOF
-  --tail                 Applicable when --source is a file
-                         brod will start from EOF and keep tailing for new bytes
-  --msg-deli=<msg-deli>  Message delimiter
-                         NOTE: A message is always delimited when reaching EOF
-                         [default: \\n]
-  --kv-deli=<kv-deli>    Key-Value delimiter.
-                         when not provided, messages are produced with
-                         only value, key is set to null. [default: none]
-  --blk-size=<size>      Block size (bytes) when reading bytes from a file.
-                         Applicable when --source is file and --msg-deli is
-                         not \\n.  [default: 1M]
-  --acks=<acks>          Required acks. [default: all]
-                         Supported values:
-                           all or -1: Require acks from all in-sync replica
-                           1: Require acks from only partition leader
-                           0: Require no acks
-  --ack-timeout=<time>   How long the partition leader should wait for replicas
-                         to ack before sending response to producer
-                         not applicable when --acks is not 'all'
-                         [default: 10s]
-  --max-linger-ms=<ms>   Max ms for messages to linger in buffer [default: 200]
-  --max-linger-cnt=<N>   Max messages to linger in buffer [default: 100]
-  --max-batch=<bytes>    Max size for one message-set (before compression)
-                         The value can be either an integer to indicate bytes
-                         or followed by K/M to indicate KBytes or MBytes
-                         [default: 1M]
-  --compression=<compr>  Supported values: none/gzip/snappy [default: none]
-"
-?COMMAND_COMMON_OPTIONS
-"NOTE: When --source is path/to/file, it by default reads from BOF
-      unless --tail is given.
-"
-).
-
--define(GROUPS_CMD, "groups").
--define(GROUPS_DOC, "usage:
-  brod groups [options]
-
-options:
-  -b,--brokers=<brokers> Comma separated host:port pairs
-                         [default: localhost:9092]
-  --ids=<group-id>       Comma separated group IDs to describe
-                         [default: all]
-"
-?COMMAND_COMMON_OPTIONS
-).
-
-
--define(COMMITS_CMD, "commits").
--define(COMMITS_DOC, "usage:
-  brod commits [options]
-
-options:
-  -b,--brokers=<brokers> Comma separated host:port pairs
-                         [default: localhost:9092]
-  -d,--describe          Describe committed offsets,
-                         otherwise reset commit history
-  -i,--id=<group-id>     Group ID to describe, or to commit offsets.
-  -t,--topic=<topic>     Topic name to commit offsets
-  -o,--offsets=<offsets> latest: commit latest offset for all partitions,
-                         earliest: commit earliest offset for all partitions,
-                         Comma separated 'partition:offset' pairs.
-  -r,--retention=<time>  An integer to indicate the retention for the commit,
-                         default time unit is seconds. Accepts one char suffix
-                         as time unit, s=second m=mintue h=hour d=day.
-                         Default value -1 is to respect kafka config.
-                         [default: -1]
-  --protocol=<protocol>  Protocol name to be used when trying to join group.
-                         [default: roundrobin]
-"
-?COMMAND_COMMON_OPTIONS
-).
-
--define(DOCS,
-        [ {?META_CMD,    ?META_DOC}
-        , {?OFFSET_CMD,  ?OFFSET_DOC}
-        , {?SEND_CMD,    ?SEND_DOC}
-        , {?FETCH_CMD,   ?FETCH_DOC}
-        , {?PIPE_CMD,    ?PIPE_DOC}
-        , {?GROUPS_CMD,  ?GROUPS_DOC}
-        , {?COMMITS_CMD, ?COMMITS_DOC}
-        ]).
-
--define(LOG_LEVEL_QUIET, 0).
--define(LOG_LEVEL_VERBOSE, 1).
--define(LOG_LEVEL_DEBUG, 2).
-
--type log_level() :: non_neg_integer().
-
--type command() :: string().
-
-main(Args) ->
-  _ = main(Args, halt).
-
--spec main([string()], halt | exit) -> no_return().
-main(["-h" | _], _Stop) ->
-  print(?MAIN_DOC);
-main(["--help" | _], _Stop) ->
-  print(?MAIN_DOC);
-main(["-v" | _], _Stop) ->
-  print_version();
-main(["--version" | _], _Stop) ->
-  print_version();
-main(["-" ++ _ = Arg | _], Stop) ->
-  logerr("Unknown option: ~s\n", [Arg]),
-  print(?MAIN_DOC),
-  ?STOP(Stop);
-main([Command | _] = Args, Stop) ->
-  case lists:keyfind(Command, 1, ?DOCS) of
-    {_, Doc} ->
-      main(Command, Doc, Args, Stop);
-    false ->
-      logerr("Unknown command: ~s\n", [Command]),
-      print(?MAIN_DOC),
-      ?STOP(Stop)
-  end;
-main(_, Stop) ->
-  print(?MAIN_DOC),
-  ?STOP(Stop).
-
--spec main(command(), string(), [string()], halt | exit) -> _ | no_return().
-main(Command, Doc, Args0, Stop) ->
-  IsHelp = lists:member("--help", Args0) orelse lists:member("-h", Args0),
-  IsVerbose = lists:member("--verbose", Args0),
-  IsDebug = lists:member("--debug", Args0),
-  Args = Args0 -- ["--verbose", "--debug"],
-  LogLevels = [{IsDebug, ?LOG_LEVEL_DEBUG},
-               {IsVerbose, ?LOG_LEVEL_VERBOSE},
-               {true, ?LOG_LEVEL_QUIET}],
-  {true, LogLevel} = lists:keyfind(true, 1, LogLevels),
-  erlang:put(brod_cli_log_level, LogLevel),
-  case IsHelp of
-    true ->
-      print(Doc);
-    false ->
-      main(Command, Doc, Args, Stop, LogLevel)
-  end.
-
--spec main(command(), string(), [string()], halt | exit,
-           log_level()) -> _ | no_return().
-main(Command, Doc, Args, Stop, LogLevel) ->
-  ParsedArgs =
-    try
-      docopt:docopt(Doc, Args, [debug || LogLevel =:= ?LOG_LEVEL_DEBUG])
-    catch
-      C1 : E1 ?BIND_STACKTRACE(Stack1) ->
-        ?GET_STACKTRACE(Stack1),
-        verbose("~p:~p\n~p\n", [C1, E1, Stack1]),
-        io:format(user, "~p~n", [{C1, E1, Stack1}]),
-        ?STOP(Stop)
-    end,
-  case LogLevel =:= ?LOG_LEVEL_QUIET of
-    true ->
-      logger:add_handler(brod_cli, logger_std_h,
-                         #{config => #{file => "brod.log"}, level => info}),
-      logger:remove_handler(default);
-    false ->
-      ok
-  end,
-  %% So the linked processes won't take me with them
-  %% This is to allow error_logger to write more logs
-  %% before stopping init process immediately
-  process_flag(trap_exit, true),
-  ok = brod:start(),
-  try
-    Brokers = parse(ParsedArgs, "--brokers", fun parse_brokers/1),
-    ConnConfig0 = parse_connection_config(ParsedArgs),
-    Paths = parse(ParsedArgs, "--ebin-paths", fun parse_paths/1),
-    NoApiQuery = parse(ParsedArgs, "--no-api-vsn-query", fun parse_boolean/1)
-                 orelse ({0, 9} =:= get_kafka_version()),
-    ok = code:add_pathsa(Paths),
-    SockOpts = [{query_api_versions, not NoApiQuery} |  ConnConfig0],
-    verbose("connection config: ~p\n", [SockOpts]),
-    run(Command, Brokers, SockOpts, ParsedArgs)
-  catch
-    throw : Reason when is_binary(Reason) ->
-      %% invalid options etc.
-      logerr([Reason, "\n"]),
-      ?STOP(Stop);
-    C2 : E2 ?BIND_STACKTRACE(Stack2) ->
-      ?GET_STACKTRACE(Stack2),
-      logerr("~p:~p\n~p\n", [C2, E2, Stack2]),
-      ?STOP(Stop)
-  end.
-
-run(?META_CMD, Brokers, Topic, SockOpts, Args) ->
-  Topics = case Topic of
-             <<"*">> -> []; %% fetch all topics
-             _       -> [Topic]
-           end,
-  IsJSON = parse(Args, "--json", fun parse_boolean/1),
-  IsText = parse(Args, "--text", fun parse_boolean/1),
-  Format = kf(true, [ {IsJSON, json}
-                    , {IsText, text}
-                    , {true, text}
-                    ]),
-  IsList = parse(Args, "--list", fun parse_boolean/1),
-  IsUrp = parse(Args, "--under-replicated", fun parse_boolean/1),
-  {ok, Metadata} = brod:get_metadata(Brokers, Topics, SockOpts),
-  format_metadata(Metadata, Format, IsList, IsUrp);
-run(?OFFSET_CMD, Brokers, Topic, SockOpts, Args) ->
-  Partition = parse(Args, "--partition", fun("all") -> all;
-                                            (Num) -> int(Num)
-                                         end),
-  IsOneLine = parse(Args, "--one-line", fun parse_boolean/1),
-  Time = parse(Args, "--time", fun parse_offset_time/1),
-  ok = start_client(Brokers, SockOpts),
-  try
-    resolve_offsets_print(Topic, Partition, Time, IsOneLine)
-  after
-    brod_client:stop(?CLIENT)
-  end;
-run(?FETCH_CMD, Brokers, Topic, ConnOpts, Args) ->
-  Partition = parse(Args, "--partition", fun int/1), %% not parse_partition/1
-  Count = parse(Args, "--count", fun int/1),
-  Offset0 = parse(Args, "--offset", fun parse_offset_time/1),
-  Wait = parse(Args, "--wait", fun parse_timeout/1),
-  KvDeli = parse(Args, "--kv-deli", fun parse_delimiter/1),
-  MsgDeli = parse(Args, "--msg-deli", fun parse_delimiter/1),
-  FmtFun = parse(Args, "--fmt",
-                 fun(FmtOption) ->
-                     parse_fmt(FmtOption, KvDeli, MsgDeli)
-                 end),
-  MaxBytes = parse(Args, "--max-bytes", fun parse_size/1),
-  {ok, Conn} = brod:connect_leader(Brokers, Topic, Partition, ConnOpts),
-  Offset = resolve_begin_offset(Conn, Topic, Partition, Offset0),
-  FetchOpts = #{max_wait_time => Wait, max_bytes => MaxBytes},
-  FoldLimits = case Count < 0 of
-                 true -> #{};
-                 false -> #{message_count => Count}
-               end,
-  FoldFun =
-    fun(M, Acc) ->
-      #kafka_message{offset = O, key = K, value = V} = M,
-      R = case is_function(FmtFun, 3) of
-            true -> FmtFun(O, K, V);
-            false -> FmtFun(M)
-          end,
-      case R of
-        ok -> ok;
-        IoData -> print(IoData)
-      end,
-      {ok, Acc + 1}
-    end,
-  {FetchedCount, NextOffset, Reason} =
-    brod:fold(Conn, Topic, Partition, Offset, FetchOpts,
-              0, FoldFun, FoldLimits),
-  logerr("Fetch loop stopped after ~p messages~n", [FetchedCount]),
-  logerr("Continue at Offset: ~p~n", [NextOffset]),
-  logerr("Reason: ~p~n", [Reason]);
-run(?SEND_CMD, Brokers, Topic, SockOpts, Args) ->
-  Partition = parse(Args, "--partition", fun parse_partition/1),
-  Acks = parse(Args, "--acks", fun parse_acks/1),
-  AckTimeout = parse(Args, "--ack-timeout", fun parse_timeout/1),
-  Compression = parse(Args, "--compression", fun parse_compression/1),
-  Key = parse(Args, "--key", fun("null") -> <<"">>;
-                                (K) -> bin(K)
-                             end),
-  Value0 = parse(Args, "--value", fun("null") -> <<"">>;
-                                     ("@" ++ F) -> {file, F};
-                                     (V) -> bin(V)
-                                  end),
-  Value =
-    case Value0 of
-      {file, File} ->
-        {ok, Bin} = file:read_file(File),
-        Bin;
-      <<_/binary>> ->
-        Value0
-    end,
-  ProducerConfig =
-    [ {required_acks, Acks}
-    , {ack_timeout, AckTimeout}
-    , {compression, Compression}
-    , {min_compression_batch_size, 0}
-    , {max_linger_ms, 0}
-    ],
-  ClientConfig =
-    [ {auto_start_producers, true}
-    , {default_producer_config, ProducerConfig}
-    ] ++ SockOpts,
-  ok = start_client(Brokers, ClientConfig),
-  Msgs = [{brod_utils:epoch_ms(), Key, Value}],
-  ok = brod:produce_sync(?CLIENT, Topic, Partition, <<>>, Msgs);
-run(?PIPE_CMD, Brokers, Topic, SockOpts, Args) ->
-  Partition = parse(Args, "--partition", fun parse_partition/1),
-  Acks = parse(Args, "--acks", fun parse_acks/1),
-  AckTimeout = parse(Args, "--ack-timeout", fun parse_timeout/1),
-  Compression = parse(Args, "--compression", fun parse_compression/1),
-  KvDeli = parse(Args, "--kv-deli", fun parse_delimiter/1),
-  MsgDeli = parse(Args, "--msg-deli", fun parse_delimiter/1),
-  MaxLingerMs = parse(Args, "--max-linger-ms", fun int/1),
-  MaxLingerCnt = parse(Args, "--max-linger-cnt", fun int/1),
-  MaxBatch = parse(Args, "--max-batch", fun parse_size/1),
-  Source = parse(Args, "--source", fun parse_source/1),
-  IsPrompt = parse(Args, "--prompt", fun parse_boolean/1),
-  IsTail = parse(Args, "--tail", fun parse_boolean/1),
-  IsNoExit = parse(Args, "--no-eof-exit", fun parse_boolean/1),
-  BlkSize = parse(Args, "--blk-size", fun parse_size/1),
-  ProducerConfig =
-    [ {required_acks, Acks}
-    , {ack_timeout, AckTimeout}
-    , {compression, Compression}
-    , {min_compression_batch_size, 0}
-    , {max_linger_ms, MaxLingerMs}
-    , {max_linger_count, MaxLingerCnt}
-    , {max_batch_size, MaxBatch}
-    ],
-  ClientConfig =
-    [ {auto_start_producers, true}
-    , {default_producer_config, ProducerConfig}
-    ] ++ SockOpts,
-  ok = start_client(Brokers, ClientConfig),
-  SendFun =
-    fun(?TKV(Ts, Key, Value), PendingAcks) ->
-        {ok, CallRef} =
-          brod:produce(?CLIENT, Topic, Partition, <<>>, [{Ts, Key, Value}]),
-        debug("sent: ~w\n", [CallRef]),
-        debug("value: ~P\n", [Value, 9]),
-        queue:in(CallRef, PendingAcks)
-    end,
-  KvDeliForReader = case none =:= KvDeli of
-                      true -> none;
-                      false -> bin(KvDeli)
-                    end,
-  ReaderArgs = [ {source, Source}
-               , {kv_deli, KvDeliForReader}
-               , {msg_deli, bin(MsgDeli)}
-               , {prompt, IsPrompt}
-               , {tail, IsTail}
-               , {no_exit, IsNoExit}
-               , {blk_size, BlkSize}
-               , {retry_delay, 100}
-               ],
-  {ok, ReaderPid} =
-    brod_cli_pipe:start_link(ReaderArgs),
-  _ = erlang:monitor(process, ReaderPid),
-  pipe(ReaderPid, SendFun, queue:new()).
-
-run(?GROUPS_CMD, Brokers, SockOpts, Args) ->
-  IDs = parse(Args, "--ids", fun parse_cg_ids/1),
-  cg(Brokers, SockOpts, IDs);
-run(?COMMITS_CMD, Brokers, SockOpts, Args) ->
-  IsDesc = parse(Args, "--describe", fun parse_boolean/1),
-  ID = parse(Args, "--id", fun bin/1),
-  Topic = parse(Args, "--topic",
-                fun(?undef) -> ?undef;
-                   (Name) -> bin(Name)
-                end),
-  ok = start_client(Brokers, SockOpts),
-  case IsDesc of
-    true -> show_commits(ID, Topic);
-    false -> reset_commits(ID, Topic, Args)
-  end;
-run(Cmd, Brokers, SockOpts, Args) ->
-  %% Clause for all per-topic commands
-  Topic = parse(Args, "--topic", fun bin/1),
-  run(Cmd, Brokers, Topic, SockOpts, Args).
-
-resolve_offsets_print(Topic, all, Time, IsOneLine) ->
-  Offsets = resolve_offsets(Topic, Time),
-  Outputs =
-    lists:map(
-      fun({Partition, Offset}) ->
-          io_lib:format("~p:~p", [Partition, Offset])
-      end, Offsets),
-  Delimiter = case IsOneLine of
-                true -> ",";
-                false -> "\n"
-              end,
-  print(infix(Outputs, Delimiter));
-resolve_offsets_print(Topic, Partition, Time, _) when is_integer(Partition) ->
-  {ok, Offset} = resolve_offset(Topic, Partition, Time),
-  print(integer_to_list(Offset)).
-
-resolve_offsets(Topic, Time) ->
-  {ok, Count} = brod_client:get_partitions_count(?CLIENT, Topic),
-  Partitions = lists:seq(0, Count - 1),
-  lists:map(
-    fun(P) ->
-        {ok, Offset} = resolve_offset(Topic, P, Time),
-        {P, Offset}
-    end, Partitions).
-
-resolve_offset(Topic, Partition, Time) ->
-  {ok, SockPid} = brod_client:get_leader_connection(?CLIENT, Topic, Partition),
-  brod_utils:resolve_offset(SockPid, Topic, Partition, Time).
-
-show_commits(GroupId, Topic) ->
-  case brod:fetch_committed_offsets(?CLIENT, GroupId) of
-    {ok, PerTopicStructs0} ->
-      Pred = fun(S) -> Topic =:= ?undef orelse Topic =:= kf(topic, S) end,
-      PerTopicStructs = lists:filter(Pred, PerTopicStructs0),
-      lists:foreach(fun print_commits/1, PerTopicStructs);
-    {error, Reason} ->
-      throw_bin("Failed to fetch committed offsets ~p\n", [Reason])
-  end.
-
-reset_commits(ID, Topic, Args) ->
-  Retention = parse(Args, "--retention", fun parse_retention/1),
-  ProtocolName = parse(Args, "--protocol", fun(X) -> X end),
-  Offsets0 = parse(Args, "--offsets", fun parse_commit_offsets_input/1),
-  Offsets =
-    case is_atom(Offsets0) of
-      true -> resolve_offsets(Topic, Offsets0);
-      false -> Offsets0
-    end,
-  Group = [ {id, ID}
-          , {topic, Topic}
-          , {retention, Retention}
-          , {protocol, ProtocolName}
-          , {offsets, Offsets}
-          ],
-  brod_cg_commits:run(?CLIENT, Group).
-
-parse_commit_offsets_input("latest") -> latest;
-parse_commit_offsets_input("earliest") -> earliest;
-parse_commit_offsets_input(PartitionOffsets) ->
-  Pairs = string:tokens(PartitionOffsets, ","),
-  F = fun(Pair) ->
-          [Partition, Offset] = string:tokens(Pair, ":"),
-          {int(Partition), parse_offset_time(Offset)}
-      end,
-  lists:map(F, Pairs).
-
-parse_retention("-1") -> -1;
-parse_retention([_|_] = R) ->
-  case lists:last(R) of
-    X when X >= $0 andalso X =< $9 ->
-      int(R);
-    Unit ->
-      int(lists:reverse(tl(lists:reverse(R)))) *
-      case Unit of
-        $s -> 1;
-        $S -> 1;
-        $m -> 60;
-        $M -> 60;
-        $h -> 60 * 60;
-        $H -> 60 * 60;
-        $d -> 60 * 60 * 24;
-        $D -> 60 * 60 * 24
-      end
-  end.
-
-print_commits(Struct) ->
-  Topic = kf(name, Struct),
-  PartRsps = kf(partitions, Struct),
-  print([Topic, ":\n"]),
-  print([pp_fmt_struct(1, P) || P <- PartRsps]).
-
-cg(BootstrapEndpoints, SockOpts, all) ->
-  %% List all groups
-  All = list_groups(BootstrapEndpoints, SockOpts),
-  lists:foreach(fun print_cg_cluster/1, All);
-cg(BootstrapEndpoints, SockOpts, IDs) ->
-  CgClusters = list_groups(BootstrapEndpoints, SockOpts),
-  describe_cgs(CgClusters, SockOpts, lists:usort(IDs)).
-
-describe_cgs(_, _SockOpts, []) -> ok;
-describe_cgs([], _SockOpts, IDs) ->
-  logerr("Unknown group IDs: ~s", [infix(IDs, ", ")]);
-describe_cgs([{Coordinator, CgList} | Rest], SockOpts, IDs) ->
-  %% Get all IDs managed by current coordinator.
-  ThisIDs = [ID || #brod_cg{id = ID} <- CgList, lists:member(ID, IDs)],
-  ok = do_describe_cgs(Coordinator, SockOpts, ThisIDs),
-  IDsRest = IDs -- ThisIDs,
-  describe_cgs(Rest, SockOpts, IDsRest).
-
-do_describe_cgs(_Coordinator, _SockOpts, []) -> ok;
-do_describe_cgs(Coordinator, SockOpts, IDs) ->
-  case brod:describe_groups(Coordinator, SockOpts, IDs) of
-    {ok, DescArray} ->
-      ok = print("~s\n", [fmt_endpoint(Coordinator)]),
-      lists:foreach(fun print_cg_desc/1, DescArray);
-    {error, Reason} ->
-      logerr("Failed to describe IDs [~s] at broker ~s\nreason:~p\n",
-             [infix(IDs, ","), fmt_endpoint(Coordinator), Reason])
-  end.
-
-print_cg_desc(Desc) ->
-  EC = kf(error_code, Desc),
-  GroupId = kf(group_id, Desc),
-  case ?IS_ERROR(EC) of
-    true ->
-      logerr("Failed to describe group id=~s\nreason:~p\n", [GroupId, EC]);
-    false ->
-      D1 = lists:keydelete(error_code, 1, ensure_list(Desc)),
-      D  = lists:keydelete(group_id, 1, ensure_list(D1)),
-      print("  ~s\n~s", [GroupId, pp_fmt_struct(_Indent = 2, D)])
-  end.
-
-ensure_list(Struct) when is_map(Struct) -> maps:to_list(Struct);
-ensure_list(List) when is_list(List) -> List.
-
-pp_fmt_struct(Indent, Map) when is_map(Map) ->
-  pp_fmt_struct(Indent, maps:to_list(Map));
-pp_fmt_struct(Indent, Fields0) when is_list(Fields0) ->
-  Fields = case Fields0 of
-             [_] -> Fields0;
-             _ -> lists:keydelete(no_error, 2, Fields0)
-           end,
-  F = fun(IsFirst, {N, V}) when is_map(V) ->
-          indent_fmt(IsFirst, Indent,
-                     "~p:\n~s", [N, pp_fmt_struct(Indent + 1, V)]);
-          (IsFirst, {N, V}) ->
-          indent_fmt(IsFirst, Indent,
-                     "~p: ~s", [N, pp_fmt_struct_value(Indent, V)])
-      end,
-  [ F(true, hd(Fields))
-  | lists:map(fun(Fi) -> F(false, Fi) end, tl(Fields))
-  ].
-
-pp_fmt_struct_value(_Indent, X) when is_integer(X) orelse
-                                     is_atom(X) orelse
-                                     is_binary(X) orelse
-                                     X =:= [] ->
-  [pp_fmt_prim(X), "\n"];
-pp_fmt_struct_value(Indent, [H | _] = Array) when is_list(Array) ->
-  case is_tuple(H) orelse is_map(H) of
-    true ->
-      %% array of sub struct
-      ["\n",
-       lists:map(fun(Item) ->
-                     pp_fmt_struct(Indent + 1, Item)
-                 end, Array)
-      ];
-    false ->
-      %% array of primitive values
-      [infix([pp_fmt_prim(V) || V <- Array], ","), "\n"]
-  end.
-
-pp_fmt_prim([]) -> "[]";
-pp_fmt_prim(N) when is_integer(N) -> integer_to_list(N);
-pp_fmt_prim(A) when is_atom(A) -> atom_to_list(A);
-pp_fmt_prim(S) when is_binary(S) ->
-  case is_printable(S) of
-    true -> S;
-    false -> io_lib:format("~w", [S])
-  end.
-
-is_printable(B) when is_binary(B) ->
-  is_printable(unicode:characters_to_list(B, utf8));
-is_printable(S) when is_list(S) ->
-  io_lib:printable_unicode_list(S);
-is_printable(_) ->
-  false.
-
-indent_fmt(true, Indent, Fmt, Args) ->
-  io_lib:format(lists:duplicate((Indent - 1) * 2, $\s) ++ "- " ++ Fmt, Args);
-indent_fmt(false, Indent, Fmt, Args) ->
-  io_lib:format(lists:duplicate(Indent * 2, $\s) ++ Fmt, Args).
-
-print_cg_cluster({Endpoint, Cgs}) ->
-  ok = print([fmt_endpoint(Endpoint), "\n"]),
-  IoData = [ io_lib:format("  ~s (~s)\n", [Id, Type])
-             || #brod_cg{id = Id, protocol_type = Type} <- Cgs
-           ],
-  print(IoData).
-
-fmt_endpoint({Host, Port}) ->
-  bin(io_lib:format("~s:~B", [Host, Port])).
-
-%% Return consumer groups clustered by group coordinator
-%% {CoordinatorEndpoint, [group_id()]}.
-list_groups(Brokers, SockOpts) ->
-  Cgs = brod:list_all_groups(Brokers, SockOpts),
-  lists:keysort(1, lists:foldl(fun do_list_groups/2, [], Cgs)).
-
-do_list_groups({_Endpoint, []}, Acc) -> Acc;
-do_list_groups({Endpoint, {error, Reason}}, Acc) ->
-  logerr("Failed to list groups at kafka ~s\nreason~p",
-         [fmt_endpoint(Endpoint), Reason]),
-  Acc;
-do_list_groups({Endpoint, Cgs}, Acc) ->
-  [{Endpoint, Cgs} | Acc].
-
-pipe(ReaderPid, SendFun, PendingAcks0) ->
-  PendingAcks1 = flush_pending_acks(PendingAcks0, _Timeout = 0),
-  receive
-    {pipe, ReaderPid, Messages} ->
-      PendingAcks = lists:foldl(SendFun, PendingAcks1, Messages),
-      pipe(ReaderPid, SendFun, PendingAcks);
-    {'DOWN', _Ref, process, ReaderPid, Reason} ->
-      %% Reader is down, flush pending acks
-      debug("reader down, reason: ~p\n", [Reason]),
-      _ = flush_pending_acks(PendingAcks1, infinity);
-    #brod_produce_reply{ call_ref = CallRef
-                       , result = brod_produce_req_acked
-                       } ->
-      {{value, CallRef}, PendingAcks} = queue:out(PendingAcks1),
-      pipe(ReaderPid, SendFun, PendingAcks)
-  end.
-
-flush_pending_acks(Queue, Timeout) ->
-  case queue:peek(Queue) of
-    empty ->
-      Queue;
-    {value, CallRef} ->
-      case brod:sync_produce_request(CallRef, Timeout) of
-        ok ->
-          debug("acked: ~w\n", [CallRef]),
-          {_, Rest} = queue:out(Queue),
-          flush_pending_acks(Rest, Timeout);
-        {error, timeout} ->
-          Queue
-      end
-  end.
-
-resolve_begin_offset(_Sock, _T, _P, Offset) when is_integer(Offset) ->
-  Offset;
-resolve_begin_offset(Sock, Topic, Partition, last) ->
-  Earliest = resolve_begin_offset(Sock, Topic, Partition, earliest),
-  Latest = resolve_begin_offset(Sock, Topic, Partition, latest),
-  case Latest =:= Earliest of
-    true  -> erlang:throw(bin("partition is empty"));
-    false -> Latest - 1
-  end;
-resolve_begin_offset(Sock, Topic, Partition, Time) ->
-  {ok, Offset} = brod_utils:resolve_offset(Sock, Topic, Partition, Time),
-  Offset.
-
-parse_source("stdin") ->
-  standard_io;
-parse_source("@" ++ Path) ->
-  parse_source(Path);
-parse_source(Path) ->
-  case filelib:is_regular(Path) of
-    true -> {file, Path};
-    false -> erlang:throw(bin(["bad file ", Path]))
-  end.
-
-parse_size(Size) ->
-  case lists:reverse(Size) of
-    "K" ++ N -> int(lists:reverse(N)) * (1 bsl 10);
-    "M" ++ N -> int(lists:reverse(N)) * (1 bsl 20);
-    N        -> int(lists:reverse(N))
-  end.
-
-format_metadata(Metadata, Format, IsList, IsToListUrp) ->
-  Brokers = kf(brokers, Metadata),
-  Topics0 = kf(topics, Metadata),
-  Cluster = kf(cluster_id, Metadata, ?undef),
-  Controller = kf(controller_id, Metadata, ?undef),
-  Topics1 = case IsToListUrp of
-              true -> lists:filter(fun is_ur_topic/1, Topics0);
-              false -> Topics0
-            end,
-  Topics = format_topics(Topics1),
-  case Format of
-    json ->
-      JSON = jsone:encode([ {brokers, Brokers}
-                          , {topics, Topics}
-                          , {cluster_id, Cluster}
-                          , {controller_id, Controller}
-                          ]),
-      print([JSON, "\n"]);
-    text ->
-      CL = case Cluster of
-             ?undef -> "";
-             _ -> io_lib:format("cluster_id: ~s\n", [Cluster])
-           end,
-
-      CT = case Controller of
-             ?undef -> "";
-             _ -> io_lib:format("controller: ~p\n", [Controller])
-           end,
-      BL = format_broker_lines(Brokers),
-      TL = format_topics_lines(Topics, IsList),
-      case IsList of
-        true -> print(TL);
-        false -> print([CL, CT, BL, TL])
-      end
-  end.
-
-format_broker_lines(Brokers) ->
-  Header = io_lib:format("brokers [~p]:\n", [length(Brokers)]),
-  F = fun(Broker) ->
-          Id = kf(node_id, Broker),
-          Host = kf(host, Broker),
-          Port = kf(port, Broker),
-          Rack = kf(rack, Broker, <<>>),
-          HostStr = fmt_endpoint({Host, Port}),
-          format_broker_line(Id, Rack, HostStr)
-      end,
-  [Header, lists:map(F, Brokers)].
-
-format_broker_line(Id, Rack, Endpoint)
- when Rack =:= ?kpro_null orelse Rack =:= <<>> ->
-  io_lib:format("  ~p: ~s\n", [Id, Endpoint]);
-format_broker_line(Id, Rack, Endpoint) ->
-  io_lib:format("  ~p(~s): ~s\n", [Id, Rack, Endpoint]).
-
-format_topics_lines(Topics, true) ->
-  Header = io_lib:format("topics [~p]:\n", [length(Topics)]),
-  [Header, lists:map(fun format_topic_list_line/1, Topics)];
-format_topics_lines(Topics, false) ->
-  Header = io_lib:format("topics [~p]:\n", [length(Topics)]),
-  [Header, lists:map(fun format_topic_lines/1, Topics)].
-
-format_topic_list_line({Name, Partitions}) when is_list(Partitions) ->
-  io_lib:format("  ~s\n", [Name]);
-format_topic_list_line({Name, ErrorCode}) ->
-  ErrorStr = format_error_code(ErrorCode),
-  io_lib:format("  ~s: [ERROR] ~s\n", [Name, ErrorStr]).
-
-format_topic_lines({Name, Partitions}) when is_list(Partitions) ->
-  Header = io_lib:format("  ~s [~p]:\n", [Name, length(Partitions)]),
-  PartitionsText = format_partitions_lines(Partitions),
-  [Header, PartitionsText];
-format_topic_lines({Name, ErrorCode}) ->
-  ErrorStr = format_error_code(ErrorCode),
-  io_lib:format("  ~s: [ERROR] ~s\n", [Name, ErrorStr]).
-
-format_error_code(E) when is_atom(E) -> atom_to_list(E);
-format_error_code(E) when is_integer(E) -> integer_to_list(E).
-
-format_partitions_lines(Partitions0) ->
-  Partitions1 =
-    lists:map(fun({Pnr, Info}) ->
-                  {binary_to_integer(Pnr), Info}
-              end, Partitions0),
-  Partitions = lists:keysort(1, Partitions1),
-  lists:map(fun format_partition_lines/1, Partitions).
-
-format_partition_lines({Partition, Info}) ->
-  LeaderNodeId = kf(leader, Info),
-  Status = kf(status, Info),
-  Isr = kf(isr, Info),
-  Osr = kf(osr, Info),
-  MaybeWarning = case ?IS_ERROR(Status) of
-                   true -> [" [", atom_to_list(Status), "]"];
-                   false -> ""
-                 end,
-  ReplicaList =
-    case Osr of
-      [] -> format_list(Isr, "");
-      _  -> [format_list(Isr, ""), ",", format_list(Osr, "*")]
-    end,
-  io_lib:format("~7s: ~2s (~s)~s\n",
-                [integer_to_list(Partition),
-                 integer_to_list(LeaderNodeId),
-                 ReplicaList, MaybeWarning]).
-
-format_list(List, Mark) ->
-  infix(lists:map(fun(I) -> [integer_to_list(I), Mark] end, List), ",").
-
-infix([], _Sep) -> [];
-infix([_] = L, _Sep) -> L;
-infix([H | T], Sep) -> [H, Sep, infix(T, Sep)].
-
-format_topics(Topics) ->
-  TL = lists:map(fun format_topic/1, Topics),
-  lists:keysort(1, TL).
-
-format_topic(Topic) ->
-  TopicName = kf(name, Topic),
-  PL = kf(partitions, Topic),
-  {TopicName, format_partitions(PL)}.
-
-format_partitions(Partitions) ->
-  PL = lists:map(fun format_partition/1, Partitions),
-  lists:keysort(1, PL).
-
-format_partition(P) ->
-  ErrorCode = kf(error_code, P),
-  PartitionNr = kf(partition_index, P),
-  LeaderNodeId = kf(leader_id, P),
-  Replicas = kf(replica_nodes, P),
-  Isr = kf(isr_nodes, P),
-  Data = [ {leader, LeaderNodeId}
-         , {status, ErrorCode}
-         , {isr, Isr}
-         , {osr, Replicas -- Isr}
-         ],
-  {integer_to_binary(PartitionNr), Data}.
-
-%% Return true if a topics is under-replicated
-is_ur_topic(Topic) ->
-  ErrorCode = kf(error_code, Topic),
-  Partitions = kf(partition_metadata, Topic),
-  %% when there is an error, we do not know if
-  %% it is under-replicated or not
-  %% return true to alert user
-  ?IS_ERROR(ErrorCode) orelse lists:any(fun is_ur_partition/1, Partitions).
-
-%% Return true if a partition is under-replicated
-is_ur_partition(Partition) ->
-  ErrorCode = kf(error_code, Partition),
-  Replicas = kf(replicas, Partition),
-  Isr = kf(isr, Partition),
-  ?IS_ERROR(ErrorCode) orelse lists:sort(Isr) =/= lists:sort(Replicas).
-
-parse_delimiter("none") -> none;
-parse_delimiter(EscappedStr) -> eval_str(EscappedStr).
-
-eval_str([]) -> [];
-eval_str([$\\, $n | Rest]) ->
-  [$\n | eval_str(Rest)];
-eval_str([$\\, $t | Rest]) ->
-  [$\t | eval_str(Rest)];
-eval_str([$\\, $s | Rest]) ->
-  [$\s | eval_str(Rest)];
-eval_str([C | Rest]) ->
-  [C | eval_str(Rest)].
-
-parse_fmt("v", _KvDel, MsgDeli) ->
-  fun(_Offset, _Key, Value) -> [Value, MsgDeli] end;
-parse_fmt("kv", KvDeli, MsgDeli) ->
-  fun(_Offset, Key, Value) -> [Key, KvDeli, Value, MsgDeli] end;
-parse_fmt("okv", KvDeli, MsgDeli) ->
-  fun(Offset, Key, Value) ->
-      [integer_to_list(Offset), KvDeli,
-       Key, KvDeli, Value, MsgDeli]
-  end;
-parse_fmt("eterm", _KvDeli, _MsgDeli) ->
-  fun(Offset, Key, Value) ->
-      io_lib:format("~p.\n", [{Offset, Key, Value}])
-  end;
-parse_fmt(FunLiteral0, _KvDeli, _MsgDeli) ->
-  FunLiteral = ensure_end_with_dot(FunLiteral0),
-  {ok, Tokens, _Line} = erl_scan:string(FunLiteral),
-  {ok, [Expr]} = erl_parse:parse_exprs(Tokens),
-  fun(#kafka_message{offset     = Offset,
-                     key        = Key,
-                     value      = Value,
-                     ts_type    = TsType,
-                     ts         = Ts,
-                     headers    = Headers
-                    }) ->
-      Bindings =
-        lists:foldl(
-          fun({VarName, VarValue}, Acc) ->
-            erl_eval:add_binding(VarName, VarValue, Acc)
-          end, erl_eval:new_bindings(),
-          [ {'Offset', Offset}
-          , {'Key', Key}
-          , {'Value', Value}
-          , {'TsType', TsType}
-          , {'Ts', Ts}
-          , {'Headers', Headers}
-          ]),
-      {value, Val, _NewBindings} = erl_eval:expr(Expr, Bindings),
-      case Val of
-        F when is_function(F, 3) ->
-          %% for backward compatibility
-          F(Offset, Key, Value);
-        V ->
-          V
-      end
-  end.
-
-%% Append a dot to the function literal.
-ensure_end_with_dot(Str0) ->
-  Str = rstrip(Str0, [$\n, $\t, $\s, $.]),
-  Str ++ ".".
-
-rstrip(Str, CharSet) ->
-  lists:reverse(lstrip(lists:reverse(Str), CharSet)).
-
-lstrip([], _) -> [];
-lstrip([C | Rest] = Str, CharSet) ->
-  case lists:member(C, CharSet) of
-    true -> lstrip(Rest, CharSet);
-    false -> Str
-  end.
-
-parse_partition("random") ->
-  fun(_Topic, PartitionsCount, _Key, _Value) ->
-      {_, _, Micro} = os:timestamp(),
-      {ok, Micro rem PartitionsCount}
-  end;
-parse_partition("hash") ->
-  fun(_Topic, PartitionsCount, Key, _Value) ->
-      Hash = erlang:phash2(Key),
-      {ok, Hash rem PartitionsCount}
-  end;
-parse_partition(I) ->
-  try
-    list_to_integer(I)
-  catch
-    _ : _ ->
-      erlang:throw(bin(["Bad partition: ", I]))
-  end.
-
-parse_acks("all") -> -1;
-parse_acks("-1") -> -1;
-parse_acks("0") -> 0;
-parse_acks("1") -> 1;
-parse_acks(X) -> erlang:throw(bin(["Bad --acks value: ", X])).
-
-parse_timeout(Str) ->
-  case lists:reverse(Str) of
-    "s" ++ R -> int(lists:reverse(R)) * 1000;
-    "m" ++ R -> int(lists:reverse(R)) * 60 * 1000;
-    _        -> int(Str)
-  end.
-
-parse_compression("none") -> no_compression;
-parse_compression("gzip") -> gzip;
-parse_compression("snappy") -> snappy;
-parse_compression(X) -> erlang:throw(bin(["Unknown --compresion value: ", X])).
-
-parse_offset_time("earliest") -> earliest;
-parse_offset_time("latest") -> latest;
-parse_offset_time("last") -> last;
-parse_offset_time(T) -> int(T).
-
-parse_connection_config(Args) ->
-  SslBool = parse(Args, "--ssl", fun parse_boolean/1),
-  SslVersions = parse(Args, "--ssl-versions", fun parse_ssl_versions/1),
-  CaCertFile = parse(Args, "--cacertfile", fun parse_file/1),
-  CertFile = parse(Args, "--certfile", fun parse_file/1),
-  KeyFile = parse(Args, "--keyfile", fun parse_file/1),
-  FilterPred = fun({_, V}) -> V =/= ?undef end,
-  SslOpt =
-    case SslBool of
-      true ->
-        Opts =
-          [{cacertfile, CaCertFile},
-           {certfile, CertFile},
-           {keyfile, KeyFile},
-           {versions, SslVersions},
-           %% TODO: verify_peer if cacertfile is provided
-           {verify, verify_none}
-          ],
-        lists:filter(FilterPred, Opts);
-      false ->
-        false
-    end,
-  SaslPlain = parse(Args, "--sasl-plain", fun parse_file/1),
-  SaslScram256 = parse(Args, "--scram256", fun parse_file/1),
-  SaslScram512 = parse(Args, "--scram512", fun parse_file/1),
-  SaslOpts0 = [ {scram_sha_512, SaslScram512}
-              , {scram_sha_256, SaslScram256}
-              , {plain, SaslPlain}
-              ],
-  SaslOpts = case lists:filter(FilterPred, SaslOpts0) of
-               [] -> [];
-               [H | _] -> [{sasl, H}]
-             end,
-  lists:filter(FilterPred, [{ssl, SslOpt} | SaslOpts]).
-
-parse_boolean(true) -> true;
-parse_boolean(false) -> false;
-parse_boolean("true") -> true;
-parse_boolean("false") -> false;
-parse_boolean(?undef) -> false.
-
-parse_cg_ids("") -> [];
-parse_cg_ids("all") -> all;
-parse_cg_ids(Str) -> [bin(I) || I <- string:tokens(Str, ",")].
-
-parse_ssl_versions(?undef) ->
-    parse_ssl_versions("");
-parse_ssl_versions(Versions) ->
-    case lists:map(fun parse_ssl_version/1, string:tokens(Versions, ", ")) of
-        [] ->
-            ['tlsv1.2'];
-        Vsns ->
-            Vsns
-    end.
-
-parse_ssl_version("1.2") ->
-    'tlsv1.2';
-parse_ssl_version("1.3") ->
-    'tlsv1.3';
-parse_ssl_version("1.1") ->
-    'tlsv1.1';
-parse_ssl_version(Other) ->
-    error({unsupported_tls_version, Other}).
-
-parse_file(?undef) ->
-  ?undef;
-parse_file(Path) ->
-  case filelib:is_regular(Path) of
-    true  -> Path;
-    false -> erlang:throw(bin(["bad file ", Path]))
-  end.
-
-parse(Args, OptName, ParseFun) ->
-  case lists:keyfind(OptName, 1, Args) of
-    {_, Arg} ->
-      try
-        ParseFun(Arg)
-      catch
-        C : E ?BIND_STACKTRACE(Stack) ->
-          ?GET_STACKTRACE(Stack),
-          verbose("~p:~p\n~p\n", [C, E, Stack]),
-          Reason =
-            case Arg of
-              ?undef -> ["Missing option ", OptName];
-              _      -> ["Failed to parse ", OptName, ": ", Arg]
-            end,
-          erlang:throw(bin(Reason))
-      end;
-    false ->
-      Reason = [OptName, " is missing"],
-      erlang:throw(bin(Reason))
-  end.
-
-print_version() ->
-  _ = application:load(brod),
-  {_, _, V} = lists:keyfind(brod, 1, application:loaded_applications()),
-  print([V, "\n"]).
-
-print(IoData) -> io:put_chars(stdio(), IoData).
-
-print(Fmt, Args) -> io:put_chars(stdio(), io_lib:format(Fmt, Args)).
-
-logerr(IoData) -> io:put_chars(stderr(), ["*** ", IoData]).
-
-logerr(Fmt, Args) ->
-  io:put_chars(stderr(), io_lib:format("*** " ++ Fmt, Args)).
-
-verbose(Fmt, Args) ->
-  case erlang:get(brod_cli_log_level) >= ?LOG_LEVEL_VERBOSE of
-    true  -> logerr("[verbo]: " ++ Fmt, Args);
-    false -> ok
-  end.
-
-debug(Fmt, Args) ->
-  case erlang:get(brod_cli_log_level) >= ?LOG_LEVEL_DEBUG of
-    true  -> logerr("[debug]: " ++ Fmt, Args);
-    false -> ok
-  end.
-
-stdio() ->
-  case get(redirect_stdio) of
-    undefined -> user;
-    Other -> Other
-  end.
-
-stderr() ->
-  case get(redirect_stderr) of
-    undefined -> standard_error;
-    Other -> Other
-  end.
-
-int(Str) -> list_to_integer(trim(Str)).
-
-trim_h([$\s | T]) -> trim_h(T);
-trim_h(X) -> X.
-
-trim(Str) -> trim_h(lists:reverse(trim_h(lists:reverse(Str)))).
-
-bin(IoData) -> iolist_to_binary(IoData).
-
-parse_brokers(HostsStr) ->
-  F = fun(HostPortStr) ->
-          Pair = string:tokens(HostPortStr, ":"),
-          case Pair of
-            [Host, PortStr] -> {Host, list_to_integer(PortStr)};
-            [Host]          -> {Host, 9092}
-          end
-      end,
-  shuffle(lists:map(F, string:tokens(HostsStr, ","))).
-
-%% Parse code paths.
-parse_paths(?undef) -> [];
-parse_paths(Str) -> string:tokens(Str, ",").
-
-%% Randomize the order.
-shuffle(L) ->
-  RandList = lists:map(fun(_) -> element(3, os:timestamp()) end, L),
-  {_, SortedL} = lists:unzip(lists:keysort(1, lists:zip(RandList, L))),
-  SortedL.
-
--spec kf(kpro:field_name(), kpro:struct()) -> kpro:field_value().
-kf(FieldName, Struct) -> kpro:find(FieldName, Struct).
-
--spec kf(kpro:field_name(), kpro:struct(), kpro:field_value()) ->
-        kpro:field_value().
-kf(FieldName, Struct, Default) ->
-  kpro:find(FieldName, Struct, Default).
-
-start_client(BootstrapEndpoints, ClientConfig) ->
-  {ok, _} = brod_client:start_link(BootstrapEndpoints, ?CLIENT, ClientConfig),
-  ok.
-
--spec throw_bin(string(), [term()]) -> no_return().
-throw_bin(Fmt, Args) ->
-  erlang:throw(bin(io_lib:format(Fmt, Args))).
-
-get_kafka_version() ->
-  case os:getenv("KAFKA_VERSION") of
-    false ->
-      ?LATEST_KAFKA_VERSION;
-    Vsn ->
-      [Major, Minor | _] = string:tokens(Vsn, "."),
-      {list_to_integer(Major), list_to_integer(Minor)}
-  end.
-
--endif.
-
-%%%_* Emacs ====================================================================
-%%% Local Variables:
-%%% allout-layout: t
-%%% erlang-indent-level: 2
-%%% End:
diff --git a/src/brod_cli_pipe.erl b/src/brod_cli_pipe.erl
deleted file mode 100644
index 54c4c033..00000000
--- a/src/brod_cli_pipe.erl
+++ /dev/null
@@ -1,390 +0,0 @@
-%%%
-%%%   Copyright (c) 2017-2021, Klarna Bank AB (publ)
-%%%
-%%%   Licensed under the Apache License, Version 2.0 (the "License");
-%%%   you may not use this file except in compliance with the License.
-%%%   You may obtain a copy of the License at
-%%%
-%%%       http://www.apache.org/licenses/LICENSE-2.0
-%%%
-%%%   Unless required by applicable law or agreed to in writing, software
-%%%   distributed under the License is distributed on an "AS IS" BASIS,
-%%%   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-%%%   See the License for the specific language governing permissions and
-%%%   limitations under the License.
-%%%
-
-%% @private The input source of brod-cli pipe command.
-%% This module implements a process that reads off the bytes
-%% from the data source (either stdin or a file)
-%% and sends the bytes to parent process.
-%% Messages sent to parent process:
-%%
-%% ```{pipe, self(), [{Ts :: integer(), Key :: binary(), Val :: binary()}]}'''
-%% @end
--module(brod_cli_pipe).
-
--ifdef(build_brod_cli).
-
--behaviour(gen_server).
-
--export([ start_link/1
-        , stop/1
-        ]).
-
--export([ code_change/3
-        , handle_call/3
-        , handle_cast/2
-        , handle_info/2
-        , init/1
-        , terminate/2
-        ]).
-
--include("brod_int.hrl").
-
--type arg_name() :: source
-                  | kv_deli
-                  | msg_deli
-                  | prompt
-                  | tail
-                  | no_exit
-                  | blk_size
-                  | retry_delay.
-
--type arg_value() :: term().
-
--define(LINE_BREAK, <<"\n">>).
--define(STDIN, standard_io).
--define(EOF_RETRY_DELAY_MS, 100).
--define(CONTINUE_MSG, continue).
--define(PARENT_BUSY_MSG_QUEUE_LEN_THRESHOLD, 100).
-
--type delimiter() :: binary().
--type epoch_ms() :: integer().
--type read_fun() ::
-        fun((?STDIN | file:io_device(), [binary()]) -> eof |
-              {[{epoch_ms(), Key :: binary(), Val :: binary()}], [binary()]}).
-
--record(state, { parent :: pid()
-               , source :: ?STDIN | {file, string()}
-               , read_fun :: read_fun()
-               , is_eof_exit :: boolean()
-               , is_tail :: boolean()
-               , io_device :: ?undef | ?STDIN | file:io_device()
-               , acc_bytes = [] :: [binary()]
-               , retry_delay :: timeout()
-               }).
-
--type state() :: #state{}.
-
-%% @doc Args explained:
-%%
-%% source:   'standard_io' | {file, "path/to/source"}
-%% kv_deli:  'none' | binary().
-%%           Delimiter bytes for message key and value
-%% msg_deli: binary(). Delimiter between kafka messages
-%%           NOTE: eof is always considered a message delimiter
-%% prompt:   boolean(). Applicable when source is standard_io AND
-%%           when kv_deli and msg_deli are both '\n'
-%%           prompts 'key> ' for key input and 'val> ' for value input
-%% tail:     boolean(). Applicable when source is a file
-%%           tell brod-cli to start reading from EOF
-%% no_exit:  boolean(). Do not exit when reaching EOF
-%% blk_size: Read block size
--spec start_link([{arg_name(), arg_value()}]) -> {ok, pid()}.
-start_link(Args) ->
-  Parent = self(),
-  Arg = fun(Name) -> {_, V} = lists:keyfind(Name, 1, Args), V end,
-  KvDeli = Arg(kv_deli),
-  MsgDeli = Arg(msg_deli),
-  Source = Arg(source),
-  IsLineMode = MsgDeli =:= ?LINE_BREAK,
-  BlkSize = Arg(blk_size),
-  IsPrompt = Arg(prompt),
-  IsTail = Arg(tail),
-  IsNoExit = Arg(no_exit),
-  IsEofExit = not (IsTail orelse IsNoExit),
-  ReadFun =
-    case IsLineMode of
-      true when IsPrompt andalso Source =:= ?STDIN ->
-        make_prompt_line_reader(KvDeli);
-      true ->
-        make_line_reader(KvDeli, _PromptStr = "");
-      false ->
-        make_stream_reader(KvDeli, MsgDeli, BlkSize, IsEofExit)
-    end,
-  State = #state{ parent = Parent
-                , source = Source
-                , read_fun = ReadFun
-                , is_tail = IsTail
-                , is_eof_exit = IsEofExit
-                , retry_delay = Arg(retry_delay)
-                },
-  gen_server:start_link({local, ?MODULE}, ?MODULE, State, []).
-
-%% @doc Stop gen_server.
-stop(Pid) -> gen_server:cast(Pid, stop).
-
-%% @doc Tell reader to continue.
-continue() -> self() ! ?CONTINUE_MSG.
-
-%% @hidden
-init(#state{source = Source, is_tail = IsTail} = State0) ->
-  IoDevice =
-    case Source of
-      ?STDIN ->
-        ?STDIN;
-      {file, File} ->
-        {ok, Fd} = file:open(File, [read, binary]),
-        IsTail andalso file:position(Fd, eof),
-        Fd
-    end,
-  State = State0#state{io_device = IoDevice},
-  _ = continue(),
-  {ok, State}.
-
-%% @hidden
-handle_info(?CONTINUE_MSG, #state{parent = Parent} = State) ->
-  case erlang:process_info(Parent, message_queue_len) of
-    {_, Len} when Len >= ?PARENT_BUSY_MSG_QUEUE_LEN_THRESHOLD ->
-      ok = delay_continue(State),
-      {noreply, State};
-    _ ->
-      handle_read(State)
-  end;
-handle_info(_Info, State) ->
-  {noreply, State}.
-
-%% @hidden
-handle_cast(stop, State) ->
-  {stop, normal, State};
-handle_cast(_Cast, State) ->
-  {noreply, State}.
-
-%% @hidden
-handle_call(Call, _From, State) ->
-  {reply, {error, {unknown_call, Call}}, State}.
-
-%% @hidden
-code_change(_OldVsn, State, _Extra) ->
-  {ok, State}.
-
-%% @hidden
-terminate(_Reason, _State) ->
-  ok.
-
-%%%_* Privates =================================================================
-
-send_to_parent(Parent, Msgs0) ->
-  FilterF = fun(?TKV(_T, K, V)) -> K =/= <<>> orelse V =/= <<>> end,
-  Msgs = lists:filter(FilterF, Msgs0),
-  Msgs =/= [] andalso erlang:send(Parent, {pipe, self(), Msgs}),
-  ok.
-
-handle_read(#state{ read_fun = ReadFun
-                  , acc_bytes = Acc0
-                  , io_device = IoDevice
-                  , parent = Parent
-                  } = State0) ->
-  case ReadFun(IoDevice, Acc0) of
-    eof ->
-      handle_eof(State0);
-    {Msgs, Acc} ->
-      _ = continue(), %% continue next trunk
-      ok = send_to_parent(Parent, Msgs),
-      State = State0#state{acc_bytes = Acc},
-      {noreply, State}
-  end.
-
-handle_eof(#state{io_device = ?STDIN} = State) ->
-  %% standard_io pipe closed
-  {stop, normal, State};
-handle_eof(#state{is_eof_exit = true} = State) ->
-  {stop, normal, State};
-handle_eof(#state{io_device = Fd} = State) ->
-  %% Get current position
-  {ok, LastPos} = file:position(Fd, {cur, 0}),
-  %% Try set position to EOF,
-  %% see if it is the current position
-  case file:position(Fd, eof) of
-    {ok, NewPos} when NewPos < LastPos ->
-      %% File has been truncated.
-      %% Don't know what to do because
-      %% we can not assume the file is truncated to empty
-      {stop, pipe_source_truncated, State};
-    {ok, _Pos} ->
-      _ = file:position(Fd, LastPos),
-      ok = delay_continue(State),
-      {noreply, State}
-  end.
-
--spec delay_continue(state()) ->
-  ok.
-delay_continue(#state{retry_delay = Delay}) ->
-  _ = erlang:send_after(Delay, self(), ?CONTINUE_MSG),
-  ok.
-
--spec make_prompt_line_reader(none | delimiter()) -> read_fun().
-make_prompt_line_reader(_KvDeli = none) ->
-  %% Read only value, no key
-  fun(?STDIN, _Acc) ->
-      case read_line(?STDIN, "VAL> ") of
-        eof   -> eof;
-        Value -> {[{_Key = <<>>, Value}], []}
-      end
-  end;
-make_prompt_line_reader(_KvDeli = ?LINE_BREAK) ->
-  fun(?STDIN, _Acc) ->
-      case read_line(?STDIN, "KEY> ") of
-        eof -> eof;
-        Key ->
-          case read_line(?STDIN, "VAL> ") of
-            eof -> {[make_msg(Key, <<>>)], []};
-            Value -> {[make_msg(Key, Value)], []}
-          end
-      end
-  end;
-make_prompt_line_reader(KvDeli) ->
-  Prompt = "KEY" ++ binary_to_list(KvDeli) ++ "VAL> ",
-  make_line_reader(KvDeli, Prompt).
-
--spec make_line_reader(none | binary(), string()) -> read_fun().
-make_line_reader(KvDeli, Prompt) ->
-  fun(IoDevice, _Acc) ->
-      case read_line(IoDevice, Prompt) of
-        eof ->
-          eof;
-        Key when KvDeli =:= <<"\n">> ->
-          case read_line(IoDevice, Prompt) of
-            eof ->
-              eof;
-            Val ->
-              {[make_msg(Key, Val)], []}
-          end;
-        Val when KvDeli =:= none ->
-          {[make_msg(<<>>, Val)], []};
-        Line ->
-          [Key, Value] = binary:split(Line, bin(KvDeli)),
-          {[make_msg(Key, Value)], []}
-      end
-  end.
-
--spec make_stream_reader(none | delimiter(), delimiter(),
-                         pos_integer(), boolean()) -> read_fun().
-make_stream_reader(KvDeli, MsgDeli, BlkSize, IsEofExit) ->
-  IsSameDeli = MsgDeli =:= KvDeli,
-  KvDeliCp = case is_binary(KvDeli) of
-               true -> binary:compile_pattern(KvDeli);
-               false -> none
-             end,
-  MsgDeliCp = binary:compile_pattern(MsgDeli),
-  fun(IoDevice, Acc) ->
-      case file:read(IoDevice, BlkSize) of
-        eof ->
-          case IsEofExit of
-            true when Acc =:= [] ->
-              %% Reached EOF
-              eof;
-            true ->
-              %% Configured to exit when reaching EOF
-              %% try split kv-pairs NOW
-              LastMsg = bin(lists:reverse(Acc)),
-              KvPairs = split_kv_pairs([LastMsg], KvDeliCp, IsSameDeli),
-              {KvPairs, []};
-            false ->
-              %% Keep looping for the next message delimiter
-              {[], Acc}
-          end;
-        {ok, Bytes} ->
-          Acc1 = add_acc(size(MsgDeli), Bytes, Acc),
-          {Messages, NewAcc} = split_messages(MsgDeliCp, Acc1),
-          KvPairs = split_kv_pairs(Messages, KvDeliCp, IsSameDeli),
-          {KvPairs, NewAcc}
-      end
-  end.
-
--spec add_acc(pos_integer(), binary(), [binary()]) -> [binary()].
-add_acc(_DeliSize = 1, Bytes, Acc) ->
-  %% Delimiter is only one byte, in no way could it be cut in half
-  [Bytes | Acc];
-add_acc(_DeliSize, Bytes, []) ->
-  [Bytes];
-add_acc(DeliSize, Bytes, [Tail | Header]) ->
-  Size = size(Tail) - DeliSize,
-  case Size =< 0 of
-    true ->
-      [<<Tail/binary, Bytes/binary>> | Header];
-    false ->
-      %% cut a DeliSize tail from acc and prepend as current head
-      %% to make sure we will not cut delimiter into two chunks
-      <<TailH:Size/binary, TailT/binary>> = Tail, %% cut
-      NewTail = <<TailT/binary, Bytes/binary>>, %% new tail
-      [NewTail, TailH | Header]
-  end.
-
--spec split_messages(binary:cp(), [binary()]) -> {[binary()], [binary()]}.
-split_messages(MsgDeliCp, [Tail | Header]) ->
-  case binary:split(Tail, MsgDeliCp, [global]) of
-    [_] ->
-      %% no delimiter found
-      {[], [Tail | Header]};
-    [First0 | More] ->
-      First = bin([lists:reverse(Header), First0]),
-      case lists:reverse(More) of
-        [<<>> | Msgs] ->
-          {[First | lists:reverse(Msgs)], []};
-        [NewTail | Msgs] ->
-          {[First | lists:reverse(Msgs)], [NewTail]}
-      end
-  end.
-
--spec split_kv_pairs([binary()], none | delimiter(), boolean()) -> brod:value().
-split_kv_pairs(Msgs, none, _IsSameDeli) ->
-  lists:map(fun(Msg) -> make_msg(<<>>, Msg) end, Msgs);
-split_kv_pairs(Msgs, _KvDeliCp, _IsSameDeli = true) ->
-  make_msgs(Msgs);
-split_kv_pairs(Msgs, KvDeliCp, _IsSameDeli = false) ->
-  lists:map(fun(Msg) ->
-                [K, V] = binary:split(Msg, KvDeliCp),
-                make_msg(K, V)
-            end, Msgs).
-
-make_msgs([]) -> [];
-make_msgs([K, V | Rest]) ->
-  [make_msg(K, V) | make_msgs(Rest)].
-
-make_msg(K, V) ->
-  CreateTs = brod_utils:epoch_ms(),
-  ?TKV(CreateTs, K, V).
-
--spec read_line(?STDIN | file:io_device(), string()) -> eof | binary().
-read_line(IoDevice, Prompt) ->
-  case io:get_line(IoDevice, Prompt) of
-    eof -> eof;
-    Line ->
-      Chars = unicode:characters_to_list(Line),
-      unicode:characters_to_binary(rstrip(Chars, "\n"))
-  end.
-
--spec rstrip(string(), string()) -> string().
-rstrip(Str, CharSet) ->
-  lists:reverse(lstrip(lists:reverse(Str), CharSet)).
-
--spec lstrip(string(), string()) -> string().
-lstrip([], _) -> [];
-lstrip([C | Rest] = Str, CharSet) ->
-  case lists:member(C, CharSet) of
-    true -> lstrip(Rest, CharSet);
-    false -> Str
-  end.
-
--spec bin(iodata()) -> binary().
-bin(X) -> iolist_to_binary(X).
-
--endif.
-
-%%%_* Emacs ====================================================================
-%%% Local Variables:
-%%% allout-layout: t
-%%% erlang-indent-level: 2
-%%% End:
diff --git a/test/brod_cli_pipe_tests.erl b/test/brod_cli_pipe_tests.erl
deleted file mode 100644
index b7568596..00000000
--- a/test/brod_cli_pipe_tests.erl
+++ /dev/null
@@ -1,162 +0,0 @@
-%%%
-%%%   Copyright (c) 2017-2021, Klarna Bank AB (publ)
-%%%
-%%%   Licensed under the Apache License, Version 2.0 (the "License");
-%%%   you may not use this file except in compliance with the License.
-%%%   You may obtain a copy of the License at
-%%%
-%%%       http://www.apache.org/licenses/LICENSE-2.0
-%%%
-%%%   Unless required by applicable law or agreed to in writing, software
-%%%   distributed under the License is distributed on an "AS IS" BASIS,
-%%%   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-%%%   See the License for the specific language governing permissions and
-%%%   limitations under the License.
-%%%
-
--module(brod_cli_pipe_tests).
-
--ifdef(build_brod_cli).
-
--include_lib("eunit/include/eunit.hrl").
--include("brod_int.hrl").
-
--define(WAIT(Pattern, Handle, Timeout),
-        fun Wait() ->
-          receive
-            {'EXIT', _Pid, normal} ->
-              %% discard normal exits of linked pid
-              Wait();
-            Pattern ->
-              Handle;
-            Msg ->
-              erlang:error({unexpected,
-                            [{line, ?LINE},
-                             {pattern, ??Pattern},
-                             {received, Msg}]})
-          after
-            Timeout ->
-              erlang:error({timeout,
-                            [{line, ?LINE},
-                             {pattern, ??Pattern}]})
-          end
-        end()).
-
--define(TEST_FILE, "pipe.testdata").
-
-line_mode_test() ->
-  Lines = ["key1\n", "val1\n", "key2\n", "val2\n", "key3\n"],
-  ok = file:write_file(?TEST_FILE, Lines),
-  Args =
-    [ {source, {file, ?TEST_FILE}}
-    , {kv_deli, <<"\n">>}
-    , {msg_deli, <<"\n">>}
-    , {prompt, false}
-    , {tail, false}
-    , {no_exit, false}
-    , {blk_size, ignore}
-    , {retry_delay, ignore}
-    ],
-  {ok, Pid} = brod_cli_pipe:start_link(Args),
-  erlang:monitor(process, Pid),
-  ?WAIT({pipe, Pid, [?TKV(_, <<"key1">>, <<"val1">>)]}, ok, 1000),
-  ?WAIT({pipe, Pid, [?TKV(_, <<"key2">>, <<"val2">>)]}, ok, 1000),
-  ?WAIT({'DOWN', _Ref, process, Pid, normal}, ok, 1000),
-  ok.
-
-line_mode_no_key_test() ->
-  Lines = ["val1\n", "val2\n", "val3\n"],
-  ok = file:write_file(?TEST_FILE, Lines),
-  Args =
-    [ {source, {file, ?TEST_FILE}}
-    , {kv_deli, none}
-    , {msg_deli, <<"\n">>}
-    , {prompt, true}
-    , {tail, false}
-    , {no_exit, false}
-    , {blk_size, ignore}
-    , {retry_delay, ignore}
-    ],
-  {ok, Pid} = brod_cli_pipe:start_link(Args),
-  erlang:monitor(process, Pid),
-  ?WAIT({pipe, Pid, [?TKV(_, <<>>, <<"val1">>)]}, ok, 1000),
-  ?WAIT({pipe, Pid, [?TKV(_, <<>>, <<"val2">>)]}, ok, 1000),
-  ?WAIT({pipe, Pid, [?TKV(_, <<>>, <<"val3">>)]}, ok, 1000),
-  ?WAIT({'DOWN', _Ref, process, Pid, normal}, ok, 1000),
-  ok.
-
-line_mode_split_test() ->
-  Lines = ["key1:val1\n", "key2:val2\n"],
-  ok = file:write_file(?TEST_FILE, Lines),
-  Args =
-    [ {source, {file, ?TEST_FILE}}
-    , {kv_deli, <<":">>}
-    , {msg_deli, <<"\n">>}
-    , {prompt, false}
-    , {tail, false}
-    , {no_exit, false}
-    , {blk_size, 1}
-    , {retry_delay, ignore}
-    ],
-  {ok, Pid} = brod_cli_pipe:start_link(Args),
-  erlang:monitor(process, Pid),
-  ?WAIT({pipe, Pid, [?TKV(_, <<"key1">>, <<"val1">>)]}, ok, 1000),
-  ?WAIT({pipe, Pid, [?TKV(_, <<"key2">>, <<"val2">>)]}, ok, 1000),
-  ?WAIT({'DOWN', _Ref, process, Pid, normal}, ok, 1000),
-  ok.
-
-stream_one_byte_delimiter_test() ->
-  Data = "key1:val1#key2:val2#key3:val3",
-  ok = file:write_file(?TEST_FILE, Data),
-  Args =
-    [ {source, {file, ?TEST_FILE}}
-    , {kv_deli, <<":">>}
-    , {msg_deli, <<"#">>}
-    , {prompt, false}
-    , {tail, false}
-    , {no_exit, false}
-    , {blk_size, 2}
-    , {retry_delay, ignore}
-    ],
-  {ok, Pid} = brod_cli_pipe:start_link(Args),
-  erlang:monitor(process, Pid),
-  ?WAIT({pipe, Pid, [?TKV(_, <<"key1">>, <<"val1">>)]}, ok, 1000),
-  ?WAIT({pipe, Pid, [?TKV(_, <<"key2">>, <<"val2">>)]}, ok, 1000),
-  ?WAIT({pipe, Pid, [?TKV(_, <<"key3">>, <<"val3">>)]}, ok, 1000),
-  ?WAIT({'DOWN', _Ref, process, Pid, normal}, ok, 1000),
-  ok.
-
-stream_test() ->
-  Data = "key1::val1###key2::val2###key3::val3##",
-  ok = file:write_file(?TEST_FILE, Data),
-  Args =
-    [ {source, {file, ?TEST_FILE}}
-    , {kv_deli, <<"::">>}
-    , {msg_deli, <<"###">>}
-    , {prompt, false}
-    , {tail, false}
-    , {no_exit, true}
-    , {blk_size, 3}
-    , {retry_delay, 10}
-    ],
-  {ok, Pid} = brod_cli_pipe:start_link(Args),
-  erlang:monitor(process, Pid),
-  ?WAIT({pipe, Pid, [?TKV(_, <<"key1">>, <<"val1">>)]}, ok, 1000),
-  ?WAIT({pipe, Pid, [?TKV(_, <<"key2">>, <<"val2">>)]}, ok, 1000),
-  ?assertError({timeout, _},
-               ?WAIT({pipe, Pid, [{<<"key3">>, <<"val3">>}]}, ok, 500)),
-  MoreData = "#key-4::val-4###",
-  ok = file:write_file(?TEST_FILE, MoreData, [append]),
-  ?WAIT({pipe, Pid, [?TKV(_, <<"key3">>, <<"val3">>)]}, ok, 1000),
-  ?WAIT({pipe, Pid, [?TKV(_, <<"key-4">>, <<"val-4">>)]}, ok, 1000),
-  ok = brod_cli_pipe:stop(Pid),
-  ?WAIT({'DOWN', _Ref, process, Pid, normal}, ok, 1000),
-  ok.
-
--endif.
-
-%%%_* Emacs ====================================================================
-%%% Local Variables:
-%%% allout-layout: t
-%%% erlang-indent-level: 2
-%%% End:
diff --git a/test/brod_cli_tests.erl b/test/brod_cli_tests.erl
deleted file mode 100644
index 014c9e1f..00000000
--- a/test/brod_cli_tests.erl
+++ /dev/null
@@ -1,253 +0,0 @@
-%%%
-%%%   Copyright (c) 2017-2021, Klarna Bank AB (publ)
-%%%
-%%%   Licensed under the Apache License, Version 2.0 (the "License");
-%%%   you may not use this file except in compliance with the License.
-%%%   You may obtain a copy of the License at
-%%%
-%%%       http://www.apache.org/licenses/LICENSE-2.0
-%%%
-%%%   Unless required by applicable law or agreed to in writing, software
-%%%   distributed under the License is distributed on an "AS IS" BASIS,
-%%%   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-%%%   See the License for the specific language governing permissions and
-%%%   limitations under the License.
-%%%
-
--module(brod_cli_tests).
-
--ifdef(build_brod_cli).
-
--include_lib("eunit/include/eunit.hrl").
--include("brod_int.hrl").
-
-%% no crash on 'help', 'version' etc commands
-informative_test() ->
-  brod:main(["--help"]), % should not halt
-  run(["--version"]),
-  run(["meta", "--help", "--debug"]).
-
-error_test() ->
-  put(redirect_stderr, standard_io),
-  ?assertExit(N when is_integer(N), run(["-unknown-opt"])).
-
-meta_test() ->
-  run(["meta", "-b", "localhost", "-L"]),
-  run(["meta", "-b", "localhost", "-t", "test-topic"]).
-
-ssl_test() ->
-  run(["meta", "-b", "localhost:9093", "-L",
-       "--ssl",
-       "--cacertfile", "test/data/ssl/ca.pem",
-       "--keyfile", "test/data/ssl/client-key.pem",
-       "--certfile", "test/data/ssl/client-crt.pem",
-       "--ssl-versions", "1.2,1.1"
-      ]).
-
-offset_test() ->
-  Args = ["offset", "-b", "localhost", "-t", "test-topic", "-p", "0"],
-  run(Args),
-  run(Args ++ ["-T", "latest"]),
-  run(Args ++ ["-T", "earliest"]),
-  run(Args ++ ["-T", "-1"]),
-  run(Args ++ ["-T", "-2"]),
-  run(Args ++ ["-T", "0"]).
-
-send_fetch_test() ->
-  K = make_ts_str(),
-  V = make_ts_str(),
-  Args = ["--brokers", "localhost:9092,localhost:9192", "-t",
-          "test-topic", "-p", "0"],
-  Output = cmd(["send", "-k", K, "-v", V] ++ Args),
-  ?assertEqual(<<"">>, Output),
-  FetchOutput = cmd(["fetch", "-c", "1", "--fmt", "kv"] ++ Args),
-  ?assertEqual(iolist_to_binary(K ++ ":" ++ V ++ "\n"), FetchOutput),
-  ok.
-
-sasl_test_() ->
-  case get_kafka_version() of
-    ?KAFKA_0_9 -> [];
-    _ -> [fun test_sasl/0]
-  end.
-
-test_sasl() ->
-  ok = file:write_file("sasl.testdata", "alice\necila\n"),
-  K = make_ts_str(),
-  V = make_ts_str(),
-  Output =
-    cmd(["send", "--brokers", "localhost:9194,localhost:9094",
-         "-t", "test-topic", "-p", "0",
-         "--ssl",
-         "--cacertfile", "test/data/ssl/ca.pem",
-         "--keyfile", "test/data/ssl/client-key.pem",
-         "--certfile", "test/data/ssl/client-crt.pem",
-         "--sasl-plain", "sasl.testdata",
-         "-k", K, "-v", V]),
-  ?assertEqual(<<"">>, Output),
-  FetchOutput =
-    cmd(["fetch", "--brokers", "localhost:9092", "-t", "test-topic",
-         "-p", "0", "-c", "1", "--fmt", "kv"]),
-  ?assertEqual(iolist_to_binary([K, ":", V, "\n"]), FetchOutput),
-  ok.
-
-fetch_format_fun_test() ->
-  T = os:timestamp(),
-  Value = term_to_binary(T),
-  file:write_file("fetch.testdata", Value),
-  run(["send", "-b", "localhost", "-t", "test-topic", "-p", "0",
-       "-v", "@fetch.testdata"]),
-  FmtFun = "io_lib:format(\"~p\", [binary_to_term(Value)])",
-  Output =
-    cmd(["fetch", "-b", "localhost", "-t", "test-topic", "-p", "0",
-         "-c", "1", "--fmt", FmtFun]),
-  Expected = iolist_to_binary(io_lib:format("~p", [T])),
-  ?assertEqual(Expected, Output).
-
-fetch_format_expr_test() ->
-  T = os:timestamp(),
-  Value = term_to_binary(T),
-  file:write_file("fetch.testdata", Value),
-  cmd(["send", "-b", "localhost", "-t", "test-topic", "-p", "0",
-       "-v", "@fetch.testdata"]),
-  FmtExpr = "io_lib:format(\"~p\", [binary_to_term(Value)])",
-  Output =
-    cmd(["fetch", "-b", "localhost", "-t", "test-topic", "-p", "0",
-         "-c", "1", "--fmt", FmtExpr]),
-  Expected = iolist_to_binary(io_lib:format("~p", [T])),
-  ?assertEqual(Expected, Output).
-
-pipe_test() ->
-  %% get last offset
-  OffsetStr = cmd(["offset", "-b", "localhost", "-t", "test-topic", "-p", "0",
-                   "-T", "latest"]),
-  %% send file
-  PipeCmdOutput =
-    cmd(["pipe", "-b", "localhost", "-t", "test-topic", "-p", "0",
-         "-s", "README.md", "--kv-deli", "none", "--msg-deli", "'\\n'"]),
-  ?assertEqual(<<"">>, PipeCmdOutput),
-  FetchedText =
-    cmd(["fetch", "-b", "localhost", "-t", "test-topic", "-p", "0",
-         "-w", "100", "-c", "-1", "-o", binary_to_list(OffsetStr)]),
-  {ok, ReadmeText} = file:read_file("README.md"),
-  Split =
-    fun(Text) ->
-        Lines = binary:split(iolist_to_binary(Text), <<"\n">>, [global]),
-        lists:filtermap(
-          fun(<<>>) -> false;
-             (Line) -> {true, iolist_to_binary([Line, "\n"])}
-          end, Lines)
-    end,
-  ExpectedLines = Split(ReadmeText),
-  FetchedLines = Split(FetchedText),
-  ?assertEqual(hd(ExpectedLines), hd(FetchedLines)),
-  ?assertEqual(lists:last(ExpectedLines), lists:last(FetchedLines)),
-  ?assertEqual(ExpectedLines, FetchedLines).
-
-groups_test() ->
-  assert_no_error(cmd(["groups"])),
-  assert_no_error(cmd(["groups", "--ids", "all"])).
-
-commits_describe_test() ->
-  assert_no_error(cmd(["commits", "--id", "test-group", "--describe"])).
-
-commits_overwrite_test_() ->
-  {timeout, 20,
-   fun() ->
-       assert_no_error(cmd(["commits", "--id", "test-group", "-t", "test-topic",
-                           "-o", "0:1", "-r", "1d", "--protocol", "range"]))
-   end}.
-
-assert_no_error(Result) ->
-  case binary:match(iolist_to_binary(Result), <<"***">>) of
-    nomatch -> ok;
-    _ -> erlang:throw(Result)
-  end.
-
-make_ts_str() ->
-  Ts = os:timestamp(),
-  {{Y,M,D}, {H,Min,Sec}} = calendar:now_to_universal_time(Ts),
-  {_, _, Micro} = Ts,
-  S = io_lib:format("~4.4.0w-~2.2.0w-~2.2.0w:~2.2.0w:~2.2.0w:~2.2.0w.~6.6.0w",
-                    [Y, M, D, H, Min, Sec, Micro]),
-  lists:flatten(S).
-
-get_kafka_version() ->
-  case os:getenv("KAFKA_VERSION") of
-    false ->
-      ?LATEST_KAFKA_VERSION;
-    Vsn ->
-      [Major, Minor | _] = string:tokens(Vsn, "."),
-      {list_to_integer(Major), list_to_integer(Minor)}
-  end.
-
-run(Args0) ->
-  Args = case kafka_test_helper:kafka_version() of
-           {0, Minor} when Minor < 11 ->
-             Args0 ++ ["--no-api-vsn-query"];
-           _ ->
-             Args0
-         end,
-  _ = cmd(Args),
-  ok.
-
-cmd(Args) ->
-  Parent = self(),
-  IO = erlang:spawn_link(fun() -> io_loop(Parent, []) end),
-  put(redirect_stdio, IO),
-  try
-    brod_cli:main(Args, exit),
-    timer:sleep(10), % avoid race
-    IO ! stop,
-    Result = receive {outputs, Outputs} -> Outputs
-             after 5000 -> throw(timeout) end,
-    catch brod:stop_client(brod_cli_client),
-    Result
-  after
-    _ = brod:stop(),
-    ensure_down(brod_kafka_apis)
-  end.
-
-ensure_down(RegName) -> ensure_down(RegName, 10).
-
-ensure_down(RegName, 0) -> error({wont_die, RegName});
-ensure_down(RegName, N) ->
-  case whereis(RegName) of
-    undefined -> ok;
-    Pid ->
-      exit(Pid, kill),
-      timer:sleep((11 - N) * 10),
-      ensure_down(RegName, N - 1)
-  end.
-
-io_loop(Parent, Acc0) ->
-  receive
-    {io_request, From, ReplyAs, Req} ->
-      Acc = io(From, ReplyAs, Req, Acc0),
-      io_loop(Parent, Acc);
-    stop ->
-      Parent ! {outputs, iolist_to_binary(Acc0)}
-  end.
-
-io(From, ReplyAs, Req, Acc0) ->
-  {Reply, Acc} = io(Req, Acc0),
-  erlang:send(From, {io_reply, ReplyAs, Reply}),
-  Acc.
-
-%% supports only a subset of io requests
-io({put_chars, Chars}, Acc) ->
-  {ok, [Chars | Acc]};
-io({put_chars, _Code, Chars}, Acc) ->
-  io({put_chars, Chars}, Acc);
-io({setopts, _Opts}, Acc) ->
-  {ok, Acc};
-io(Unknown, _Acc) ->
-  io:format(standard_error, "unknown_io_request: ~p\n", [Unknown]),
-  exit({unknown_io_request, Unknown}).
-
--endif.
-
-%%%_* Emacs ====================================================================
-%%% Local Variables:
-%%% allout-layout: t
-%%% erlang-indent-level: 2
-%%% End: