Check Kafka Topics

Check Kafka Topics

Verify Kafka topic existence and partition configuration using exact names or wildcard patterns.

Description

This skill checks whether specific Kafka topics exist and retrieves their partition counts. It supports wildcard pattern matching (e.g., agent.*) to check multiple related topics at once. Useful for verifying topic creation, checking partition configuration, and validating topic naming conventions.

Current Features

• ✅ Topic existence: Verify topics exist by exact name
• ✅ Wildcard matching: Pattern-based topic discovery (e.g., agent.*, test-*)
• ✅ Partition counts: View partition configuration (with --include-partitions)
• ✅ Bulk checking: Check multiple topics in a single call
• ✅ Total topic count: View all topics on the broker

When to Use

• Topic verification: Confirm expected topics exist after deployment
• Pattern validation: Check all topics matching a naming convention
• Partition planning: Review partition configuration for scaling
• Troubleshooting: Verify topic creation and naming

Usage

# List all topics (count only)
python3 ${CLAUDE_PLUGIN_ROOT}/skills/system-status/check-kafka-topics/execute.py

# Check specific topics
python3 ${CLAUDE_PLUGIN_ROOT}/skills/system-status/check-kafka-topics/execute.py \
  --topics onex.evt.omniclaude.agent-actions.v1,agent.routing.requested.v1

# Check topics with wildcard patterns
python3 ${CLAUDE_PLUGIN_ROOT}/skills/system-status/check-kafka-topics/execute.py \
  --topics "agent.*,test-*"

# Include partition details
python3 ${CLAUDE_PLUGIN_ROOT}/skills/system-status/check-kafka-topics/execute.py \
  --topics onex.evt.omniclaude.agent-actions.v1 --include-partitions

Arguments

• --topics: Comma-separated list of topic names or patterns
  • Supports exact names: onex.evt.omniclaude.agent-actions.v1
  • Supports wildcards: agent.*, test-*, *.v1
  • Default: None (returns only total topic count)
• --include-partitions: Include partition counts for each topic
  • Default: False (existence check only)

Output Format

Without --topics (default):

{
  "broker": "<kafka-bootstrap-servers>:9092",
  "status": "healthy",
  "total_topics": 15
}

With --topics (exact match):

{
  "broker": "<kafka-bootstrap-servers>:9092",
  "status": "healthy",
  "total_topics": 15,
  "topics": {
    "onex.evt.omniclaude.agent-actions.v1": {
      "exists": true
    },
    "nonexistent-topic": {
      "exists": false
    }
  }
}

With --topics and --include-partitions:

{
  "broker": "<kafka-bootstrap-servers>:9092",
  "status": "healthy",
  "total_topics": 15,
  "topics": {
    "onex.evt.omniclaude.agent-actions.v1": {
      "exists": true,
      "partitions": 3
    },
    "agent.routing.requested.v1": {
      "exists": true,
      "partitions": 1
    }
  }
}

With wildcard pattern (--topics "agent.*"):

{
  "broker": "<kafka-bootstrap-servers>:9092",
  "status": "healthy",
  "total_topics": 15,
  "topics": {
    "onex.evt.omniclaude.agent-actions.v1": {
      "exists": true
    },
    "agent.routing.requested.v1": {
      "exists": true
    },
    "agent.routing.completed.v1": {
      "exists": true
    }
  }
}

Unmatched wildcard pattern:

{
  "broker": "<kafka-bootstrap-servers>:9092",
  "status": "healthy",
  "total_topics": 15,
  "topics": {
    "nonexistent-*": {
      "exists": false,
      "matched": 0,
      "message": "No topics match pattern: nonexistent-*"
    }
  }
}

Stats query failure (e.g., permission denied):

{
  "broker": "<kafka-bootstrap-servers>:9092",
  "status": "healthy",
  "total_topics": 15,
  "topics": {
    "onex.evt.omniclaude.agent-actions.v1": {
      "exists": true,
      "error": "Stats query failed: Permission denied"
    }
  }
}

Exit Codes

• 0 - Kafka reachable and topics checked successfully
• 1 - Kafka unreachable or error occurred (see error field in output)

Examples

Check if agent routing topics exist:

python3 ${CLAUDE_PLUGIN_ROOT}/skills/system-status/check-kafka-topics/execute.py \
  --topics "agent.routing.requested.v1,agent.routing.completed.v1"

Find all agent-related topics with partition counts:

python3 ${CLAUDE_PLUGIN_ROOT}/skills/system-status/check-kafka-topics/execute.py \
  --topics "agent*" --include-partitions

Verify topic creation in CI/CD pipeline:

python3 ${CLAUDE_PLUGIN_ROOT}/skills/system-status/check-kafka-topics/execute.py \
  --topics "my-new-topic" | jq -e '.topics["my-new-topic"].exists'
# Exit code 0 if exists, 1 if not

Get all topics count:

python3 ${CLAUDE_PLUGIN_ROOT}/skills/system-status/check-kafka-topics/execute.py | jq '.total_topics'

Future Enhancements

• ⏳ Consumer group status: Consumer lag, group state, and member counts
• ⏳ Recent message activity: Message rate (messages/sec) and throughput (bytes/sec)
• ⏳ Topic health metrics: Replication factor, under-replicated partitions, leader distribution
• ⏳ Offset monitoring: Earliest/latest offsets per partition, total message count
• ⏳ Retention policy: TTL and size-based retention configuration
• ⏳ Consumer lag analysis: Per-consumer lag and lag trends
• ⏳ Topic configuration: Compression, cleanup policy, segment size
• ⏳ Partition rebalancing: Identify unbalanced partitions across brokers

Troubleshooting

Connection failures ("status": "unreachable"):

• Verify Kafka is running: docker ps | grep kafka
• Check bootstrap servers: echo $KAFKA_BOOTSTRAP_SERVERS
• Test connectivity: nc -zv <kafka-bootstrap-servers> 19092

Stats query failures ("error": "Stats query failed"):

• Verify kafkacat/kcat is installed: which kcat
• Check permissions: Ensure user has topic describe permissions
• Try manual query: kcat -L -b <kafka-bootstrap-servers>:9092 -t onex.evt.omniclaude.agent-actions.v1

Wildcard not matching expected topics:

• List all topics manually: kcat -L -b <kafka-bootstrap-servers>:9092
• Verify pattern syntax: Use shell glob patterns (*, ?, [])
• Check topic naming: Topics might use different delimiter (e.g., - vs .)

Performance issues with large topic lists:

• Use --include-partitions sparingly (requires additional API calls per topic)
• Filter with specific patterns instead of broad wildcards
• Consider caching results if checking frequently