Kafka Admin Client for AWS MSK (Node.js)

A Node.js Kafka admin client designed to connect to AWS Managed Streaming for Apache Kafka (MSK) and fetch topic information. This client provides a simple interface for managing Kafka topics and cluster operations.

Features

• 🔐 AWS MSK Authentication: Uses IAM credentials for secure connection to MSK clusters
• 📋 Topic Management: List, create, delete, and get detailed information about Kafka topics
• 🏢 Cluster Information: Retrieve cluster metadata and broker information
• 🚀 Easy to Use: Simple API for common Kafka admin operations
• 🔧 Configurable: Support for custom configuration options

Prerequisites

• Node.js 14.x or higher
• AWS credentials configured (via AWS CLI, environment variables, or IAM roles)
• Access to an AWS MSK cluster

Installation

1. Clone or download this project
2. Install dependencies:

npm install

3. Copy the environment configuration:

cp .env.example .env

4. Update the .env file with your AWS and Kafka configuration:

AWS_REGION=us-east-1
KAFKA_BOOTSTRAP_SERVERS=your-msk-cluster-endpoint:9092

Configuration

Environment Variables

Variable	Description	Required	Default
AWS_REGION	AWS region where your MSK cluster is located	Yes	us-east-1
KAFKA_BOOTSTRAP_SERVERS	Comma-separated list of MSK bootstrap servers	Yes	localhost:9092
AWS_ACCESS_KEY_ID	AWS access key ID	No*	-
AWS_SECRET_ACCESS_KEY	AWS secret access key	No*	-
AWS_SESSION_TOKEN	AWS session token (for temporary credentials)	No*	-

*If not provided, the SDK will use IAM roles, AWS CLI credentials, or other credential providers.

AWS Credentials Setup

You can configure AWS credentials in several ways:

1. Environment Variables (recommended for development):

   export AWS_ACCESS_KEY_ID=your_access_key
   export AWS_SECRET_ACCESS_KEY=your_secret_key
   export AWS_REGION=us-east-1

2. AWS CLI:

   aws configure

3. IAM Roles (recommended for production):

   • Attach an IAM role to your EC2 instance or ECS task
   • The SDK will automatically use the role credentials

Usage

Basic Usage

const { KafkaAdminClient } = require('./src');

async function main() {
  const kafkaAdmin = new KafkaAdminClient(
    'your-msk-cluster-endpoint:9092',
    'us-east-1'
  );
  
  try {
    await kafkaAdmin.connect();
    
    // List all topics
    const topics = await kafkaAdmin.listKafkaTopics();
    console.log('Topics:', Object.keys(topics));
    
    // Get cluster information
    const clusterInfo = await kafkaAdmin.getClusterInfo();
    console.log('Cluster ID:', clusterInfo.clusterId);
    
  } finally {
    await kafkaAdmin.disconnect();
  }
}

main().catch(console.error);

Advanced Usage

const { KafkaAdminClient } = require('./src');

async function advancedExample() {
  const kafkaAdmin = new KafkaAdminClient(
    'your-msk-cluster-endpoint:9092',
    'us-east-1',
    {
      // Additional Kafka configuration
      requestTimeout: 30000,
      retry: {
        retries: 3
      }
    }
  );
  
  try {
    await kafkaAdmin.connect();
    
    // Create a new topic
    await kafkaAdmin.createTopic('my-new-topic', {
      numPartitions: 3,
      replicationFactor: 2
    });
    
    // Get detailed information about a topic
    const topicDetails = await kafkaAdmin.getTopicDetails('my-new-topic');
    console.log('Topic details:', topicDetails);
    
    // Delete a topic
    await kafkaAdmin.deleteTopic('my-new-topic');
    
  } finally {
    await kafkaAdmin.disconnect();
  }
}

advancedExample().catch(console.error);

API Reference

KafkaAdminClient

Constructor

new KafkaAdminClient(bootstrapServers, region, optionalConfigs)

• bootstrapServers (string): Comma-separated list of MSK bootstrap servers
• region (string): AWS region where the MSK cluster is located
• optionalConfigs (object): Additional Kafka configuration options

Methods

connect()

Connects to the Kafka cluster.

await kafkaAdmin.connect();

disconnect()

Disconnects from the Kafka cluster.

await kafkaAdmin.disconnect();

listKafkaTopics()

Lists all topics in the cluster.

const topics = await kafkaAdmin.listKafkaTopics();
// Returns: { 'topic1': {}, 'topic2': {}, ... }

getTopicDetails(topicName)

Gets detailed information about a specific topic.

const details = await kafkaAdmin.getTopicDetails('my-topic');
// Returns: { name: 'my-topic', partitions: [...], isInternal: false }

getClusterInfo()

Gets cluster information including brokers and controller.

const clusterInfo = await kafkaAdmin.getClusterInfo();
// Returns: { clusterId: '...', brokers: [...], controller: {...} }

createTopic(topicName, options)

Creates a new topic.

await kafkaAdmin.createTopic('my-topic', {
  numPartitions: 3,
  replicationFactor: 2,
  configEntries: [
    { name: 'cleanup.policy', value: 'compact' }
  ]
});

deleteTopic(topicName)

Deletes a topic.

await kafkaAdmin.deleteTopic('my-topic');

Running the Application

Development Mode

npm run dev

Production Mode

npm start

Error Handling

The client includes comprehensive error handling. All methods throw descriptive errors that can be caught and handled appropriately:

try {
  const topics = await kafkaAdmin.listKafkaTopics();
} catch (error) {
  console.error('Failed to list topics:', error.message);
  // Handle the error appropriately
}

Security Considerations

• Always use IAM roles in production environments
• Ensure your AWS credentials have the minimum required permissions
• Use environment variables or secure credential stores for sensitive information
• Consider using AWS Secrets Manager for credential management

Required IAM Permissions

Your AWS credentials need the following permissions to work with MSK:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "kafka:GetBootstrapBrokers",
        "kafka:DescribeCluster"
      ],
      "Resource": "arn:aws:kafka:region:account:cluster/cluster-name/*"
    }
  ]
}

Troubleshooting

Common Issues

1. Connection Timeout: Check your network connectivity and MSK cluster endpoint
2. Authentication Failed: Verify your AWS credentials and permissions
3. Topic Not Found: Ensure the topic exists and you have proper permissions

Debug Mode

Enable debug logging by setting the environment variable:

DEBUG=kafkajs* npm start

Contributing

1. Fork the repository
2. Create a feature branch
3. Make your changes
4. Add tests if applicable
5. Submit a pull request

License

ISC License - see LICENSE file for details.

Support

For issues and questions, please create an issue in the repository or contact the development team.