错误 WSO2 - Siddhi - “MarkIn 连续调用而没有调用 markOut in”

如何解决错误 WSO2 - Siddhi - “MarkIn 连续调用而没有调用 markOut in”

我是 WSO2 Siddhi 的新手，我希望您能帮助我。我一直在尝试修复它，但我不能。

我正在尝试“通过监控面板监控流处理器”，并查看图表，为此我将我的 siddhi 代码保存在：

C:\...\wso2\server\deployment\siddhi-files 和当执行 WSO2\wso2si-4.0.0\bin>。 \服务器

我正在运行 siddhi 规则，但出现以下错误：

错误 {io.siddhi.core.stream.StreamJunction} - 'SubmitSolutionPRUEBA'中的错误在使用来自 Stream 的事件后 'OutputShareSomeKNowledgeStream'，在 io.siddhi.SiddhiApps.SubmitSolutionPRUEBA.Siddhi.Tables.reward_basicTable.updateOrInsert.latency. 中连续调用 MarkIn 而不调用 markOut。因此，丢弃事件 'StreamEvent{时间戳=1626790380632,beforeWindowData=null,onAfterWind],outputData=[论坛,2017-May-10 09:51:48,0f2f5191-5515-402d-bba2-73aa280285d3,Forum12,4 type=CURRENT,4 type null}' java.lang.IllegalStateException：在 io.siddhi.SiddhiApps.SubmitSolutionPRUEBA.Siddhi.Tables.reward_basicTable.updateOrInsert.latency 中连续调用 MarkIn 而不调用 markOut

    at org.wso2.carbon.si.metrics.core.LatencyMetric.markIn(LatencyMetric.java:62)
    at io.siddhi.core.table.Table.updateOrAddEvents(Table.java:498)
    at io.siddhi.core.query.output.callback.UpdateOrInsertTableCallback.send(UpdateOrInsertTableCallback.java:78)
    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.selector.QuerySelector.process(QuerySelector.java:97)
    at io.siddhi.core.query.input.ProcessstreamReceiver.processAndClear(ProcessstreamReceiver.java:182)
    at io.siddhi.core.query.input.ProcessstreamReceiver.process(ProcessstreamReceiver.java:84)
    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.selector.QuerySelector.process(QuerySelector.java:97)
    at io.siddhi.core.query.processor.filter.FilterProcessor.process(FilterProcessor.java:58)
    at io.siddhi.core.query.input.ProcessstreamReceiver.processAndClear(ProcessstreamReceiver.java:182)
    at io.siddhi.core.query.input.ProcessstreamReceiver.process(ProcessstreamReceiver.java:84)
    at io.siddhi.core.query.input.ProcessstreamReceiver.receive(ProcessstreamReceiver.java:145)
    at io.siddhi.core.stream.StreamJunction.sendEvent(StreamJunction.java:227)
    at io.siddhi.core.stream.StreamJunction.access$000(StreamJunction.java:64)
    at io.siddhi.core.stream.StreamJunction$Publisher.send(StreamJunction.java:514)
    at io.siddhi.core.stream.input.Inputdistributor.send(Inputdistributor.java:39)
    at io.siddhi.core.stream.input.InputEntryValve.send(InputEntryValve.java:55)
    at io.siddhi.core.stream.input.InputHandler.send(InputHandler.java:91)
    at io.siddhi.core.stream.input.source.PassthroughSourceHandler.sendEvents(PassthroughSourceHandler.java:40)
    at io.siddhi.core.stream.input.source.InputEventHandler.sendEvents(InputEventHandler.java:111)
    at io.siddhi.extension.map.text.sourcemapper.TextSourceMapper.onEventHandler(TextSourceMapper.java:307)
    at io.siddhi.extension.map.text.sourcemapper.TextSourceMapper.mapAndProcess(TextSourceMapper.java:254)
    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.file.processors.FileProcessor.receive(FileProcessor.java:176)
    at org.wso2.transport.file.connector.server.FileConsumer$EventListener.fileUpdated(FileConsumer.java:328)
    at org.wso2.transport.file.connector.server.FileConsumer$EventListener.access$100(FileConsumer.java:302)
    at org.wso2.transport.file.connector.server.FileConsumer.readLines(FileConsumer.java:272)
    at org.wso2.transport.file.connector.server.FileConsumer.processFile(FileConsumer.java:231)
    at org.wso2.transport.file.connector.server.FileConsumer.consume(FileConsumer.java:146)
    at org.wso2.transport.file.connector.server.FileServerConnector.poll(FileServerConnector.java:73)
    at org.wso2.carbon.connector.framework.server.polling.PollingJob.execute(PollingJob.java:51)
    at org.quartz.core.JobRunShell.run(JobRunShell.java:202)
    at org.quartz.simpl.SimpleThreadPool$WorkerThread.run(SimpleThreadPool.java:573)

---

我在 WSO2 Siddhi 中的代码是：

-- 输入数据

@app:statistics(reporter = 'jdbc')

define stream InputDataStream (activityType string,weekdayTS string,monthTS string,dayTS string,yearTS string,hourTS string,cestTS string,studentId string,activityTypeId string,resto string);

-- 退出

define stream OutputShareSomeKNowledgeStream (activityType string,timeStamp string,totalTopics int,totalAnswers int);

@store(type='rdbms',jdbc.url="jdbc:MysqL://localhost:3306/PruebaDB_1?&serverTimezone=UTC&autoReconnect=true&useSSL=false",username="root",password="root",jdbc.driver.name="com.MysqL.cj.jdbc.Driver")
@primaryKey('studentId','activityTypeId')
define table OutputShareSomeKNowledgeTable (activityType string,totalAnswers int);

-- 表格

@store(type='rdbms',jdbc.url="jdbc:MysqL://localhost:3306/MEdit4CEPDB_1?&serverTimezone=UTC&autoReconnect=true&useSSL=false",jdbc.driver.name="com.MysqL.cj.jdbc.Driver")
@primaryKey('playerId','activityTypeId')    
define table reward_basicTable (rewardName string,description string,count bool,type string,complexEventName string,playerId string,value int);

-- 3. 查询 --

@info(name='ShareSomeKNowledge')
from InputDataStream[activityType == 'Forum' and 
                    (convert(str:split(resto,',0),'int') + convert(str:split(resto,'([\s\,]+)',1),'int')) >= 5]  -- if totalTopics + totalAnswers
select  activityType,str:concat(yearTS,'-',monthTS,dayTS,' ',hourTS) as timeStamp,studentId,activityTypeId,convert(str:split(resto,'int') as totalTopics,'int') as totalAnswers
insert into OutputShareSomeKNowledgeStream;
  
@info(name='ShareSomeKNowledge_Basic')   -- Guardo puntuacion
from OutputShareSomeKNowledgeStream
select  'Point' as rewardName,'' as description,true as count,'FixAction' as type,'ShareSomeKNowledge' as complexEventName,timeStamp,studentId as playerId,10 as value
update or insert into reward_basicTable
on reward_basicTable.playerId==playerId and reward_basicTable.activityTypeId==activityTypeId;

-- I keep in table BD

from OutputShareSomeKNowledgeStream
select *
update or insert into OutputShareSomeKNowledgeTable
on OutputShareSomeKNowledgeTable.studentId==studentId  and OutputShareSomeKNowledgeTable.activityTypeId==activityTypeId;

我的输入文件包含数据：

Forum,Sun May 07 06:07:37 CEST 2017,Forum1,3,0     
Forum,Mon May 08 04:59:25 CEST 2017,1     
Forum,Wed May 10 03:14:10 CEST 2017,4,2     
Forum,3    
Forum,Wed May 10 09:51:48 CEST 2017,12

测试时程序运行良好，但在部署时执行它以获得延迟图，使用内存......监控，文件的第一行输入数据将其保存在表格中，但其余部分没有不是，它给出了上面公开的错误，我没有得到图表。

我该如何解决错误？我需要查看跟踪图表。 谢谢

解决方法

[已解决]只需切换到Siddhi siddhi-runner-5.1.2和siddhi-tooling-5.1.2的原生版本