WindowGenerator extends MuxReservoir implements Stream, Runnable { @Override public final void run() { try { resetBeginNewWindow(); } catch (InterruptedException ie) { handleException(ie); } } WindowGenerator(ScheduledExecutorService service, int capacity); final void advanceWindow(); @Override final void run(); void setFirstWindow(long millis); void setResetWindow(long millis); void setWindowWidth(int millis); void setCheckpointCount(int streamingWindowCount, int offset); @Override void setup(StreamContext context); @Override void activate(StreamContext context); @Override void deactivate(); @Override void teardown(); @Override void put(Object tuple); @Override boolean putControl(ControlTuple payload); @Override int getCount(boolean reset); static long getWindowCount(long millis, long firstMillis, long widthMillis); static long getWindowId(long millis, long firstWindowMillis, long windowWidthMillis); static long getNextWindowMillis(long windowId, long firstWindowMillis, long windowWidthMillis); static long getNextWindowId(long windowId, long firstWindowMillis, long windowWidthMillis); static long getAheadWindowId(long windowId, long firstWindowMillis, long windowWidthMillis, int ahead); static long compareWindowId(long windowIdA, long windowIdB, long windowWidthMillis); static long getWindowMillis(long windowId, long firstWindowMillis, long windowWidthMillis); static long getBaseSecondsFromWindowId(long windowId); static final int WINDOW_MASK; static final int MIN_WINDOW_ID; static final int MAX_WINDOW_ID; static final int MAX_WINDOW_WIDTH; }
@Test public void testOutofSequenceError() throws Exception { logger.info("Testing Out of Sequence Error"); LogicalPlan dag = new LogicalPlan(); String workingDir = new File("target/testOutofSequenceError").getAbsolutePath(); dag.setAttribute(Context.OperatorContext.STORAGE_AGENT, new AsyncFSStorageAgent(workingDir, null)); RandomNumberGenerator rng = dag.addOperator("random", new RandomNumberGenerator()); MyLogger ml = dag.addOperator("logger", new MyLogger()); dag.addStream("stream", rng.output, ml.input); StramLocalCluster lc = new StramLocalCluster(dag); lc.run(10000); }