create-kafka-topic.js

/**
 * Create Kafka Topic for OpsGuardian
 * 
 * Usage: Configure .env file with Kafka credentials, then run:
 * node create-kafka-topic.js
 */

import { Kafka } from 'kafkajs';
import dotenv from 'dotenv';

dotenv.config();

const kafka = new Kafka({
  clientId: 'opsguardian-admin',
  brokers: (process.env.KAFKA_BROKERS || 'localhost:9092').split(','),
  ssl: process.env.KAFKA_SSL === 'true' ? {
    rejectUnauthorized: false,
  } : false,
  sasl: process.env.KAFKA_SASL_USERNAME ? {
    mechanism: 'plain',
    username: process.env.KAFKA_SASL_USERNAME,
    password: process.env.KAFKA_SASL_PASSWORD,
  } : undefined,
  connectionTimeout: 10000,
  requestTimeout: 30000,
});

async function createTopic() {
  const admin = kafka.admin();
  
  try {
    console.log('🔌 Connecting to Kafka...');
    await admin.connect();
    console.log('✅ Connected to Kafka');
    
    console.log('📋 Checking existing topics...');
    const topics = await admin.listTopics();
    console.log('Existing topics:', topics);
    
    const topicName = 'system-logs';
    
    if (topics.includes(topicName)) {
      console.log(`✅ Topic "${topicName}" already exists`);
    } else {
      console.log(`📝 Creating topic "${topicName}"...`);
      await admin.createTopics({
        topics: [
          {
            topic: topicName,
            numPartitions: 3,
            replicationFactor: 2,
          },
        ],
      });
      console.log(`✅ Topic "${topicName}" created successfully`);
    }
    
    await admin.disconnect();
    console.log('👋 Disconnected from Kafka');
  } catch (error) {
    console.error('❌ Error:', error.message);
    process.exit(1);
  }
}

createTopic();