[Integration][Kafka] Add consumer groups

integrations/kafka/.port/resources/blueprints.json

@@ -92,5 +92,102 @@
         "many": true
       }
     }
+  },
+  {
+    "identifier": "kafkaConsumerGroup",
+    "title": "Consumer Group",
+    "icon": "Kafka",
+    "schema": {
+      "properties": {
+        "state": {
+          "title": "State",
+          "type": "string",
+          "description": "The current state of the consumer group."
+        },
+        "members": {
+          "title": "Members",
+          "type": "array",
+          "description": "List of members in the consumer group.",
+          "items": {
+            "type": "object",
+            "properties": {
+              "id": {
+                "title": "Member ID",
+                "type": "string",
+                "description": "Unique identifier for the consumer member."
+              },
+              "client_id": {
+                "title": "Client ID",
+                "type": "string",
+                "description": "Client ID of the consumer member."
+              },
+              "host": {
+                "title": "Host",
+                "type": "string",
+                "description": "Host address of the consumer member."
+              },
+              "assignment": {
+                "title": "Assignment",
+                "type": "object",
+                "description": "Details of the topic partitions assigned to the member.",
+                "properties": {
+                  "topic_partitions": {
+                    "title": "Topic Partitions",
+                    "type": "array",
+                    "description": "List of topic-partition pairs assigned to the member.",
+                    "items": {
+                      "type": "object",
+                      "properties": {
+                        "topic": {
+                          "title": "Topic",
+                          "type": "string",
+                          "description": "Name of the topic."
+                        },
+                        "partition": {
+                          "title": "Partition",
+                          "type": "number",
+                          "description": "Partition number within the topic."
+                        }
+                      }
+                    }
+                  }
+                }
+              }
+            }
+          }
+        },
+        "coordinator": {
+          "title": "Coordinator",
+          "type": "number",
+          "description": "Broker ID of the coordinator for the consumer group."
+        },
+        "partition_assignor": {
+          "title": "Partition Assignor",
+          "type": "string",
+          "description": "Strategy used to assign partitions to consumers."
+        },
+        "is_simple_consumer_group": {
+          "title": "Is Simple Consumer Group",
+          "type": "boolean",
+          "description": "Indicates if the group is a simple consumer group."
+        },
+        "authorized_operations": {
+          "title": "Authorized Operations",
+          "type": "array",
+          "description": "List of operations authorized for the consumer group.",
+          "items": {
+            "type": "string"
+          }
+        }
+      }
+    },
+    "calculationProperties": {},
+    "relations": {
+      "cluster": {
+        "target": "kafkaCluster",
+        "required": true,
+        "many": false
+      }
+    }
   }
 ]

integrations/kafka/.port/resources/port-app-config.yaml

@@ -46,3 +46,21 @@ resources:
           relations:
             cluster: .cluster_name
             brokers: '[.cluster_name + "_" + (.partitions[].replicas[] | tostring)] | unique'
+  - kind: consumer_group
+    selector:
+      query: 'true'
+    port:
+      entity:
+        mappings:
+          identifier: .cluster_name + "_" + .group_id
+          title: .group_id
+          blueprint: '"kafkaConsumerGroup"'
+          properties:
+            state: .state
+            members: .members
+            coordinator: .coordinator.id
+            partition_assignor: .partition_assignor
+            is_simple_consumer_group: .is_simple_consumer_group
+            authorized_operations: .authorized_operations
+          relations:
+            cluster: .cluster_name

integrations/kafka/.port/spec.yaml

@@ -9,6 +9,7 @@ features:
       - kind: cluster
       - kind: broker
       - kind: topic
+      - kind: consumer_group
 saas:
   enabled: false
 configurations:

integrations/kafka/examples/consumer_groups.entity.json

@@ -0,0 +1,36 @@
+{
+    "blueprint": "kafkaConsumerGroup",
+    "identifier": "local-cluster_conduktor_gateway",
+    "createdAt": "2025-01-17T14:23:38.182Z",
+    "updatedBy": "<port-client-id>",
+    "createdBy": "<port-client-id>",
+    "icon": null,
+    "team": [],
+    "title": "conduktor_gateway",
+    "relations": {
+        "cluster": "local-cluster"
+    },
+    "properties": {
+        "partition_assignor": "range",
+        "coordinator": null,
+        "authorized_operations": null,
+        "is_simple_consumer_group": false,
+        "members": [
+            {
+                "assignment": {
+                    "topic_partitions": [
+                        {
+                            "partition": 0,
+                            "topic": "_conduktor_gateway_license"
+                        }
+                    ]
+                },
+                "host": "172.23.0.6",
+                "id": "conduktor-gateway_6969-707254dd-d7d6-4e34-9ac9-d868d5fd5f56",
+                "client_id": "conduktor-gateway_6969"
+            }
+        ],
+        "state": "STABLE"
+    },
+    "updatedAt": "2025-01-17T14:23:38.182Z"
+}

integrations/kafka/examples/consumer_groups.response.json

@@ -0,0 +1,24 @@
+{
+    "group_id": "_confluent-ksql-default_query_CTAS_PURCHASE_PER_PRODUCT_0",
+    "state": "EMPTY",
+    "members": [
+        {
+            "assignment": {
+                "topic_partitions": [
+                    {
+                        "partition": 0,
+                        "topic": "_conduktor_gateway_license"
+                    }
+                ]
+            },
+            "host": "172.23.0.6",
+            "id": "conduktor-gateway_6969-707254dd-d7d6-4e34-9ac9-d868d5fd5f56",
+            "client_id": "conduktor-gateway_6969"
+        }
+    ],
+    "cluster_name": "local-cluster",
+    "coordinator": 0,
+    "partition_assignor": "range",
+    "is_simple_consumer_group": false,
+    "authorized_operations": null
+}

integrations/kafka/kafka_integration/client.py

@@ -86,3 +86,55 @@ def describe_topics(self) -> list[dict[str, Any]]:
                 logger.error(f"Failed to describe topic {topic_name}: {e}")
                 raise e
         return result_topics
+
+    def describe_consumer_groups(self) -> list[dict[str, Any]]:
+        """Describe all consumer groups in the cluster."""
+        result_groups: list[dict[str, Any]] = []
+
+        # List all consumer groups and wait for the future to complete
+        groups_metadata = self.kafka_admin_client.list_consumer_groups()
+        groups_result = groups_metadata.result()
+        group_ids = [group.group_id for group in groups_result.valid]
+
+        logger.info(f"Found {len(group_ids)} consumer groups")
+        if not group_ids:
+            return result_groups
+
+        # Describe the consumer groups
+        groups_description = self.kafka_admin_client.describe_consumer_groups(group_ids)
+
+        for group_id, future in groups_description.items():
+            try:
+                group_info = future.result()
+                members = [
+                    {
+                        "id": member.member_id,
+                        "client_id": member.client_id,
+                        "host": member.host,
+                        "assignment": {
+                            "topic_partitions": [
+                                {"topic": tp.topic, "partition": tp.partition}
+                                for tp in member.assignment.topic_partitions
+                            ]
+                        },
+                    }
+                    for member in group_info.members
+                ]
+
+                result_groups.append(
+                    {
+                        "group_id": group_id,
+                        "state": group_info.state.name,
+                        "members": members,
+                        "cluster_name": self.cluster_name,
+                        "coordinator": group_info.coordinator.id,
+                        "partition_assignor": group_info.partition_assignor,
+                        "is_simple_consumer_group": group_info.is_simple_consumer_group,
+                        "authorized_operations": group_info.authorized_operations,
+                    }
+                )
+            except Exception as e:
+                logger.error(f"Failed to describe consumer group {group_id}: {e}")
+                raise e
+
+        return result_groups

integrations/kafka/main.py

@@ -33,3 +33,10 @@ async def resync_topics(kind: str) -> ASYNC_GENERATOR_RESYNC_TYPE:
     kafka_clients = init_clients()
     for kafka_client in kafka_clients:
         yield kafka_client.describe_topics()
+
+
+@ocean.on_resync("consumer_group")
+async def resync_consumer_groups(kind: str) -> ASYNC_GENERATOR_RESYNC_TYPE:
+    kafka_clients = init_clients()
+    for kafka_client in kafka_clients:
+        yield kafka_client.describe_consumer_groups()