Admin: list_config_resources (requires 4.1+)

kafka/admin/_configs.py

@@ -14,6 +14,7 @@
 from kafka.protocol.admin import (
     AlterConfigsRequest,
     DescribeConfigsRequest,
+    ListConfigResourcesRequest,
 )
 
 if TYPE_CHECKING:
@@ -130,6 +131,49 @@ def describe_configs(self, config_resources, include_synonyms=False, config_filt
         return self._manager.run(self._async_describe_configs, config_resources,
                                  include_synonyms, config_filter)
 
+    @staticmethod
+    def _list_config_resources_process_response(response):
+        error_type = Errors.for_code(response.error_code)
+        if error_type is not Errors.NoError:
+            raise error_type(
+                "ListConfigResourcesRequest failed with response '{}'.".format(response))
+        ret = defaultdict(list)
+        for resource in response.config_resources:
+            resource_type = ConfigResourceType(resource.resource_type)
+            ret[resource_type.name.lower()].append(resource.resource_name)
+        return dict(ret)
+
+    async def _async_list_config_resources(self, resource_types=None):
+        wire_types = []
+        for rt in resource_types or []:
+            if not isinstance(rt, ConfigResourceType):
+                try:
+                    rt = ConfigResourceType[str(rt).upper().replace('-', '_')]
+                except KeyError:
+                    raise ValueError(f'Unrecognized ConfigResourceType: {rt}')
+            wire_types.append(rt.value)
+        request = ListConfigResourcesRequest(resource_types=wire_types)
+        response = await self._manager.send(request)
+        return self._list_config_resources_process_response(response)
+
+    def list_config_resources(self, resource_types=None):
+        """List config resources known to the cluster.
+
+        Useful for discovering resource types that have no separate enumeration
+        API (e.g. ``CLIENT_METRICS``, ``GROUP``). For ``TOPIC`` and ``BROKER``
+        the data is also available via ``Metadata`` / cluster descriptions.
+
+        Keyword Arguments:
+            resource_types (list, optional): Filter by resource type. Each entry
+                may be a :class:`ConfigResourceType` or its name (e.g. ``'TOPIC'``).
+                If None or empty, the broker returns all supported types.
+                Requires broker >= 4.1 for anything other than ``CLIENT_METRICS``.
+
+        Returns:
+            dict of {resource_type (str): [resource_name (str)]}
+        """
+        return self._manager.run(self._async_list_config_resources, resource_types)
+
     async def _get_missing_dynamic_configs(self, config_resources):
         resource_lookups = [ConfigResource(resource.resource_type, resource.name) for resource in config_resources]
         dynamic_configs = await self._async_describe_configs(resource_lookups, config_filter='modified', flat=True)

kafka/cli/admin/__init__.py

@@ -113,7 +113,6 @@ def run_cli(args=None):
     # --trace ?
 
     # [configs]
-    # ListConfigResources
     # IncrementalAlterConfigs (not supported yet)
 
     # [client-quotas]

kafka/cli/admin/configs/__init__.py

@@ -2,6 +2,7 @@
 
 from .alter import AlterConfigs
 from .describe import DescribeConfigs
+from .list import ListConfigResources
 from .reset import ResetConfigs
 
 
@@ -11,6 +12,6 @@ class ConfigsSubCommand:
     def add_subparser(cls, subparsers):
         parser = subparsers.add_parser('configs', help='Manage Kafka Configuration')
         commands = parser.add_subparsers()
-        for cmd in [DescribeConfigs, AlterConfigs, ResetConfigs]:
+        for cmd in [DescribeConfigs, AlterConfigs, ListConfigResources, ResetConfigs]:
             cmd.add_subparser(commands)
         parser.set_defaults(command=lambda *_args: parser.print_help() or sys.exit(2))

kafka/cli/admin/configs/alter.py

@@ -1,40 +1,24 @@
-from kafka.admin import ConfigResource
+from .common import add_resource_arguments, parse_resources
 
 
 class AlterConfigs:
 
     @classmethod
     def add_subparser(cls, subparsers):
         parser = subparsers.add_parser('alter', help='Alter 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.add_argument('--broker-logger', type=str, action='append', dest='broker_loggers', default=[])
-        parser.add_argument('-g', '--group', type=str, action='append', dest='groups', default=[])
+        add_resource_arguments(parser)
         parser.add_argument('-c', '--config', type=str, action='append', dest='configs', required=True, help='key=value to alter')
         parser.add_argument('-v', '--validate-only', action='store_true', default=False)
         parser.add_argument('--allow-unknown', action='store_false', dest='raise_on_unknown', default=True)
         parser.set_defaults(command=cls.command)
 
     @classmethod
     def command(cls, client, args):
-        configs = dict(config.split('=') for config in args.configs)
-        resources = []
-        for topic in args.topics:
-            if resources:
-                raise ValueError('Only one resource type per request')
-            resources.append(ConfigResource('TOPIC', topic, configs))
-        for broker in args.brokers:
-            if resources:
-                raise ValueError('Only one resource type per request')
-            resources.append(ConfigResource('BROKER', broker, configs))
-        for broker in args.broker_loggers:
-            if resources:
-                raise ValueError('Only one resource type per request')
-            resources.append(ConfigResource('BROKER_LOGGER', broker, configs))
-        for group in args.groups:
-            if resources:
-                raise ValueError('Only one resource type per request')
-            resources.append(ConfigResource('GROUP', group, configs))
+        try:
+            configs = dict(config.split('=') for config in args.configs)
+        except ValueError:
+            raise ValueError(f'Unable to parse configs! {args.configs}')
+        resources = parse_resources(args, configs=configs)
         return client.alter_configs(resources,
                                     validate_only=args.validate_only,
                                     raise_on_unknown=args.raise_on_unknown)

kafka/cli/admin/configs/common.py

@@ -0,0 +1,17 @@
+from kafka.admin import ConfigResource
+
+
+def add_resource_arguments(parser):
+    """Add arguments for specifying ConfigResource"""
+    parser.add_argument(
+        '-r', '--resource-type', type=str, required=True,
+        help='Type of resource to describe: topic, broker, broker_logger, '
+             'client_metrics, group.')
+    parser.add_argument(
+        '-n', '--resource-name', type=str, action='append', dest='resource_names', required=True,
+        help='Name of resource(s) to describe. May be repeated.')
+
+
+def parse_resources(args, configs=None):
+    return [ConfigResource(args.resource_type.upper().replace('-', '_'), resource_name, configs)
+            for resource_name in args.resource_names]

kafka/cli/admin/configs/describe.py

@@ -1,16 +1,13 @@
-from kafka.admin import ConfigResource
+from .common import add_resource_arguments, parse_resources
 
 
 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.add_argument('--broker-logger', type=str, action='append', dest='broker_loggers', default=[])
-        parser.add_argument('-g', '--group', type=str, action='append', dest='groups', default=[])
-        parser.add_argument('-k', '--key', type=str, action='append', dest='keys', default=None)
+        add_resource_arguments(parser)
+        parser.add_argument('-c', '--config', type=str, action='append', dest='configs', default=None)
         parser.add_argument('--dynamic', action='store_true', default=False)
         parser.add_argument('--modified', action='store_true', default=False)
         parser.add_argument('--static', action='store_true', default=False)
@@ -19,16 +16,7 @@ def add_subparser(cls, subparsers):
 
     @classmethod
     def command(cls, client, args):
-        resources = []
-        for topic in args.topics:
-            resources.append(ConfigResource('TOPIC', topic, args.keys))
-        for broker in args.brokers:
-            resources.append(ConfigResource('BROKER', broker, args.keys))
-        for broker in args.broker_loggers:
-            resources.append(ConfigResource('BROKER_LOGGER', broker, args.keys))
-        for group in args.groups:
-            resources.append(ConfigResource('GROUP', group, args.keys))
-
+        resources = parse_resources(args, configs=args.configs)
         if args.modified:
             config_filter = 'modified'
         elif args.dynamic:

kafka/cli/admin/configs/list.py

@@ -0,0 +1,19 @@
+class ListConfigResources:
+
+    @classmethod
+    def add_subparser(cls, subparsers):
+        parser = subparsers.add_parser(
+            'list',
+            help='List config resources known to the cluster (requires broker >= 4.1 '
+                 'for non client_metrics types)')
+        parser.add_argument(
+            '-r', '--resource-type', type=str, action='append', dest='resource_types', default=[],
+            help='Filter by resource type (repeatable): topic, broker, '
+                  'broker_logger, client_metrics, group. Omit to list all '
+                  'supported types.')
+        parser.set_defaults(command=cls.command)
+
+    @classmethod
+    def command(cls, client, args):
+        return client.list_config_resources(
+            resource_types=args.resource_types or None)

kafka/cli/admin/configs/reset.py

@@ -1,31 +1,20 @@
-from kafka.admin import ConfigResource
+from .common import add_resource_arguments, parse_resources
 
 
 class ResetConfigs:
 
     @classmethod
     def add_subparser(cls, subparsers):
         parser = subparsers.add_parser('reset', help='Reset 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.add_argument('--broker-logger', type=str, action='append', dest='broker_loggers', default=[])
-        parser.add_argument('-g', '--group', type=str, action='append', dest='groups', default=[])
-        parser.add_argument('-c', '--config', type=str, action='append', dest='configs', default=[], help='key=value to reset')
+        add_resource_arguments(parser)
+        parser.add_argument('-c', '--config', type=str, action='append', dest='configs', default=[], help='key to reset')
         parser.add_argument('-v', '--validate-only', action='store_true', default=False)
         parser.add_argument('--allow-unknown', action='store_false', dest='raise_on_unknown', default=True)
         parser.set_defaults(command=cls.command)
 
     @classmethod
     def command(cls, client, args):
-        resources = []
-        for topic in args.topics:
-            resources.append(ConfigResource('TOPIC', topic, args.configs))
-        for broker in args.brokers:
-            resources.append(ConfigResource('BROKER', broker, args.configs))
-        for broker in args.broker_loggers:
-            resources.append(ConfigResource('BROKER_LOGGER', broker, args.configs))
-        for group in args.groups:
-            resources.append(ConfigResource('GROUP', group, args.configs))
+        resources = parse_resources(args, configs=args.configs)
         return client.reset_configs(resources,
                                     validate_only=args.validate_only,
                                     raise_on_unknown=args.raise_on_unknown)

kafka/protocol/admin/configs.py

@@ -7,8 +7,12 @@ class AlterConfigsResponse(ApiMessage): pass
 class DescribeConfigsRequest(ApiMessage): pass
 class DescribeConfigsResponse(ApiMessage): pass
 
+class ListConfigResourcesRequest(ApiMessage): pass
+class ListConfigResourcesResponse(ApiMessage): pass
+
 
 __all__ = [
     'AlterConfigsRequest', 'AlterConfigsResponse',
     'DescribeConfigsRequest', 'DescribeConfigsResponse',
+    'ListConfigResourcesRequest', 'ListConfigResourcesResponse',
 ]

kafka/protocol/admin/configs.pyi

@@ -5,7 +5,7 @@ from typing import Any, Self
 from kafka.protocol.api_message import ApiMessage
 from kafka.protocol.data_container import DataContainer
 
-__all__ = ['AlterConfigsRequest', 'AlterConfigsResponse', 'DescribeConfigsRequest', 'DescribeConfigsResponse']
+__all__ = ['AlterConfigsRequest', 'AlterConfigsResponse', 'DescribeConfigsRequest', 'DescribeConfigsResponse', 'ListConfigResourcesRequest', 'ListConfigResourcesResponse']
 
 class AlterConfigsRequest(ApiMessage):
     class AlterConfigsResource(DataContainer):
@@ -255,3 +255,74 @@ class DescribeConfigsResponse(ApiMessage):
     def is_request(cls) -> bool: ...
     def expect_response(self) -> bool: ...
     def with_header(self, correlation_id: int = 0, client_id: str = "kafka-python") -> None: ...
+
+class ListConfigResourcesRequest(ApiMessage):
+    resource_types: list[int]
+    def __init__(
+        self,
+        *args: Any,
+        resource_types: list[int] = ...,
+        version: int | None = None,
+        **kwargs: Any,
+    ) -> None: ...
+    @property
+    def version(self) -> int | None: ...
+    def to_dict(self, meta: bool = False, json: bool = True) -> dict: ...
+    name: str
+    type: str
+    API_KEY: int
+    API_VERSION: int
+    valid_versions: tuple[int, int]
+    min_version: int
+    max_version: int
+    @property
+    def header(self) -> Any: ...
+    @classmethod
+    def is_request(cls) -> bool: ...
+    def expect_response(self) -> bool: ...
+    def with_header(self, correlation_id: int = 0, client_id: str = "kafka-python") -> None: ...
+
+class ListConfigResourcesResponse(ApiMessage):
+    class ConfigResource(DataContainer):
+        resource_name: str
+        resource_type: int
+        def __init__(
+            self,
+            *args: Any,
+            resource_name: str = ...,
+            resource_type: int = ...,
+            version: int | None = None,
+            **kwargs: Any,
+        ) -> None: ...
+        @property
+        def version(self) -> int | None: ...
+        def to_dict(self, meta: bool = False, json: bool = True) -> dict: ...
+
+    throttle_time_ms: int
+    error_code: int
+    config_resources: list[ConfigResource]
+    def __init__(
+        self,
+        *args: Any,
+        throttle_time_ms: int = ...,
+        error_code: int = ...,
+        config_resources: list[ConfigResource] = ...,
+        version: int | None = None,
+        **kwargs: Any,
+    ) -> None: ...
+    @property
+    def version(self) -> int | None: ...
+    def to_dict(self, meta: bool = False, json: bool = True) -> dict: ...
+    name: str
+    type: str
+    API_KEY: int
+    API_VERSION: int
+    valid_versions: tuple[int, int]
+    min_version: int
+    max_version: int
+    @property
+    def header(self) -> Any: ...
+    @classmethod
+    def is_request(cls) -> bool: ...
+    def expect_response(self) -> bool: ...
+    def with_header(self, correlation_id: int = 0, client_id: str = "kafka-python") -> None: ...

kafka/protocol/schemas/resources/ListConfigResourcesRequest.json

@@ -0,0 +1,31 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements.  See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You 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.
+
+{
+  "apiKey": 74,
+  "type": "request",
+  "listeners": ["broker"],
+  "name": "ListConfigResourcesRequest",
+  // Version 0 is used as ListClientMetricsResourcesRequest which only lists client metrics resources.
+  // Version 1 adds ResourceTypes field (KIP-1142). If there is no specified ResourceTypes, it should return all configuration resources.
+  "validVersions": "0-1",
+  "flexibleVersions": "0+",
+  "fields": [
+    { "name": "ResourceTypes", "type": "[]int8", "versions": "1+",
+      "about": "The list of resource type. If the list is empty, it uses default supported config resource types."
+    }
+  ]
+}
+