anchor.diff:  [[MicroProfile_Reactive_Messaging_SmallRye]]
   = MicroProfile Reactive Messaging Subsystem Configuration
   ----
  -:smallrye-reactive-messaging-version:    3.4
  -:eclipse-mp-reactive-messaging-api-version:    2.0
  +:smallrye-reactive-messaging-version:       3.6
  +:smallrye-reactive-messaging-tag:           {smallrye-reactive-messaging-version}.0
  +:eclipse-mp-reactive-messaging-api-version: 2.0
   ----
   
   
  
positive.code: 
  ==== Kafka Connector
  
  An example of a minimal microprofile-config.properties file for Kafka for the example application shown previously:
  
  ```
  kafka.bootstrap.servers=kafka:9092
  
negative.code: In addition to the above, Apache Kafka, and SmallRye Reactive Messaging's Kafka connector understand a lot more properties. These can be found in the SmallRye Reactive Messaging Kafka connector https://smallrye.io/smallrye-reactive-messaging/smallrye-reactive-messaging/{smallrye-reactive-messaging-version}/kafka/kafka.html[documentation], and in the Apache Kafka documentation for the https://kafka.apache.org/documentation/#producerconfigs[producers] and the https://kafka.apache.org/documentation/#consumerconfigs[consumers].
  
  The prefixes discussed above are stripped off before passing the property to Kafka.
  
  ===== Kafka User API
  In order to be able to get more information about messages received from Kafka, and to be able to influence how Kafka handles messages, there is a user API for Kafka. This API lives in the https://github.com/smallrye/smallrye-reactive-messaging/tree/{smallrye-reactive-messaging-tag}/smallrye-reactive-messaging-kafka-api/src/main/java/io/smallrye/reactive/messaging/kafka/api[`io/smallrye/reactive/messaging/kafka/api`] package.
  
  The API consists of the following classes:
  
  * https://github.com/smallrye/smallrye-reactive-messaging/tree/{smallrye-reactive-messaging-tag}/smallrye-reactive-messaging-kafka-api/src/main/java/io/smallrye/reactive/messaging/kafka/api/IncomingKafkaRecordMetadata.java[`IncomingKafkaRecordMetadata`] - This metadata contains information such as:
  ** the `key` of the Kafka record represented by a `Message`
  ** the Kafka `topic` and `partition` used for the `Message`, and the `offset` within those
  ** the `Message` `timestamp` and `timestampType`
  ** the `Message` `headers` - these are pieces of information the application can attach on the producing side, and receive on the consuming side. They are stored and forwarded on by Kafka but have no meaning to Kafka itself.
  * https://github.com/smallrye/smallrye-reactive-messaging/tree/{smallrye-reactive-messaging-tag}/smallrye-reactive-messaging-kafka-api/src/main/java/io/smallrye/reactive/messaging/kafka/api/OutgoingKafkaRecordMetadata.java[`OutgoingKafkaRecordMetadata`] - This is constructed via the builder returned via the `builder()` method, and allows you to specify/override how Kafka will handle the messages. Similar to the `IncomingKafkaRecordMetadata` case, you can set:
  ** the `key`. Kafka will then treat this entry as the key of the message
  ** the `topic`, as already seen we typically use the `microprofile-config.properties` configuration to specify the topic to use for a channel backed by Kafka. However, in some cases the code sending the message might need to make some choices (for example depending on values contained in the data) about which topic to send to. Specifying this here will make Kafka use that topic.
  ** the `partition`. Generally, it is best to let Kafka's partitioner choose the partition, but for cases where it is essential to be able to specify it this can be done
  ** the `timestamp` if you don't want the one auto-generated by Kafka
  ** `headers` - you can attach headers for the consumer, as mentioned for `IncomingKafkaRecordMetadata`
  * https://github.com/smallrye/smallrye-reactive-messaging/tree/{smallrye-reactive-messaging-tag}/smallrye-reactive-messaging-kafka-api/src/main/java/io/smallrye/reactive/messaging/kafka/api/KafkaMetadataUtil.java[`KafkaMetadataUtil`] contains utility methods to write `OutgoingKafkaRecordMetadata` to a `Message`, and to read `IncomingKafkaRecordMetadata` from a `Message`. Note that if you write `OutgoingKafkaRecordMetadata` to a `Message` which is sent to a channel not handled by Kafka it will be ignored, and if you attempt to read `IncomingKafkaRecordMetadata` from a `Message` arriving from a channel no handled by Kafka it will be `null`.
  
  The following example shows how to write and read the `key` from a message:
  
  [source, java, options="nowrap"]
  ----
  @Inject
  @Channel("from-user")
  Emitter<Integer> emitter;
  
  @Incoming("from-user")
  @Outgoing("to-kafka")
  public Message<Integer> send(Message<Integer> msg) {
      // Set the key in the metadata
      OutgoingKafkaRecordMetadata<String> md =
              OutgoingKafkaRecordMetadata.<String>builder()
                  .withKey("KEY-" + i)
                  .build();
      // Note that Message is immutable so the copy returned by this method
      // call is not the same as the parameter to the method
      return KafkaMetadataUtil.writeOutgoingKafkaMetadata(msg, md);
  }
  
  @Incoming("from-kafka")
  public CompletionStage<Void> receive(Message<Integer> msg) {
      IncomingKafkaRecordMetadata<String, Integer> metadata =
          KafkaMetadataUtil.readIncomingKafkaMetadata(msg).get();
  
      // We can now read the Kafka record key
      String key = metadata.getKey();
  
      // When using the Message wrapper around the payload we need to explicitly ack
      // them
      return msg.ack();
  }
  ----
  To configure the Kafka mapping we need a `microprofile-config.properties`
  ```
  kafka.bootstrap.servers=kafka:9092
  
  mp.messaging.outgoing.to-kafka.connector=smallrye-kafka
  mp.messaging.outgoing.to-kafka.topic=some-topic
  mp.messaging.outgoing.to-kafka.value.serializer=org.apache.kafka.common.serialization.IntegerSerializer
  mp.messaging.outgoing.to-kafka.key.serializer=org.apache.kafka.common.serialization.StringSerializer
  
  mp.messaging.incoming.from-kafka.connector=smallrye-kafka
  mp.messaging.incoming.from-kafka.topic=some-topic
  mp.messaging.incoming.from-kafka.value.deserializer=org.apache.kafka.common.serialization.IntegerDeserializer
  mp.messaging.incoming.from-kafka.key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
  ```
  This configuration looks a lot like the previous configuration that we saw, but note that we need to specify the `key.serializer` for the outgoing channel, and the `key.deserializer` for the incoming channel. As before, they are implementations of `org.apache.kafka.common.serialization.Serializer` and `org.apache.kafka.common.serialization.Deserializer` respectively. Kafka provides implementations for basic types, and you may write your own and include them in the deployment.
  
  ===== A note on `org.apache.kafka` classes
  While we do expose the Kafka Clients jar in our BOMs, its usage is limited to
  
  * Classes/interfaces exposed via the Kafka User API, e.g.:
  ** `org.apache.kafka.common.header.Header` and `org.apache.kafka.common.header.Headers` and implementations of those that are considered public API as per the Apache Kafka documentation.
  ** `org.apache.kafka.clients.consumer.ConsumerRecord`
  ** `org.apache.kafka.common.record.TimestampType`
  * Classes/interfaces needed for serialization and deserialization:
  ** `org.apache.kafka.common.serialization.Deserializer`
  ** `org.apache.kafka.common.serialization.Serializer`
  ** Implementatations of `org.apache.kafka.common.serialization.Deserializer` and `org.apache.kafka.common.serialization.Serializer` in the `org.apache.kafka.common.serialization` package
  
  
  == Component Reference
  
  