Pipeline restarting due to NPE when field with NULL value passed to rex function

Applicability

Data Stream Processor

Description

The pipeline was restarting regularly:
With the following exception :

Caused by: org.apache.flink.streaming.runtime.tasks.ExceptionInChainedOperatorException: Could not forward element to next operator
at org.apache.flink.streaming.runtime.tasks.OperatorChain$ChainingOutput.pushToOperator(OperatorChain.java:549)
at org.apache.flink.streaming.runtime.tasks.OperatorChain$ChainingOutput.collect(OperatorChain.java:523)
at org.apache.flink.streaming.runtime.tasks.OperatorChain$ChainingOutput.collect(OperatorChain.java:483)
at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:727)
at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:705)
at org.apache.flink.streaming.api.operators.StreamMap.processElement(StreamMap.java:41)
at org.apache.flink.streaming.runtime.tasks.OperatorChain$ChainingOutput.pushToOperator(OperatorChain.java:546)
... 58 more
Caused by: java.lang.NullPointerException
at java.util.regex.Matcher.getTextLength(Matcher.java:1283)
at java.util.regex.Matcher.reset(Matcher.java:309)
at java.util.regex.Matcher.<init>(Matcher.java:229)
at java.util.regex.Pattern.matcher(Pattern.java:1093)
at com.google.code.regexp.Matcher.<init>(Unknown Source)
at com.google.code.regexp.Pattern.matcher(Unknown Source)
at com.splunk.streaming.functions.streaming.function.AbstractRegexParseWithOptionsFunction$ApplyRegex.map(AbstractRegexParseWithOptionsFunction.java:315)
at com.splunk.streaming.functions.streaming.function.AbstractRegexParseWithOptionsFunction$ApplyRegex.map(AbstractRegexParseWithOptionsFunction.java:283)
at com.splunk.streaming.functions.streaming.function.RichMapFunctionWithMetrics.map(RichMapFunctionWithMetrics.java:104)
at com.splunk.streaming.functions.streaming.function.RichMapFunctionWithMetrics.map(RichMapFunctionWithMetrics.java:24)
at org.apache.flink.streaming.api.operators.StreamMap.processElement(StreamMap.java:41)
at org.apache.flink.streaming.runtime.tasks.OperatorChain$ChainingOutput.pushToOperator(OperatorChain.java:546)
... 64 more

 

Investigation

This part of the stack was helpful :

Caused by: java.lang.NullPointerException
at java.util.regex.Matcher.getTextLength(Matcher.java:1283)
at java.util.regex.Matcher.reset(Matcher.java:309)
at java.util.regex.Matcher.<init>(Matcher.java:229)
at java.util.regex.Pattern.matcher(Pattern.java:1093)

It's saying that a field with null value was passed to the regex(Parse with regex)
and matcher does not like that and throws a NullPointerException

In the problematic pipeline, There is this (snippet)

eval body=tostring(body)
| rex field=body "\\\"EventName\\\":\\\"(?<EventName>[^\\\"]+)\\\","
| where EventName IN ("Requestor","Approver","System","Administrators","Assign","Add")
| rex field=body "\\\"Target\\\":\\\"(?<Target>[^\\\"]+)\\\","
| rex field=Target "(Asset|Domain|ManagedSystem)[:=](?:[^ ]+) (M|Managed)?Account[:=](?:[^\\\\]+\\\\\\\\)?(?<userName1>.+)"
| rex field=Target "(?:[^\\/]+)\\/(?<userName2>[^,]+)"

So

| rex field=body "\\\"Target\\\":\\\"(?<Target>[^\\\"]+)\\\","

It does not find a match. Target will be null
So when a rex then is run on that field... The NPE can occur

| rex field=Target "(Asset|Domain|ManagedSystem)[:=](?:[^ ]+) (M|Managed)?Account[:=](?:[^\\\\]+\\\\\\\\)?(?<userName1>.+)"

 

Resolution

The NPE no longer occurs in the scenario of when a Null is passed to the rex function in DSP 1.3.1 .

To workaround in earlier versions of DSP , can. implement the following workarounds:

Add eval to set the field to a space(' ') before passing to rex 

fieldname=if(isnull(fieldname)," ",fieldname)

 

If the events with null fields are not required to be processed:

where fieldname  IS NOT NULL

Related Links

DSP: Parse regex

Revision History

Revision	Publish Date	Comments
1.0	29-Sep-2026	Initial Release