Repository: dpkp/kafka-python
Pull request #2650 (2025-06-23 21:47:41)

python -m cli interfaces for kafka.admin, kafka.consumer, kafka.producer
Builds on @rdiomar's work in #119 :

```
$ python -m kafka.producer -b 127.0.0.1:9092 -t "hello"
yo dude
what's up?

# Meanwhile, in another terminal...
$ python -m kafka.consumer -b 127.0.0.1:9092 -t "hello" -g "analytics"
yo dude
what's up?
```

Adding kafka.admin to the mix:

```
$ python -m kafka.admin -b 127.0.0.1:9092 cluster describe
{'brokers': [{'host': 'localhost', 'node_id': 0, 'port': 9092, 'rack': None}],
 'cluster_id': 'xxx-yyy',
 'controller_id': 0,
 'throttle_time_ms': 0}
```


Patch:
diff --git a/kafka/admin/__main__.py b/kafka/admin/__main__.py
new file mode 100644
index 0000000..7760638
--- /dev/null
+++ b/kafka/admin/__main__.py
@@ -0,0 +1,7 @@
+from __future__ import absolute_import
+
+import sys
+
+from kafka.cli.admin import run_cli
+
+sys.exit(run_cli())
diff --git a/kafka/admin/config_resource.py b/kafka/admin/config_resource.py
index e3294c9..06754ba 100644
--- a/kafka/admin/config_resource.py
+++ b/kafka/admin/config_resource.py
@@ -34,3 +34,9 @@ class ConfigResource(object):
         self.resource_type = resource_type
         self.name = name
         self.configs = configs
+
+    def __str__(self):
+        return "ConfigResource %s=%s" % (self.resource_type, self.name)
+
+    def __repr__(self):
+        return "ConfigResource(%s, %s, %s)" % (self.resource_type, self.name, self.configs)
diff --git a/kafka/admin/new_topic.py b/kafka/admin/new_topic.py
index 645ac38..e43c522 100644
--- a/kafka/admin/new_topic.py
+++ b/kafka/admin/new_topic.py
@@ -1,7 +1,5 @@
 from __future__ import absolute_import
 
-from kafka.errors import IllegalArgumentError
-
 
 class NewTopic(object):
     """ A class for new topic creation
@@ -16,17 +14,14 @@ class NewTopic(object):
         topic_configs (dict of str: str): A mapping of config key
             and value for the topic.
     """
-
     def __init__(
             self,
             name,
-            num_partitions,
-            replication_factor,
+            num_partitions=-1,
+            replication_factor=-1,
             replica_assignments=None,
             topic_configs=None,
     ):
-        if not (num_partitions == -1 or replication_factor == -1) ^ (replica_assignments is None):
-            raise IllegalArgumentError('either num_partitions/replication_factor or replica_assignment must be specified')
         self.name = name
         self.num_partitions = num_partitions
         self.replication_factor = replication_factor
diff --git a/kafka/cli/__init__.py b/kafka/cli/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/kafka/cli/admin/__init__.py b/kafka/cli/admin/__init__.py
new file mode 100644
index 0000000..02971e3
--- /dev/null
+++ b/kafka/cli/admin/__init__.py
@@ -0,0 +1,149 @@
+from __future__ import absolute_import
+
+import argparse
+import json
+import logging
+from pprint import pprint
+
+from kafka.admin.client import KafkaAdminClient
+from .cluster import ClusterSubCommand
+from .configs import ConfigsSubCommand
+from .consumer_groups import ConsumerGroupsSubCommand
+from .log_dirs import LogDirsSubCommand
+from .topics import TopicsSubCommand
+
+def main_parser():
+    parser = argparse.ArgumentParser(
+        prog='python -m kafka.admin',
+        description='Kafka admin client',
+    )
+    parser.add_argument(
+        '-b', '--bootstrap-servers', type=str, action='append', required=True,
+        help='host:port for cluster bootstrap servers')
+    parser.add_argument(
+        '-c', '--extra-config', type=str, action='append',
+        help='additional configuration properties for admin client')
+    parser.add_argument(
+        '-l', '--log-level', type=str,
+        help='logging level, passed to logging.basicConfig')
+    parser.add_argument(
+        '-f', '--format', type=str, default='raw',
+        help='output format: raw|json')
+    return parser
+
+
+_LOGGING_LEVELS = {'NOTSET': 0, 'DEBUG': 10, 'INFO': 20, 'WARNING': 30, 'ERROR': 40, 'CRITICAL': 50}
+
+
+def build_kwargs(props):
+    kwargs = {}
+    for prop in props or []:
+        k, v = prop.split('=')
+        try:
+            v = int(v)
+        except ValueError:
+            pass
+        if v == 'None':
+            v = None
+        elif v == 'False':
+            v = False
+        elif v == 'True':
+            v = True
+        kwargs[k] = v
+    return kwargs
+
+
+def run_cli(args=None):
+    parser = main_parser()
+    subparsers = parser.add_subparsers(help='subcommands')
+    for cmd in [ClusterSubCommand, ConfigsSubCommand, LogDirsSubCommand,
+                TopicsSubCommand, ConsumerGroupsSubCommand]:
+        cmd.add_subparser(subparsers)
+
+    config = parser.parse_args(args)
+    if config.log_level:
+        logging.basicConfig(level=_LOGGING_LEVELS[config.log_level.upper()])
+    if config.format not in ('raw', 'json'):
+        raise ValueError('Unrecognized format: %s' % config.format)
+    logger = logging.getLogger(__name__)
+
+    kwargs = build_kwargs(config.extra_config)
+    client = KafkaAdminClient(bootstrap_servers=config.bootstrap_servers, **kwargs)
+    try:
+        result = config.command(client, config)
+        if config.format == 'raw':
+            pprint(result)
+        elif config.format == 'json':
+            print(json.dumps(result))
+        return 0
+    except AttributeError:
+        parser.print_help()
+        return 2
+    except Exception:
+        logger.exception('Error!')
+        return 1
+
+if __name__ == '__main__':
+    import sys
+    sys.exit(run_cli())
+
+
+# Commands TODO:
+    # [acls]
+    # describe
+    # create
+    # delete
+
+    # [configs]
+    # alter
+    # IncrementalAlterConfigs (not supported yet)
+
+    # [partitions]
+    # create
+    # alter-reassignments (AlterPartitionReassignments - not supported yet)
+    # list-reassignments (ListPartitionReassignments - not supported yet)
+
+    # [records]
+    # delete
+
+    # [consumer-groups]
+    # remove-members (not supported yet)
+    # delete-offsets (not supported yet)
+    # alter-offsets (not supported yet)
+
+    # [offsets]
+    # list (not supported yet)
+    # delete (OffsetDelete - not supported yet)
+
+    # leader-election
+    # perform_leader_election
+
+    # [log-dirs]
+    # describe (currently broken)
+    # alter (AlterReplicaLogDirs - not supported yet)
+
+    # [client-quotas]
+    # describe (DescribeClientQuotas - not supported yet)
+    # alter (AlterClientQuotas - not supported yet)
+
+    # DescribeQuorum (not supported yet)
+
+    # [producers]
+    # describe (DescribeProducers - not supported yet)
+
+    # [transactions]
+    # describe (DescribeTransactions - not supported yet)
+    # list (ListTransactions - not supported yet)
+    # abort (not supported yet)
+
+    # [topics]
+    # describe-partitions (DescribeTopicPartitions - not supported yet)
+
+    # [cluster]
+    # describe-features (DescribeFeatures - not supported yet)
+    # update-features (UpdateFeatures - not supported yet)
+    # version
+    # api-versions
+
+
+
diff --git a/kafka/cli/admin/cluster/__init__.py b/kafka/cli/admin/cluster/__init__.py
new file mode 100644
index 0000000..7352285
--- /dev/null
+++ b/kafka/cli/admin/cluster/__init__.py
@@ -0,0 +1,16 @@
+from __future__ import absolute_import
+
+import sys
+
+from .describe import DescribeCluster
+
+
+class ClusterSubCommand:
+
+    @classmethod
+    def add_subparser(cls, subparsers):
+        parser = subparsers.add_parser('cluster', help='Manage Kafka Cluster')
+        commands = parser.add_subparsers()
+        for cmd in [DescribeCluster]:
+            cmd.add_subparser(commands)
+        parser.set_defaults(command=lambda *_args: parser.print_help() or sys.exit(2))
diff --git a/kafka/cli/admin/cluster/describe.py b/kafka/cli/admin/cluster/describe.py
new file mode 100644
index 0000000..6a2ff06
--- /dev/null
+++ b/kafka/cli/admin/cluster/describe.py
@@ -0,0 +1,9 @@
+from __future__ import absolute_import
+
+
+class DescribeCluster:
+
+    @classmethod
+    def add_subparser(cls, subparsers):
+        parser = subparsers.add_parser('describe', help='Describe Kafka Cluster')
+        parser.set_defaults(command=lambda cli, _args: cli.describe_cluster())
diff --git a/kafka/cli/admin/configs/__init__.py b/kafka/cli/admin/configs/__init__.py
new file mode 100644
index 0000000..7ec6d10
--- /dev/null
+++ b/kafka/cli/admin/configs/__init__.py
@@ -0,0 +1,16 @@
+from __future__ import absolute_import
+
+import sys
+
+from .describe import DescribeConfigs
+
+
+class ConfigsSubCommand:
+
+    @classmethod
+    def add_subparser(cls, subparsers):
+        parser = subparsers.add_parser('configs', help='Manage Kafka Configuration')
+        commands = parser.add_subparsers()
+        for cmd in [DescribeConfigs]:
+            cmd.add_subparser(commands)
+        parser.set_defaults(command=lambda *_args: parser.print_help() or sys.exit(2))
diff --git a/kafka/cli/admin/configs/describe.py b/kafka/cli/admin/configs/describe.py
new file mode 100644
index 0000000..3ff3666
--- /dev/null
+++ b/kafka/cli/admin/configs/describe.py
@@ -0,0 +1,24 @@
+from __future__ import absolute_import
+
+from kafka.admin.config_resource import ConfigResource
+
+
+class DescribeConfigs:
+
+    @classmethod
+    def add_subparser(cls, subparsers):
+        parser = subparsers.add_parser('describe', help='Describe Kafka Configs')
+        parser.add_argument('-t', '--topic', type=str, action='append', dest='topics', default=[])
+        parser.add_argument('-b', '--broker', type=str, action='append', dest='brokers', default=[])
+        parser.set_defaults(command=cls.command)
+
+    @classmethod
+    def command(cls, client, args):
+        resources = []
+        for topic in args.topics:
+            resources.append(ConfigResource('TOPIC', topic))
+        for broker in args.brokers:
+            resources.append(ConfigResource('BROKER', broker))
+
+        response = client.describe_configs(resources)
+        return list(zip([(r.resource_type.name, r.name) for r in resources], [{str(vals[0]): vals[1] for vals in r.resources[0][4]} for r in response]))
diff --git a/kafka/cli/admin/consumer_groups/__init__.py b/kafka/cli/admin/consumer_groups/__init__.py
new file mode 100644
index 0000000..cfb1bdb
--- /dev/null
+++ b/kafka/cli/admin/consumer_groups/__init__.py
@@ -0,0 +1,19 @@
+from __future__ import absolute_import
+
+import sys
+
+from .delete import DeleteConsumerGroups
+from .describe import DescribeConsumerGroups
+from .list import ListConsumerGroups
+from .list_offsets import ListConsumerGroupOffsets
+
+
+class ConsumerGroupsSubCommand:
+
+    @classmethod
+    def add_subparser(cls, subparsers):
+        parser = subparsers.add_parser('consumer-groups', help='Manage Kafka Consumer Groups')
+        commands = parser.add_subparsers()
+        for cmd in [ListConsumerGroups, DescribeConsumerGroups, ListConsumerGroupOffsets, DeleteConsumerGroups]:
+            cmd.add_subparser(commands)
+        parser.set_defaults(command=lambda *_args: parser.print_help() or sys.exit(2))
diff --git a/kafka/cli/admin/consumer_groups/delete.py b/kafka/cli/admin/consumer_groups/delete.py
new file mode 100644
index 0000000..5724ae5
--- /dev/null
+++ b/kafka/cli/admin/consumer_groups/delete.py
@@ -0,0 +1,10 @@
+from __future__ import absolute_import
+
+
+class DeleteConsumerGroups:
+
+    @classmethod
+    def add_subparser(cls, subparsers):
+        parser = subparsers.add_parser('delete', help='Delete Consumer Groups')
+        parser.add_argument('-g', '--group-id', type=str, action='append', dest='groups', required=True)
+        parser.set_defaults(command=lambda cli, args: cli.delete_consumer_groups(args.groups))
diff --git a/kafka/cli/admin/consumer_groups/describe.py b/kafka/cli/admin/consumer_groups/describe.py
new file mode 100644
index 0000000..02298e9
--- /dev/null
+++ b/kafka/cli/admin/consumer_groups/describe.py
@@ -0,0 +1,10 @@
+from __future__ import absolute_import
+
+
+class DescribeConsumerGroups:
+
+    @classmethod
+    def add_subparser(cls, subparsers):
+        parser = subparsers.add_parser('describe', help='Describe Consumer Groups')
+        parser.add_argument('-g', '--group-id', type=str, action='append', dest='groups', required=True)
+        parser.set_defaults(command=lambda cli, args: cli.describe_consumer_groups(args.groups))
diff --git a/kafka/cli/admin/consumer_groups/list.py b/kafka/cli/admin/consumer_groups/list.py
new file mode 100644
index 0000000..6c02f3b
--- /dev/null
+++ b/kafka/cli/admin/consumer_groups/list.py
@@ -0,0 +1,9 @@
+from __future__ import absolute_import
+
+
+class ListConsumerGroups:
+
+    @classmethod
+    def add_subparser(cls, subparsers):
+        parser = subparsers.add_parser('list', help='List Consumer Groups')
+        parser.set_defaults(command=lambda cli, _args: cli.list_consumer_groups())
diff --git a/kafka/cli/admin/consumer_groups/list_offsets.py b/kafka/cli/admin/consumer_groups/list_offsets.py
new file mode 100644
index 0000000..7c05c5a
--- /dev/null
+++ b/kafka/cli/admin/consumer_groups/list_offsets.py
@@ -0,0 +1,10 @@
+from __future__ import absolute_import
+
+
+class ListConsumerGroupOffsets:
+
+    @classmethod
+    def add_subparser(cls, subparsers):
+        parser = subparsers.add_parser('list-offsets', help='List Offsets for Consumer Group')
+        parser.add_argument('-g', '--group-id', type=str, required=True)
+        parser.set_defaults(command=lambda cli, args: cli.list_consumer_group_offsets(args.group_id))
diff --git a/kafka/cli/admin/log_dirs/__init__.py b/kafka/cli/admin/log_dirs/__init__.py
new file mode 100644
index 0000000..3a4a943
--- /dev/null
+++ b/kafka/cli/admin/log_dirs/__init__.py
@@ -0,0 +1,16 @@
+from __future__ import absolute_import
+
+import sys
+
+from .describe import DescribeLogDirs
+
+
+class LogDirsSubCommand:
+
+    @classmethod
+    def add_subparser(cls, subparsers):
+        parser = subparsers.add_parser('log-dirs', help='Manage Kafka Topic/Partition Log Directories')
+        commands = parser.add_subparsers()
+        for cmd in [DescribeLogDirs]:
+            cmd.add_subparser(commands)
+        parser.set_defaults(command=lambda *_args: parser.print_help() or sys.exit(2))
diff --git a/kafka/cli/admin/log_dirs/describe.py b/kafka/cli/admin/log_dirs/describe.py
new file mode 100644
index 0000000..a0c301f
--- /dev/null
+++ b/kafka/cli/admin/log_dirs/describe.py
@@ -0,0 +1,9 @@
+from __future__ import absolute_import
+
+
+class DescribeLogDirs:
+
+    @classmethod
+    def add_subparser(cls, subparsers):
+        parser = subparsers.add_parser('describe', help='Get topic log directories for brokers')
+        parser.set_defaults(command=lambda cli, _args: cli.describe_log_dirs())
diff --git a/kafka/cli/admin/topics/__init__.py b/kafka/cli/admin/topics/__init__.py
new file mode 100644
index 0000000..dbf5694
--- /dev/null
+++ b/kafka/cli/admin/topics/__init__.py
@@ -0,0 +1,19 @@
+from __future__ import absolute_import
+
+import sys
+
+from .create import CreateTopic
+from .delete import DeleteTopic
+from .describe import DescribeTopics
+from .list import ListTopics
+
+
+class TopicsSubCommand:
+
+    @classmethod
+    def add_subparser(cls, subparsers):
+        parser = subparsers.add_parser('topics', help='List/Describe/Create/Delete Kafka Topics')
+        commands = parser.add_subparsers()
+        for cmd in [ListTopics, DescribeTopics, CreateTopic, DeleteTopic]:
+            cmd.add_subparser(commands)
+        parser.set_defaults(command=lambda *_args: parser.print_help() or sys.exit(2))
diff --git a/kafka/cli/admin/topics/create.py b/kafka/cli/admin/topics/create.py
new file mode 100644
index 0000000..d033f6b
--- /dev/null
+++ b/kafka/cli/admin/topics/create.py
@@ -0,0 +1,18 @@
+from __future__ import absolute_import
+
+from kafka.admin.new_topic import NewTopic
+
+
+class CreateTopic:
+
+    @classmethod
+    def add_subparser(cls, subparsers):
+        parser = subparsers.add_parser('create', help='Create a Kafka Topic')
+        parser.add_argument('-t', '--topic', type=str, required=True)
+        parser.add_argument('--num-partitions', type=int, default=-1)
+        parser.add_argument('--replication-factor', type=int, default=-1)
+        parser.set_defaults(command=cls.command)
+
+    @classmethod
+    def command(cls, client, args):
+        return client.create_topics([NewTopic(args.topic, args.num_partitions, args.replication_factor)])
diff --git a/kafka/cli/admin/topics/delete.py b/kafka/cli/admin/topics/delete.py
new file mode 100644
index 0000000..a88400e
--- /dev/null
+++ b/kafka/cli/admin/topics/delete.py
@@ -0,0 +1,10 @@
+from __future__ import absolute_import
+
+
+class DeleteTopic:
+
+    @classmethod
+    def add_subparser(cls, subparsers):
+        parser = subparsers.add_parser('delete', help='Delete Kafka Topic')
+        parser.add_argument('-t', '--topic', type=str, required=True)
+        parser.set_defaults(command=lambda cli, args: cli.delete_topics([args.topic]))
diff --git a/kafka/cli/admin/topics/describe.py b/kafka/cli/admin/topics/describe.py
new file mode 100644
index 0000000..2e96871
--- /dev/null
+++ b/kafka/cli/admin/topics/describe.py
@@ -0,0 +1,10 @@
+from __future__ import absolute_import
+
+
+class DescribeTopics:
+
+    @classmethod
+    def add_subparser(cls, subparsers):
+        parser = subparsers.add_parser('describe', help='Describe Kafka Topics')
+        parser.add_argument('-t', '--topic', type=str, action='append', dest='topics')
+        parser.set_defaults(command=lambda cli, args: cli.describe_topics(args.topics or None))
diff --git a/kafka/cli/admin/topics/list.py b/kafka/cli/admin/topics/list.py
new file mode 100644
index 0000000..2dbf382
--- /dev/null
+++ b/kafka/cli/admin/topics/list.py
@@ -0,0 +1,9 @@
+from __future__ import absolute_import
+
+
+class ListTopics:
+
+    @classmethod
+    def add_subparser(cls, subparsers):
+        parser = subparsers.add_parser('list', help='List Kafka Topics')
+        parser.set_defaults(command=lambda cli, _args: cli.list_topics())
diff --git a/kafka/cli/consumer/__init__.py b/kafka/cli/consumer/__init__.py
new file mode 100644
index 0000000..7b19910
--- /dev/null
+++ b/kafka/cli/consumer/__init__.py
@@ -0,0 +1,90 @@
+from __future__ import absolute_import, print_function
+
+import argparse
+import logging
+
+from kafka import KafkaConsumer
+
+
+def main_parser():
+    parser = argparse.ArgumentParser(
+        prog='python -m kafka.consumer',
+        description='Kafka console consumer',
+    )
+    parser.add_argument(
+        '-b', '--bootstrap-servers', type=str, action='append', required=True,
+        help='host:port for cluster bootstrap servers')
+    parser.add_argument(
+        '-t', '--topic', type=str, action='append', dest='topics', required=True,
+        help='subscribe to topic')
+    parser.add_argument(
+        '-g', '--group', type=str, required=True,
+        help='consumer group')
+    parser.add_argument(
+        '-c', '--extra-config', type=str, action='append',
+        help='additional configuration properties for kafka consumer')
+    parser.add_argument(
+        '-l', '--log-level', type=str,
+        help='logging level, passed to logging.basicConfig')
+    parser.add_argument(
+        '-f', '--format', type=str, default='str',
+        help='output format: str|raw|full')
+    parser.add_argument(
+        '--encoding', type=str, default='utf-8', help='encoding to use for str output decode()')
+    return parser
+
+
+_LOGGING_LEVELS = {'NOTSET': 0, 'DEBUG': 10, 'INFO': 20, 'WARNING': 30, 'ERROR': 40, 'CRITICAL': 50}
+
+
+def build_kwargs(props):
+    kwargs = {}
+    for prop in props or []:
+        k, v = prop.split('=')
+        try:
+            v = int(v)
+        except ValueError:
+            pass
+        if v == 'None':
+            v = None
+        elif v == 'False':
+            v = False
+        elif v == 'True':
+            v = True
+        kwargs[k] = v
+    return kwargs
+
+
+def run_cli(args=None):
+    parser = main_parser()
+    config = parser.parse_args(args)
+    if config.log_level:
+        logging.basicConfig(level=_LOGGING_LEVELS[config.log_level.upper()])
+    if config.format not in ('str', 'raw', 'full'):
+        raise ValueError('Unrecognized format: %s' % config.format)
+    logger = logging.getLogger(__name__)
+
+    kwargs = build_kwargs(config.extra_config)
+    consumer = KafkaConsumer(bootstrap_servers=config.bootstrap_servers, group_id=config.group, **kwargs)
+    consumer.subscribe(config.topics)
+    try:
+        for m in consumer:
+            if config.format == 'str':
+                print(m.value.decode(config.encoding))
+            elif config.format == 'full':
+                print(m)
+            else:
+                print(m.value)
+    except KeyboardInterrupt:
+        logger.info('Bye!')
+        return 0
+    except Exception:
+        logger.exception('Error!')
+        return 1
+    finally:
+        consumer.close()
+
+
+if __name__ == '__main__':
+    import sys
+    sys.exit(run_cli())
diff --git a/kafka/cli/producer/__init__.py b/kafka/cli/producer/__init__.py
new file mode 100644
index 0000000..186eae1
--- /dev/null
+++ b/kafka/cli/producer/__init__.py
@@ -0,0 +1,85 @@
+from __future__ import absolute_import, print_function
+
+import argparse
+import logging
+
+from kafka import KafkaProducer
+
+
+def main_parser():
+    parser = argparse.ArgumentParser(
+        prog='python -m kafka.producer',
+        description='Kafka console producer',
+    )
+    parser.add_argument(
+        '-b', '--bootstrap-servers', type=str, action='append', required=True,
+        help='host:port for cluster bootstrap servers')
+    parser.add_argument(
+        '-t', '--topic', type=str, required=True,
+        help='publish to topic')
+    parser.add_argument(
+        '-c', '--extra-config', type=str, action='append',
+        help='additional configuration properties for kafka producer')
+    parser.add_argument(
+        '-l', '--log-level', type=str,
+        help='logging level, passed to logging.basicConfig')
+    parser.add_argument(
+        '--encoding', type=str, default='utf-8',
+        help='byte encoding for produced messages')
+    return parser
+
+
+_LOGGING_LEVELS = {'NOTSET': 0, 'DEBUG': 10, 'INFO': 20, 'WARNING': 30, 'ERROR': 40, 'CRITICAL': 50}
+
+
+def build_kwargs(props):
+    kwargs = {}
+    for prop in props or []:
+        k, v = prop.split('=')
+        try:
+            v = int(v)
+        except ValueError:
+            pass
+        if v == 'None':
+            v = None
+        elif v == 'False':
+            v = False
+        elif v == 'True':
+            v = True
+        kwargs[k] = v
+    return kwargs
+
+
+def run_cli(args=None):
+    parser = main_parser()
+    config = parser.parse_args(args)
+    if config.log_level:
+        logging.basicConfig(level=_LOGGING_LEVELS[config.log_level.upper()])
+    logger = logging.getLogger(__name__)
+
+    kwargs = build_kwargs(config.extra_config)
+    producer = KafkaProducer(bootstrap_servers=config.bootstrap_servers, **kwargs)
+
+    def log_result(res_or_err):
+        if isinstance(res_or_err, Exception):
+            logger.error("Error producing message", exc_info=res_or_err)
+        else:
+            logger.info("Message produced: %s", res_or_err)
+
+    try:
+        while True:
+            value = input()
+            producer.send(config.topic, value=value.encode(config.encoding)).add_both(log_result)
+    except KeyboardInterrupt:
+        logger.info('Bye!')
+        return 0
+    except Exception:
+        logger.exception('Error!')
+        return 1
+    finally:
+        producer.close()
+
+
+if __name__ == '__main__':
+    import sys
+    sys.exit(run_cli())
diff --git a/kafka/consumer/__main__.py b/kafka/consumer/__main__.py
new file mode 100644
index 0000000..0356a1a
--- /dev/null
+++ b/kafka/consumer/__main__.py
@@ -0,0 +1,7 @@
+from __future__ import absolute_import
+
+import sys
+
+from kafka.cli.consumer import run_cli
+
+sys.exit(run_cli())
diff --git a/kafka/producer/__main__.py b/kafka/producer/__main__.py
new file mode 100644
index 0000000..e5fd1b1
--- /dev/null
+++ b/kafka/producer/__main__.py
@@ -0,0 +1,7 @@
+from __future__ import absolute_import
+
+import sys
+
+from kafka.cli.producer import run_cli
+
+sys.exit(run_cli())


Test patch:
diff --git a/test/test_admin.py b/test/test_admin.py
index cdb7424..e6f7937 100644
--- a/test/test_admin.py
+++ b/test/test_admin.py
@@ -58,18 +58,34 @@ def test_acl_resource():
         )
 
 def test_new_topic():
-    with pytest.raises(IllegalArgumentError):
-        _bad_topic = kafka.admin.NewTopic('foo', -1, -1)
-    with pytest.raises(IllegalArgumentError):
-        _bad_topic = kafka.admin.NewTopic('foo', 1, -1)
-    with pytest.raises(IllegalArgumentError):
-        _bad_topic = kafka.admin.NewTopic('foo', 1, 1, {1: [1, 1, 1]})
+    good_topic = kafka.admin.NewTopic('foo')
+    assert good_topic.name == 'foo'
+    assert good_topic.num_partitions == -1
+    assert good_topic.replication_factor == -1
+    assert good_topic.replica_assignments == {}
+    assert good_topic.topic_configs == {}
+
+    good_topic = kafka.admin.NewTopic('foo', 1)
+    assert good_topic.name == 'foo'
+    assert good_topic.num_partitions == 1
+    assert good_topic.replication_factor == -1
+    assert good_topic.replica_assignments == {}
+    assert good_topic.topic_configs == {}
+
+    good_topic = kafka.admin.NewTopic('foo', 1, 1, {1: [1, 1, 1]})
+    assert good_topic.name == 'foo'
+    assert good_topic.num_partitions == 1
+    assert good_topic.replication_factor == 1
+    assert good_topic.replica_assignments == {1: [1, 1, 1]}
+    assert good_topic.topic_configs == {}
+
     good_topic = kafka.admin.NewTopic('foo', 1, 2)
     assert good_topic.name == 'foo'
     assert good_topic.num_partitions == 1
     assert good_topic.replication_factor == 2
     assert good_topic.replica_assignments == {}
     assert good_topic.topic_configs == {}
+
     good_topic = kafka.admin.NewTopic('bar', -1, -1, {1: [1, 2, 3]}, {'key': 'value'})
     assert good_topic.name == 'bar'
     assert good_topic.num_partitions == -1