npm install kafkajs

// kafka.config.ts
import { Kafka } from 'kafkajs';

export const kafka = new Kafka({
  clientId: 'my-app',
  brokers: ['localhost:9092'],
});

// kafka.producer.service.ts
import { Injectable, OnModuleInit } from '@nestjs/common';
import { Producer } from 'kafkajs';
import { kafka } from './kafka.config';

@Injectable()
export class KafkaProducerService implements OnModuleInit {
  private producer: Producer;

  constructor() {
    this.producer = kafka.producer();
  }

  async onModuleInit() {
    await this.producer.connect();
  }

  async sendMessage(topic: string, message: any) {
    await this.producer.send({
      topic,
      messages: [
        { 
          value: JSON.stringify(message) 
        },
      ],
    });
  }

  async onModuleDestroy() {
    await this.producer.disconnect();
  }
}

// kafka.consumer.service.ts
import { Injectable, OnModuleInit } from '@nestjs/common';
import { Consumer } from 'kafkajs';
import { kafka } from './kafka.config';

@Injectable()
export class KafkaConsumerService implements OnModuleInit {
  private consumer: Consumer;

  constructor() {
    this.consumer = kafka.consumer({ 
      groupId: 'my-consumer-group' 
    });
  }

  async onModuleInit() {
    await this.consumer.connect();
  }

  async subscribe(topic: string, callback: (message: any) => Promise<void>) {
    await this.consumer.subscribe({ topic, fromBeginning: false });
    
    await this.consumer.run({
      eachMessage: async ({ message }) => {
        const value = message.value?.toString();
        if (value) {
          await callback(JSON.parse(value));
        }
      },
    });
  }

  async onModuleDestroy() {
    await this.consumer.disconnect();
  }
}

// email.service.ts
import { Injectable, OnModuleInit } from '@nestjs/common';
import { KafkaConsumerService } from './kafka.consumer.service';

@Injectable()
export class EmailService implements OnModuleInit {
  constructor(private kafkaConsumerService: KafkaConsumerService) {}

  async onModuleInit() {
    // BullMQ의 기존 큐 이름을 Kafka 토픽으로 사용
    await this.kafkaConsumerService.subscribe('email-notification', 
      async (message) => {
        await this.processEmail(message);
      }
    );
  }

  private async processEmail(message: any) {
    // 기존 BullMQ의 프로세스 로직을 그대로 활용
    console.log('Processing email:', message);
    // 이메일 전송 로직...
  }
}

// kafka.module.ts
import { Module } from '@nestjs/common';
import { KafkaProducerService } from './kafka.producer.service';
import { KafkaConsumerService } from './kafka.consumer.service';

@Module({
  providers: [KafkaProducerService, KafkaConsumerService],
  exports: [KafkaProducerService, KafkaConsumerService],
})
export class KafkaModule {}

// message.interface.ts
interface KafkaMessage {
  id: string;
  type: string;
  data: any;
  timestamp: number;
  metadata?: {
    retry?: number;
    originalQueue?: string;
  };
}

// retry.strategy.ts
import { Injectable } from '@nestjs/common';
import { KafkaMessage } from './message.interface';

@Injectable()
export class RetryStrategy {
  async handleError(message: KafkaMessage, error: Error) {
    const maxRetries = 3;
    const retryCount = message.metadata?.retry || 0;

    if (retryCount < maxRetries) {
      // 재시도 로직
      return this.retryMessage({
        ...message,
        metadata: {
          ...message.metadata,
          retry: retryCount + 1
        }
      });
    }

    // 최대 재시도 횟수 초과 시 처리
    await this.handleFailedMessage(message, error);
  }
}

// health.controller.ts
import { Controller, Get } from '@nestjs/common';
import { HealthCheck, HealthCheckService } from '@nestjs/terminus';
import { KafkaHealthIndicator } from './kafka.health';

@Controller('health')
export class HealthController {
  constructor(
    private health: HealthCheckService,
    private kafkaHealth: KafkaHealthIndicator,
  ) {}

  @Get()
  @HealthCheck()
  check() {
    return this.health.check([
      () => this.kafkaHealth.isHealthy('kafka'),
    ]);
  }
}

// metrics.service.ts
import { Injectable } from '@nestjs/common';
import { PrometheusService } from './prometheus.service';

@Injectable()
export class MetricsService {
  private messageCounter;
  private processingTimeHistogram;

  constructor(private prometheusService: PrometheusService) {
    this.messageCounter = this.prometheusService.createCounter({
      name: 'kafka_messages_total',
      help: 'Total number of Kafka messages processed'
    });

    this.processingTimeHistogram = this.prometheusService.createHistogram({
      name: 'kafka_message_processing_time',
      help: 'Time spent processing Kafka messages'
    });
  }

  recordMessage(topic: string) {
    this.messageCounter.inc({ topic });
  }

  recordProcessingTime(topic: string, time: number) {
    this.processingTimeHistogram.observe({ topic }, time);
  }
}