anchor.diff:  [[kafka]]
   = Apache Kafka support
   
  -Citrus provides support for sending and receiving messages on a Kafka topic. We have to separate between synchronous and asynchronous communication. So in this chapter we explain how to setup Kafka message endpoints for synchronous and asynchronous outbound and inbound communication
  +Kafka is a distributed streaming platform that enables you to publish and subscribe to streams of records, similar to a message queue or enterprise messaging system. Citrus provides support for sending/receiving records to/from a Kafka topic. Citrus
  +acts as producer or consumer where the Citrus Kafka endpoint can be used bidirectional. In the current version Citrus supports asynchronous communication. Synchronous communication is about to be added, too.
   
   NOTE: The Kafka components in Citrus are kept in a separate Maven module. If not already done so you have to include the module as Maven dependency to your project
   
  
positive.code: </beans>
  ----
  
  After that you are able to use customized Citrus XML elements in order to define the Spring beans.
  
  [[kafka-endpoints]]
  == Kafka endpoints
  
  By default Citrus Kafka endpoints are asynchronous. So let us first of all deal with asynchronous messaging which means that we will not wait for any response message after sending or receiving a message.
  
  The test case itself should not know about Kafka transport details like queue names or connection credentials. This information is stored in the endpoint component configuration that lives in the basic Spring configuration file in Citrus. So let us have a look at a simple Kafka message endpoint configuration in Citrus.
  
  [source,xml]
  ----
  <citrus-kafka:endpoint id="helloServiceQueueEndpoint"
            destination-name="Citrus.HelloService.Request.Queue"
            timeout="10000"/>
  ----
  
  The endpoint component receives an unique id and a Kafka destination name. This can be a queue or topic destination. We will deal with Kafka topics later on. For now the timeout setting completes our first Kafka endpoint component definition.
  
  The endpoint needs a Kafka connection factory for connecting to a Kafka message broker. The connection factory is also added as Spring bean to the Citrus Spring application context.
  
  [source,xml]
  ----
  <bean id="connectionFactory" 
           class="org.apache.activemq.ActiveMQConnectionFactory">
      <property name="brokerURL" value="tcp://localhost:61616" />
  </bean>
  ----
  
  The Kafka connection factory receives the Kafka message broker URL and is able to hold many other connection specific options. In this example we use the Apache ActiveMQ connection factory implementation as we want to use the ActiveMQ message broker. Citrus works by default with a bean id *connectionFactory* . All Citrus Kafka component will automatically recognize this connection factory.
  
  TIP: Spring makes it very easy to connect to other Kafka broker implementations too (e.g. Apache ActiveMQ, TIBCO Enterprise Messaging Service, IBM Websphere MQ). Just add the required connection factory implementation as *connectionFactory* bean.
  
  NOTE: All of the Citrus Kafka endpoint components will automatically look for a bean named *connectionFactory* by default. You can use the *connection-factory* endpoint attribute in order to use another connection factory instance with different bean names.
  
  [source,xml]
  ----
  <citrus-kafka:endpoint id="helloServiceQueueEndpoint"
        destination-name="Citrus.HelloService.Request.Queue"
        connection-factory="myConnectionFacotry"/>
  ----
  
  As an alternative to that you may want to use a special Spring kafka template implementation as custom bean in your endpoint.
  
  [source,xml]
  ----
  <citrus-kafka:endpoint id="helloServiceQueueEndpoint"
                destination-name="Citrus.HelloService.Request.Queue"
                kafka-template="myJmsTemplate"/>
  ----
  
  The endpoint is now ready to be used inside a test case. Inside a test case you can send or receive messages using this endpoint. The test actions can reference the Kafka endpoint using its identifier. When sending a message the message endpoint creates a Kafka message producer and will simply publish the message to the defined Kafka destination. As the communication is asynchronous by default producer does not wait for a synchronous response.
  
  When receiving a messages with this endpoint the endpoint creates a Kafka consumer on the Kafka destination. The endpoint then acts as a message driven listener. This means that the message consumer connects to the given destination and waits for messages to arrive.
  
  NOTE: Besides the destination-name attribute you can also provide a reference to a destination implementation.
  
  [source,xml]
  ----
  <citrus-kafka:endpoint id="helloServiceQueueEndpoint"
                                  destination="helloServiceQueue"/>
  
  <amq:queue id="helloServiceQueue" physicalName="Citrus.HelloService.Request.Queue"/>
  ----
  
  The destination attribute references to a Kafka destination object in the Spring application context. In the example above we used the ActiveMQ queue destination component. The destination reference can also refer to a JNDI lookup for instance.
  
  [[kafka-synchronous-endpoints]]
  == Kafka synchronous endpoints
  
  When using synchronous message endpoints Citrus will manage a reply destination for receiving a synchronous response message on the reply destination. The following figure illustrates that we now have two destinations in our communication scenario.
  
  image:figure_006.jpg[figure_006.jpg]
  
  The synchronous message endpoint component is similar to the asynchronous brother that we have discussed before. The only difference is that the endpoint will automatically manage a reply destination behind the scenes. By default Citrus uses temporary reply destinations that get automatically deleted after the communication handshake is done. Again we need to use a Kafka connection factory in the Spring XML configuration as the component need to connect to a Kafka message broker.
  
  [source,xml]
  ----
  <citrus-kafka:sync-endpoint id="helloServiceSyncEndpoint"
            destination-name="Citrus.HelloService.InOut.Queue"
            timeout="10000"/>
  ----
  
  The synchronous component defines a target destination which again is either a queue or topic destination. If nothing else is defined the endpoint will create temporary reply destinations on its own. When the endpoint has sent a message it waits synchronously for the response message to arrive on the reply destination. You can receive this reply message in your test case by referencing this same endpoint in a receive test action. In case no reply message arrives in time a message timeout error is raised respectively.
  
  See the following example test case which references the synchronous message endpoint in its send and receive test action in order to send out a message and wait for the synchronous response.
  
  [source,xml]
  ----
  <testcase name="synchronousMessagingTest">
      <actions>
          <send endpoint="helloServiceSyncEndpoint">
              <message>
                  <data>
                    [...]
                  </data>
              </message>
          </send>
          
          <receive endpoint="helloServiceSyncEndpoint">
              <message>
                  <data>
                    [...]
                  </data>
              </message>
          </receive>
      </actions>
  </testcase>
  ----
  
  We initiated the synchronous communication by sending a message on the synchronous endpoint. The second step then receives the synchronous message on the temporary reply destination that was automatically created for us.
  
  If you rather want to define a static reply destination you can do so, too. The static reply destination is not deleted after communication handshake. You may need to work with message selectors then in order to pick the right response message that belongs to a specific communication handshake. You can define a static reply destination on the synchronous endpoint component as follows.
  
  [source,xml]
  ----
  <citrus-kafka:sync-endpoint id="helloServiceSyncEndpoint"
            destination-name="Citrus.HelloService.InOut.Queue"
            reply-destination-name="Citrus.HelloService.Reply.Queue"
            timeout="10000"/>
  ----
  
  Instead of using the *reply-destination-name* feel free to use the destination reference with *reply-destination* attribute. Again you can use a JNDI lookup then to reference a destination object.
  
  IMPORTANT: Be aware of permissions that are mandatory for creating temporary destinations. Citrus tries to create temporary queues on the Kafka message broker. Following from that the Citrus Kafka user has to have the permission to do so. Be sure that the user has the sufficient rights when using temporary reply destinations.
  
  Up to now we have sent a message and waited for a synchronous response in the next step. Now it is also possible to switch the directions of send and receive actions. Then we have the situation where Citrus receives a Kafka message first and then Citrus is in charge of providing a proper synchronous response message to the initial sender.
  
  image:figure_007.jpg[figure_007.jpg]
  
  In this scenario the foreign message producer has stored a dynamic Kafka reply queue destination to the Kafka header. So Citrus has to send the reply message to this specific reply destination, which is dynamic of course. Fortunately the heavy lift is done with the Kafka message endpoint and we do not have to change anything in our configuration. Again we just define a synchronous message endpoint in the application context.
  
  [source,xml]
  ----
  <citrus-kafka:sync-endpoint id="helloServiceSyncEndpoint"
        destination-name="Citrus.HelloService.InOut.Queue"
        timeout="10000"/>
  ----
  
  Now the only thing that changes here is that we first receive a message in our test case on this endpoint. The second step is a send message action that references this same endpoint and we are done. Citrus automatically manages the reply destinations for us.
  
  [source,xml]
  ----
  <testcase name="synchronousMessagingTest">
    <actions>
          <receive endpoint="helloServiceSyncEndpoint">
              <message>
                  <data>
                    [...]
                  </data>
              </message>
          </receive>
  
          <send endpoint="helloServiceSyncEndpoint">
              <message>
                  <data>
                    [...]
                  </data>
              </message>
          </send>
      </actions>
  </testcase>
  ----
  
  [[kafka-topics]]
  == Kafka topics
  
  Up to now we have used Kafka queue destinations on our endpoints. Citrus is also able to connect to Kafka topic destinations. In contrary to Kafka queues which represents the *point-to-point* communication Kafka topics use *publish-subscribe* mechanism in order to spread messages over Kafka.
  A Kafka topic producer publishes messages to the topic, while the topic accepts multiple message subscriptions and delivers the message to all subscribers.
  
  The Citrus Kafka endpoints offer the attribute *'pub-sub-domain'*. Once this attribute is set to *true* Citrus will use Kafka topics instead of queue destinations.
  
  NOTE: When using Kafka topics in your project you may want to configure a `javax.kafka.TopicConnectionFactory` instead of a `javax.kafka.QueueConnectionFactory`.
  
  See the following example where the publish-subscribe attribute is set to true in Kafka message endpoint components.
  
  [source,xml]
  ----
  <citrus-kafka:endpoint id="helloServiceTopicEndpoint"
              destination="helloServiceTopic"
              pub-sub-domain="true"/>
  ----
  
  When using Kafka topics you will be able to subscribe several test actions to the topic destination and receive a message multiple times as all subscribers will receive the message. Also other applications besides Citrus are also able to consume
  messages with a topic subscription. This allows Citrus and other software components to coexist in a test environment.
  
  [[kafka-topic-subscriber]]
  === Kafka topic subscriber
  
  By default Citrus does not deal with durable subscribers when using Kafka topics. This means that messages that were sent in advance to the message subscription are not delivered to the Citrus message endpoint. Following from that racing conditions may cause problems when using Kafka topic endpoints in Citrus.
  Be sure to start the Citrus subscription before messages are sent to the topic. Otherwise you may loose some messages that were sent in advance to the subscription. By default Citrus will use a subscription per receive action using the Kafka endpoint in the test cases. This means that the topic subscription is
  started and stopped per receive action when the action is performed inside a test case.
  
  In order to solve racing conditions for messages that are sent prior to the subscription you can also use a `auto-start` setting on the Kafka endpoint component. This causes Citrus to start/stop the subscription based on the endpoint lifecycle instead of linking the subscription to the receive action. When the endpoint is ready
  the subscription is started and all incoming message events are cached and stored to a internal in memory message channel for later consumption in the tests. Here is the endpoint configuration with `auto-start` enabled.
  
  [source,xml]
  ----
  <citrus-kafka:endpoint id="helloServiceTopicEndpoint"
              destination="helloServiceTopic"
              pub-sub-domain="true"
              auto-start="true"/>
  ----
  
  NOTE: The `auto-start` option is only valid in combination with `pub-sub-domain` enabled. Other combinations may be ignored or lead to configuration failure at startup.
  
  Now with `auto-start` set to `true` the Citrus Kafka endpoint will setup a subscription at the very beginning when the endpoint is loaded in the project. The internal message channel name is derived from the Kafka endpoint id and follows the pattern:
  
  `{citrus-kafka:endpoint@id}":subscriber.inbound"`
  
  The in memory channel id is the combined result of the Kafka endpoint id and the prefix `:subscriber.inbound`. In our example this would be `helloServiceTopicEndpoint:subscriber.inbound`. Now all messages sent to the topic in advance to the tests are cached and ready for consumption and verification in the test.
  In the test nothing really changes for you. You simply use a receive test action on the Kafka endpoint as you would have done before. In the background Citrus will automatically receive the messages from the in memory cache. This mechanism enables us to not loose any messages that were sent to the topic in prior
  to Citrus firing up the test cases.
  
  IMPORTANT: There is a small downside of the `auto-start` topic subscriber. As incoming events are cached internally you will not be able to receive the same topic event in multiple receive actions within the Citrus project. If you need to receive the topic message in several places within Citrus you need
  to set up several Kafka topic endpoints with `auto-start` enabled. In case you just have one receive action at a time you are good to go with the `auto-start` subscriber as it is described here.
  
  [[kafka-topic-durable-subscription]]
  == Kafka topic durable subscription
  
  When using durable subscriptions on Kafka message brokers the message events on a topic are preserved for a subscriber even if the subscriber is inactive. This means that the subscriber may not loose any message events on that particular topic as the subscription is durable and all events are stored for later consumption.
  In case you want to activate durable subscriptions on the Citrus Kafka endpoint use the `durable-subscription` setting in the configuration:
  
  [source,xml]
  ----
  <citrus-kafka:endpoint id="helloServiceTopicEndpoint"
              connection-factory="topicConnectionFactory"
              destination="helloServiceTopic"
              pub-sub-domain="true"
              durable-subscription="true"
              auto-start="true"/>
  
  <bean id="topicConnectionFactory" class="org.springframework.kafka.connection.SingleConnectionFactory">
    <constructor-arg>
      <bean class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="brokerURL" value="${kafka.broker.url}" />
        <property name="watchTopicAdvisories" value="false"/>
        <property name="clientID" value="citrusDurableConnectionFactory"/>
      </bean>
    </constructor-arg>
  </bean>
  ----
  
  The durable subscription in Citrus implies that the subscriber is started when the endpoint configuration is done. All messages received on that subscription are cached internally until the receive action in the test case is performed for actual message consumption. The `auto-start` setting is required to be enabled
  for this reason when using durable subscriptions.
  
  By default Citrus is using the Kafka endpoint subscriber name as durable subscription name (e.g. *helloServiceTopicEndpoint:subscriber*). You can overwrite the durable subscriber name with `durable-subscriber-name` setting on the endpoint.
  
  In addition to that you need to add a client id on the connection factory so the message broker is able to identify the durable subscription with the client address. Also we use the `SingleConnectionFactory` implementation of Spring as a connection factory wrapper so we do not fail because of multiple connections with the
  same durable subscriber id.
  
  [[kafka-message-headers]]
  == Kafka message headers
  
  The Kafka specification defines a set of special message header entries that can go into your Kafka message. These Kafka headers are stored differently in a Kafka message header than other custom header entries do. Therefore these special header values should be set in a special syntax that we discuss in the next paragraphs.
  
  [source,xml]
  ----
  <header>
      <element name="citrus_kafka_correlationId" value="${correlationId}"/>
      <element name="citrus_kafka_messageId" value="${messageId}"/>
      <element name="citrus_kafka_redelivered" value="${redelivered}"/>
      <element name="citrus_kafka_timestamp" value="${timestamp}"/>
  </header>
  ----
  
  As you see all Kafka specific message headers use the `citrus_kafka_` prefix. This prefix comes from Spring Integration message header mappers that take care of setting those headers in the Kafka message header properly.
  
  Typing of message header entries may also be of interest in order to meet the Kafka standards of typed message headers. For instance the following message header is of type double and is therefore transferred via Kafka as a double value.
  
  [source,xml]
  ----
  <header>
      <element name="amount" value="19.75" type="double"/>
  </header>
  ----
  
  [[dynamic-destination-names]]
  == Dynamic destination names
  
  Usually you set the target destination as property on the Kafka endpoint component. In some cases it might be useful to set the target destination in a more dynamic way during the test run. You can do this
  by adding a special message header named *citrus_kafka_topic*. This header is automatically interpreted by the Citrus Kafka endpoint and is set as the target destination before a message is sent.
  
  [source,xml]
  ----
  <send endpoint="kafkaEndpoint">
      <message>
          ...
      </message>
      <header>
          <element name="citrus_kafka_topic" value="dynamic.destination.name"/>
      </header>
  </send>
  ----
  
  This action above will send the message to the destination "_dynamic.destination.name_" no matter what default destination is set on the referenced endpoint component named _kafkaEndpoint_. The dynamic destination name setting also supports test variables so you
  can use variables and functions in the destination name, too.
  
  Another possibility for dynamic Kafka destinations is given with the link:#dynamic-endpoint-components[dynamic endpoints].
negative.code: </dependency>
  ----
  
  Citrus provides a "citrus-kafka" configuration namespace and schema definition for Kafka related components and features. Include this namespace into your Spring XML configuration in order to use the Citrus Kafka configuration elements. The namespace URI and
  schema location are added to the Spring bean root element.
  
  [source,xml]
  ----
  