Comprehensive comparison for technology in applications

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.*;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.util.*;
import java.time.Duration;

public class OrderProcessingService {
    private final KafkaProducer<String, String> producer;
    private final KafkaConsumer<String, String> consumer;
    private final ObjectMapper objectMapper;
    private static final String ORDER_TOPIC = "orders";
    private static final String NOTIFICATION_TOPIC = "order-notifications";

    public OrderProcessingService(String bootstrapServers, String groupId) {
        this.objectMapper = new ObjectMapper();
        this.producer = createProducer(bootstrapServers);
        this.consumer = createConsumer(bootstrapServers, groupId);
    }

    private KafkaProducer<String, String> createProducer(String bootstrapServers) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        props.put(ProducerConfig.RETRIES_CONFIG, 3);
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);
        return new KafkaProducer<>(props);
    }

    private KafkaConsumer<String, String> createConsumer(String bootstrapServers, String groupId) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100);
        return new KafkaConsumer<>(props);
    }

    public void publishOrder(Order order) {
        try {
            String orderJson = objectMapper.writeValueAsString(order);
            ProducerRecord<String, String> record = new ProducerRecord<>(ORDER_TOPIC, order.getOrderId(), orderJson);
            
            producer.send(record, (metadata, exception) -> {
                if (exception != null) {
                    System.err.println("Error publishing order: " + order.getOrderId() + ", Error: " + exception.getMessage());
                } else {
                    System.out.println("Order published successfully: " + order.getOrderId() + ", Partition: " + metadata.partition() + ", Offset: " + metadata.offset());
                }
            });
        } catch (Exception e) {
            System.err.println("Failed to serialize order: " + e.getMessage());
            throw new RuntimeException("Order publishing failed", e);
        }
    }

    public void processOrders() {
        consumer.subscribe(Collections.singletonList(ORDER_TOPIC));
        
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
                
                for (ConsumerRecord<String, String> record : records) {
                    try {
                        Order order = objectMapper.readValue(record.value(), Order.class);
                        processOrder(order);
                        sendNotification(order);
                        consumer.commitSync();
                    } catch (Exception e) {
                        System.err.println("Error processing order at offset " + record.offset() + ": " + e.getMessage());
                    }
                }
            }
        } finally {
            consumer.close();
            producer.close();
        }
    }

    private void processOrder(Order order) {
        System.out.println("Processing order: " + order.getOrderId() + " for customer: " + order.getCustomerId());
    }

    private void sendNotification(Order order) throws Exception {
        String notification = objectMapper.writeValueAsString(Map.of("orderId", order.getOrderId(), "status", "processed"));
        producer.send(new ProducerRecord<>(NOTIFICATION_TOPIC, order.getCustomerId(), notification));
    }

    static class Order {
        private String orderId;
        private String customerId;
        private double amount;

        public String getOrderId() { return orderId; }
        public void setOrderId(String orderId) { this.orderId = orderId; }
        public String getCustomerId() { return customerId; }
        public void setCustomerId(String customerId) { this.customerId = customerId; }
        public double getAmount() { return amount; }
        public void setAmount(double amount) { this.amount = amount; }
    }
}