Difference between Kafka and JMS

Producer --> [ msg3 | msg2 | msg1 ] --> Consumer A  (gets msg1)
                                   --> Consumer B  (gets msg2)
After acknowledgement, msg1 and msg2 are deleted from the queue.

Partition 0:  [0] [1] [2] [3] [4] [5] [6]  <-- producer appends here
                       ^           ^
                       |           |
           Billing group (offset 2)
                                   |
                      Analytics group (offset 5)

Both groups read the same records independently. Nothing is deleted on read.

<dependency>
    <groupId>org.apache.activemq</groupId>
    <artifactId>artemis-jakarta-client</artifactId>
    <version>2.37.0</version>
</dependency>

import jakarta.jms.ConnectionFactory;
import jakarta.jms.JMSContext;
import jakarta.jms.Queue;
import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory;

public class JmsOrderSender {

    public static void main(String[] args) {
        ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");

        // try-with-resources closes the context (connection + session) automatically
        try (JMSContext context = factory.createContext()) {
            Queue queue = context.createQueue("OrderQueue");

            context.createProducer()
                    .setProperty("priority", "HIGH")   // custom property for selectors
                    .send(queue, "New Order: 12345");

            System.out.println("Order sent to OrderQueue");
        }
    }
}

import jakarta.jms.ConnectionFactory;
import jakarta.jms.JMSConsumer;
import jakarta.jms.JMSContext;
import jakarta.jms.Queue;
import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory;

public class JmsOrderReceiver {

    public static void main(String[] args) {
        ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");

        try (JMSContext context = factory.createContext(JMSContext.AUTO_ACKNOWLEDGE)) {
            Queue queue = context.createQueue("OrderQueue");

            // Message selector: only receive messages whose "priority" is HIGH
            JMSConsumer consumer = context.createConsumer(queue, "priority = 'HIGH'");

            // Wait up to 5 seconds for a message
            String order = consumer.receiveBody(String.class, 5_000);
            if (order != null) {
                System.out.println("Received: " + order);
                // AUTO_ACKNOWLEDGE: the broker now deletes this message from the queue
            } else {
                System.out.println("No message within 5 seconds");
            }
        }
    }
}

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.8.0</version>
</dependency>

import java.util.Properties;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

public class KafkaOrderProducer {

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        // Wait until all in-sync replicas have the record
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        // Retries never create duplicates
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");

        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
            // Key "order-12345": all events of this order go to one partition, in order
            ProducerRecord<String, String> record =
                    new ProducerRecord<>("orders", "order-12345", "New Order: 12345");

            producer.send(record, (metadata, exception) -> {
                if (exception == null) {
                    System.out.printf("Sent to partition %d at offset %d%n",
                            metadata.partition(), metadata.offset());
                } else {
                    exception.printStackTrace();
                }
            });
        } // close() flushes any records still waiting to be sent
    }
}

import java.time.Duration;
import java.util.List;
import java.util.Properties;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

public class KafkaOrderConsumer {

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        // Each consumer group keeps its own offsets
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "billing-service");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        // A new group starts reading from the beginning of the topic
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        // Commit offsets ourselves, only after processing
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
            consumer.subscribe(List.of("orders"));

            while (true) {
                ConsumerRecords<String, String> records =
                        consumer.poll(Duration.ofMillis(500));
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("partition=%d offset=%d key=%s value=%s%n",
                            record.partition(), record.offset(),
                            record.key(), record.value());
                }
                // Save progress; the records themselves stay in Kafka
                consumer.commitSync();
            }
        }
    }
}

package com.javabytechie.orders.jms;

import org.springframework.jms.annotation.JmsListener;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.stereotype.Component;

@Component
public class OrderMessaging {

    private final JmsTemplate jmsTemplate;

    public OrderMessaging(JmsTemplate jmsTemplate) {
        this.jmsTemplate = jmsTemplate;
    }

    public void sendOrder(String order) {
        jmsTemplate.convertAndSend("OrderQueue", order);
    }

    @JmsListener(destination = "OrderQueue")
    public void receiveOrder(String order) {
        System.out.println("JMS received: " + order); // removed from the queue after success
    }
}

package com.javabytechie.orders.kafka;

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

@Component
public class OrderEvents {

    private final KafkaTemplate<String, String> kafkaTemplate;

    public OrderEvents(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void publishOrder(String orderId, String order) {
        kafkaTemplate.send("orders", orderId, order); // key = orderId keeps events ordered
    }

    @KafkaListener(topics = "orders", groupId = "billing-service")
    public void onOrder(String order) {
        System.out.println("Kafka received: " + order); // record stays in the topic
    }
}

spring:
  artemis:
    broker-url: tcp://localhost:61616
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      auto-offset-reset: earliest
    producer:
      acks: all

Order Service --(JMS queue: ChargeCard)--> Payment Service   (one consumer, then deleted)

Order Service --(Kafka topic: orders)--+--> Inventory Service
                                       +--> Analytics Service
                                       +--> Fraud Detection
                                       +--> Loyalty Service (added later, replays history)