Check whether eventChunk hasNext event while batching
@@ -76,7 +76,6 @@ public class BatchWindowProcessor extends WindowProcessor implements FindablePro
private SiddhiAppContext siddhiAppContext;
private StreamEvent resetEvent = null;
-
@Override
protected void init(ExpressionExecutor[] attributeExpressionExecutors, ConfigReader configReader,
boolean outputExpectsExpiredEvents, SiddhiAppContext siddhiAppContext) {
@@ -111,7 +110,8 @@ public class BatchWindowProcessor extends WindowProcessor implements FindablePro
outputStreamEventChunk.add(resetEvent);
}
- while (outputExpectsExpiredEvents && streamEventChunk.hasNext()) {
+ if (streamEventChunk.hasNext()) {
+ while (outputExpectsExpiredEvents) {
StreamEvent streamEvent = streamEventChunk.next();
StreamEvent clonedStreamEvent = streamEventCloner.copyStreamEvent(streamEvent);
clonedStreamEvent.setType(StreamEvent.Type.EXPIRED);
@@ -122,10 +122,10 @@ public class BatchWindowProcessor extends WindowProcessor implements FindablePro
resetEvent.setType(ComplexEvent.Type.RESET);
outputStreamEventChunk.add(streamEventChunk.getFirst());
}
+ }
nextProcessor.process(outputStreamEventChunk);
}
-
@Override
public void start() {
//Do nothing
