anchor.diff:  * link:configuring-camel.html[Configuring Camel]
   * link:message-endpoint.html[Message Endpoint] pattern
   * link:uris.html[URIs]
  -* link:writing-components.html[Writing Components]+* link:writing-components.html[Writing Components]
  
positive.code: {% endraw %}
  // endpoint options: END
  
  
  
  
  For more information about Producer/Consumer configuration:
  
  http://kafka.apache.org/documentation.html#newconsumerconfigs[http://kafka.apache.org/documentation.html#newconsumerconfigs]
  http://kafka.apache.org/documentation.html#producerconfigs[http://kafka.apache.org/documentation.html#producerconfigs]
  
  ### Samples
  
  #### Camel 2.17 or newer
  
  Consuming messages:
  
  [source,java]
  -------------------------------------------------------------------------------------------------
  from("kafka:localhost:9092?topic=test&groupId=testing&autoOffsetReset=earliest&consumersCount=1")
                          .process(new Processor() {
                              @Override
                              public void process(Exchange exchange)
                                      throws Exception {
                                  String messageKey = "";
                                  if (exchange.getIn() != null) {
                                      Message message = exchange.getIn();
                                      Integer partitionId = (Integer) message
                                              .getHeader(KafkaConstants.PARTITION);
                                      String topicName = (String) message
                                              .getHeader(KafkaConstants.TOPIC);
                                      if (message.getHeader(KafkaConstants.KEY) != null)
                                          messageKey = (String) message
                                                  .getHeader(KafkaConstants.KEY);
                                      Object data = message.getBody();
  
  
                                      System.out.println("topicName :: "
                                              + topicName + " partitionId :: "
                                              + partitionId + " messageKey :: "
                                              + messageKey + " message :: "
                                              + data + "\n");
                                  }
                              }
                          }).to("log:input");
  -------------------------------------------------------------------------------------------------
  
  
  When consuming messages from Kafka you can use your own offset management and not delegate this management to Kafka.
  In order to keep the offsets the component needs a `StateRepository` implementation such as `FileStateRepository`.
  This bean should be available in the registry.
  Here how to use it :
  [source,java]
  -------------------------------------------------------------------------------------------------------------------
  // Create the repository in which the Kafka offsets will be persisted
  FileStateRepository repository = FileStateRepository.fileStateRepository(new File("/path/to/repo.dat"));
  
  
negative.code:                 .to("mock:result");
      }
  });
  ----------------------------------------------------------------------------------------------------------------------------
   
  
  #### Producing messages to Kafka
  
  Here is the minimal route you need in order to write messages to Kafka.
  [source,java]
  ----------------------------------------------------------------------------
  from("direct:start")
      .setBody(constant("Message from Camel"))          // Message to send
      .setHeader(KafkaConstants.KEY, constant("Camel")) // Key of the message
      .to("kafka:localhost:9092?topic=test");
  ----------------------------------------------------------------------------
  
  ### Endpoints
  
  