ad-manager crashes when wrong metrics sent to mapped-metrics

[tkamenov-expedia]

The ad-manager shuts down when wrong metrics sent to mapped-metrics kafka topic.

Way to reproduce:
Make ad-manager running locally in Kubernetes and from a command prompt run fakespans to push to mapped-metrics topic, i.e.
fakespans --kafka-broker="192.168.99.100:9092" -topic mapped-metrics
ad-manager breaks and shuts down (and is restarted by kubernetes after)

Log:

2018-10-31 00:58:37 INFO  StateChangeListener:40 - State change event called with newState=ERROR and oldState=RUNNING
2018-10-31 00:58:37 WARN  KafkaStreams:413 - stream-client [ad-manager-df8e12fe-23d3-4c9e-a912-268dbd99a7dc] All stream threads have died. The instance will be in error state and should be closed.
2018-10-31 00:58:37 INFO  StreamThread:1128 - stream-thread [ad-manager-df8e12fe-23d3-4c9e-a912-268dbd99a7dc-StreamThread-1] Shutdown complete
2018-10-31 00:58:37 ERROR StateChangeListener:50 - uncaught exception occurred running kafka streams for thread=ad-manager-df8e12fe-23d3-4c9e-a912-268dbd99a7dc-StreamThread-1
org.apache.kafka.streams.errors.StreamsException: Deserialization exception handler is set to fail upon a deserialization error. If you would rather have the streaming pipeline continue after a deserialization error, please set the default.deserialization.exception.handler appropriately.
    at org.apache.kafka.streams.processor.internals.RecordDeserializer.deserialize(RecordDeserializer.java:74)
    at org.apache.kafka.streams.processor.internals.RecordQueue.addRawRecords(RecordQueue.java:91)
    at org.apache.kafka.streams.processor.internals.PartitionGroup.addRawRecords(PartitionGroup.java:117)
    at org.apache.kafka.streams.processor.internals.StreamTask.addRecords(StreamTask.java:567)
    at org.apache.kafka.streams.processor.internals.StreamThread.addRecordsToTasks(StreamThread.java:900)
    at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:801)
    at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:749)
    at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:719)
Caused by: org.apache.kafka.common.errors.SerializationException: com.fasterxml.jackson.core.JsonParseException: Unrecognized token '$a05d37fd': was expecting ('true', 'false' or 'null')
 at [Source: (byte[])"
$a05d37fd-539e-44b9-ab6c-78574623b39b$e2ec24b3-07ad-4fc1-9f09-bcc31fa00381"some-service*  some-span0�Њ����8����"; line: 2, column: 11]
Caused by: com.fasterxml.jackson.core.JsonParseException: Unrecognized token '$a05d37fd': was expecting ('true', 'false' or 'null')
 at [Source: (byte[])"
$a05d37fd-539e-44b9-ab6c-78574623b39b$e2ec24b3-07ad-4fc1-9f09-bcc31fa00381"some-service*  some-span0�Њ����8����"; line: 2, column: 11]
    at com.fasterxml.jackson.core.JsonParser._constructError(JsonParser.java:1804)
    at com.fasterxml.jackson.core.base.ParserMinimalBase._reportError(ParserMinimalBase.java:679)
    at com.fasterxml.jackson.core.json.UTF8StreamJsonParser._reportInvalidToken(UTF8StreamJsonParser.java:3526)
    at com.fasterxml.jackson.core.json.UTF8StreamJsonParser._handleUnexpectedValue(UTF8StreamJsonParser.java:2621)
    at com.fasterxml.jackson.core.json.UTF8StreamJsonParser._nextTokenNotInObject(UTF8StreamJsonParser.java:826)
    at com.fasterxml.jackson.core.json.UTF8StreamJsonParser.nextToken(UTF8StreamJsonParser.java:723)
    at com.fasterxml.jackson.databind.ObjectMapper._initForReading(ObjectMapper.java:4141)
    at com.fasterxml.jackson.databind.ObjectMapper._readMapAndClose(ObjectMapper.java:4000)
    at com.fasterxml.jackson.databind.ObjectMapper.readValue(ObjectMapper.java:3091)
    at com.expedia.adaptivealerting.kafka.serde.JsonPojoDeserializer.deserialize(JsonPojoDeserializer.java:65)
    at org.apache.kafka.common.serialization.ExtendedDeserializer$Wrapper.deserialize(ExtendedDeserializer.java:65)
    at org.apache.kafka.common.serialization.ExtendedDeserializer$Wrapper.deserialize(ExtendedDeserializer.java:55)
    at org.apache.kafka.streams.processor.internals.SourceNode.deserializeValue(SourceNode.java:56)
    at org.apache.kafka.streams.processor.internals.RecordDeserializer.deserialize(RecordDeserializer.java:61)
    at org.apache.kafka.streams.processor.internals.RecordQueue.addRawRecords(RecordQueue.java:91)
    at org.apache.kafka.streams.processor.internals.PartitionGroup.addRawRecords(PartitionGroup.java:117)
    at org.apache.kafka.streams.processor.internals.StreamTask.addRecords(StreamTask.java:567)
    at org.apache.kafka.streams.processor.internals.StreamThread.addRecordsToTasks(StreamThread.java:900)
    at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:801)
    at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:749)
    at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:719)
2018-10-31 00:58:37 ERROR HealthStatusController:47 - Setting the app status as 'UNHEALTHY'
2018-10-31 00:58:37 INFO  Application:91 - Shutting down topology
2018-10-31 00:58:37 INFO  KafkaStreams:261 - stream-client [ad-manager-df8e12fe-23d3-4c9e-a912-268dbd99a7dc] State transition from ERROR to PENDING_SHUTDOWN
2018-10-31 00:58:37 INFO  StateChangeListener:40 - State change event called with newState=PENDING_SHUTDOWN and oldState=ERROR
2018-10-31 00:58:37 INFO  StreamThread:1094 - stream-thread [ad-manager-df8e12fe-23d3-4c9e-a912-268dbd99a7dc-StreamThread-1] Informed to shut down
2018-10-31 00:58:37 INFO  KafkaStreams:261 - stream-client [ad-manager-df8e12fe-23d3-4c9e-a912-268dbd99a7dc] State transition from PENDING_SHUTDOWN to NOT_RUNNING
2018-10-31 00:58:37 INFO  StateChangeListener:40 - State change event called with newState=NOT_RUNNING and oldState=PENDING_SHUTDOWN
2018-10-31 00:58:37 INFO  KafkaStreams:867 - stream-client [ad-manager-df8e12fe-23d3-4c9e-a912-268dbd99a7dc] Streams client stopped completely
2018-10-31 00:58:37 INFO  Application:94 - Shutting down jmxReporter
2018-10-31 00:58:37 INFO  Application:97 - Shutting down logger. Bye!```

[williewheeler]

See StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG