title: WSO2 Kafka Inbound Endpoint Retry on Error
body: <p>We try to use Kafka with WSO2 ESB server.</p>
  
  <p>We've implemented and API which puts incoming messages to Kafka.</p>
  
  <p>Then we've implemented an inbound endpoint that retrieves messages from Kafka and transfer these messages to other external sytems. </p>
  
  <p>Everything works very well in happy path but when we test the "external systems down" scenario failed messages are not delivered when external systems up again.</p>
  
  <p>How can we send failed messages to external systems? </p>
  
  <p>API config:</p>
  
  <pre><code>&lt;?xml version="1.0" encoding="UTF-8"?&gt;
  &lt;api context="/api/event" name="EventAPI" xmlns="http://ws.apache.org/ns/synapse"&gt;
      &lt;resource methods="POST"&gt;
          &lt;inSequence&gt;
              &lt;log category="DEBUG" description="" level="full"/&gt;
              &lt;kafkaTransport.init&gt;
                  &lt;bootstrapServers&gt;localhost:9092&lt;/bootstrapServers&gt;
                  &lt;keySerializerClass&gt;org.apache.kafka.common.serialization.StringSerializer&lt;/keySerializerClass&gt;
                  &lt;valueSerializerClass&gt;org.apache.kafka.common.serialization.StringSerializer&lt;/valueSerializerClass&gt;
                  &lt;maxPoolSize&gt;20&lt;/maxPoolSize&gt;
              &lt;/kafkaTransport.init&gt;
              &lt;kafkaTransport.publishMessages&gt;
                  &lt;topic&gt;event_topic&lt;/topic&gt;
              &lt;/kafkaTransport.publishMessages&gt;
              &lt;loopback/&gt;
          &lt;/inSequence&gt;
          &lt;outSequence&gt;
              &lt;payloadFactory media-type="json"&gt;
                  &lt;format&gt;{"result" : "OK"}&lt;/format&gt;
              &lt;/payloadFactory&gt;
              &lt;property name="messageType" scope="axis2" type="STRING" value="application/json"/&gt;
              &lt;send/&gt;
          &lt;/outSequence&gt;            
      &lt;/resource&gt;
  &lt;/api&gt;
  </code></pre>
  
  <p>Inbound config:</p>
  
  <pre><code>&lt;?xml version="1.0" encoding="UTF-8"?&gt;
  &lt;inboundEndpoint name="EventTransmitter" protocol="kafka"
      sequence="transmit_sequence" suspend="false" onError="fault"
      xmlns="http://ws.apache.org/ns/synapse"&gt;
      &lt;parameters&gt;
  
          &lt;parameter name="interval"&gt;10&lt;/parameter&gt;
          &lt;parameter name="coordination"&gt;true&lt;/parameter&gt;
          &lt;parameter name="sequential"&gt;true&lt;/parameter&gt;
          &lt;parameter name="zookeeper.connect"&gt;localhost:2181&lt;/parameter&gt;
          &lt;parameter name="consumer.type"&gt;highlevel&lt;/parameter&gt;
          &lt;parameter name="content.type"&gt;application/json&lt;/parameter&gt;
          &lt;parameter name="topics"&gt;event_topic&lt;/parameter&gt;
          &lt;parameter name="group.id"&gt;myconsumer&lt;/parameter&gt;
          &lt;parameter name="consumer.id"&gt;myconsumer&lt;/parameter&gt;
  
          &lt;parameter name="dual.commit.enabled"&gt;true&lt;/parameter&gt;
          &lt;parameter name="auto.offset.reset"&gt;largest&lt;/parameter&gt;
  
      &lt;/parameters&gt;
  &lt;/inboundEndpoint&gt;
  </code></pre>
  
  <p>Sequence:</p>
  
  <pre><code>&lt;?xml version="1.0" encoding="UTF-8"?&gt;
  &lt;sequence name="transmit_sequence" onError="fault" trace="disable" xmlns="http://ws.apache.org/ns/synapse"&gt;
  
      &lt;send receive="event_transmit_out_sequence"&gt;
          &lt;endpoint key="gov:endpoints/HandlerEndpoint.xml"/&gt;
      &lt;/send&gt;
  
  &lt;/sequence&gt;
  </code></pre>