Fix kafka header with null values

[againom]

To reproduce:
Send a kafka message with null header values

Sample error:

java.lang.NullPointerException: null
	at java.lang.String.<init>(String.java:600)
	at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
	at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
	at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
	at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
	at clojure.lang.Reflector.invokeConstructor(Reflector.java:305)
	at ziggurat.messaging.producer$record_headers__GT_map$fn__9507.invoke(producer.clj:91)
	at clojure.core.protocols$iter_reduce.invokeStatic(protocols.clj:49)
	at clojure.core.protocols$fn__8162.invokeStatic(protocols.clj:75)
	at clojure.core.protocols$fn__8162.invoke(protocols.clj:75)
	at clojure.core.protocols$fn__8110$G__8105__8123.invoke(protocols.clj:13)
	at clojure.core$reduce.invokeStatic(core.clj:6830)
	at clojure.core$reduce.invoke(core.clj:6812)
	at ziggurat.messaging.producer$record_headers__GT_map.invokeStatic(producer.clj:90)
	at ziggurat.messaging.producer$record_headers__GT_map.invoke(producer.clj:89)
	at ziggurat.messaging.producer$properties_for_publish.invokeStatic(producer.clj:99)
	at ziggurat.messaging.producer$properties_for_publish.invoke(producer.clj:95)
	at ziggurat.messaging.producer$publish_internal$fn__9514.invoke(producer.clj:115)
	at ziggurat.messaging.producer$publish_internal.invokeStatic(producer.clj:113)
	at ziggurat.messaging.producer$publish_internal.invoke(producer.clj:110)
	at ziggurat.messaging.producer$publish.invokeStatic(producer.clj:130)
	at ziggurat.messaging.producer$publish.invoke(producer.clj:126)
	at ziggurat.messaging.producer$publish_to_delay_queue.invokeStatic(producer.clj:231)
	at ziggurat.messaging.producer$publish_to_delay_queue.invoke(producer.clj:227)
	at ziggurat.messaging.producer$retry.invokeStatic(producer.clj:266)
	at ziggurat.messaging.producer$retry.invoke(producer.clj:263)
	at ziggurat.mapper$mapper_func$fn__10010$fn__10012.invoke(mapper.clj:55)
	at new_reliquary.core$wrap_with_named_transaction$fn__9238.invoke(core.clj:28)
	at new_reliquary.core.NewRelicTracer.trace(core.clj:32)
	at new_reliquary.core.NewRelicTracer.doTransaction(core.clj:35)
	at new_reliquary.core$with_newrelic_transaction.invokeStatic(core.clj:45)
	at new_reliquary.core$with_newrelic_transaction.invoke(core.clj:43)
	at new_reliquary.core$with_newrelic_transaction.invokeStatic(core.clj:47)
	at new_reliquary.core$with_newrelic_transaction.invoke(core.clj:43)
	at ziggurat.mapper$mapper_func$fn__10010.invoke(mapper.clj:42)
	at ziggurat.streams$traced_handler_fn.invokeStatic(streams.clj:140)
	at ziggurat.streams$traced_handler_fn.invoke(streams.clj:129)
	at ziggurat.streams$topology$fn__16575.invoke(streams.clj:201)
	at ziggurat.streams$value_mapper$reify__16509.apply(streams.clj:72)
	at org.apache.kafka.streams.kstream.internals.AbstractStream.lambda$withKey$1(AbstractStream.java:106)
	at org.apache.kafka.streams.kstream.internals.KStreamMapValues$KStreamMapProcessor.process(KStreamMapValues.java:40)
	at org.apache.kafka.streams.processor.internals.ProcessorAdapter.process(ProcessorAdapter.java:71)
	at org.apache.kafka.streams.processor.internals.ProcessorNode.lambda$process$2(ProcessorNode.java:181)
	at org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.maybeMeasureLatency(StreamsMetricsImpl.java:884)
	at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:181)
	at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forwardInternal(ProcessorContextImpl.java:281)
	at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:260)
	at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:219)
	at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:172)
	at org.apache.kafka.streams.kstream.internals.KStreamMapValues$KStreamMapProcessor.process(KStreamMapValues.java:41)
	at org.apache.kafka.streams.processor.internals.ProcessorAdapter.process(ProcessorAdapter.java:71)
	at org.apache.kafka.streams.processor.internals.ProcessorNode.lambda$process$2(ProcessorNode.java:181)
	at org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.maybeMeasureLatency(StreamsMetricsImpl.java:884)
	at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:181)
	at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forwardInternal(ProcessorContextImpl.java:281)
	at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:260)
	at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:219)
	at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:172)
	at org.apache.kafka.streams.kstream.internals.KStreamTransformValues$KStreamTransformValuesProcessor.process(KStreamTransformValues.java:64)
	at org.apache.kafka.streams.processor.internals.ProcessorAdapter.process(ProcessorAdapter.java:71)
	at org.apache.kafka.streams.processor.internals.ProcessorNode.lambda$process$2(ProcessorNode.java:181)
	at org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.maybeMeasureLatency(StreamsMetricsImpl.java:884)
	at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:181)
	at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forwardInternal(ProcessorContextImpl.java:281)
	at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:260)
	at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:219)
	at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:172)
	at org.apache.kafka.streams.kstream.internals.KStreamFlatTransform$KStreamFlatTransformProcessor.process(KStreamFlatTransform.java:67)
	at org.apache.kafka.streams.processor.internals.ProcessorAdapter.process(ProcessorAdapter.java:71)
	at org.apache.kafka.streams.processor.internals.ProcessorNode.lambda$process$2(ProcessorNode.java:181)
	at org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.maybeMeasureLatency(StreamsMetricsImpl.java:884)
	at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:181)
	at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forwardInternal(ProcessorContextImpl.java:281)
	at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:260)
	at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:219)
	at org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:86)
	at org.apache.kafka.streams.processor.internals.StreamTask.lambda$process$1(StreamTask.java:731)
	at org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.maybeMeasureLatency(StreamsMetricsImpl.java:879)
	at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:731)
	at org.apache.kafka.streams.processor.internals.TaskManager.process(TaskManager.java:1177)
	at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:753)
	at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:583)
	at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:556)
	

[anmol1vw13]

LGTM