Replace 'expiredEventQueue' nullity check with
outputExpectsExpiredEvents flag
@@ -105,8 +105,6 @@ public class BatchWindowProcessor extends WindowProcessor implements FindablePro
}
outputStreamEventChunk.add(expiredEventQueue.getFirst());
}
- }
- if (expiredEventQueue != null) {
expiredEventQueue.clear();
}
if (resetEvent != null) {
@@ -142,7 +140,9 @@ public class BatchWindowProcessor extends WindowProcessor implements FindablePro
public Map<String, Object> currentState() {
Map<String, Object> state = new HashMap<>();
synchronized (this) {
- state.put("ExpiredEventQueue", expiredEventQueue != null ? expiredEventQueue.getSnapshot() : null);
+ if (outputExpectsExpiredEvents) {
+ state.put("ExpiredEventQueue", expiredEventQueue.getSnapshot());
+ }
state.put("ResetEvent", resetEvent);
}
return state;
@@ -150,7 +150,7 @@ public class BatchWindowProcessor extends WindowProcessor implements FindablePro
@Override
public synchronized void restoreState(Map<String, Object> state) {
- if (expiredEventQueue != null) {
+ if (outputExpectsExpiredEvents) {
expiredEventQueue.clear();
expiredEventQueue.restore((SnapshotStateList) state.get("ExpiredEventQueue"));
}
