python -m cli interfaces for kafka.admin, kafka.consumer, kafka.producer (#2650)

Author: dpkp

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())

kafka/admin/config_resource.py

@@ -34,3 +34,9 @@ def __init__(
         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)

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

kafka/cli/__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
+
+
+

kafka/cli/admin/__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))

kafka/cli/admin/cluster/__init__.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())

kafka/cli/admin/cluster/describe.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))

kafka/cli/admin/configs/__init__.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]))

kafka/cli/admin/configs/describe.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))