title: How to use a custom Kafka Streams ProductionExceptionHandler?
body: <p>Im developing a kafka-streams application but am having trouble overriding the default ProductionExceptionHandler. <br/> 
  I have created a class implementing the ProductionExceptionHandler </p>
  
  <pre><code>public class RtaCustomProcessingExceptionHandler implements ProductionExceptionHandler {
      private static final Logger log = LoggerFactory.getLogger(RtaCustomProcessingExceptionHandler.class);
      private RtaHandlerClient handlerClient;
  
      @Override
      public void configure(Map&lt;String, ?&gt; map) {
          handlerClient = RtaHandlerClient.getInstance();
      }
  
      @Override
      public ProductionExceptionHandlerResponse handle(final ProducerRecord&lt;byte[], byte[]&gt; record,
                                                       final Exception exception) {
          log.debug("PASSING");
  
          return ProductionExceptionHandlerResponse.CONTINUE;
      }
  }
  </code></pre>
  
  <p>and added that to my properties</p>
  
  <pre><code>        properties.put(StreamsConfig.DEFAULT_PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG,
              RtaCustomProcessingExceptionHandler.class);
  </code></pre>
  
  <p>which I use to instanciate KafkaStreams</p>
  
  <pre><code>this.streams = new KafkaStreams(BasicTopology.createTopology(config), config.asProperties());
  </code></pre>
  
  <p><br/>
  When the application is starting up I can see in the log that my handler is beeing picked up</p>
  
  <pre><code>[2018-10-10 07:58:40,471] INFO StreamsConfig values: 
      application.id = xdr-0
      application.server = 
      bootstrap.servers = [kafka-1:9092]
      buffered.records.per.partition = 1000
      cache.max.bytes.buffering = 10485760
      client.id = 
      commit.interval.ms = 1000
      connections.max.idle.ms = 540000
      default.deserialization.exception.handler = class org.apache.kafka.streams.errors.LogAndFailExceptionHandler
      default.key.serde = class org.apache.kafka.common.serialization.Serdes$ByteArraySerde
      default.production.exception.handler = class com.ericsson.dcp.rtang.kafka.streams.xdr.error.handler.RtaCustomProcessingExceptionHandler
      default.timestamp.extractor = class org.apache.kafka.streams.processor.WallclockTimestampExtractor
      default.value.serde = class org.apache.kafka.common.serialization.Serdes$ByteArraySerde
      metadata.max.age.ms = 300000
  </code></pre>
  
  <p>But then later in the log it seems to be overwriten for some odd reason that I cant understand</p>
  
  <pre><code> [2018-10-10 07:58:40,958] INFO StreamsConfig values: 
      application.id = xdr-0
      application.server = 
      bootstrap.servers = [kafka-1:9092]
      buffered.records.per.partition = 1000
      cache.max.bytes.buffering = 10485760
      client.id = xdr-0-99215001-f1fd-43ae-8c3f-026cbd11d013-StreamThread-1-consumer
      commit.interval.ms = 30000
      connections.max.idle.ms = 540000
      default.deserialization.exception.handler = class org.apache.kafka.streams.errors.LogAndFailExceptionHandler
      default.key.serde = class org.apache.kafka.common.serialization.Serdes$ByteArraySerde
      default.production.exception.handler = class org.apache.kafka.streams.errors.DefaultProductionExceptionHandler
      default.timestamp.extractor = class org.apache.kafka.streams.processor.FailOnInvalidTimestamp
      default.value.serde = class org.apache.kafka.common.serialization.Serdes$ByteArraySerde
  </code></pre>
  
  <p>Does anyone have a clue why this is happening?<br/><br/>
  BR<br/>
  -Jiinxy </p>
  
  <p><strong>EDIT</strong>: Updated with actual handler implementation by request.</p>
  
  <p><strong>EDIT2</strong>: Ive set up my testing so I enque 10 items for processing wheres the 6th item contains a value that should cause an exception typically my own <code>public class RtaRecoverableProcessingException extends ApiException</code> but also <code>RecordTooLargeException</code> which both should be caught by a custom handler as far as I understand. Ive also added break-points to <a href="https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java#L164-L202" rel="nofollow noreferrer">RecordCollectorImpl.java</a> at line 162(<code>producer.send(..</code> and 166 (<code>if(exception..</code> I can see that for the first five items the process passes correctly both line 162 and 166. However when the exception is thrown line 166 is not passed. From what I can tell It should be passing line 166 even if an exceptions is thrown since it seems to be handled at line 191(<code>else if (productionExceptionHandler.handle..</code>).</p>