By Version

Edit this Page

Getting Started with Quarkus Messaging and Apache Kafka

In this guide, you will build two applications that exchange messages through Apache Kafka using Quarkus Messaging: a producer that sends quote requests and a processor that replies with prices.

Pré-requisitos

Para concluir este guia, você precisa:

Cerca de 15 minutos
Um IDE
JDK 17+ instalado com JAVA_HOME configurado corretamente
Apache Maven 3.9.16
Docker e Docker Compose ou Podman e Docker Compose
Opcionalmente, o Quarkus CLI se você quiser usá-lo
Opcionalmente, Mandrel ou GraalVM instalado e configurado apropriadamente se você quiser criar um executável nativo (ou Docker se você usar uma compilação de contêiner nativo)

Arquitetura

The two applications communicate via Kafka. The first application sends a quote request to Kafka and consumes Kafka messages from the quote topic. The second application receives the quote request and sends a quote back.

A primeira aplicação, o producer, permitirá que o usuário solicite algumas cotações por meio de um endpoint HTTP. Para cada solicitação de cotação, um identificador aleatório é gerado e retornado ao usuário, para marcar a solicitação de cotação como pending. Ao mesmo tempo, o id de solicitação gerado é enviado por um tópico Kafka quote-requests.

A segunda aplicação, o processor, lerá do tópico quote-requests, colocará um preço aleatório na cotação e a enviará para um tópico Kafka chamado quotes.

Por fim, o producer lerá as cotações e as enviará para o navegador usando Server-Sent Events. Assim, o usuário verá o preço da cotação atualizado de pending para o preço recebido em tempo real.

Solução

Follow the instructions below to create the applications step by step. You can also go directly to the completed example.

Clone o repositório Git: git clone https://github.com/quarkusio/quarkus-quickstarts.git, ou baixe um arquivo.

A solução está localizada no diretório kafka-quickstart .

Criando o projeto Maven

First, create two projects: the producer and the processor.

Para criar o projeto producer, em um terminal, execute:

CLI

quarkus create app org.acme:kafka-quickstart-producer \
    --extension='rest-jackson,messaging-kafka' \
    --no-code

Para criar um projeto Gradle, adicione a opção --gradle ou --gradle-kotlin-dsl.

Para obter mais informações sobre como instalar e usar a CLI do Quarkus, consulte o guia Quarkus CLI.

Maven

mvn io.quarkus.platform:quarkus-maven-plugin:3.37.2:create \
    -DprojectGroupId=org.acme \
    -DprojectArtifactId=kafka-quickstart-producer \
    -Dextensions='rest-jackson,messaging-kafka' \
    -DnoCode

Para criar um projeto Gradle, adicione a opção '-DbuildTool=gradle' ou '-DbuildTool=gradle-kotlin-dsl'.

Para usuários do Windows:

Se estiver usando cmd, (não use barra invertida '\' e coloque tudo na mesma linha)
Se estiver usando o Powershell, envolva os parâmetros '-D' entre aspas duplas, por exemplo, '"-DprojectArtifactId=kafka-quickstart-producer"'

This command creates the project structure and selects two Quarkus extensions:

Quarkus REST (anteriormente RESTEasy Reactive) e seu suporte Jackson (para manipular JSON) para servir o endpoint HTTP.
O conector Kafka para Mensageria Reativa

Para criar o projeto processor, a partir do mesmo diretório, execute:

CLI

quarkus create app org.acme:kafka-quickstart-processor \
    --extension='messaging-kafka' \
    --no-code

Para criar um projeto Gradle, adicione a opção --gradle ou --gradle-kotlin-dsl.

Para obter mais informações sobre como instalar e usar a CLI do Quarkus, consulte o guia Quarkus CLI.

Maven

mvn io.quarkus.platform:quarkus-maven-plugin:3.37.2:create \
    -DprojectGroupId=org.acme \
    -DprojectArtifactId=kafka-quickstart-processor \
    -Dextensions='messaging-kafka' \
    -DnoCode

Para criar um projeto Gradle, adicione a opção '-DbuildTool=gradle' ou '-DbuildTool=gradle-kotlin-dsl'.

Para usuários do Windows:

Se estiver usando cmd, (não use barra invertida '\' e coloque tudo na mesma linha)
Se estiver usando o Powershell, envolva os parâmetros '-D' entre aspas duplas, por exemplo, '"-DprojectArtifactId=kafka-quickstart-processor"'

Nesse ponto, você deve ter a seguinte estrutura:

.
├── kafka-quickstart-processor
│  ├── README.md
│  ├── mvnw
│  ├── mvnw.cmd
│  ├── pom.xml
│  └── src
│     └── main
│        ├── docker
│        ├── java
│        └── resources
│           └── application.properties
└── kafka-quickstart-producer
   ├── README.md
   ├── mvnw
   ├── mvnw.cmd
   ├── pom.xml
   └── src
      └── main
         ├── docker
         ├── java
         └── resources
            └── application.properties

Open the two projects in your IDE.

Serviços de Desenvolvimento

No need to start a Kafka broker in dev mode or for tests. Quarkus starts one automatically. See Dev Services for Kafka for details.

O objeto Quote

The Quote class is used in both the producer and processor projects. For simplicity, duplicate the class. In both projects, create the src/main/java/org/acme/kafka/model/Quote.java file, with the following content:

package org.acme.kafka.model;

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 +
                '}';
    }
}

JSON representation of Quote objects will be used in messages sent to the Kafka topic and also in the server-sent events sent to web browsers. Quarkus has built-in capabilities to deal with JSON Kafka messages and automatically generates the required serializers and deserializers.

Enviando pedido de cotação

Dentro do projeto producer, crie o arquivo src/main/java/org/acme/kafka/producer/QuotesResource.java e adicione o seguinte conteúdo:

package org.acme.kafka.producer;

import java.util.UUID;

import jakarta.ws.rs.POST;
import jakarta.ws.rs.Path;
import jakarta.ws.rs.Produces;
import jakarta.ws.rs.core.MediaType;

import org.eclipse.microprofile.reactive.messaging.Channel;
import org.eclipse.microprofile.reactive.messaging.Emitter;

@Path("/quotes")
public class QuotesResource {

    @Channel("quote-requests")
    Emitter<String> quoteRequestEmitter; (1)

    /**
     * Endpoint to generate a new quote request id and send it to "quote-requests" Kafka topic using the emitter.
     */
    @POST
    @Path("/request")
    @Produces(MediaType.TEXT_PLAIN)
    public String createRequest() {
        UUID uuid = UUID.randomUUID();
        quoteRequestEmitter.send(uuid.toString()); (2)
        return uuid.toString(); (3)
    }
}

1	Injete um `Emitter` de Mensageria Reativa para enviar mensagens para o canal `quote-requests`.
2	Em uma solicitação post, gere um UUID aleatório e envie-o para o tópico Kafka usando o emissor.
3	Retorne o mesmo UUID para o cliente.

The quote-requests channel is managed as a Kafka topic, as that’s the only connector on the classpath. If not indicated otherwise, like in this example, Quarkus uses the channel name as topic name. So, in this example, the application writes into the quote-requests topic. Quarkus also configures the serializer automatically, because it finds that the Emitter produces String values.

When you have multiple connectors, you need to indicate which connector to use in the application configuration.

Processando solicitações de cotação

Now consume the quote request and give out a price. Inside the processor project, create the src/main/java/org/acme/kafka/processor/QuotesProcessor.java file and add the following content:

package org.acme.kafka.processor;

import java.util.Random;

import jakarta.enterprise.context.ApplicationScoped;

import org.acme.kafka.model.Quote;
import org.eclipse.microprofile.reactive.messaging.Incoming;
import org.eclipse.microprofile.reactive.messaging.Outgoing;

import io.smallrye.reactive.messaging.annotations.Blocking;

/**
 * A bean consuming data from the "quote-requests" Kafka topic (mapped to "requests" channel) and giving out a random quote.
 * The result is pushed to the "quotes" Kafka topic.
 */
@ApplicationScoped
public class QuotesProcessor {

    private Random random = new Random();

    @Incoming("requests") (1)
    @Outgoing("quotes")   (2)
    @Blocking             (3)
    public Quote process(String quoteRequest) throws InterruptedException {
        // simulate some hard working task
        Thread.sleep(200);
        return new Quote(quoteRequest, random.nextInt(100));
    }
}

1	Indica que o método consome os itens do canal `requests`.
2	Indica que os objetos retornados pelo método são enviados para o canal `quotes`.
3	Indica que o processamento é bloqueante e não pode ser executado na thread chamadora.

For every Kafka record from the quote-requests topic, Reactive Messaging calls the process method, and sends the returned Quote object to the quotes channel. In this case, configure the channels in the application.properties file:

%dev.quarkus.http.port=8081

# Configure the incoming `quote-requests` Kafka topic
mp.messaging.incoming.requests.topic=quote-requests
mp.messaging.incoming.requests.auto.offset.reset=earliest

The configuration properties are structured as follows:

mp.messaging.[outgoing|incoming].{channel-name}.property=value

O segmento channel-name deve corresponder ao valor definido nas anotações @Incoming e @Outgoing:

quote-requests → Kafka topic from which the quote requests are read
quotes → Kafka topic to which the quotes are written

Mais detalhes sobre esta configuração estão disponíveis na seção Producer configuration e Consumer configuration da documentação do Kafka. Essas propriedades são configuradas com o prefixo kafka. Uma lista exaustiva de propriedades de configuração está disponível em Kafka Reference Guide - Configuration.

mp.messaging.incoming.requests.auto.offset.reset=earliest instructs the application to start reading the topics from the first offset, when there is no committed offset for the consumer group. In other words, it will also process messages sent before the processor application started.

Não é necessário definir serializadores ou desserializadores. O Quarkus os detecta e, se nenhum for encontrado, gera-os usando serialização JSON.

Recebendo cotações

Back to the producer project. Modify the QuotesResource to consume quotes from Kafka and send them back to the client via Server-Sent Events:

import io.smallrye.mutiny.Multi;

...

@Channel("quotes")
Multi<Quote> quotes; (1)

/**
 * Endpoint retrieving the "quotes" Kafka topic and sending the items to a server sent event.
 */
@GET
@Produces(MediaType.SERVER_SENT_EVENTS) (2)
public Multi<Quote> stream() {
    return quotes; (3)
}

1	Injecta o canal `quotes` utilizando o qualificador `@Channel`
2	Indica que o conteúdo é enviado usando `Server-Sent Events`
3	Retorna o fluxo (Reactive Stream)

Não há necessidade de configurar nada, pois o Quarkus associará automaticamente o canal quotes ao tópico quotes do Kafka. Ele também gerará um desserializador para a classe Quote.

Serialização de mensagens no Kafka

Neste exemplo, usamos Jackson para serializar/desserializar mensagens do Kafka. Para mais opções de serialização de mensagens, veja Guia de Referência do Apache Kafka - Serialização JSON.

A contract-first approach using a schema registry is strongly recommended. See the Using Apache Kafka with Schema Registry and Avro guide or the Using Apache Kafka with Schema Registry and JSON Schema guide.

A página HTML

The final piece is an HTML page that requests quotes and displays the prices received over SSE.

Dentro do projeto producer, crie o arquivo src/main/resources/META-INF/resources/quotes.html com o seguinte conteúdo:

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Prices</title>

    <link rel="stylesheet" type="text/css"
          href="https://cdnjs.cloudflare.com/ajax/libs/patternfly/3.24.0/css/patternfly.min.css">
    <link rel="stylesheet" type="text/css"
          href="https://cdnjs.cloudflare.com/ajax/libs/patternfly/3.24.0/css/patternfly-additions.min.css">
</head>
<body>
<div class="container">
    <div class="card">
        <div class="card-body">
            <h2 class="card-title">Quotes</h2>
            <button class="btn btn-info" id="request-quote">Request Quote</button>
            <div class="quotes"></div>
        </div>
    </div>
</div>
</body>
<script src="https://code.jquery.com/jquery-3.6.0.min.js"></script>
<script>
    $("#request-quote").click((event) => {
        fetch("/quotes/request", {method: "POST"})
        .then(res => res.text())
        .then(qid => {
            var row = $(`<h4 class='col-md-12' id='${qid}'>Quote # <i>${qid}</i> | <strong>Pending</strong></h4>`);
            $(".quotes").prepend(row);
        });
    });

    var source = new EventSource("/quotes");
    source.onmessage = (event) => {
      var json = JSON.parse(event.data);
      $(`#${json.id}`).html((index, html) => {
        return html.replace("Pending", `\$\xA0${json.price}`);
      });
    };
</script>
</html>

When the user clicks the button, an HTTP request is made to request a quote, and a pending quote is added to the list. On each quote received over SSE, the corresponding item in the list is updated.

Executando a aplicação

Run both applications. In one terminal, run:

mvn -f producer quarkus:dev

Em outro terminal, execute:

mvn -f processor quarkus:dev

O Quarkus inicia um broker Kafka automaticamente, configura a aplicação e compartilha a instância do broker Kafka entre diferentes aplicações. Veja Dev Services para o Kafka para mais detalhes.

Abra http://localhost:8080/quotes.html no seu navegador e solicite algumas cotações clicando no botão.

Execução na JVM ou modo nativo

Quando não estiver executando em modo de desenvolvimento ou teste, você precisará iniciar seu broker Kafka. Você pode seguir as instruções Site do Apache Kafka ou criar um arquivo docker-compose.yaml com o seguinte conteúdo:

services:

  kafka:
    image: quay.io/strimzi/kafka:latest-kafka-4.1.0
    command: [
      "sh", "-c",
      "./bin/kafka-storage.sh format --standalone -t $$(./bin/kafka-storage.sh random-uuid) -c ./config/server.properties && ./bin/kafka-server-start.sh ./config/server.properties --override advertised.listeners=$${KAFKA_ADVERTISED_LISTENERS}"
    ]
    ports:
      - "9092:9092"
    environment:
      LOG_DIR: "/tmp/logs"
      KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://kafka:9092'
    networks:
      - kafka-quickstart-network

  producer:
    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:
      - "8080:8080"
    networks:
      - kafka-quickstart-network

  processor:
    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-network

networks:
  kafka-quickstart-network:
    name: kafkaquickstart

Certifique-se de construir primeiro ambas as aplicações no modo JVM com:

mvn -f producer package
mvn -f processor package

Uma vez empacotado, execute docker-compose up.

Este é um cluster de desenvolvimento, não use em produção.

You can also build and run the applications as native executables. First, compile both applications as native:

mvn -f producer package -Dnative -Dquarkus.native.container-build=true
mvn -f processor package -Dnative -Dquarkus.native.container-build=true

Execute o sistema com:

export QUARKUS_MODE=native
docker-compose up --build

Avançando mais

This guide demonstrated how to interact with Kafka using Quarkus and SmallRye Reactive Messaging to build data streaming applications.

For the full list of features and configuration options, see the Reference guide for Apache Kafka Extension.

The Quarkus Kafka extension also supports using Kafka clients directly.

Pré-requisitos
Arquitetura
Solução
Criando o projeto Maven
O objeto Quote
Enviando pedido de cotação
Processando solicitações de cotação
Recebendo cotações
A página HTML
Executando a aplicação
Execução na JVM ou modo nativo
Avançando mais

Getting Started with Quarkus Messaging and Apache Kafka

Pré-requisitos

Arquitetura

Solução

Criando o projeto Maven

O objeto Quote

Enviando pedido de cotação

Processando solicitações de cotação

Recebendo cotações

A página HTML

Executando a aplicação

Execução na JVM ou modo nativo

Avançando mais

Related content

On the same extensions

On the same topics