siddhi-io / siddhi-execution-json Goto Github PK

I have encountered a concurrent problem when using the toObject function. I spent a few hours checking the source code. And I realize a potential concurrency issue in class ToJSONObjectFunctionExtension.

JSONParser is created as a global variable in class ToJSONObjectFunctionExtension. When executing the execute(Object data, State state) function, the parse(data.toString()) function will be called with the input parameter. Everything looks good

    private static final JSONParser jsonParser = new JSONParser(JSONParser.DEFAULT_PERMISSIVE_MODE);  //global variable

    //...
    
    @Override
    protected Object execute(Object data, State state) {
        Object returnValue = null;
        try {
            returnValue = jsonParser.parse(data.toString());
        } catch (ParseException e) {
            log.error(siddhiQueryContext.getSiddhiAppContext().getName() + ":" + siddhiQueryContext.getName() +
                    ": Cannot parse the given string into JSON. Hence returning null");
        }
        return returnValue;
    }

Unfortunately, If you check function parse(String input) of class JSONParser, you will see the input parameter will be set to a global variable in class JSONParserString. In the case of multiple threads execution function execute(Object data, State state) at the same time, it will cause the concurrency issue.

Version: siddhi-execution-json 2.0.9

Description:

The Following issue can be observed when the same function is used by the two Siddhi applications parallel.

ERROR {io.siddhi.core.stream.StreamJunction} - Error in 'chunkAPP' after consuming events from Stream 'SweetProductionStream', null. Exception on class 'io.siddhi.extension.execution.json.function.ToJSONObjectFunctionExtension'. Hence, dropping event 'StateEvent{streamEvents=[null, null], timestamp=-1, type=CURRENT, outputData=[["Asia","Asia Pacific"], ["CO2","Refined products","Petroleum","Middle distillates","Diesel fuel"], a], next=StateEvent{streamEvents=[null, null], timestamp=-1, type=CURRENT, outputData=[["Asia","Asia Pacific"], ["CO2","Refined products","Petroleum","Middle distillates","Diesel fuel"], a], next=StateEvent{streamEvents=[null, null], timestamp=-1, type=CURRENT, outputData=[["Asia","Asia Pacific"], ["CO2","Refined products","Petroleum","Middle distillates","Diesel fuel"], a], next=StateEvent{streamEvents=[null, null], timestamp=-1, type=CURRENT, outputData=[["Asia","Asia Pacific"], ["CO2","Refined products","Petroleum","Middle distillates","Diesel fuel"], a], next=StateEvent{streamEvents=[null, null], timestamp=-1, type=CURRENT, outputData=[["Asia","Asia Pacific"], ["CO2","Refined products","Petroleum","Middle distillates","Diesel fuel"], a], next=StateEvent{streamEvents=[null, null], timestamp=-1, type=CURRENT, outputData=[["Asia","Asia Pacific"], ["CO2","Refined products","Petroleum","Middle distillates","Diesel fuel"], a], next=StateEvent{streamEvents=[null, null], timestamp=-1, type=CURRENT, outputData=[["Asia","Asia Pacific"], ["CO2","Refined products","Petroleum","Middle distillates","Diesel fuel"], a], next=StateEvent{streamEvents=[null, null], timestamp=-1, type=CURRENT, outputData=[["Asia","Asia Pacific"], ["CO2","Refined products","Petroleum","Middle distillates","Diesel fuel"], a], next=StateEvent{streamEvents=[null, null], timestamp=-1, type=CURRENT, outputData=[["Asia","Asia Pacific"], ["CO2","Refined products","Petroleum","Middle distillates","Diesel fuel"], a], next=StateEvent{streamEvents=[null, null], timestamp=-1, type=CURRENT, outputData=[["Asia","Asia Pacific"], ["CO2","Refined products","Petroleum","Middle distillates","Diesel fuel"], a], next=StateEvent{streamEvents=[null, null], timestamp=-1, type=CURRENT, outputData=[["Asia","Asia Pacific"], ["CO2","Refined products","Petroleum","Middle distillates","Diesel fuel"], a], next=StateEvent{streamEvents=[null, null], timestamp=-1, type=CURRENT, outputData=[["Asia","Asia Pacific"], ["CO2","Refined products","Petroleum","Middle distillates","Diesel fuel"], a], next=StateEvent{streamEvents=[null, null], timestamp=-1, type=CURRENT, outputData=[["Asia","Asia Pacific"], ["CO2","Refined products","Petroleum","Middle distillates","Diesel fuel"], a], next=StateEvent{streamEvents=[null, null], timestamp=-1, type=CURRENT, outputData=[["Asia","Asia Pacific"], ["CO2","Refined products","Petroleum","Middle distillates","Diesel fuel"], a], next=StateEvent{streamEvents=[null, null], timestamp=-1, type=CURRENT, outputData=[["Asia","Asia Pacific"], ["CO2","Refined products","Petroleum","Middle distillates","Diesel fuel"], a], next=StateEvent{streamEvents=[null, null], timestamp=-1, type=CURRENT, outputData=[["Asia","Asia Pacific"], ["CO2","Refined products","Petroleum","Middle distillates","Diesel fuel"], a], next=StateEvent{streamEvents=[null, null], timestamp=-1, type=CURRENT, outputData=[["Asia","Asia Pacific"], ["CO2","Refined products","Petroleum","Middle distillates","Diesel fuel"], a], next=StateEvent{streamEvents=[null, null], timestamp=-1, type=CURRENT, outputData=[["Asia","Asia Pacific"], ["CO2","Refined products","Petroleum","Middle distillates","Diesel fuel"], a], next=null}}}}}}}}}}}}}}}}}}' io.siddhi.core.exception.SiddhiAppRuntimeException: null. Exception on class 'io.siddhi.extension.execution.json.function.ToJSONObjectFunctionExtension' at io.siddhi.core.executor.function.FunctionExecutor.execute(FunctionExecutor.java:105) at io.siddhi.core.query.selector.attribute.processor.AttributeProcessor.process(AttributeProcessor.java:41) at io.siddhi.core.query.selector.QuerySelector.processNoGroupBy(QuerySelector.java:172) at io.siddhi.core.query.selector.QuerySelector.process(QuerySelector.java:87) at io.siddhi.core.query.input.ProcessStreamReceiver.processAndClear(ProcessStreamReceiver.java:182) at io.siddhi.core.query.input.ProcessStreamReceiver.process(ProcessStreamReceiver.java:89) at io.siddhi.core.query.input.ProcessStreamReceiver.receive(ProcessStreamReceiver.java:115) at io.siddhi.core.stream.StreamJunction.sendEvent(StreamJunction.java:179) at io.siddhi.core.stream.StreamJunction$Publisher.send(StreamJunction.java:496) at io.siddhi.core.query.output.callback.InsertIntoStreamCallback.send(InsertIntoStreamCallback.java:56) at io.siddhi.core.query.output.ratelimit.OutputRateLimiter.sendToCallBacks(OutputRateLimiter.java:104) at io.siddhi.core.query.output.ratelimit.PassThroughOutputRateLimiter.process(PassThroughOutputRateLimiter.java:44) at io.siddhi.core.query.output.ratelimit.OutputRateLimiter.process(OutputRateLimiter.java:131) at io.siddhi.core.query.selector.QuerySelector.process(QuerySelector.java:118) at io.siddhi.core.query.input.stream.join.JoinProcessor.process(JoinProcessor.java:98) at io.siddhi.core.query.processor.stream.window.EmptyWindowProcessor.process(EmptyWindowProcessor.java:93) at io.siddhi.core.query.processor.stream.window.EmptyWindowProcessor.process(EmptyWindowProcessor.java:50) at io.siddhi.core.query.processor.stream.window.BatchingWindowProcessor.processEventChunk(BatchingWindowProcessor.java:79) at io.siddhi.core.query.processor.stream.AbstractStreamProcessor.process(AbstractStreamProcessor.java:132) at io.siddhi.core.query.input.stream.join.JoinProcessor.process(JoinProcessor.java:86) at io.siddhi.core.query.input.ProcessStreamReceiver.processAndClear(ProcessStreamReceiver.java:182) at io.siddhi.core.query.input.ProcessStreamReceiver.process(ProcessStreamReceiver.java:89) at io.siddhi.core.query.input.ProcessStreamReceiver.receive(ProcessStreamReceiver.java:127) at io.siddhi.core.stream.StreamJunction.sendEvent(StreamJunction.java:202) at io.siddhi.core.stream.StreamJunction$Publisher.send(StreamJunction.java:505) at io.siddhi.core.stream.input.InputDistributor.send(InputDistributor.java:34) at io.siddhi.core.stream.input.InputEntryValve.send(InputEntryValve.java:45) at io.siddhi.core.stream.input.InputHandler.send(InputHandler.java:78) at io.siddhi.core.stream.input.source.PassThroughSourceHandler.sendEvent(PassThroughSourceHandler.java:35) at io.siddhi.core.stream.input.source.InputEventHandler.sendEvent(InputEventHandler.java:82) at io.siddhi.extension.map.json.sourcemapper.JsonSourceMapper.mapAndProcess(JsonSourceMapper.java:255) at io.siddhi.core.stream.input.source.SourceMapper.onEvent(SourceMapper.java:200) at io.siddhi.core.stream.input.source.SourceMapper.onEvent(SourceMapper.java:144) at io.siddhi.extension.io.http.source.HttpWorkerThread.run(HttpWorkerThread.java:62) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748)Caused by: java.lang.NullPointerException at io.siddhi.extension.execution.json.function.ToJSONObjectFunctionExtension.execute(ToJSONObjectFunctionExtension.java:131) at io.siddhi.core.executor.function.FunctionExecutor.execute(FunctionExecutor.java:93) ... 36 more

It was observed this was happening due to the static object "jsonParser"[1]. After making this object non-static this issue was not observed.

[1].

Affected Product Version:
WSO2 SI 1.1.0

Description:
Currently, it supports tokenization, but we need to use str: library to group the JSONs its better if we have a JSON grouping feature in json: itself.

Is it possible to add null values in json using json:Element() function?

Class : SetElementJSONFunctionExtensio.java line no: 236

        if (key != null) {
            Object parsedJsonElement = gson.fromJson(jsonElement.toString(), Object.class);
            if (jsonElement instanceof String && (parsedJsonElement instanceof Map
                    || parsedJsonElement instanceof List)) {
                documentContext.put(path, key, gson.fromJson(jsonElement.toString(), Object.class));
            } else {
                documentContext.put(path, key, jsonElement);
            }

I am getting null pointer exception here Object parsedJsonElement = gson.fromJson(jsonElement.toString(), Object.class);
in my case, value for jsonElement itself is null.

Is addition of null values is avoided purposely?

Description:
Issue observed when feeding the output of json:setElement() fucntion to json:toString().

Affected Product Version:
siddhi-execution-json-2.0.7