Comprehensive comparison for technology in applications

const { Kafka, CompressionTypes, logLevel } = require('kafkajs');
const { v4: uuidv4 } = require('uuid');

// Production-grade Kafka order processing service
class OrderProcessingService {
  constructor() {
    this.kafka = new Kafka({
      clientId: 'order-service',
      brokers: process.env.KAFKA_BROKERS?.split(',') || ['localhost:9092'],
      logLevel: logLevel.INFO,
      retry: {
        initialRetryTime: 300,
        retries: 8
      }
    });

    this.producer = this.kafka.producer({
      allowAutoTopicCreation: false,
      transactionTimeout: 30000
    });

    this.consumer = this.kafka.consumer({
      groupId: 'order-processing-group',
      sessionTimeout: 30000,
      heartbeatInterval: 3000
    });
  }

  async initialize() {
    await this.producer.connect();
    await this.consumer.connect();
    await this.consumer.subscribe({
      topics: ['orders.created'],
      fromBeginning: false
    });
  }

  async publishOrder(orderData) {
    const orderId = uuidv4();
    const message = {
      key: orderId,
      value: JSON.stringify({
        orderId,
        customerId: orderData.customerId,
        items: orderData.items,
        totalAmount: orderData.totalAmount,
        timestamp: new Date().toISOString()
      }),
      headers: {
        'correlation-id': uuidv4(),
        'source': 'order-api'
      }
    };

    try {
      const result = await this.producer.send({
        topic: 'orders.created',
        compression: CompressionTypes.GZIP,
        messages: [message]
      });
      console.log(`Order published successfully: ${orderId}`, result);
      return { success: true, orderId };
    } catch (error) {
      console.error('Failed to publish order:', error);
      throw new Error(`Order publication failed: ${error.message}`);
    }
  }

  async processOrders() {
    await this.consumer.run({
      eachMessage: async ({ topic, partition, message }) => {
        try {
          const order = JSON.parse(message.value.toString());
          console.log(`Processing order: ${order.orderId}`);

          // Simulate order validation and processing
          await this.validateOrder(order);
          await this.processPayment(order);
          await this.updateInventory(order);

          // Publish success event
          await this.producer.send({
            topic: 'orders.processed',
            messages: [{
              key: order.orderId,
              value: JSON.stringify({
                ...order,
                status: 'processed',
                processedAt: new Date().toISOString()
              })
            }]
          });

          console.log(`Order processed successfully: ${order.orderId}`);
        } catch (error) {
          console.error('Order processing failed:', error);
          // Send to dead letter queue
          await this.handleFailedOrder(message, error);
        }
      }
    });
  }

  async validateOrder(order) {
    if (!order.customerId || !order.items || order.items.length === 0) {
      throw new Error('Invalid order data');
    }
  }

  async processPayment(order) {
    // Simulate payment processing
    return new Promise(resolve => setTimeout(resolve, 100));
  }

  async updateInventory(order) {
    // Simulate inventory update
    return new Promise(resolve => setTimeout(resolve, 50));
  }

  async handleFailedOrder(message, error) {
    await this.producer.send({
      topic: 'orders.failed',
      messages: [{
        key: message.key?.toString(),
        value: message.value,
        headers: {
          'error-message': error.message,
          'failed-at': new Date().toISOString()
        }
      }]
    });
  }

  async shutdown() {
    await this.consumer.disconnect();
    await this.producer.disconnect();
  }
}

// Usage example
const service = new OrderProcessingService();

process.on('SIGTERM', async () => {
  await service.shutdown();
  process.exit(0);
});

module.exports = OrderProcessingService;