CONCAT should handle null columns gracefully

[rodesai]

If a query has a CONCAT UDF and receives a record where one of the columns is null, no result is emitted and the following is printed in the log:

[2018-02-27 13:33:55,269] ERROR Error calculating column with index 0 : V3 (io.confluent.ksql.structured.SelectValueMapper:72) java.lang.reflect.InvocationTargetException at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) at java.lang.reflect.Method.invoke(Method.java:498) at org.codehaus.janino.ScriptEvaluator.evaluate(ScriptEvaluator.java:756) at org.codehaus.janino.ScriptEvaluator.evaluate(ScriptEvaluator.java:748) at io.confluent.ksql.structured.SelectValueMapper.apply(SelectValueMapper.java:70) at io.confluent.ksql.structured.SelectValueMapper.apply(SelectValueMapper.java:32) at org.apache.kafka.streams.kstream.internals.AbstractStream$2.apply(AbstractStream.java:127) at org.apache.kafka.streams.kstream.internals.KStreamMapValues$KStreamMapProcessor.process(KStreamMapValues.java:40) at org.apache.kafka.streams.processor.internals.ProcessorNode$1.run(ProcessorNode.java:46) at org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:208) at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:124) at org.apache.kafka.streams.processor.internals.AbstractProcessorContext.forward(AbstractProcessorContext.java:174) at org.apache.kafka.streams.kstream.internals.KStreamTransformValues$KStreamTransformValuesProcessor.process(KStreamTransformValues.java:169) at org.apache.kafka.streams.processor.internals.ProcessorNode$1.run(ProcessorNode.java:46) at org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:208) at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:124) at org.apache.kafka.streams.processor.internals.AbstractProcessorContext.forward(AbstractProcessorContext.java:174) at org.apache.kafka.streams.kstream.internals.KStreamMapValues$KStreamMapProcessor.process(KStreamMapValues.java:41) at org.apache.kafka.streams.processor.internals.ProcessorNode$1.run(ProcessorNode.java:46) at org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:208) at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:124) at org.apache.kafka.streams.processor.internals.AbstractProcessorContext.forward(AbstractProcessorContext.java:174) at org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:80) at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:221) at org.apache.kafka.streams.processor.internals.AssignedStreamsTasks.process(AssignedStreamsTasks.java:94) at org.apache.kafka.streams.processor.internals.TaskManager.process(TaskManager.java:422) at org.apache.kafka.streams.processor.internals.StreamThread.processAndMaybeCommit(StreamThread.java:924) at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:804) at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:756) at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:726) Caused by: java.lang.NullPointerException at io.confluent.ksql.function.udf.string.ConcatKudf.evaluate(ConcatKudf.java:34) at SC.eval0(Unknown Source) ... 32 more

[big-andy-coates]

Hey @rodesai, what do you think should happen?

[rodesai]

I think we should return a null result, and not spam the log with these error messages. Turns out we actually do return null, so this bug would just cover not spamming the log.

[rmoff]

Related: #750