anchor.diff:  
   == Creating the Maven Project
   
  -First, we need a new project.
  -Create a new project with the following command:
  +First, we need to create two projects.
  +Create the `producer` project with the following command:
   
   [source,bash,subs=attributes+]
   ----
   mvn io.quarkus:quarkus-maven-plugin:{quarkus-version}:create \
       -DprojectGroupId=org.acme \
  -    -DprojectArtifactId=kafka-quickstart \
  -    -DclassName="org.acme.kafka.PriceResource" \
  -    -Dpath="/prices" \
  +    -DprojectArtifactId=kafka-quickstart-producer \
  +    -DclassName="org.acme.kafka.producer.QuotesResource" \
  +    -Dpath="/quotes" \
       -Dextensions="resteasy-reactive,resteasy-reactive-jackson,smallrye-reactive-messaging-kafka"
  -cd kafka-quickstart
   ----
   
   This command generates a Maven project, importing the Reactive Messaging and Kafka connector extensions.
  
positive.code: 
  == Get it running
  
  You just need to run the application using:
  
  [source,bash]
  ----
  ./mvnw quarkus:dev
  ----
  
  Open `http://localhost:8080/prices.html` in your browser.
  
  == Running in JVM or Native mode
  
  
negative.code: See xref:kafka-dev-services.adoc[Dev Services for Kafka] for details.
  ====
  
  == The Quote object
  
  The quote class will be used in both `producer` and `processor` projects.
  For the sake of simplicity we will duplicate the class.
  In both projects, create the `src/main/java/org/acme/kafka/model/Quote.java` file, with the following content:
  
  [source,java]
  ----
  package org.acme.kafka.model;
  
  public class Quote {
  
      public String id;
      public int price;
  
      public Quote() { }
  
      public Quote(String id, int price) {
          this.id = id;
          this.price = price;
      }
  }
  ----
  
  JSON representation of `Quote` objects will be used in messages sent to the Kafka topic
  and also server-sent events sent to browser clients.
  
  Quarkus has built-in capabilities to deal with JSON Kafka messages.
  In a following section we will create serializer/deserializer classes for Jackson.
  
  == Sending quote request
  
  Inside the `producer` project locate the generated  `src/main/java/org/acme/kafka/producer/QuotesResource.java` file, and add the following content:
  
  [source,java]
  ----
  package org.acme.kafka.producer;
  
  import java.util.UUID;
  
  import javax.inject.Inject;
  import javax.ws.rs.GET;
  import javax.ws.rs.POST;
  import javax.ws.rs.Path;
  import javax.ws.rs.Produces;
  import javax.ws.rs.core.MediaType;
  
  import org.acme.kafka.model.Quote;
  import org.eclipse.microprofile.reactive.messaging.Channel;
  import org.eclipse.microprofile.reactive.messaging.Emitter;
  
  import io.smallrye.mutiny.Multi;
  
  @Path("/quotes")
  public class QuotesResource {
  
      @Inject
      @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();
      }
  }
  ----
  <1> Inject a Reactive Messaging `Emitter` to send messsages to the `quote-requests` channel.
  <2> On a post request, generate a random UUID and send it to the Kafka topic using the emitter.
  
  This channel is mapped to Kafka using the configuration we will add to the `application.properties` file:
  
  [source, properties]
  ----
  # Configure the outgoing Kafka topic quote-requests
  mp.messaging.outgoing.quote-requests.connector=smallrye-kafka
  ----
  
  All we need to specify is the `smallrye-kafka` connector.
  By default reactive messaging will map the channel name `quote-requests` to the Kafka topic name.
  The serializer for `String` is automatically configured.
  
  == Processing quote requests
  
  Now let's consume the quote request and give out a price.
  Inside the `processor` project locate the `src/main/java/org/acme/kafka/processor/QuoteProcessor.java` file and add the following:
  
  [source, java]
  ----
  package org.acme.kafka.processor;
  
  import java.util.Random;
  
  import javax.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 "request" Kafka topic 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> Indicates that the method consumes the items from the `requests` channel
  <2> Indicates that the objects returned by the method are sent to the `quotes` channel
  <3> Indicates that the processing is _blocking_ and cannot be run on the caller thread.
  
  The `process` method is called for every Kafka _record_ from the `quote-requests` topic, and will send a `Quote` object to the `quotes` topic.
  
  As with the previous example we need to configure the connectors in the `application.properties` file:
  
  [source, properties]
  ----
  %dev.quarkus.http.port=8081
  
  # Configure the incoming Kafka topic `quote-requests`
  mp.messaging.incoming.requests.connector=smallrye-kafka
  mp.messaging.incoming.requests.topic=quote-requests
  mp.messaging.incoming.requests.auto.offset.reset=earliest
  mp.messaging.incoming.requests.health-readiness-enabled=false
  
  # Configure the outgoing Kafka topic `quotes`
  mp.messaging.outgoing.quotes.connector=smallrye-kafka
  mp.messaging.outgoing.quotes.value.serializer=io.quarkus.kafka.client.serialization.ObjectMapperSerializer
  ----
  
  Note that in this case we have one incoming and one outgoing connector configuration, each one distinctly named.
  The configuration keys are structured as follows:
  
  `mp.messaging.[outgoing|incoming].{channel-name}.property=value`
  
  The `channel-name` segment must match the value set in the `@Incoming` and `@Outgoing` annotation:
  
  * `quote-requests` -> Kafka topic from which we read the quote requests
  * `quotes` -> Kafka topic in which we write the quotes
  
  [NOTE]
  ====
  More details about this configuration is available on the https://kafka.apache.org/documentation/#producerconfigs[Producer configuration] and https://kafka.apache.org/documentation/#consumerconfigs[Consumer configuration] section from the Kafka documentation. These properties are configured with the prefix `kafka`. An exhaustive list of configuration properties is available in <<kafka-configuration>>.
  ====
  
  Also for the outgoing configuration we specified the serializer because we are sending a `Quote` object as the message payload.
  
  Quarkus provides default implementations for Kafka serializer/deserializer pairs using Jackson `ObjectMapper`.
  `ObjectMapperSerializer` can be used to serialize all objects via Jackson.
  
  == Receiving quotes
  
  Back to our `producer` project.
  Let's modify the `QuotesResource` to consume quotes, bind it to a JAX-RS resource and send events to clients:
  
  [source,java]
  ----
  import io.smallrye.mutiny.Multi;
  [...]
  
  @Inject
  @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> Injects the `quotes` channel using the `@Channel` qualifier
  <2> Indicates that the content is sent using `Server Sent Events`
  <3> Returns the stream (_Reactive Stream_)
  
  Again we need to configure the incoming `quotes` channel inside `producer` project.
  Add the following inside `application.properties` file:
  
  [source, properties]
  ----
  # Configure the outgoing Kafka topic quote-requests
  mp.messaging.outgoing.quote-requests.connector=smallrye-kafka
  
  # Configure the incoming Kafka topic quotes
  mp.messaging.incoming.quotes.connector=smallrye-kafka
  mp.messaging.incoming.quotes.value.deserializer=org.acme.kafka.model.QuoteDeserializer
  mp.messaging.incoming.quotes.health-readiness-enabled=false
  ----
  
  == JSON serialization via Jackson
  
  Finally, we will configure JSON serialization for messages using Jackson.
  Previously we've seen the usage of `ObjectMapperSerializer` to serialize objects via Jackson.
  For the corresponding deserializer class, we need to create a subclass of `ObjectMapperDeserializer`.
  
  You may have noticed the deserializer configuration on the previous step specifying `org.acme.kafka.model.QuoteDeserializer` class.
  So, let's create it inside `producer` project on `src/main/java/org/acme/kafka/model/QuoteDeserializer.java`
  
  [source,java]
  ----
  package org.acme.kafka.model;
  
  import io.quarkus.kafka.client.serialization.ObjectMapperDeserializer;
  
  public class QuoteDeserializer extends ObjectMapperDeserializer<Quote> {
      public QuoteDeserializer() {
          // pass the class to the parent.
          super(Quote.class);
      }
  }
  ----
  