Comprehensive comparison for Backend technology in applications

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

const app = express();
app.use(express.json());

// Kafka configuration
const kafka = new Kafka({
  clientId: 'order-service',
  brokers: process.env.KAFKA_BROKERS?.split(',') || ['localhost:9092'],
  retry: {
    initialRetryTime: 100,
    retries: 8
  }
});

const producer = kafka.producer({
  createPartitioner: Partitioners.LegacyPartitioner,
  idempotent: true,
  maxInFlightRequests: 5,
  transactionalId: 'order-service-producer'
});

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

let isProducerConnected = false;

// Initialize Kafka producer
async function initializeKafka() {
  try {
    await producer.connect();
    isProducerConnected = true;
    console.log('Kafka producer connected successfully');
  } catch (error) {
    console.error('Failed to connect Kafka producer:', error);
    process.exit(1);
  }
}

// Order creation endpoint
app.post('/api/orders', async (req, res) => {
  if (!isProducerConnected) {
    return res.status(503).json({ error: 'Service temporarily unavailable' });
  }

  try {
    const { userId, items, totalAmount, shippingAddress } = req.body;

    // Validate required fields
    if (!userId || !items || !totalAmount || !shippingAddress) {
      return res.status(400).json({ error: 'Missing required fields' });
    }

    const orderId = uuidv4();
    const orderEvent = {
      orderId,
      userId,
      items,
      totalAmount,
      shippingAddress,
      status: 'PENDING',
      createdAt: new Date().toISOString()
    };

    // Send order event to Kafka with retry logic
    await producer.send({
      topic: 'orders.created',
      messages: [
        {
          key: userId,
          value: JSON.stringify(orderEvent),
          headers: {
            'correlation-id': uuidv4(),
            'event-type': 'OrderCreated'
          }
        }
      ],
      compression: 1 // GZIP compression
    });

    console.log(`Order ${orderId} published to Kafka successfully`);
    res.status(201).json({ orderId, status: 'PENDING', message: 'Order received and processing' });

  } catch (error) {
    console.error('Error processing order:', error);
    res.status(500).json({ error: 'Failed to process order' });
  }
});

// Consumer for order processing
async function startOrderConsumer() {
  try {
    await consumer.connect();
    await consumer.subscribe({ topic: 'orders.created', fromBeginning: false });

    await consumer.run({
      eachMessage: async ({ topic, partition, message }) => {
        try {
          const order = JSON.parse(message.value.toString());
          const correlationId = message.headers['correlation-id']?.toString();

          console.log(`Processing order: ${order.orderId}, correlation-id: ${correlationId}`);

          // Simulate order processing logic
          await processOrder(order);

          // Publish order processed event
          await producer.send({
            topic: 'orders.processed',
            messages: [
              {
                key: order.userId,
                value: JSON.stringify({ ...order, status: 'PROCESSED', processedAt: new Date().toISOString() }),
                headers: { 'correlation-id': correlationId || uuidv4() }
              }
            ]
          });

        } catch (error) {
          console.error('Error processing message:', error);
          // Send to dead letter queue for failed messages
          await producer.send({
            topic: 'orders.dlq',
            messages: [{ value: message.value, headers: message.headers }]
          });
        }
      }
    });

    console.log('Order consumer started successfully');
  } catch (error) {
    console.error('Failed to start consumer:', error);
    process.exit(1);
  }
}

async function processOrder(order) {
  // Simulate processing time
  await new Promise(resolve => setTimeout(resolve, 1000));
  console.log(`Order ${order.orderId} processed successfully`);
}

// Graceful shutdown
process.on('SIGTERM', async () => {
  console.log('SIGTERM received, shutting down gracefully');
  await consumer.disconnect();
  await producer.disconnect();
  process.exit(0);
});

// Start the application
(async () => {
  await initializeKafka();
  await startOrderConsumer();
  app.listen(3000, () => console.log('Order service listening on port 3000'));
})();