# vi-kafka-producer

A configurable Kafka producer package for Node.js applications with NFL event examples and AWS MSK integration. This package provides an easy-to-use interface for sending messages to Apache Kafka with support for authentication, SSL, batching, and multiple topics.

## Features

- 🚀 Easy to use API with Promise support
- 🔧 Configurable connection settings with environment variables
- 🔐 Support for SASL authentication and SSL
- 📦 Batch message sending
- 🎯 Multiple topics support in single operation
- ⚡ High-performance message production
- 🛡️ Built-in error handling and retry logic
- 📊 Connection status monitoring
- 🎛️ Customizable producer settings
- 🏈 NFL event examples with real-world data structures
- ☁️ AWS MSK cluster integration ready

## Installation

```bash
npm install kafka-producer-js
```

## Requirements

- Node.js >= 14.0.0
- Apache Kafka cluster (local or AWS MSK)
- dotenv package for environment configuration

## Quick Start

### Basic Usage

```javascript
const { createConnectedProducer } = require('kafka-producer-js');

async function sendMessage() {
  // Configuration
  const config = {
    brokers: ['localhost:9092'],
    clientId: 'my-app'
  };

  try {
    // Create and connect producer
    const producer = await createConnectedProducer(config);

    // Send a message
    await producer.send('my-topic', {
      key: 'user-123',
      value: {
        userId: 123,
        action: 'login',
        timestamp: Date.now()
      }
    });

    // Disconnect when done
    await producer.disconnect();
    console.log('Message sent successfully!');
  } catch (error) {
    console.error('Error:', error.message);
  }
}

sendMessage();
```

### Environment Variable Configuration

Create a `.env` file in your project root:

```bash
KAFKA_BROKERS=your-broker:9092
KAFKA_CLIENT_ID=your-app
KAFKA_LOG_LEVEL=info
```

```javascript
require('dotenv').config();

const config = {
  brokers: process.env.KAFKA_BROKERS?.split(',') || ['localhost:9092'],
  clientId: process.env.KAFKA_CLIENT_ID || 'default-client',
  logLevel: process.env.KAFKA_LOG_LEVEL || 'info'
};
```

## NFL Event Examples

The package includes comprehensive examples with real NFL event data structures:

### User Connection Event
```javascript
const userEvent = {
  key: 'user-175551803607995',
  value: {
    event: 'user_connected',
    user_id: 175551803607995,
    source: 'NFL',
    timestamp: new Date().toISOString(),
    connection: {
      user_id: 175551973627120,
      source: 'NFL'
    }
  },
  headers: {
    source: 'NFL',
    version: '1.0'
  }
};
```

### Batch Events
```javascript
const batchEvents = [
  {
    key: 'user-175551803607995',
    value: {
      event: 'user_connected',
      user_id: 175551803607995,
      source: 'NFL',
      timestamp: new Date().toISOString()
    }
  },
  {
    key: 'user-175551803607996',
    value: {
      event: 'user_disconnected',
      user_id: 175551803607996,
      source: 'NFL',
      timestamp: new Date().toISOString(),
      session_duration: 1800
    }
  }
];
```

## Testing and Examples

The package includes comprehensive examples and validation scripts:

### Basic Usage Examples
Run the basic functionality examples (requires Kafka cluster):

```bash
npm run basic
# or
node examples/basic-usage.js
```

### Advanced Configuration Examples
Test advanced features and configurations:

```bash
npm run advanced
# or
node examples/advanced-usage.js
```

### Run All Examples
Execute all examples in sequence:

```bash
npm run examples
```

### Example Files
- **`examples/basic-usage.js`** - Core functionality demonstration
- **`examples/advanced-usage.js`** - Advanced configuration examples

## Configuration

### Basic Configuration

```javascript
const config = {
  brokers: ['broker1:9092', 'broker2:9092'], // Required: Array of broker addresses
  clientId: 'my-app',                        // Optional: Client identifier
  logLevel: 'warn',                          // Optional: error, warn, info, debug
  connectionTimeout: 3000,                   // Optional: Connection timeout in ms
  requestTimeout: 30000,                     // Optional: Request timeout in ms
  retry: {                                   // Optional: Retry configuration
    initialRetryTime: 100,
    retries: 8
  }
};
```

### AWS MSK Configuration

```javascript
const config = {
  brokers: [
    'b-1.your-cluster.region.amazonaws.com:9094',
    'b-2.your-cluster.region.amazonaws.com:9094'
  ],
  clientId: 'aws-msk-producer',
  logLevel: 'info'
};
```

## API Reference

### createConnectedProducer(config)

Creates a producer instance and automatically initializes and connects it.

**Parameters:**
- `config` (Object): Kafka configuration object

**Returns:** Promise<KafkaProducer>

### KafkaProducer Methods

#### send(topic, message)
Sends a single message to a topic.

#### sendBatch(topic, messages)
Sends multiple messages to a single topic.

#### sendToMultipleTopics(topicMessages)
Sends messages to multiple topics in a single operation.

#### disconnect()
Disconnects from the Kafka cluster.

#### isProducerConnected()
Checks if the producer is connected.

## Environment Variables

| Variable | Description | Default |
|----------|-------------|---------|
| `KAFKA_BROKERS` | Comma-separated broker addresses | `localhost:9092` |
| `KAFKA_CLIENT_ID` | Client identifier | `vi-kafka-producer` |
| `KAFKA_LOG_LEVEL` | Log level | `info` |
| `KAFKA_RETRY_INITIAL_TIME` | Initial retry delay (ms) | `100` |
| `KAFKA_RETRY_COUNT` | Maximum retry attempts | `3` |

## Best Practices

1. **Environment Configuration**: Use `.env` files for configuration
2. **Connection Management**: Reuse producer instances
3. **Batch Processing**: Use `sendBatch()` for multiple messages
4. **Error Handling**: Implement proper error handling and retry logic
5. **Graceful Shutdown**: Disconnect producers before application shutdown
6. **Monitoring**: Check connection status using `isProducerConnected()`

## License

MIT

## Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

## Support

If you encounter any issues or have questions, please file an issue on the GitHub repository.