Kafka quickstart review

Author: Ladicek

kafka-quickstart/README.md

@@ -20,9 +20,9 @@ and in another terminal:
 mvn -f processor quarkus:dev
 ```
 
-_NOTE_: Quarkus Dev Services starts a Kafka broker for you automatically. 
+_NOTE_: Quarkus Dev Services starts a Kafka broker for you automatically.
 
-Then, open your browser to `http://localhost:8080/quotes.html`.
+Then, open your browser at `http://localhost:8080/quotes.html`.
 You can send quote requests and observe received quotes.
 
 ## Anatomy
@@ -31,7 +31,7 @@ The application is composed of the following components:
 
 #### Producer
 
-The _producer_ application receive requests from the user (via HTTP) and sends data to the Kafka broker.
+The _producer_ application receive requests from the user (via HTTP) and sends _quote requests_ to the Kafka broker.
 Two main components compose the application:
 
 * `QuoteProducer` generates uniquely identified quote requests and sends them to the Kafka topic `quote-requests`.
@@ -40,26 +40,25 @@ It also consumes the Kafka topic `quotes` and relays received messages to the br
 
 #### Processor
 
-The _processor_ application receives data from Kafka, processes them, and writes the result into the `quotes` Kafka topic.
+The _processor_ application receives quote requests from Kafka, processes them, and writes results into the `quotes` Kafka topic.
+The application has one main class:
 
-Two main classes compose the application:
-
-* `QuoteProcessor` consumes quote request id's from the `quote-requests` Kafka topic and responds back to the `quotes` topic with a `Quote` object containing a random price.
+* `QuoteProcessor` consumes quote request ids from the `quote-requests` Kafka topic and responds back to the `quotes` topic with a `Quote` object containing a random price.
 
 The connection to Kafka is configured in the `src/main/resources/application.properties` file.
 
 ## Running the application in Docker
 
-To run the application on Docker:
+To run the application in Docker, first make sure that both services are built:
 
-First make sure that both services are packaged:
 ```bash
 mvn package
 ```
 
-Then launch the Docker compose:
+Then launch Docker Compose:
+
 ```bash
-docker compose up
+docker-compose up
 ```
 
 This will create a single-node Kafka cluster and launch both applications.
@@ -69,10 +68,19 @@ This will create a single-node Kafka cluster and launch both applications.
 You can compile the application into a native binary using:
 
 ```bash
-./mvn package -Pnative
+mvn package -Dnative
+```
+
+As you are running in _prod_ mode, you need a Kafka cluster.
+
+If you have Docker installed, you can simply run:
+
+```bash
+export QUARKUS_MODE=native
+docker-compose up --build
 ```
 
-As you are running in _prod_ mode, you need a Kafka cluster. You can follow the instructions from the [Apache Kafka web site](https://kafka.apache.org/quickstart) or run `docker-compose up` if you have docker installed on your machine.
+Alternatively, you can follow the instructions from the [Apache Kafka web site](https://kafka.apache.org/quickstart).
 
 Then run both applications respectively with:
 

kafka-quickstart/docker-compose.yaml

@@ -34,10 +34,12 @@ services:
       - kafka-quickstart-network
 
   producer:
-    image: quarkus-quickstarts/kafka-quickstart-producer:1.0
+    image: quarkus-quickstarts/kafka-quickstart-producer:1.0-${QUARKUS_MODE:-jvm}
     build:
       context: producer
       dockerfile: src/main/docker/Dockerfile.${QUARKUS_MODE:-jvm}
+    depends_on:
+      - kafka
     environment:
       KAFKA_BOOTSTRAP_SERVERS: kafka:9092
     ports:
@@ -46,10 +48,12 @@ services:
       - kafka-quickstart-network
 
   processor:
-    image: quarkus-quickstarts/kafka-quickstart-processor:1.0
+    image: quarkus-quickstarts/kafka-quickstart-processor:1.0-${QUARKUS_MODE:-jvm}
     build:
       context: processor
       dockerfile: src/main/docker/Dockerfile.${QUARKUS_MODE:-jvm}
+    depends_on:
+      - kafka
     environment:
       KAFKA_BOOTSTRAP_SERVERS: kafka:9092
     networks:

kafka-quickstart/processor/src/main/java/org/acme/kafka/model/Quote.java

@@ -6,12 +6,20 @@ public class Quote {
     public int price;
 
     /**
-     * Default constructor needed for Jackson serialization
+     * Default constructor required for Jackson serializer
      */
     public Quote() { }
 
     public Quote(String id, int price) {
         this.id = id;
         this.price = price;
     }
-}
+
+    @Override
+    public String toString() {
+        return "Quote{" +
+                "id='" + id + '\'' +
+                ", price=" + price +
+                '}';
+    }
+}

kafka-quickstart/processor/src/main/resources/application.properties

@@ -1,10 +1,10 @@
 %dev.quarkus.http.port=8081
 
-# Configure the incoming Kafka topic quote-requests (we read from it)
+# Configure the incoming `quote-requests` Kafka topic
 mp.messaging.incoming.requests.connector=smallrye-kafka
 mp.messaging.incoming.requests.topic=quote-requests
 mp.messaging.incoming.requests.auto.offset.reset=earliest
 
-# Configure the outgoing Kafka topic quotes (we write to it)
+# Configure the outgoing `quotes` Kafka topic
 mp.messaging.outgoing.quotes.connector=smallrye-kafka
 mp.messaging.outgoing.quotes.value.serializer=io.quarkus.kafka.client.serialization.ObjectMapperSerializer

kafka-quickstart/processor/src/test/java/org/acme/kafka/processor/QuoteProcessorTest.java

@@ -18,6 +18,7 @@
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.serialization.StringDeserializer;
 import org.apache.kafka.common.serialization.StringSerializer;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
@@ -36,12 +37,17 @@ public class QuoteProcessorTest {
     KafkaConsumer<String, Quote> quoteConsumer;
 
     @BeforeEach
-    void beforeAll() {
-        System.out.println(kafkaConfig);
+    void setUp() {
         quoteConsumer = new KafkaConsumer<>(consumerConfig(), new StringDeserializer(), new ObjectMapperDeserializer<>(Quote.class));
         quoteRequestProducer = new KafkaProducer<>(kafkaConfig, new StringSerializer(), new StringSerializer());
     }
 
+    @AfterEach
+    void tearDown() {
+        quoteRequestProducer.close();
+        quoteConsumer.close();
+    }
+
     Properties consumerConfig() {
         Properties properties = new Properties();
         properties.putAll(kafkaConfig);

kafka-quickstart/producer/src/main/java/org/acme/kafka/model/Quote.java

@@ -5,11 +5,21 @@ public class Quote {
     public String id;
     public int price;
 
+    /**
+     * Default constructor required for Jackson serializer
+     */
+    public Quote() { }
+
+    public Quote(String id, int price) {
+        this.id = id;
+        this.price = price;
+    }
+
     @Override
     public String toString() {
         return "Quote{" +
                 "id='" + id + '\'' +
                 ", price=" + price +
                 '}';
     }
-}
+}

kafka-quickstart/producer/src/main/java/org/acme/kafka/model/QuoteDeserializer.java

@@ -4,7 +4,6 @@
 
 public class QuoteDeserializer extends ObjectMapperDeserializer<Quote> {
     public QuoteDeserializer() {
-        // pass the class to the parent.
         super(Quote.class);
     }
 }

kafka-quickstart/producer/src/main/resources/META-INF/resources/quotes.html

@@ -27,14 +27,14 @@ <h2 class="card-title">Quotes</h2>
         .then(res => res.text())
         .then(qid => {
             var row = $(`<h4 class='col-md-12' id='${qid}'>Quote # <i>${qid}</i> | <strong>Pending</strong></h4>`);
-            $(".quotes").append(row);
+            $(".quotes").prepend(row);
         });
     });
 
     var source = new EventSource("/quotes");
     source.onmessage = (event) => {
       var json = JSON.parse(event.data);
-      $(`#${json.id}`).html(function(index, html) {
+      $(`#${json.id}`).html((index, html) => {
         return html.replace("Pending", `\$\xA0${json.price}`);
       });
     };