engine/src/test/java/com/datatorrent/stram/engine/WindowGeneratorTest.java - apex-core - Git at Google

 /**
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *   http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing,
  * software distributed under the License is distributed on an
  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  * KIND, either express or implied.  See the License for the
  * specific language governing permissions and limitations
  * under the License.
  */
 package com.datatorrent.stram.engine;

 import java.io.File;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;

 import org.junit.Assert;
 import org.junit.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 import com.datatorrent.api.Context;
 import com.datatorrent.api.Context.OperatorContext;
 import com.datatorrent.api.DefaultInputPort;
 import com.datatorrent.api.DefaultOutputPort;
 import com.datatorrent.api.InputOperator;
 import com.datatorrent.api.Sink;
 import com.datatorrent.bufferserver.packet.MessageType;
 import com.datatorrent.common.util.AsyncFSStorageAgent;
 import com.datatorrent.common.util.BaseOperator;
 import com.datatorrent.common.util.ScheduledThreadPoolExecutor;
 import com.datatorrent.stram.StramLocalCluster;
 import com.datatorrent.stram.plan.logical.LogicalPlan;
 import com.datatorrent.stram.support.ManualScheduledExecutorService;
 import com.datatorrent.stram.tuple.ResetWindowTuple;
 import com.datatorrent.stram.tuple.Tuple;

 public class WindowGeneratorTest
 {
   @Test
   public void test2ndResetWindow() throws InterruptedException
   {
     logger.info("Testing 2nd Reset Window");

     ManualScheduledExecutorService msse = new ManualScheduledExecutorService(1);
     WindowGenerator generator = new WindowGenerator(msse, (WindowGenerator.MAX_WINDOW_ID << 1) + 1024);

     generator.setFirstWindow(0L);
     generator.setResetWindow(0L);
     generator.setWindowWidth(1);

     SweepableReservoir reservoir = generator.acquireReservoir(Node.OUTPUT, (WindowGenerator.MAX_WINDOW_ID << 1) + 1024);
     final AtomicBoolean loggingEnabled = new AtomicBoolean(true);
     reservoir.setSink(new Sink<Object>()
     {
       @Override
       public void put(Object payload)
       {
         if (loggingEnabled.get()) {
           logger.debug(payload.toString());
         }

       }

       @Override
       public int getCount(boolean reset)
       {
         return 0;
       }

     });

     generator.activate(null);

     msse.tick(1); /* reset window and begin window */
     msse.tick(1); /* end window and begin window */
     loggingEnabled.set(false);
     for (int i = 0; i < WindowGenerator.MAX_WINDOW_ID - 2; i++) {
       msse.tick(1); /* end window and begin window */
     }
     loggingEnabled.set(true);
     msse.tick(1); /* end window, reset window, begin window */

     final AtomicInteger beginWindowCount = new AtomicInteger(0);
     final AtomicInteger endWindowCount = new AtomicInteger(0);
     final AtomicInteger resetWindowCount = new AtomicInteger(0);
     Tuple t;
     reservoir.sweep();
     while ((t = reservoir.sweep()) != null) {
       reservoir.remove();
       switch (t.getType()) {
         case BEGIN_WINDOW:
           beginWindowCount.incrementAndGet();
           break;

         case END_WINDOW:
           endWindowCount.incrementAndGet();
           break;

         case RESET_WINDOW:
           resetWindowCount.incrementAndGet();
           break;

         default:
           break;
       }
     }

     Assert.assertEquals("begin windows", WindowGenerator.MAX_WINDOW_ID + 1 + 1, beginWindowCount.get());
     Assert.assertEquals("end windows", WindowGenerator.MAX_WINDOW_ID + 1, endWindowCount.get());
     Assert.assertEquals("reset windows", 2, resetWindowCount.get());
   }

   /**
    * Test of resetWindow functionality of WindowGenerator.
    */
   @Test
   public void testResetWindow()
   {
     ManualScheduledExecutorService msse = new ManualScheduledExecutorService(1);
     msse.setCurrentTimeMillis(0x0afebabe * 1000L);
     WindowGenerator generator = new WindowGenerator(msse, WindowGenerator.MAX_WINDOW_ID + 1024);

     final long currentTIme = msse.getCurrentTimeMillis();
     final int windowWidth = 0x1234abcd;
     generator.setFirstWindow(currentTIme);
     generator.setResetWindow(currentTIme);
     generator.setWindowWidth(windowWidth);
     SweepableReservoir reservoir = generator.acquireReservoir(Node.OUTPUT, 1024);
     reservoir.setSink(new Sink<Object>()
     {
       boolean firsttime = true;

       @Override
       public int getCount(boolean reset)
       {
         return 0;
       }

       @Override
       public void put(Object payload)
       {
         assert (false);
         if (firsttime) {
           assert (payload instanceof ResetWindowTuple);
           firsttime = false;
         } else {
           assert (payload instanceof Tuple);
         }
       }

     });
     generator.activate(null);
     msse.tick(1);

     Assert.assertNull(reservoir.sweep());
     ResetWindowTuple rwt = (ResetWindowTuple)reservoir.sweep();
     reservoir.remove();
     assert (rwt.getWindowId() == 0x0afebabe00000000L);
     assert (rwt.getBaseSeconds() * 1000L == currentTIme);
     assert (rwt.getIntervalMillis() == windowWidth);

     Tuple t = reservoir.sweep();
     reservoir.remove();
     assert (t.getType() == MessageType.BEGIN_WINDOW);
     assert (t.getWindowId() == 0x0afebabe00000000L);

     assert (reservoir.sweep() == null);
   }

   @Test
   public void testWindowGen() throws Exception
   {
     final AtomicLong currentWindow = new AtomicLong();
     final AtomicInteger beginWindowCount = new AtomicInteger();
     final AtomicInteger endWindowCount = new AtomicInteger();

     final AtomicLong windowXor = new AtomicLong();

     Sink<Object> s = new Sink<Object>()
     {
       @Override
       public int getCount(boolean reset)
       {
         return 0;
       }

       @Override
       public void put(Object payload)
       {
         logger.debug("unexpected payload {}", payload);
       }

     };

     ScheduledThreadPoolExecutor stpe = new ScheduledThreadPoolExecutor(1, "WindowGenerator");
     int windowWidth = 200;
     long firstWindowMillis = stpe.getCurrentTimeMillis();
     firstWindowMillis -= firstWindowMillis % 1000L;

     WindowGenerator wg = new WindowGenerator(new ScheduledThreadPoolExecutor(1, "WindowGenerator"), WindowGenerator.MAX_WINDOW_ID + 1024);
     wg.setResetWindow(firstWindowMillis);
     wg.setFirstWindow(firstWindowMillis);
     wg.setWindowWidth(windowWidth);
     SweepableReservoir reservoir = wg.acquireReservoir("GeneratorTester", windowWidth);
     reservoir.setSink(s);

     wg.activate(null);
     Thread.sleep(200);
     wg.deactivate();

     reservoir.sweep(); /* just transfer over all the control tuples */
     Tuple t;
     while ((t = reservoir.sweep()) != null) {
       reservoir.remove();
       long windowId = t.getWindowId();

       switch (t.getType()) {
         case BEGIN_WINDOW:
           currentWindow.set(windowId);
           beginWindowCount.incrementAndGet();
           windowXor.set(windowXor.get() ^ windowId);
           break;

         case END_WINDOW:
           endWindowCount.incrementAndGet();
           windowXor.set(windowXor.get() ^ windowId);
           break;

         case RESET_WINDOW:
           break;

         default:
           currentWindow.set(0);
           break;
       }
     }
     long lastWindowMillis = System.currentTimeMillis();

     Assert.assertEquals("only last window open", currentWindow.get(), windowXor.get());

     long expectedCnt = (lastWindowMillis - firstWindowMillis) / windowWidth;

     Assert.assertTrue("Minimum begin window count", expectedCnt + 1 <= beginWindowCount.get());
     Assert.assertEquals("end window count", beginWindowCount.get() - 1, endWindowCount.get());
   }

   static class RandomNumberGenerator implements InputOperator
   {
     public final transient DefaultOutputPort<Integer> output = new DefaultOutputPort<>();

     @Override
     public void emitTuples()
     {
       try {
         Thread.sleep(500);
       } catch (InterruptedException ex) {
         logger.debug("interrupted!", ex);
       }

       output.emit(++count);
     }

     @Override
     public void beginWindow(long windowId)
     {
     }

     @Override
     public void endWindow()
     {
     }

     @Override
     public void setup(OperatorContext context)
     {
     }

     @Override
     public void teardown()
     {
     }

     int count;
   }

   static class MyLogger extends BaseOperator
   {
     public final transient DefaultInputPort<Integer> input = new DefaultInputPort<Integer>()
     {
       @Override
       public void process(Integer tuple)
       {
         logger.debug("received {}", tuple);
       }

     };
   }

   @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);
   }

   @Test
   public void testWindowToTime()
   {
     long first = 1431714014000L;

     for (int windowWidthMillis : new int[]{500, 123}) {
       long time1 = WindowGenerator.getWindowMillis(6149164867354886271L, first, windowWidthMillis);
       long time2 = WindowGenerator.getWindowMillis(6149164867354886272L, first, windowWidthMillis);

       long window1 = WindowGenerator.getWindowId(time1, first, windowWidthMillis);
       long window2 = WindowGenerator.getWindowId(time2, first, windowWidthMillis);

       Assert.assertEquals("window 1", 6149164867354886271L, window1);
       Assert.assertEquals("window 2", 6149164867354886272L, window2);

       Assert.assertEquals("window millis difference", windowWidthMillis, time2 - time1);
     }
   }

   @Test
   public void testWindowToTimeBaseSecondRollover()
   {
     long first = 1431714014123L;

     for (int windowWidthMillis : new int[]{500, 123}) {
       long window1 = WindowGenerator.getWindowId(first, first, windowWidthMillis);
       window1 |= WindowGenerator.MAX_WINDOW_ID;
       long window2 = WindowGenerator.getNextWindowId(window1, first, windowWidthMillis);
       Assert.assertTrue("base seconds should be greater during an rollover", (window2 >> 32) > (window1 >> 32));
       long time1 = WindowGenerator.getWindowMillis(window1, first, windowWidthMillis);
       long time2 = WindowGenerator.getWindowMillis(window2, first, windowWidthMillis);

       Assert.assertEquals("max window id", WindowGenerator.MAX_WINDOW_ID, window1 & WindowGenerator.WINDOW_MASK);
       Assert.assertEquals("rollover after max", 0, window2 & WindowGenerator.WINDOW_MASK);
       Assert.assertEquals("window millis difference", windowWidthMillis, time2 - time1);
     }
   }

   @Test
   public void testWindowIdAhead()
   {
     long first = 1431714014123L;
     int ahead = 678;
     for (int windowWidthMillis : new int[]{500, 123}) {
       long window1 = WindowGenerator.getWindowId(first, first, windowWidthMillis);
       long window2 = WindowGenerator.getAheadWindowId(window1, first, windowWidthMillis, ahead);
       for (int i = 0; i < ahead; i++) {
         window1 = WindowGenerator.getNextWindowId(window1, first, windowWidthMillis);
       }
       Assert.assertEquals(window2, window1);
     }
   }

   @Test
   public void testWindowIdCompare()
   {
     long first = 1431714014123L;
     int ahead = 341;
     for (int windowWidthMillis : new int[]{500, 123}) {
       long window1 = WindowGenerator.getWindowId(first, first, windowWidthMillis);
       long window2 = WindowGenerator.getAheadWindowId(window1, first, windowWidthMillis, ahead);
       Assert.assertEquals(ahead, WindowGenerator.compareWindowId(window2, window1, windowWidthMillis));
     }
   }

   public static final Logger logger = LoggerFactory.getLogger(WindowGeneratorTest.class);
 }
	/**
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing,
	* software distributed under the License is distributed on an
	* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	* KIND, either express or implied. See the License for the
	* specific language governing permissions and limitations
	* under the License.
	*/
	package com.datatorrent.stram.engine;

	import java.io.File;
	import java.util.concurrent.atomic.AtomicBoolean;
	import java.util.concurrent.atomic.AtomicInteger;
	import java.util.concurrent.atomic.AtomicLong;

	import org.junit.Assert;
	import org.junit.Test;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	import com.datatorrent.api.Context;
	import com.datatorrent.api.Context.OperatorContext;
	import com.datatorrent.api.DefaultInputPort;
	import com.datatorrent.api.DefaultOutputPort;
	import com.datatorrent.api.InputOperator;
	import com.datatorrent.api.Sink;
	import com.datatorrent.bufferserver.packet.MessageType;
	import com.datatorrent.common.util.AsyncFSStorageAgent;
	import com.datatorrent.common.util.BaseOperator;
	import com.datatorrent.common.util.ScheduledThreadPoolExecutor;
	import com.datatorrent.stram.StramLocalCluster;
	import com.datatorrent.stram.plan.logical.LogicalPlan;
	import com.datatorrent.stram.support.ManualScheduledExecutorService;
	import com.datatorrent.stram.tuple.ResetWindowTuple;
	import com.datatorrent.stram.tuple.Tuple;

	public class WindowGeneratorTest
	{
	@Test
	public void test2ndResetWindow() throws InterruptedException
	{
	logger.info("Testing 2nd Reset Window");

	ManualScheduledExecutorService msse = new ManualScheduledExecutorService(1);
	WindowGenerator generator = new WindowGenerator(msse, (WindowGenerator.MAX_WINDOW_ID << 1) + 1024);

	generator.setFirstWindow(0L);
	generator.setResetWindow(0L);
	generator.setWindowWidth(1);

	SweepableReservoir reservoir = generator.acquireReservoir(Node.OUTPUT, (WindowGenerator.MAX_WINDOW_ID << 1) + 1024);
	final AtomicBoolean loggingEnabled = new AtomicBoolean(true);
	reservoir.setSink(new Sink<Object>()
	{
	@Override
	public void put(Object payload)
	{
	if (loggingEnabled.get()) {
	logger.debug(payload.toString());
	}

	}

	@Override
	public int getCount(boolean reset)
	{
	return 0;
	}

	});

	generator.activate(null);

	msse.tick(1); /* reset window and begin window */
	msse.tick(1); /* end window and begin window */
	loggingEnabled.set(false);
	for (int i = 0; i < WindowGenerator.MAX_WINDOW_ID - 2; i++) {
	msse.tick(1); /* end window and begin window */
	}
	loggingEnabled.set(true);
	msse.tick(1); /* end window, reset window, begin window */

	final AtomicInteger beginWindowCount = new AtomicInteger(0);
	final AtomicInteger endWindowCount = new AtomicInteger(0);
	final AtomicInteger resetWindowCount = new AtomicInteger(0);
	Tuple t;
	reservoir.sweep();
	while ((t = reservoir.sweep()) != null) {
	reservoir.remove();
	switch (t.getType()) {
	case BEGIN_WINDOW:
	beginWindowCount.incrementAndGet();
	break;

	case END_WINDOW:
	endWindowCount.incrementAndGet();
	break;

	case RESET_WINDOW:
	resetWindowCount.incrementAndGet();
	break;

	default:
	break;
	}
	}

	Assert.assertEquals("begin windows", WindowGenerator.MAX_WINDOW_ID + 1 + 1, beginWindowCount.get());
	Assert.assertEquals("end windows", WindowGenerator.MAX_WINDOW_ID + 1, endWindowCount.get());
	Assert.assertEquals("reset windows", 2, resetWindowCount.get());
	}

	/**
	* Test of resetWindow functionality of WindowGenerator.
	*/
	@Test
	public void testResetWindow()
	{
	ManualScheduledExecutorService msse = new ManualScheduledExecutorService(1);
	msse.setCurrentTimeMillis(0x0afebabe * 1000L);
	WindowGenerator generator = new WindowGenerator(msse, WindowGenerator.MAX_WINDOW_ID + 1024);

	final long currentTIme = msse.getCurrentTimeMillis();
	final int windowWidth = 0x1234abcd;
	generator.setFirstWindow(currentTIme);
	generator.setResetWindow(currentTIme);
	generator.setWindowWidth(windowWidth);
	SweepableReservoir reservoir = generator.acquireReservoir(Node.OUTPUT, 1024);
	reservoir.setSink(new Sink<Object>()
	{
	boolean firsttime = true;

	@Override
	public int getCount(boolean reset)
	{
	return 0;
	}

	@Override
	public void put(Object payload)
	{
	assert (false);
	if (firsttime) {
	assert (payload instanceof ResetWindowTuple);
	firsttime = false;
	} else {
	assert (payload instanceof Tuple);
	}
	}

	});
	generator.activate(null);
	msse.tick(1);

	Assert.assertNull(reservoir.sweep());
	ResetWindowTuple rwt = (ResetWindowTuple)reservoir.sweep();
	reservoir.remove();
	assert (rwt.getWindowId() == 0x0afebabe00000000L);
	assert (rwt.getBaseSeconds() * 1000L == currentTIme);
	assert (rwt.getIntervalMillis() == windowWidth);

	Tuple t = reservoir.sweep();
	reservoir.remove();
	assert (t.getType() == MessageType.BEGIN_WINDOW);
	assert (t.getWindowId() == 0x0afebabe00000000L);

	assert (reservoir.sweep() == null);
	}

	@Test
	public void testWindowGen() throws Exception
	{
	final AtomicLong currentWindow = new AtomicLong();
	final AtomicInteger beginWindowCount = new AtomicInteger();
	final AtomicInteger endWindowCount = new AtomicInteger();

	final AtomicLong windowXor = new AtomicLong();

	Sink<Object> s = new Sink<Object>()
	{
	@Override
	public int getCount(boolean reset)
	{
	return 0;
	}

	@Override
	public void put(Object payload)
	{
	logger.debug("unexpected payload {}", payload);
	}

	};

	ScheduledThreadPoolExecutor stpe = new ScheduledThreadPoolExecutor(1, "WindowGenerator");
	int windowWidth = 200;
	long firstWindowMillis = stpe.getCurrentTimeMillis();
	firstWindowMillis -= firstWindowMillis % 1000L;

	WindowGenerator wg = new WindowGenerator(new ScheduledThreadPoolExecutor(1, "WindowGenerator"), WindowGenerator.MAX_WINDOW_ID + 1024);
	wg.setResetWindow(firstWindowMillis);
	wg.setFirstWindow(firstWindowMillis);
	wg.setWindowWidth(windowWidth);
	SweepableReservoir reservoir = wg.acquireReservoir("GeneratorTester", windowWidth);
	reservoir.setSink(s);

	wg.activate(null);
	Thread.sleep(200);
	wg.deactivate();

	reservoir.sweep(); /* just transfer over all the control tuples */
	Tuple t;
	while ((t = reservoir.sweep()) != null) {
	reservoir.remove();
	long windowId = t.getWindowId();

	switch (t.getType()) {
	case BEGIN_WINDOW:
	currentWindow.set(windowId);
	beginWindowCount.incrementAndGet();
	windowXor.set(windowXor.get() ^ windowId);
	break;

	case END_WINDOW:
	endWindowCount.incrementAndGet();
	windowXor.set(windowXor.get() ^ windowId);
	break;

	case RESET_WINDOW:
	break;

	default:
	currentWindow.set(0);
	break;
	}
	}
	long lastWindowMillis = System.currentTimeMillis();

	Assert.assertEquals("only last window open", currentWindow.get(), windowXor.get());

	long expectedCnt = (lastWindowMillis - firstWindowMillis) / windowWidth;

	Assert.assertTrue("Minimum begin window count", expectedCnt + 1 <= beginWindowCount.get());
	Assert.assertEquals("end window count", beginWindowCount.get() - 1, endWindowCount.get());
	}

	static class RandomNumberGenerator implements InputOperator
	{
	public final transient DefaultOutputPort<Integer> output = new DefaultOutputPort<>();

	@Override
	public void emitTuples()
	{
	try {
	Thread.sleep(500);
	} catch (InterruptedException ex) {
	logger.debug("interrupted!", ex);
	}

	output.emit(++count);
	}

	@Override
	public void beginWindow(long windowId)
	{
	}

	@Override
	public void endWindow()
	{
	}

	@Override
	public void setup(OperatorContext context)
	{
	}

	@Override
	public void teardown()
	{
	}

	int count;
	}

	static class MyLogger extends BaseOperator
	{
	public final transient DefaultInputPort<Integer> input = new DefaultInputPort<Integer>()
	{
	@Override
	public void process(Integer tuple)
	{
	logger.debug("received {}", tuple);
	}

	};
	}

	@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);
	}

	@Test
	public void testWindowToTime()
	{
	long first = 1431714014000L;

	for (int windowWidthMillis : new int[]{500, 123}) {
	long time1 = WindowGenerator.getWindowMillis(6149164867354886271L, first, windowWidthMillis);
	long time2 = WindowGenerator.getWindowMillis(6149164867354886272L, first, windowWidthMillis);

	long window1 = WindowGenerator.getWindowId(time1, first, windowWidthMillis);
	long window2 = WindowGenerator.getWindowId(time2, first, windowWidthMillis);

	Assert.assertEquals("window 1", 6149164867354886271L, window1);
	Assert.assertEquals("window 2", 6149164867354886272L, window2);

	Assert.assertEquals("window millis difference", windowWidthMillis, time2 - time1);
	}
	}

	@Test
	public void testWindowToTimeBaseSecondRollover()
	{
	long first = 1431714014123L;

	for (int windowWidthMillis : new int[]{500, 123}) {
	long window1 = WindowGenerator.getWindowId(first, first, windowWidthMillis);
	window1 \|= WindowGenerator.MAX_WINDOW_ID;
	long window2 = WindowGenerator.getNextWindowId(window1, first, windowWidthMillis);
	Assert.assertTrue("base seconds should be greater during an rollover", (window2 >> 32) > (window1 >> 32));
	long time1 = WindowGenerator.getWindowMillis(window1, first, windowWidthMillis);
	long time2 = WindowGenerator.getWindowMillis(window2, first, windowWidthMillis);

	Assert.assertEquals("max window id", WindowGenerator.MAX_WINDOW_ID, window1 & WindowGenerator.WINDOW_MASK);
	Assert.assertEquals("rollover after max", 0, window2 & WindowGenerator.WINDOW_MASK);
	Assert.assertEquals("window millis difference", windowWidthMillis, time2 - time1);
	}
	}

	@Test
	public void testWindowIdAhead()
	{
	long first = 1431714014123L;
	int ahead = 678;
	for (int windowWidthMillis : new int[]{500, 123}) {
	long window1 = WindowGenerator.getWindowId(first, first, windowWidthMillis);
	long window2 = WindowGenerator.getAheadWindowId(window1, first, windowWidthMillis, ahead);
	for (int i = 0; i < ahead; i++) {
	window1 = WindowGenerator.getNextWindowId(window1, first, windowWidthMillis);
	}
	Assert.assertEquals(window2, window1);
	}
	}

	@Test
	public void testWindowIdCompare()
	{
	long first = 1431714014123L;
	int ahead = 341;
	for (int windowWidthMillis : new int[]{500, 123}) {
	long window1 = WindowGenerator.getWindowId(first, first, windowWidthMillis);
	long window2 = WindowGenerator.getAheadWindowId(window1, first, windowWidthMillis, ahead);
	Assert.assertEquals(ahead, WindowGenerator.compareWindowId(window2, window1, windowWidthMillis));
	}
	}

	public static final Logger logger = LoggerFactory.getLogger(WindowGeneratorTest.class);
	}