SIADIK Kafka Consumer

A Kafka consumer package for SIADIK integration. This package provides event-driven architecture for handling domain events from SIADIK.

Features

• 🚀 Event-driven architecture with Kafka
• 🔌 Multi-database support (MySQL, Oracle, Progress)
• 📦 Easy initialization with CLI
• 🎯 Automatic handler generation from event names
• 🛡️ Type-safe event handling

Installation

npm install dcktrp-siadik-consumer

Quick Start

1. Initialize your project

npx dcktrp-siadik-kafka init

This will:

• Create the necessary directory structure (kafka/handlers, kafka/config, etc.)
• Ask you to select a database (MySQL, Oracle, or Progress)
• Generate .env.kafka configuration file
• Create database configuration file
• Auto-generate handler stubs from event definitions

2. Configure environment variables

Edit .env.kafka:

KAFKA_BROKERS=localhost:9092
KAFKA_CLIENT_ID=siadik-consumer
KAFKA_USERNAME=
KAFKA_PASSWORD=
KAFKA_SSL=false
KAFKA_TOPIC=siadik-events
KAFKA_GROUP_ID=your_app-siadik
KAFKA_FROM_BEGINNING=false

# Database configuration (automatically set by init)
DB_CONNECTION=mysql
DB_HOST=127.0.0.1
DB_PORT=3306
DB_DATABASE=test
DB_USERNAME=root
DB_PASSWORD=

3. Implement event handlers

Generated handlers are located in kafka/handlers/{service}/{event}.js:

// kafka/handlers/pegawai/updated.js
module.exports = async ({ event, data, meta }) => {
  console.log('[pegawai.updated] handler triggered')
  console.log('Data:', data)
  console.log('Meta:', meta)

  try {
    // Implement your business logic here
    // data: payload from Kafka message
    // meta: metadata (topic, partition, offset)
    
  } catch (err) {
    console.error('[pegawai.updated] error:', err)
    throw err
  }
}

4. Start the consumer

npm start
# or
node bin/cli.js start

Usage as Library

You can also import and use this as a library in your application:

const { dispatch } = require('dcktrp-siadik-consumer')

// Dispatch an event manually
await dispatch({
  event: 'pegawai.updated',
  data: {
    id: 1,
    nama: 'John Doe'
  },
  meta: {
    source: 'manual-trigger'
  }
})

Supported Events

Event names are defined in src/events/EventNames.js:

• lokasi_kerja.updated - Lokasi kerja updated
• pegawai.updated - Employee updated
• pegawai.mutasi - Employee mutation
• pegawai.pensiun - Employee retirement
• pegawai.meninggal-dunia - Employee deceased
• pegawai.cuti - Employee leave
• pegawai.pindah-tugas - Employee reassignment
• pegawai.plt-plh - Employee acting/temporary assignment

CLI Commands

• dcktrp-siadik-kafka init - Initialize the consumer project
• dcktrp-siadik-kafka start - Start the Kafka consumer
• dcktrp-siadik-generate - Generate handler stubs from events

Environment Variables

| Variable | Default | Description | |----------|---------|-------------| | KAFKA_BROKERS | localhost:9092 | Kafka broker addresses (comma-separated) | | KAFKA_CLIENT_ID | siadik-consumer | Kafka client ID | | KAFKA_USERNAME | - | Kafka SASL username | | KAFKA_PASSWORD | - | Kafka SASL password | | KAFKA_SSL | false | Enable SSL for Kafka | | KAFKA_TOPIC | siadik-events | Kafka topic to subscribe | | KAFKA_GROUP_ID | siadik-consumer-group | Kafka consumer group | | KAFKA_FROM_BEGINNING | true | Read from beginning of topic | | DB_CONNECTION | - | Database type (mysql, oracle, progress) |

Project Structure

your-project/
├── kafka/
│   ├── config/
│   │   └── database.js          # Database connection pool
│   └── handlers/
│       ├── lokasi_kerja/
│       │   └── updated.js
│       └── pegawai/
│           ├── updated.js
│           ├── mutasi.js
│           └── ...
├── .env.kafka                   # Kafka configuration
└── package.json

Error Handling

Handlers should throw errors on failure. The consumer will log the error and continue processing. Make sure to:

1. Log errors appropriately
2. Implement retry logic if needed
3. Handle dead-letter queues for failed messages

Database Support

MySQL

const pool = require('./kafka/config/database')
const [rows] = await pool.query('SELECT * FROM users')

Oracle

const db = require('./kafka/config/database')
const conn = await db.getConnection()
const result = await conn.execute('SELECT * FROM users')
await conn.close()

Progress

const db = require('./kafka/config/database')
const conn = await db.connect(process.env.DSN)
const result = await conn.query('SELECT * FROM users')

Performance Tips

1. Batch processing: Group similar events for database operations
2. Connection pooling: Reuse database connections
3. Error handling: Implement proper logging and monitoring
4. Offset management: Kafka automatically manages consumer offsets

Troubleshooting

Handler not found

• Ensure the handler file exists in the correct location
• Follow the naming convention: {service}/{event}.js
• Event names must match defined events

Connection refused

• Check Kafka broker is running
• Verify KAFKA_BROKERS environment variable
• Test connectivity: telnet {broker} {port}

Database connection issues

• Verify database credentials in .env.kafka
• Check database service is running
• Review kafka/config/database.js configuration

Contributing

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

License

MIT

Support

For issues and questions, please open an issue on GitHub.