From 5d0566c9ff5ed8752ad344cbff93be994f3a8f69 Mon Sep 17 00:00:00 2001
From: Ryar Nyah <ryarnyah@gmail.com>
Date: Tue, 24 Aug 2021 11:27:12 +0200
Subject: [PATCH] Add topics-configuration to kafka_info

---
 README.md                              | 20 ++++++++++++++++
 library/kafka_info.py                  | 11 +++++++--
 module_utils/kafka_manager.py          | 15 ++++++++++++
 molecule/default/tests/test_default.py | 33 ++++++++++++++++++++++++++
 4 files changed, 77 insertions(+), 2 deletions(-)

diff --git a/README.md b/README.md
index c9e590f9..e00b12cc 100644
--- a/README.md
+++ b/README.md
@@ -368,6 +368,26 @@ Playbook:
     }
 }
 ```
+#### Topics-Configuration
+Playbook:
+```yaml
+- name: get topics configuration
+  kafka_info:
+    resource: "topic-config"
+    bootstrap_servers: "{{ ansible_ssh_host }}"
+    api_version: "{{ kafka_api_version }}"
+  register: topics
+```
+`topics` will be:
+```json
+{
+    "<results_key>": {
+        "test_1600292339": {
+            "retention.ms": "66574936"
+        }
+    }
+}
+```
 #### Consumer groups
 Playbook:
 ```yaml
diff --git a/library/kafka_info.py b/library/kafka_info.py
index 6dcad8d5..44e05799 100644
--- a/library/kafka_info.py
+++ b/library/kafka_info.py
@@ -34,7 +34,8 @@
     description:
       - 'the type of resource to get information about'
     required: True
-    choices: [topic, broker, consumer_group, acl]
+    choices: [topic, broker, consumer_group, acl,
+         topic-config]
 ''' + DOCUMENTATION_COMMON
 
 EXAMPLES = '''
@@ -58,7 +59,13 @@ def main():
     module = AnsibleModule(
         argument_spec=dict(
             resource=dict(
-                choices=['topic', 'broker', 'consumer_group', 'acl'],
+                choices=[
+                    'topic',
+                    'broker',
+                    'consumer_group',
+                    'acl',
+                    'topic-config'
+                ],
                 required=True
             ),
             **module_commons
diff --git a/module_utils/kafka_manager.py b/module_utils/kafka_manager.py
index 3f576973..2e832c42 100644
--- a/module_utils/kafka_manager.py
+++ b/module_utils/kafka_manager.py
@@ -1035,6 +1035,20 @@ def get_brokers_resource(self):
             brokers[broker.nodeId] = broker._asdict()
         return brokers
 
+    def get_topics_config(self):
+        """
+Return a dict object containing information about configuration
+of topics and partitions, and following this structure:
+{
+    "test_1600378061": {
+        "config-key": "config-value"
+    }
+}
+        """
+        topics = self.get_topics()
+        return self.get_config_for_topics({
+            topic: {} for topic in topics})
+
     def get_topics_resource(self):
         """
 Return a dict object containing information about topics and partitions,
@@ -1215,6 +1229,7 @@ def get_acls_resource(self):
     def resource_to_func(self):
         return {
             'topic': self.get_topics_resource,
+            'topic-config': self.get_topics_config,
             'broker': self.get_brokers_resource,
             'consumer_group': self.get_consumer_groups_resource,
             'acl': self.get_acls_resource
diff --git a/molecule/default/tests/test_default.py b/molecule/default/tests/test_default.py
index 52892659..370c655e 100644
--- a/molecule/default/tests/test_default.py
+++ b/molecule/default/tests/test_default.py
@@ -511,6 +511,39 @@ def test_kafka_info_topic(host):
                         else partition_info['latest_offset'] >= 0)
 
 
+def test_kafka_info_topics_config(host):
+    """
+    Check if can get config on topic.
+    """
+    # Given
+    topic_name = get_topic_name()
+    test_topic_configuration = topic_defaut_configuration.copy()
+    test_topic_configuration.update({
+        'options': {
+            'retention.ms': 66574936
+        }
+    })
+    ensure_topic(
+        host,
+        test_topic_configuration,
+        topic_name
+    )
+    time.sleep(0.3)
+    # When
+    results = call_kafka_info(
+        host,
+        {
+            'resource': 'topic-config'
+        }
+    )
+    # Then
+    for r in results:
+        assert topic_name in r['ansible_module_results']
+        for name, topic_config in r['ansible_module_results'].items():
+            if name == topic_name:
+                assert int(topic_config['retention.ms']) == 66574936
+
+
 def test_kafka_info_brokers(host):
     """
     Check if can get info on brokers