diff --git a/heron/api/src/java/com/twitter/heron/api/bolt/BaseWindowedBolt.java b/heron/api/src/java/com/twitter/heron/api/bolt/BaseWindowedBolt.java
index a25722a..182a965 100644
--- a/heron/api/src/java/com/twitter/heron/api/bolt/BaseWindowedBolt.java
+++ b/heron/api/src/java/com/twitter/heron/api/bolt/BaseWindowedBolt.java
@@ -83,6 +83,9 @@
   }
 
   private BaseWindowedBolt withWindowLength(Count count) {
+    if (count == null) {
+      throw new IllegalArgumentException("Window length cannot be set null");
+    }
     if (count.value <= 0) {
       throw new IllegalArgumentException("Window length must be positive [" + count + "]");
     }
@@ -91,7 +94,10 @@
   }
 
   private BaseWindowedBolt withWindowLength(Duration duration) {
-    if (duration.isNegative()) {
+    if (duration == null) {
+      throw new IllegalArgumentException("Window length cannot be set null");
+    }
+    if (duration.isNegative() || duration.isZero()) {
       throw new IllegalArgumentException("Window length must be positive [" + duration + "]");
     }
 
@@ -101,6 +107,9 @@
   }
 
   private BaseWindowedBolt withSlidingInterval(Count count) {
+    if (count == null) {
+      throw new IllegalArgumentException("Sliding interval cannot be set null");
+    }
     if (count.value <= 0) {
       throw new IllegalArgumentException("Sliding interval must be positive [" + count + "]");
     }
@@ -109,7 +118,10 @@
   }
 
   private BaseWindowedBolt withSlidingInterval(Duration duration) {
-    if (duration.isNegative()) {
+    if (duration == null) {
+      throw new IllegalArgumentException("Sliding interval cannot be set null");
+    }
+    if (duration.isNegative() || duration.isZero()) {
       throw new IllegalArgumentException("Sliding interval must be positive [" + duration + "]");
     }
     windowConfiguration.put(WindowingConfigs.TOPOLOGY_BOLTS_SLIDING_INTERVAL_DURATION_MS,
@@ -194,8 +206,9 @@
   }
 
   /**
-   * Specify a field in the tuple that represents the timestamp as a long value. If this
-   * field is not present in the incoming tuple, an {@link IllegalArgumentException} will be thrown.
+   * Specify a field in the tuple that represents the timestamp as a long value. The timestamp
+   * should also be in milliseconds. If this field is not present in the
+   * incoming tuple, an {@link IllegalArgumentException} will be thrown.
    *
    * @param fieldName the name of the field that contains the timestamp
    */
@@ -210,6 +223,9 @@
    */
   @SuppressWarnings("HiddenField")
   public BaseWindowedBolt withTimestampExtractor(TimestampExtractor timestampExtractor) {
+    if (timestampExtractor == null) {
+      throw new IllegalArgumentException("Timestamp extractor cannot be set to null");
+    }
     if (this.timestampExtractor != null) {
       throw new IllegalArgumentException(
           "Window is already configured with a timestamp " + "extractor: " + timestampExtractor);
@@ -234,6 +250,9 @@
    * @param streamId the name of the stream used to emit late tuples on
    */
   public BaseWindowedBolt withLateTupleStream(String streamId) {
+    if (streamId == null) {
+      throw new IllegalArgumentException("Cannot set late tuple stream id to null");
+    }
     windowConfiguration.put(WindowingConfigs.TOPOLOGY_BOLTS_LATE_TUPLE_STREAM, streamId);
     return this;
   }
@@ -247,6 +266,12 @@
    * @param duration the max lag duration
    */
   public BaseWindowedBolt withLag(Duration duration) {
+    if (duration == null) {
+      throw new IllegalArgumentException("Lag duration cannot be set null");
+    }
+    if (duration.isNegative() || duration.isZero()) {
+      throw new IllegalArgumentException("Lag duration must be positive [" + duration + "]");
+    }
     windowConfiguration.put(WindowingConfigs.TOPOLOGY_BOLTS_TUPLE_TIMESTAMP_MAX_LAG_MS,
         duration.toMillis());
     return this;
@@ -259,6 +284,12 @@
    * @param interval the interval at which watermark events are generated
    */
   public BaseWindowedBolt withWatermarkInterval(Duration interval) {
+    if (interval == null) {
+      throw new IllegalArgumentException("Watermark interval cannot be set null");
+    }
+    if (interval.isNegative() || interval.isZero()) {
+      throw new IllegalArgumentException("Watermark interval must be positive [" + interval + "]");
+    }
     windowConfiguration.put(WindowingConfigs.TOPOLOGY_BOLTS_WATERMARK_EVENT_INTERVAL_MS,
         interval.toMillis());
     return this;
diff --git a/heron/api/src/java/com/twitter/heron/api/bolt/WindowedBoltExecutor.java b/heron/api/src/java/com/twitter/heron/api/bolt/WindowedBoltExecutor.java
index 2dd1e15..d027d89 100644
--- a/heron/api/src/java/com/twitter/heron/api/bolt/WindowedBoltExecutor.java
+++ b/heron/api/src/java/com/twitter/heron/api/bolt/WindowedBoltExecutor.java
@@ -71,7 +71,7 @@
   private transient EvictionPolicy<Tuple> evictionPolicy;
   private transient Long windowLengthDurationMs;
   // package level for unit tests
-  private transient WaterMarkEventGenerator<Tuple> waterMarkEventGenerator;
+  protected transient WaterMarkEventGenerator<Tuple> waterMarkEventGenerator;
 
   public WindowedBoltExecutor(IWindowedBolt bolt) {
     this.bolt = bolt;
@@ -150,7 +150,6 @@
                                                      lifecycleListener, Map<String, Object>
       topoConf, TopologyContext context, Collection<Event<Tuple>> queue) {
 
-
     WindowManager<Tuple> manager = new WindowManager<>(lifecycleListener, queue);
 
     Count windowLengthCount = null;
diff --git a/heron/api/src/java/com/twitter/heron/api/windowing/WindowingConfigs.java b/heron/api/src/java/com/twitter/heron/api/windowing/WindowingConfigs.java
index fd71ba5..d60de33 100644
--- a/heron/api/src/java/com/twitter/heron/api/windowing/WindowingConfigs.java
+++ b/heron/api/src/java/com/twitter/heron/api/windowing/WindowingConfigs.java
@@ -14,6 +14,7 @@
 package com.twitter.heron.api.windowing;
 
 import java.util.HashMap;
+import java.util.Map;
 
 public class WindowingConfigs extends HashMap<String, Object> {
 
@@ -69,4 +70,62 @@
      */
   public static final String TOPOLOGY_BOLTS_WATERMARK_EVENT_INTERVAL_MS = "topology.bolts"
       + ".watermark.event.interval.ms";
+
+  public void setTopologyBoltsWindowLengthCount(long value) {
+    setTopologyBoltsWindowLengthCount(this, value);
+  }
+
+  public static void setTopologyBoltsWindowLengthCount(Map<String, Object> conf, long value) {
+    conf.put(TOPOLOGY_BOLTS_WINDOW_LENGTH_COUNT, value);
+  }
+
+  public void setTopologyBoltsWindowLengthDurationMs(long value) {
+    setTopologyBoltsWindowLengthDurationMs(this, value);
+  }
+
+  public static void setTopologyBoltsWindowLengthDurationMs(Map<String, Object> conf, long value) {
+    conf.put(TOPOLOGY_BOLTS_WINDOW_LENGTH_DURATION_MS, value);
+  }
+
+  public void setTopologyBoltsSlidingIntervalCount(long value) {
+    setTopologyBoltsSlidingIntervalCount(this, value);
+  }
+
+  public static void setTopologyBoltsSlidingIntervalCount(Map<String, Object> conf, long value) {
+    conf.put(TOPOLOGY_BOLTS_SLIDING_INTERVAL_COUNT, value);
+  }
+
+  public void setTopologyBoltsSlidingIntervalDurationMs(long value) {
+    setTopologyBoltsSlidingIntervalDurationMs(this, value);
+  }
+
+  public static void setTopologyBoltsSlidingIntervalDurationMs(
+      Map<String, Object> conf, long value) {
+    conf.put(TOPOLOGY_BOLTS_SLIDING_INTERVAL_DURATION_MS, value);
+  }
+
+  public void setTopologyBoltsLateTupleStream(String value) {
+    setTopologyBoltsLateTupleStream(this, value);
+  }
+
+  public static void setTopologyBoltsLateTupleStream(Map<String, Object> conf, String value) {
+    conf.put(TOPOLOGY_BOLTS_LATE_TUPLE_STREAM, value);
+  }
+
+  public void setTopologyBoltsTupleTimestampMaxLagMs(long value) {
+    setTopologyBoltsTupleTimestampMaxLagMs(this, value);
+  }
+
+  public static void setTopologyBoltsTupleTimestampMaxLagMs(Map<String, Object> conf, long value) {
+    conf.put(TOPOLOGY_BOLTS_TUPLE_TIMESTAMP_MAX_LAG_MS, value);
+  }
+
+  public void setTopologyBoltsWatermarkEventIntervalMs(long value) {
+    setTopologyBoltsWatermarkEventIntervalMs(this, value);
+  }
+
+  public static void setTopologyBoltsWatermarkEventIntervalMs(
+      Map<String, Object> conf, long value) {
+    conf.put(TOPOLOGY_BOLTS_WATERMARK_EVENT_INTERVAL_MS, value);
+  }
 }
diff --git a/heron/api/tests/java/BUILD b/heron/api/tests/java/BUILD
index 9919e11..fcff382 100644
--- a/heron/api/tests/java/BUILD
+++ b/heron/api/tests/java/BUILD
@@ -3,7 +3,9 @@
 
 api_deps_files = [
     "//heron/api/src/java:api-java",
+    "//heron/common/src/java:utils-java",
     "//third_party/java:junit4",
+    "@org_mockito_mockito_all//jar"
 ]
 
 api_deps_files = \
@@ -19,7 +21,9 @@
 java_tests(
   test_classes = [
     "com.twitter.heron.api.windowing.WindowManagerTest",
-    "com.twitter.heron.api.windowing.WaterMarkEventGeneratorTest"
+    "com.twitter.heron.api.windowing.WaterMarkEventGeneratorTest",
+    "com.twitter.heron.api.bolt.WindowedBoltExecutorTest",
+    "com.twitter.heron.api.bolt.BaseWindowedBoltTest"
   ],
   runtime_deps = [ ":api-tests" ],
   size = "small",
diff --git a/heron/api/tests/java/com/twitter/heron/api/bolt/BaseWindowedBoltTest.java b/heron/api/tests/java/com/twitter/heron/api/bolt/BaseWindowedBoltTest.java
new file mode 100644
index 0000000..222e7d0
--- /dev/null
+++ b/heron/api/tests/java/com/twitter/heron/api/bolt/BaseWindowedBoltTest.java
@@ -0,0 +1,276 @@
+//  Copyright 2017 Twitter. All rights reserved.
+//
+//  Licensed 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.twitter.heron.api.bolt;
+
+import java.time.Duration;
+
+import org.junit.Test;
+
+import com.twitter.heron.api.topology.TopologyBuilder;
+import com.twitter.heron.api.windowing.TupleWindow;
+
+import static org.junit.Assert.fail;
+
+/**
+ * Unit tests for {@link BaseWindowedBolt}
+ */
+public class BaseWindowedBoltTest {
+
+  public static class TestBolt extends BaseWindowedBolt {
+
+    private static final long serialVersionUID = -7224073487836212922L;
+
+    @Override
+    public void execute(TupleWindow inputWindow) {
+
+    }
+  }
+
+  @Test
+  public void testSettingSlidingCountWindow() {
+    final Object[][] args = new Object[][]{
+        {-1, 10},
+        {10, -1},
+        {0, 10},
+        {10, 0},
+        {0, 0},
+        {-1, -1},
+        {5, 10},
+        {1, 1},
+        {10, 5},
+        {100, 10},
+        {100, 100},
+        {200, 100},
+        {500, 100},
+        {null, null},
+        {null, 1},
+        {1, null},
+        {null, -1},
+        {-1, null}
+    };
+
+    for (Object[] arg : args) {
+      TopologyBuilder builder = new TopologyBuilder();
+      Object arg0 = arg[0];
+      Object arg1 = arg[1];
+      try {
+
+        BaseWindowedBolt.Count windowLengthCount = null;
+        if (arg0 != null) {
+          windowLengthCount = BaseWindowedBolt.Count.of((Integer) arg0);
+        }
+        BaseWindowedBolt.Count slidingIntervalCount = null;
+
+        if (arg1 != null) {
+          slidingIntervalCount = BaseWindowedBolt.Count.of((Integer) arg1);
+        }
+
+        builder.setBolt("testBolt", new TestBolt().withWindow(windowLengthCount,
+            slidingIntervalCount));
+        if (arg0 == null || arg1 == null) {
+          fail(String.format("Window length or sliding window length cannot be null -- "
+              + "windowLengthCount: %s slidingIntervalCount: %s", arg0, arg1));
+        }
+        if ((Integer) arg0 <= 0 || (Integer) arg1 <= 0) {
+          fail(String.format("Window length or sliding window length cannot be zero or less -- "
+              + "windowLengthCount: %s slidingIntervalCount: %s", arg0, arg1));
+        }
+      } catch (IllegalArgumentException e) {
+        if (arg0 != null && arg1 != null && (Integer) arg0 > 0 && (Integer) arg1 > 0) {
+          fail(String.format("Exception: %s thrown on valid input -- windowLengthCount: %s "
+              + "slidingIntervalCount: %s", e.getMessage(), arg0, arg1));
+        }
+      }
+    }
+  }
+
+  @Test
+  public void testSettingSlidingTimeWindow() {
+    final Object[][] args = new Object[][]{
+        {-1L, 10L},
+        {10L, -1L},
+        {0L, 10L},
+        {10L, 0L},
+        {0L, 0L},
+        {-1L, -1L},
+        {5L, 10L},
+        {1L, 1L},
+        {10L, 5L},
+        {100L, 10L},
+        {100L, 100L},
+        {200L, 100L},
+        {500L, 100L},
+        {null, null},
+        {null, 1L},
+        {1L, null},
+        {null, -1L},
+        {-1L, null}
+    };
+
+    for (Object[] arg : args) {
+      TopologyBuilder builder = new TopologyBuilder();
+      Object arg0 = arg[0];
+      Object arg1 = arg[1];
+      try {
+
+        Duration windowLengthDuration = null;
+        if (arg0 != null) {
+          windowLengthDuration = Duration.ofMillis((Long) arg0);
+        }
+        Duration slidingIntervalDuration = null;
+
+        if (arg1 != null) {
+          slidingIntervalDuration = Duration.ofMillis((Long) arg1);
+        }
+
+        builder.setBolt("testBolt", new TestBolt().withWindow(windowLengthDuration,
+            slidingIntervalDuration));
+        if (arg0 == null || arg1 == null) {
+          fail(String.format("Window length or sliding window length cannot be null -- "
+              + "windowLengthDuration: %s slidingIntervalDuration: %s", arg0, arg1));
+        }
+        if ((Long) arg0 <= 0 || (Long) arg1 <= 0) {
+          fail(String.format("Window length or sliding window length cannot be zero or less -- "
+              + "windowLengthDuration: %s slidingIntervalDuration: %s", arg0, arg1));
+        }
+      } catch (IllegalArgumentException e) {
+        if (arg0 != null && arg1 != null && (Long) arg0 > 0 && (Long) arg1 > 0) {
+          fail(String.format("Exception: %s thrown on valid input -- windowLengthDuration: %s "
+              + "slidingIntervalDuration: %s", e.getMessage(), arg0, arg1));
+        }
+      }
+    }
+  }
+
+  @Test
+  public void testSettingTumblingCountWindow() {
+    final Object[] args = new Object[] {-1, 0, 1, 2, 5, 10, null};
+
+    for (Object arg : args) {
+      TopologyBuilder builder = new TopologyBuilder();
+      Object arg0 = arg;
+      try {
+
+        BaseWindowedBolt.Count windowLengthCount = null;
+        if (arg0 != null) {
+          windowLengthCount = BaseWindowedBolt.Count.of((Integer) arg0);
+        }
+
+        builder.setBolt("testBolt", new TestBolt().withTumblingWindow(windowLengthCount));
+        if (arg0 == null) {
+          fail(String.format("Window length cannot be null -- windowLengthCount: %s", arg0));
+        }
+        if ((Integer) arg0 <= 0) {
+          fail(String.format("Window length cannot be zero or less -- windowLengthCount: %s",
+              arg0));
+        }
+      } catch (IllegalArgumentException e) {
+        if (arg0 != null && (Integer) arg0 > 0) {
+          fail(String.format("Exception: %s thrown on valid input -- windowLengthCount: %s", e
+              .getMessage(), arg0));
+        }
+      }
+    }
+  }
+
+  @Test
+  public void testSettingTumblingTimeWindow() {
+    final Object[] args = new Object[]{-1L, 0L, 1L, 2L, 5L, 10L, null};
+    for (Object arg : args) {
+      TopologyBuilder builder = new TopologyBuilder();
+      Object arg0 = arg;
+      try {
+
+        Duration windowLengthDuration = null;
+        if (arg0 != null) {
+          windowLengthDuration = Duration.ofMillis((Long) arg0);
+        }
+
+        builder.setBolt("testBolt", new TestBolt().withTumblingWindow(windowLengthDuration));
+        if (arg0 == null) {
+          fail(String.format("Window count duration cannot be null -- windowLengthDuration: %s",
+              arg0));
+        }
+        if ((Long) arg0 <= 0) {
+          fail(String.format("Window length cannot be zero or less -- windowLengthDuration: %s",
+              arg0));
+        }
+      } catch (IllegalArgumentException e) {
+        if (arg0 != null && (Long) arg0 > 0) {
+          fail(String.format("Exception: %s thrown on valid input -- windowLengthDuration: %s", e
+              .getMessage(), arg0));
+        }
+      }
+    }
+  }
+
+  @Test
+  public void testSettingLagTime() {
+    final Object[] args = new Object[]{-1L, 0L, 1L, 2L, 5L, 10L, null};
+    for (Object arg : args) {
+      TopologyBuilder builder = new TopologyBuilder();
+      Object arg0 = arg;
+      try {
+
+        Duration lagTime = null;
+        if (arg0 != null) {
+          lagTime = Duration.ofMillis((Long) arg0);
+        }
+
+        builder.setBolt("testBolt", new TestBolt().withLag(lagTime));
+        if (arg0 == null) {
+          fail(String.format("Window lag duration cannot be null -- lagTime: %s", arg0));
+        }
+        if ((Long) arg0 <= 0) {
+          fail(String.format("Window lag cannot be zero or less -- lagTime: %s", arg0));
+        }
+      } catch (IllegalArgumentException e) {
+        if (arg0 != null && (Long) arg0 > 0) {
+          fail(String.format("Exception: %s thrown on valid input -- lagTime: %s",
+              e.getMessage(), arg0));
+        }
+      }
+    }
+  }
+
+  @Test
+  public void testSettingWaterMarkInterval() {
+    final Object[] args = new Object[]{-1L, 0L, 1L, 2L, 5L, 10L, null};
+    for (Object arg : args) {
+      TopologyBuilder builder = new TopologyBuilder();
+      Object arg0 = arg;
+      try {
+
+        Duration watermarkInterval = null;
+        if (arg0 != null) {
+          watermarkInterval = Duration.ofMillis((Long) arg0);
+        }
+
+        builder.setBolt("testBolt", new TestBolt().withWatermarkInterval(watermarkInterval));
+        if (arg0 == null) {
+          fail(String.format("Watermark interval cannot be null -- watermarkInterval: %s", arg0));
+        }
+        if ((Long) arg0 <= 0) {
+          fail(String.format("Watermark interval cannot be zero or less -- watermarkInterval: "
+              + "%s", arg0));
+        }
+      } catch (IllegalArgumentException e) {
+        if (arg0 != null && (Long) arg0 > 0) {
+          fail(String.format("Exception: %s thrown on valid input -- watermarkInterval: %s", e
+              .getMessage(), arg0));
+        }
+      }
+    }
+  }
+}
diff --git a/heron/api/tests/java/com/twitter/heron/api/bolt/WindowedBoltExecutorTest.java b/heron/api/tests/java/com/twitter/heron/api/bolt/WindowedBoltExecutorTest.java
new file mode 100644
index 0000000..52a21d3
--- /dev/null
+++ b/heron/api/tests/java/com/twitter/heron/api/bolt/WindowedBoltExecutorTest.java
@@ -0,0 +1,241 @@
+//  Copyright 2017 Twitter. All rights reserved.
+//
+//  Licensed 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.twitter.heron.api.bolt;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import com.twitter.heron.api.Config;
+import com.twitter.heron.api.generated.TopologyAPI;
+import com.twitter.heron.api.topology.TopologyBuilder;
+import com.twitter.heron.api.topology.TopologyContext;
+import com.twitter.heron.api.tuple.Fields;
+import com.twitter.heron.api.tuple.Tuple;
+import com.twitter.heron.api.tuple.Values;
+import com.twitter.heron.api.windowing.TupleWindow;
+import com.twitter.heron.api.windowing.WindowingConfigs;
+import com.twitter.heron.common.utils.topology.TopologyContextImpl;
+import com.twitter.heron.common.utils.tuple.TupleImpl;
+
+import static org.junit.Assert.*;
+
+/**
+ * Unit tests for {@link WindowedBoltExecutor}
+ */
+public class WindowedBoltExecutorTest {
+
+  private WindowedBoltExecutor executor;
+  private TestWindowedBolt testWindowedBolt;
+
+  @SuppressWarnings("VisibilityModifier")
+  private static class TestWindowedBolt extends BaseWindowedBolt {
+    private static final long serialVersionUID = -8934326157586387333L;
+    List<TupleWindow> tupleWindows = new ArrayList<>();
+
+    @Override
+    public void execute(TupleWindow input) {
+      //System.out.println(input);
+      tupleWindows.add(input);
+    }
+  }
+
+  @SuppressWarnings({"rawtypes", "unchecked"})
+  private TopologyContext getContext(final Fields fields) {
+    TopologyBuilder builder = new TopologyBuilder();
+    return new TopologyContextImpl(new Config(),
+        builder.createTopology()
+            .setConfig(new Config())
+            .setName("test")
+            .setState(TopologyAPI.TopologyState.RUNNING)
+            .getTopology(),
+        new HashMap(), 1, null) {
+      @Override
+      public Fields getComponentOutputFields(String componentId, String streamId) {
+        return fields;
+      }
+
+    };
+  }
+
+  private Tuple getTuple(String streamId, final Fields fields, Values values) {
+
+    TopologyContext topologyContext = getContext(fields);
+    return new TupleImpl(topologyContext, TopologyAPI.StreamId.newBuilder()
+        .setId(streamId).setComponentName("s1")
+        .build(), 0, null, values, 1) {
+      @Override
+      public TopologyAPI.StreamId getSourceGlobalStreamId() {
+        return TopologyAPI.StreamId.newBuilder().setComponentName("s1").setId("default").build();
+      }
+    };
+  }
+
+  private OutputCollector getOutputCollector() {
+    return Mockito.mock(OutputCollector.class);
+  }
+
+  private TopologyContext getTopologyContext() {
+    TopologyContext context = Mockito.mock(TopologyContext.class);
+
+    Map<TopologyAPI.StreamId, TopologyAPI.Grouping> sources =
+        Collections.singletonMap(TopologyAPI.StreamId.newBuilder()
+            .setComponentName("s1").setId("default").build(), null);
+    Mockito.when(context.getThisSources()).thenReturn(sources);
+    return context;
+  }
+
+  @Before
+  public void setUp() {
+    testWindowedBolt = new TestWindowedBolt();
+    testWindowedBolt.withTimestampField("ts");
+    executor = new WindowedBoltExecutor(testWindowedBolt);
+    Map<String, Object> conf = new HashMap<>();
+    conf.put(Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS, 100000);
+    conf.put(WindowingConfigs.TOPOLOGY_BOLTS_WINDOW_LENGTH_DURATION_MS, 20L);
+    conf.put(WindowingConfigs.TOPOLOGY_BOLTS_SLIDING_INTERVAL_DURATION_MS, 10L);
+    conf.put(WindowingConfigs.TOPOLOGY_BOLTS_TUPLE_TIMESTAMP_MAX_LAG_MS, 5L);
+    // trigger manually to avoid timing issues
+    conf.put(WindowingConfigs.TOPOLOGY_BOLTS_WATERMARK_EVENT_INTERVAL_MS, 100000L);
+    executor.prepare(conf, getTopologyContext(), getOutputCollector());
+  }
+
+  @Test(expected = IllegalArgumentException.class)
+  public void testExecuteWithoutTs() throws Exception {
+    executor.execute(getTuple("s1", new Fields("a"), new Values(1)));
+  }
+
+  @Test
+  public void testExecuteWithTs() throws Exception {
+    long[] timestamps = {603, 605, 607, 618, 626, 636};
+    for (long ts : timestamps) {
+      executor.execute(getTuple("s1", new Fields("ts"), new Values(ts)));
+    }
+    //Thread.sleep(120);
+    executor.waterMarkEventGenerator.run();
+    //System.out.println(testWindowedBolt.tupleWindows);
+    assertEquals(3, testWindowedBolt.tupleWindows.size());
+    TupleWindow first = testWindowedBolt.tupleWindows.get(0);
+    assertArrayEquals(
+        new long[]{603, 605, 607},
+        new long[]{(long) first.get().get(0).getValue(0),
+            (long) first.get().get(1).getValue(0),
+            (long) first.get().get(2).getValue(0)});
+
+    TupleWindow second = testWindowedBolt.tupleWindows.get(1);
+    assertArrayEquals(
+        new long[]{603, 605, 607, 618},
+        new long[]{(long) second.get().get(0).getValue(0),
+            (long) second.get().get(1).getValue(0),
+            (long) second.get().get(2).getValue(0),
+            (long) second.get().get(3).getValue(0)});
+
+    TupleWindow third = testWindowedBolt.tupleWindows.get(2);
+    assertArrayEquals(new long[]{618, 626}, new long[]{(long) third.get().get(0).getValue(0),
+        (long) third.get().get(1).getValue(0)});
+  }
+
+  @Test
+  public void testPrepareLateTupleStreamWithoutTs() throws Exception {
+    Map<String, Object> conf = new HashMap<>();
+    conf.put(Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS, 100000);
+    conf.put(WindowingConfigs.TOPOLOGY_BOLTS_WINDOW_LENGTH_DURATION_MS, 20L);
+    conf.put(WindowingConfigs.TOPOLOGY_BOLTS_SLIDING_INTERVAL_DURATION_MS, 10L);
+    conf.put(WindowingConfigs.TOPOLOGY_BOLTS_LATE_TUPLE_STREAM, "$late");
+    conf.put(WindowingConfigs.TOPOLOGY_BOLTS_TUPLE_TIMESTAMP_MAX_LAG_MS, 5L);
+    conf.put(WindowingConfigs.TOPOLOGY_BOLTS_WATERMARK_EVENT_INTERVAL_MS, 10L);
+
+    testWindowedBolt = new TestWindowedBolt();
+    executor = new WindowedBoltExecutor(testWindowedBolt);
+    TopologyContext context = getTopologyContext();
+    // emulate the call of withLateTupleStream method
+    Mockito.when(context.getThisStreams()).thenReturn(new HashSet<>(Arrays.asList("default",
+        "$late")));
+    try {
+      executor.prepare(conf, context, getOutputCollector());
+      fail();
+    } catch (IllegalArgumentException e) {
+      assertEquals(e.getMessage(), "Late tuple stream can be defined only when specifying a "
+          + "timestamp field");
+    }
+  }
+
+  @Test
+  public void testPrepareLateTupleStreamWithoutBuilder() throws Exception {
+    Map<String, Object> conf = new HashMap<>();
+    conf.put(Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS, 100000);
+    conf.put(WindowingConfigs.TOPOLOGY_BOLTS_WINDOW_LENGTH_DURATION_MS, 20L);
+    conf.put(WindowingConfigs.TOPOLOGY_BOLTS_SLIDING_INTERVAL_DURATION_MS, 10L);
+    conf.put(WindowingConfigs.TOPOLOGY_BOLTS_LATE_TUPLE_STREAM, "$late");
+    conf.put(WindowingConfigs.TOPOLOGY_BOLTS_TUPLE_TIMESTAMP_MAX_LAG_MS, 5L);
+    conf.put(WindowingConfigs.TOPOLOGY_BOLTS_WATERMARK_EVENT_INTERVAL_MS, 10L);
+
+    testWindowedBolt = new TestWindowedBolt();
+    testWindowedBolt.withTimestampField("ts");
+    executor = new WindowedBoltExecutor(testWindowedBolt);
+    TopologyContext context = getTopologyContext();
+    try {
+      executor.prepare(conf, context, getOutputCollector());
+      fail();
+    } catch (IllegalArgumentException e) {
+      assertEquals(e.getMessage(), "Stream for late tuples must be defined with the builder "
+          + "method withLateTupleStream");
+    }
+  }
+
+
+  @Test
+  public void testExecuteWithLateTupleStream() throws Exception {
+    testWindowedBolt = new TestWindowedBolt();
+    testWindowedBolt.withTimestampField("ts");
+    executor = new WindowedBoltExecutor(testWindowedBolt);
+    TopologyContext context = getTopologyContext();
+    Mockito.when(context.getThisStreams()).thenReturn(new HashSet<>(Arrays.asList("default",
+        "$late")));
+
+    OutputCollector outputCollector = Mockito.mock(OutputCollector.class);
+    Map<String, Object> conf = new HashMap<>();
+    conf.put(Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS, 100000);
+    conf.put(WindowingConfigs.TOPOLOGY_BOLTS_WINDOW_LENGTH_DURATION_MS, 20L);
+    conf.put(WindowingConfigs.TOPOLOGY_BOLTS_SLIDING_INTERVAL_DURATION_MS, 10L);
+    conf.put(WindowingConfigs.TOPOLOGY_BOLTS_LATE_TUPLE_STREAM, "$late");
+    conf.put(WindowingConfigs.TOPOLOGY_BOLTS_TUPLE_TIMESTAMP_MAX_LAG_MS, 5L);
+    //Trigger manually to avoid timing issues
+    conf.put(WindowingConfigs.TOPOLOGY_BOLTS_WATERMARK_EVENT_INTERVAL_MS, 1_000_000L);
+    executor.prepare(conf, context, outputCollector);
+
+    long[] timestamps = {603, 605, 607, 618, 626, 636, 600};
+    List<Tuple> tuples = new ArrayList<>(timestamps.length);
+
+    for (long ts : timestamps) {
+      Tuple tuple = getTuple("s1", new Fields("ts"), new Values(ts));
+      tuples.add(tuple);
+      executor.execute(tuple);
+
+      //Update the watermark to this timestamp
+      executor.waterMarkEventGenerator.run();
+    }
+    System.out.println(testWindowedBolt.tupleWindows);
+    Tuple tuple = tuples.get(tuples.size() - 1);
+    Mockito.verify(outputCollector).emit("$late", Arrays.asList(tuple), new Values(tuple));
+  }
+}
diff --git a/heron/common/src/cpp/zookeeper/zkclient.cpp b/heron/common/src/cpp/zookeeper/zkclient.cpp
index 0fbfae6..6633520 100644
--- a/heron/common/src/cpp/zookeeper/zkclient.cpp
+++ b/heron/common/src/cpp/zookeeper/zkclient.cpp
@@ -136,32 +136,14 @@
 }
 
 void ZKClient::Init() {
-  zkaction_responses_ = new PCQueue<CallBack*>();
-  auto zkaction_response_cb = [this](EventLoop::Status status) {
-    this->OnZkActionResponse(status);
-  };
-
-  if (pipe(pipers_) < 0) {
-    LOG(FATAL) << "Pipe failed in ZKClient";
-  }
-  sp_int32 flags;
-  if ((flags = fcntl(pipers_[0], F_GETFL, 0)) < 0 ||
-      fcntl(pipers_[0], F_SETFL, flags | O_NONBLOCK) < 0 ||
-      eventLoop_->registerForRead(pipers_[0], std::move(zkaction_response_cb), true) != 0) {
-    LOG(FATAL) << "fcntl failed in ZKClient";
-  }
+  piper_ = new Piper(eventLoop_);
   zoo_deterministic_conn_order(0);  // even distribution of clients on the server
   InitZKHandle();
 }
 
 // Destructor.
 ZKClient::~ZKClient() {
-  if (eventLoop_) {
-    CHECK_EQ(eventLoop_->unRegisterForRead(pipers_[0]), 0);
-  }
-  close(pipers_[0]);
-  close(pipers_[1]);
-  delete zkaction_responses_;
+  delete piper_;
   zookeeper_close(zk_handle_);
 }
 
@@ -350,55 +332,17 @@
   }
 }
 
-void ZKClient::SignalMainThread() {
-  // This need not be protected by any mutex.
-  // The os will take care of that.
-  int rc = write(pipers_[1], "a", 1);
-  if (rc != 1) {
-    LOG(FATAL) << "Write to pipe failed in ZkClient with return code:  " << rc;
-  }
-}
-
-void ZKClient::OnZkActionResponse(EventLoop::Status _status) {
-  if (_status == EventLoop::READ_EVENT) {
-    char buf[1];
-    ssize_t readcount = read(pipers_[0], buf, 1);
-    if (readcount == 1) {
-      bool dequeued = false;
-      CallBack* cb = zkaction_responses_->trydequeue(dequeued);
-      if (cb) {
-        cb->Run();
-      }
-    } else {
-      LOG(ERROR) << "In Server read from pipers returned " << readcount << " errno " << errno
-                 << std::endl;
-      if (readcount < 0 && (errno == EAGAIN || errno == EINTR)) {
-        // Never mind. we will try again
-        return;
-      } else {
-        // We really don't know what to do here.
-        // TODO(kramasamy): Figure out a way to get the hell out of here
-        return;
-      }
-    }
-  }
-  return;
-}
-
 void ZKClient::ZkActionCb(sp_int32 rc, VCallback<sp_int32> cb) {
-  zkaction_responses_->enqueue(CreateCallback(&RunUserCb, rc, std::move(cb)));
-  SignalMainThread();
+  piper_->ExecuteInEventLoop(std::bind(&RunUserCb, rc, std::move(cb)));
 }
 
 void ZKClient::ZkWatcherCb(VCallback<> cb) {
-  zkaction_responses_->enqueue(CreateCallback(&RunWatcherCb, std::move(cb)));
-  SignalMainThread();
+  piper_->ExecuteInEventLoop(std::bind(&RunWatcherCb, std::move(cb)));
 }
 
 void ZKClient::SendWatchEvent(const ZkWatchEvent& event) {
   CHECK(client_global_watcher_cb_);
-  zkaction_responses_->enqueue(CreateCallback(&RunWatchEventCb, client_global_watcher_cb_, event));
-  SignalMainThread();
+  piper_->ExecuteInEventLoop(std::bind(&RunWatchEventCb, client_global_watcher_cb_, event));
 }
 
 const std::string ZKClient::state2String(sp_int32 _state) {
diff --git a/heron/common/src/cpp/zookeeper/zkclient.h b/heron/common/src/cpp/zookeeper/zkclient.h
index f076294..2ae6803 100644
--- a/heron/common/src/cpp/zookeeper/zkclient.h
+++ b/heron/common/src/cpp/zookeeper/zkclient.h
@@ -118,7 +118,6 @@
   ZKClient()
       : zk_handle_(NULL),
         eventLoop_(NULL),
-        zkaction_responses_(NULL),
         client_global_watcher_cb_(VCallback<ZkWatchEvent>()) {}
 
  private:
@@ -128,17 +127,10 @@
   // The function that actually inits the handle
   void InitZKHandle();
 
-  // This is the function used to signal the main thread
-  void SignalMainThread();
-
-  // When the zk callback wants to wake the main thread, it uses the SignalMainThread function.
-  // This function will get executed in the main thread.
-  void OnZkActionResponse(EventLoop::Status _status);
-
   // We wrap all user zk calls with this completion function
   // This completion function runs in the context of the
-  // zk completion thread. It basically appends to
-  // zkaction_responses_ and calls SignalMainThread
+  // zk completion thread. It basically calls the piper
+  // to execute the cb in eventLoop thread
   void ZkActionCb(sp_int32 rc, VCallback<sp_int32> cb);
 
   // This is the watcher function that gets called
@@ -157,9 +149,8 @@
 
   // We use libzookeeper_mt as our zk library. This means that
   // zk callbacks are all executed in the context of a zk thread.
-  // These pipers are how they communicate it accross to our thread
-  sp_int32 pipers_[2];
-  PCQueue<CallBack*>* zkaction_responses_;
+  // Piper are how they communicate it accross to our main thread
+  Piper* piper_;
   // A callback to notify the clients of this class about global session events.
   VCallback<ZkWatchEvent> client_global_watcher_cb_;
 };
diff --git a/heron/common/src/java/com/twitter/heron/common/basics/NIOLooper.java b/heron/common/src/java/com/twitter/heron/common/basics/NIOLooper.java
index b5e175e..3b31eb1 100644
--- a/heron/common/src/java/com/twitter/heron/common/basics/NIOLooper.java
+++ b/heron/common/src/java/com/twitter/heron/common/basics/NIOLooper.java
@@ -208,6 +208,7 @@
                            int operation,
                            ISelectHandler callback)
       throws ClosedChannelException {
+
     SelectionKey key = channel.keyFor(selector);
 
     if (key == null) {
diff --git a/heron/stmgr/tests/cpp/server/dummy_instance.cpp b/heron/stmgr/tests/cpp/server/dummy_instance.cpp
index 09d9fbc..d7134bb 100644
--- a/heron/stmgr/tests/cpp/server/dummy_instance.cpp
+++ b/heron/stmgr/tests/cpp/server/dummy_instance.cpp
@@ -40,7 +40,8 @@
       stmgr_id_(_stmgr_id),
       recvd_stmgr_pplan_(NULL),
       register_response_status(heron::proto::system::STMGR_DIDNT_REGISTER) {
-  InstallMessageHandler(&DummyInstance::HandleInstanceResponse);
+  InstallResponseHandler(new heron::proto::stmgr::RegisterInstanceRequest(),
+                         &DummyInstance::HandleInstanceResponse);
   InstallMessageHandler(&DummyInstance::HandleTupleMessage);
   InstallMessageHandler(&DummyInstance::HandleNewInstanceAssignmentMsg);
 
@@ -62,14 +63,18 @@
   }
 }
 
-void DummyInstance::HandleClose(NetworkErrorCode) {}
+void DummyInstance::HandleClose(NetworkErrorCode) {
+  AddTimer(retry_cb_, 100);
+}
 
 heron::proto::system::StatusCode DummyInstance::GetRegisterResponseStatus() {
   return register_response_status;
 }
 
-void DummyInstance::HandleInstanceResponse(
-    heron::proto::stmgr::RegisterInstanceResponse* _message) {
+void DummyInstance::HandleInstanceResponse(void*,
+    heron::proto::stmgr::RegisterInstanceResponse* _message,
+    NetworkErrorCode status) {
+  CHECK_EQ(status, OK);
   if (_message->has_pplan()) {
     if (recvd_stmgr_pplan_) {
       delete recvd_stmgr_pplan_;
@@ -87,21 +92,19 @@
     heron::proto::stmgr::NewInstanceAssignmentMessage*) {}
 
 void DummyInstance::CreateAndSendInstanceRequest() {
-  heron::proto::stmgr::RegisterInstanceRequest message;
-  heron::proto::system::Instance* instance = message.mutable_instance();
+  auto request = new heron::proto::stmgr::RegisterInstanceRequest();
+  heron::proto::system::Instance* instance = request->mutable_instance();
   instance->set_instance_id(instance_id_);
   instance->set_stmgr_id(stmgr_id_);
   instance->mutable_info()->set_task_id(task_id_);
   instance->mutable_info()->set_component_index(component_index_);
   instance->mutable_info()->set_component_name(component_name_);
-  message.set_topology_name(topology_name_);
-  message.set_topology_id(topology_id_);
-  SendMessage(message);
+  request->set_topology_name(topology_name_);
+  request->set_topology_id(topology_id_);
+  SendRequest(request, nullptr);
   return;
 }
 
-void DummyInstance::CreateAndSendTupleMessages() {}
-
 //////////////////////////////////////// DummySpoutInstance ////////////////////////////////////
 DummySpoutInstance::DummySpoutInstance(EventLoopImpl* eventLoop, const NetworkOptions& _options,
                                        const sp_string& _topology_name,
@@ -116,12 +119,8 @@
       max_msgs_to_send_(max_msgs_to_send),
       total_msgs_sent_(0),
       batch_size_(1000),
-      do_custom_grouping_(_do_custom_grouping) {}
-
-void DummySpoutInstance::HandleInstanceResponse(
-    heron::proto::stmgr::RegisterInstanceResponse* _message) {
-  DummyInstance::HandleInstanceResponse(_message);
-}
+      do_custom_grouping_(_do_custom_grouping),
+      under_backpressure_(false) {}
 
 void DummySpoutInstance::HandleNewInstanceAssignmentMsg(
     heron::proto::stmgr::NewInstanceAssignmentMessage* _msg) {
@@ -140,23 +139,25 @@
 }
 
 void DummySpoutInstance::CreateAndSendTupleMessages() {
-  for (int i = 0; (i < batch_size_) && (total_msgs_sent_ < max_msgs_to_send_);
-       ++total_msgs_sent_, ++i) {
-    heron::proto::system::HeronTupleSet tuple_set;
-    heron::proto::system::HeronDataTupleSet* data_set = tuple_set.mutable_data();
-    heron::proto::api::StreamId* tstream = data_set->mutable_stream();
-    tstream->set_id(stream_id_);
-    tstream->set_component_name(component_name_);
-    heron::proto::system::HeronDataTuple* tuple = data_set->add_tuples();
-    tuple->set_key(0);
-    // Add lots of data
-    for (size_t i = 0; i < 500; ++i) *(tuple->add_values()) = "dummy data";
+  if (!under_backpressure_) {
+    for (int i = 0; (i < batch_size_) && (total_msgs_sent_ < max_msgs_to_send_);
+         ++total_msgs_sent_, ++i) {
+      heron::proto::system::HeronTupleSet tuple_set;
+      heron::proto::system::HeronDataTupleSet* data_set = tuple_set.mutable_data();
+      heron::proto::api::StreamId* tstream = data_set->mutable_stream();
+      tstream->set_id(stream_id_);
+      tstream->set_component_name(component_name_);
+      heron::proto::system::HeronDataTuple* tuple = data_set->add_tuples();
+      tuple->set_key(0);
+      // Add lots of data
+      for (size_t i = 0; i < 500; ++i) *(tuple->add_values()) = "dummy data";
 
-    // Add custom grouping if need be
-    if (do_custom_grouping_) {
-      tuple->add_dest_task_ids(custom_grouping_dest_task_);
+      // Add custom grouping if need be
+      if (do_custom_grouping_) {
+        tuple->add_dest_task_ids(custom_grouping_dest_task_);
+      }
+      SendMessage(tuple_set);
     }
-    SendMessage(tuple_set);
   }
   if (total_msgs_sent_ != max_msgs_to_send_) {
     AddTimer([this]() { this->CreateAndSendTupleMessages(); }, 1000);
@@ -175,11 +176,6 @@
       expected_msgs_to_recv_(_expected_msgs_to_recv),
       msgs_recvd_(0) {}
 
-void DummyBoltInstance::HandleInstanceResponse(
-    heron::proto::stmgr::RegisterInstanceResponse* _message) {
-  DummyInstance::HandleInstanceResponse(_message);
-}
-
 void DummyBoltInstance::HandleTupleMessage(heron::proto::system::HeronTupleSet2* msg) {
   if (msg->has_data()) msgs_recvd_ += msg->mutable_data()->tuples_size();
   if (msgs_recvd_ >= expected_msgs_to_recv_) getEventLoop()->loopExit();
diff --git a/heron/stmgr/tests/cpp/server/dummy_instance.h b/heron/stmgr/tests/cpp/server/dummy_instance.h
index 9139d19..919413a 100644
--- a/heron/stmgr/tests/cpp/server/dummy_instance.h
+++ b/heron/stmgr/tests/cpp/server/dummy_instance.h
@@ -36,11 +36,11 @@
   void Retry() { Start(); }
 
   // Handle incoming message
-  virtual void HandleInstanceResponse(heron::proto::stmgr::RegisterInstanceResponse* _message);
+  virtual void HandleInstanceResponse(void* ctx,
+                                      heron::proto::stmgr::RegisterInstanceResponse* _message,
+                                      NetworkErrorCode status);
   // Handle incoming tuples
   virtual void HandleTupleMessage(heron::proto::system::HeronTupleSet2* _message);
-  // Send tuples
-  virtual void CreateAndSendTupleMessages();
   // Handle the instance assignment message
   virtual void HandleNewInstanceAssignmentMsg(heron::proto::stmgr::NewInstanceAssignmentMessage*);
 
@@ -76,10 +76,15 @@
 
  protected:
   // Handle incoming message
-  virtual void HandleInstanceResponse(heron::proto::stmgr::RegisterInstanceResponse* _message);
   virtual void HandleNewInstanceAssignmentMsg(
       heron::proto::stmgr::NewInstanceAssignmentMessage* _msg);
-  virtual void CreateAndSendTupleMessages();
+  void CreateAndSendTupleMessages();
+  virtual void StartBackPressureConnectionCb(Connection* connection) {
+    under_backpressure_ = true;
+  }
+  virtual void StopBackPressureConnectionCb(Connection* _connection) {
+    under_backpressure_ = false;
+  }
 
  private:
   sp_string stream_id_;
@@ -87,6 +92,7 @@
   sp_int32 total_msgs_sent_;
   sp_int32 batch_size_;
   bool do_custom_grouping_;
+  bool under_backpressure_;
   // only valid when the above is true
   sp_int32 custom_grouping_dest_task_;
 };
@@ -103,7 +109,6 @@
 
  protected:
   // Handle incoming message
-  virtual void HandleInstanceResponse(heron::proto::stmgr::RegisterInstanceResponse* _message);
   // Handle incoming tuples
   virtual void HandleTupleMessage(heron::proto::system::HeronTupleSet2* _message);
   virtual void HandleNewInstanceAssignmentMsg(
diff --git a/heron/storm/src/java/BUILD b/heron/storm/src/java/BUILD
index a825c7e..91ae2d7 100644
--- a/heron/storm/src/java/BUILD
+++ b/heron/storm/src/java/BUILD
@@ -8,6 +8,7 @@
     "//heron/api/src/java:api-java",
     "//heron/common/src/java:basics-java",
     "//heron/simulator/src/java:simulator-java",
+    "//heron/proto:proto_topology_java",
     "@com_googlecode_json_simple_json_simple//jar",
     "//third_party/java:kryo-neverlink",
 ]
diff --git a/heron/storm/src/java/org/apache/storm/Config.java b/heron/storm/src/java/org/apache/storm/Config.java
index e226a1c..25b781c 100644
--- a/heron/storm/src/java/org/apache/storm/Config.java
+++ b/heron/storm/src/java/org/apache/storm/Config.java
@@ -307,6 +307,55 @@
    */
   public static final String TRANSACTIONAL_ZOOKEEPER_PORT = "transactional.zookeeper.port";
 
+  /*
+   * Bolt-specific configuration for windowed bolts to specify the window length as a count of number of tuples
+   * in the window.
+   */
+  public static final String TOPOLOGY_BOLTS_WINDOW_LENGTH_COUNT
+      = "topology.bolts.window.length.count";
+
+  /*
+   * Bolt-specific configuration for windowed bolts to specify the window length in time duration.
+   */
+  public static final String TOPOLOGY_BOLTS_WINDOW_LENGTH_DURATION_MS
+      = "topology.bolts.window.length.duration.ms";
+
+  /*
+   * Bolt-specific configuration for windowed bolts to specify the sliding interval as a count of number of tuples.
+   */
+  public static final String TOPOLOGY_BOLTS_SLIDING_INTERVAL_COUNT
+      = "topology.bolts.window.sliding.interval.count";
+
+  /*
+   * Bolt-specific configuration for windowed bolts to specify the sliding interval in time duration.
+   */
+  public static final String TOPOLOGY_BOLTS_SLIDING_INTERVAL_DURATION_MS
+      = "topology.bolts.window.sliding.interval.duration.ms";
+
+  /**
+   * Bolt-specific configuration for windowed bolts to specify the name of the stream on which late tuples are
+   * going to be emitted. This configuration should only be used from the BaseWindowedBolt.withLateTupleStream builder
+   * method, and not as global parameter, otherwise IllegalArgumentException is going to be thrown.
+   */
+  public static final String TOPOLOGY_BOLTS_LATE_TUPLE_STREAM
+      = "topology.bolts.late.tuple.stream";
+
+  /**
+   * Bolt-specific configuration for windowed bolts to specify the maximum time lag of the tuple timestamp
+   * in milliseconds. It means that the tuple timestamps cannot be out of order by more than this amount.
+   * This config will be effective only if {@link org.apache.storm.windowing.TimestampExtractor} is specified.
+   */
+  public static final String TOPOLOGY_BOLTS_TUPLE_TIMESTAMP_MAX_LAG_MS
+      = "topology.bolts.tuple.timestamp.max.lag.ms";
+
+  /*
+   * Bolt-specific configuration for windowed bolts to specify the time interval for generating
+   * watermark events. Watermark event tracks the progress of time when tuple timestamp is used.
+   * This config is effective only if {@link org.apache.storm.windowing.TimestampExtractor} is specified.
+   */
+  public static final String TOPOLOGY_BOLTS_WATERMARK_EVENT_INTERVAL_MS
+      = "topology.bolts.watermark.event.interval.ms";
+
   /**
    * ----  DO NOT USE -----
    * This variable is used to rewrite the TOPOLOGY_AUTO_TASK_HOOKS variable.
diff --git a/heron/storm/src/java/org/apache/storm/spout/ISpout.java b/heron/storm/src/java/org/apache/storm/spout/ISpout.java
index f0c2cb6..e2ee367 100644
--- a/heron/storm/src/java/org/apache/storm/spout/ISpout.java
+++ b/heron/storm/src/java/org/apache/storm/spout/ISpout.java
@@ -57,7 +57,7 @@
    * @param collector The collector is used to emit tuples from this spout. Tuples can be emitted at any time, including the open and close methods. The collector is thread-safe and should be saved as an instance variable of this spout object.
    */
   @SuppressWarnings("rawtypes")
-  void open(Map conf, TopologyContext context, SpoutOutputCollector collector);
+  void open(Map<String, Object> conf, TopologyContext context, SpoutOutputCollector collector);
 
   /**
    * Called when an ISpout is going to be shutdown. There is no guarentee that close
diff --git a/heron/storm/src/java/org/apache/storm/topology/IRichSpoutDelegate.java b/heron/storm/src/java/org/apache/storm/topology/IRichSpoutDelegate.java
index b6a8ac7..6232e8a 100644
--- a/heron/storm/src/java/org/apache/storm/topology/IRichSpoutDelegate.java
+++ b/heron/storm/src/java/org/apache/storm/topology/IRichSpoutDelegate.java
@@ -45,7 +45,7 @@
 
   @Override
   @SuppressWarnings("rawtypes")
-  public void open(Map conf, com.twitter.heron.api.topology.TopologyContext context,
+  public void open(Map<String, Object> conf, com.twitter.heron.api.topology.TopologyContext context,
                    SpoutOutputCollector collector) {
     topologyContextImpl = new TopologyContext(context);
     spoutOutputCollectorImpl = new SpoutOutputCollectorImpl(collector);
diff --git a/heron/storm/src/java/org/apache/storm/topology/IWindowedBolt.java b/heron/storm/src/java/org/apache/storm/topology/IWindowedBolt.java
new file mode 100644
index 0000000..d848c43
--- /dev/null
+++ b/heron/storm/src/java/org/apache/storm/topology/IWindowedBolt.java
@@ -0,0 +1,53 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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 org.apache.storm.topology;
+
+import java.util.Map;
+
+import org.apache.storm.task.OutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.windowing.TimestampExtractor;
+import org.apache.storm.windowing.TupleWindow;
+
+/**
+ * A bolt abstraction for supporting time and count based sliding & tumbling windows.
+ */
+public interface IWindowedBolt extends IComponent {
+  /**
+   * This is similar to the
+   * {@link org.apache.storm.task.IBolt#prepare(Map, TopologyContext, OutputCollector)} except
+   * that while emitting, the tuples are automatically anchored to the tuples in the inputWindow.
+   */
+  void prepare(Map<String, Object> topoConf, TopologyContext context, OutputCollector collector);
+
+  /**
+   * Process the tuple window and optionally emit new tuples based on the tuples in the input
+   * window.
+   */
+  void execute(TupleWindow inputWindow);
+
+  void cleanup();
+
+  /**
+   * Return a {@link TimestampExtractor} for extracting timestamps from a
+   * tuple for event time based processing, or null for processing time.
+   *
+   * @return the timestamp extractor
+   */
+  TimestampExtractor getTimestampExtractor();
+}
diff --git a/heron/storm/src/java/org/apache/storm/topology/IWindowedBoltDelegate.java b/heron/storm/src/java/org/apache/storm/topology/IWindowedBoltDelegate.java
new file mode 100644
index 0000000..46be02a
--- /dev/null
+++ b/heron/storm/src/java/org/apache/storm/topology/IWindowedBoltDelegate.java
@@ -0,0 +1,80 @@
+/**
+ * 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 org.apache.storm.topology;
+
+import java.util.Map;
+
+import org.apache.storm.task.OutputCollectorImpl;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.tuple.TupleImpl;
+import org.apache.storm.windowing.TupleWindowImpl;
+
+public class IWindowedBoltDelegate implements com.twitter.heron.api.bolt.IWindowedBolt {
+
+  private static final long serialVersionUID = 8753943395082633132L;
+  private final IWindowedBolt delegate;
+  private TopologyContext topologyContextImpl;
+  private OutputCollectorImpl outputCollectorImpl;
+
+  public IWindowedBoltDelegate(IWindowedBolt iWindowedBolt) {
+    this.delegate = iWindowedBolt;
+  }
+
+  @Override
+  public void declareOutputFields(com.twitter.heron.api.topology.OutputFieldsDeclarer declarer) {
+    OutputFieldsGetter getter = new OutputFieldsGetter(declarer);
+    delegate.declareOutputFields(getter);
+  }
+
+  @Override
+  public Map<String, Object> getComponentConfiguration() {
+    return this.delegate.getComponentConfiguration();
+  }
+
+  @Override
+  public void prepare(
+      Map<String, Object> conf,
+      com.twitter.heron.api.topology.TopologyContext context,
+      com.twitter.heron.api.bolt.OutputCollector collector) {
+    topologyContextImpl = new TopologyContext(context);
+    outputCollectorImpl = new OutputCollectorImpl(collector);
+    delegate.prepare(conf, topologyContextImpl, outputCollectorImpl);
+  }
+
+  @Override
+  public void execute(com.twitter.heron.api.windowing.TupleWindow inputWindow) {
+    this.delegate.execute(new TupleWindowImpl(inputWindow));
+  }
+
+  @Override
+  public void cleanup() {
+    this.delegate.cleanup();
+  }
+
+  @Override
+  public com.twitter.heron.api.windowing.TimestampExtractor getTimestampExtractor() {
+
+    return (this.delegate.getTimestampExtractor() == null) ? null
+        : new com.twitter.heron.api.windowing.TimestampExtractor() {
+          @Override
+          public long extractTimestamp(com.twitter.heron.api.tuple.Tuple tuple) {
+            return delegate.getTimestampExtractor().extractTimestamp(new TupleImpl(tuple));
+          }
+        };
+  }
+}
diff --git a/heron/storm/src/java/org/apache/storm/topology/TopologyBuilder.java b/heron/storm/src/java/org/apache/storm/topology/TopologyBuilder.java
index dd9ec5b..a9594a0 100644
--- a/heron/storm/src/java/org/apache/storm/topology/TopologyBuilder.java
+++ b/heron/storm/src/java/org/apache/storm/topology/TopologyBuilder.java
@@ -24,6 +24,7 @@
 import org.apache.storm.generated.StormTopology;
 
 import com.twitter.heron.api.HeronTopology;
+import com.twitter.heron.api.bolt.WindowedBoltExecutor;
 
 public class TopologyBuilder {
   private com.twitter.heron.api.topology.TopologyBuilder delegate =
@@ -53,6 +54,19 @@
     return setBolt(id, new BasicBoltExecutor(bolt), parallelismHint);
   }
 
+  public BoltDeclarer setBolt(String id, IWindowedBolt bolt) throws IllegalArgumentException {
+    return setBolt(id, bolt, null);
+  }
+
+  public BoltDeclarer setBolt(String id, IWindowedBolt bolt, Number parallelismHint) throws
+      IllegalArgumentException {
+    return new BoltDeclarerImpl(
+        this.delegate.setBolt(
+            id, new WindowedBoltExecutor(new IWindowedBoltDelegate(bolt)),
+            parallelismHint)
+    );
+  }
+
   public SpoutDeclarer setSpout(String id, IRichSpout spout) {
     return setSpout(id, spout, null);
   }
diff --git a/heron/storm/src/java/org/apache/storm/topology/TupleFieldTimestampExtractor.java b/heron/storm/src/java/org/apache/storm/topology/TupleFieldTimestampExtractor.java
new file mode 100644
index 0000000..9f3a045
--- /dev/null
+++ b/heron/storm/src/java/org/apache/storm/topology/TupleFieldTimestampExtractor.java
@@ -0,0 +1,47 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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 org.apache.storm.topology;
+
+import org.apache.storm.tuple.Tuple;
+import org.apache.storm.windowing.TimestampExtractor;
+
+/**
+ * A {@link TimestampExtractor} that extracts timestamp from a specific field in the tuple.
+ */
+public final class TupleFieldTimestampExtractor implements TimestampExtractor {
+  private static final long serialVersionUID = 5212225232807940572L;
+  private final String fieldName;
+
+  private TupleFieldTimestampExtractor(String fieldName) {
+    this.fieldName = fieldName;
+  }
+
+  @Override
+  public long extractTimestamp(Tuple tuple) {
+    return tuple.getLongByField(fieldName);
+  }
+
+  public static TupleFieldTimestampExtractor of(String fieldName) {
+    return new TupleFieldTimestampExtractor(fieldName);
+  }
+
+  @Override
+  public String toString() {
+    return "TupleFieldTimestampExtractor{" + "fieldName='" + fieldName + '\'' + '}';
+  }
+}
diff --git a/heron/storm/src/java/org/apache/storm/topology/base/BaseWindowedBolt.java b/heron/storm/src/java/org/apache/storm/topology/base/BaseWindowedBolt.java
new file mode 100644
index 0000000..b0b5389
--- /dev/null
+++ b/heron/storm/src/java/org/apache/storm/topology/base/BaseWindowedBolt.java
@@ -0,0 +1,547 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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 org.apache.storm.topology.base;
+
+import java.io.Serializable;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.storm.task.OutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.IWindowedBolt;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.topology.TupleFieldTimestampExtractor;
+import org.apache.storm.tuple.TupleImpl;
+import org.apache.storm.windowing.TimestampExtractor;
+
+import com.twitter.heron.api.generated.TopologyAPI;
+import com.twitter.heron.api.tuple.Fields;
+import com.twitter.heron.api.tuple.Tuple;
+
+public abstract class BaseWindowedBolt implements IWindowedBolt {
+  private static final long serialVersionUID = -3998164228343123590L;
+  protected final transient com.twitter.heron.api.windowing.WindowingConfigs windowConfiguration;
+  protected com.twitter.heron.api.windowing.TimestampExtractor timestampExtractor;
+
+  /**
+   * Holds a count value for count based windows and sliding intervals.
+   */
+  public static class Count implements Serializable {
+    private static final long serialVersionUID = -2290882388716246812L;
+    public final int value;
+
+    public Count(int value) {
+      this.value = value;
+    }
+
+    /**
+     * Returns a {@link Count} of given value.
+     *
+     * @param value the count value
+     * @return the Count
+     */
+    public static Count of(int value) {
+      return new Count(value);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) {
+        return true;
+      }
+      if (o == null || getClass() != o.getClass()) {
+        return false;
+      }
+
+      Count count = (Count) o;
+
+      return value == count.value;
+
+    }
+
+    @Override
+    public int hashCode() {
+      return value;
+    }
+
+    @Override
+    public String toString() {
+      return "Count{" + "value=" + value + '}';
+    }
+  }
+
+  /**
+   * Holds a Time duration for time based windows and sliding intervals.
+   */
+  public static class Duration implements Serializable {
+    private static final long serialVersionUID = 5654070568075477148L;
+    public final int value;
+
+    public Duration(int value, TimeUnit timeUnit) {
+      this.value = (int) timeUnit.toMillis(value);
+    }
+
+    /**
+     * Returns a {@link Duration} corresponding to the the given value in milli seconds.
+     *
+     * @param milliseconds the duration in milliseconds
+     * @return the Duration
+     */
+    public static Duration of(int milliseconds) {
+      return new Duration(milliseconds, TimeUnit.MILLISECONDS);
+    }
+
+    /**
+     * Returns a {@link Duration} corresponding to the the given value in days.
+     *
+     * @param days the number of days
+     * @return the Duration
+     */
+    public static Duration days(int days) {
+      return new Duration(days, TimeUnit.DAYS);
+    }
+
+    /**
+     * Returns a {@link Duration} corresponding to the the given value in hours.
+     *
+     * @param hours the number of hours
+     * @return the Duration
+     */
+    public static Duration hours(int hours) {
+      return new Duration(hours, TimeUnit.HOURS);
+    }
+
+    /**
+     * Returns a {@link Duration} corresponding to the the given value in minutes.
+     *
+     * @param minutes the number of minutes
+     * @return the Duration
+     */
+    public static Duration minutes(int minutes) {
+      return new Duration(minutes, TimeUnit.MINUTES);
+    }
+
+    /**
+     * Returns a {@link Duration} corresponding to the the given value in seconds.
+     *
+     * @param seconds the number of seconds
+     * @return the Duration
+     */
+    public static Duration seconds(int seconds) {
+      return new Duration(seconds, TimeUnit.SECONDS);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) {
+        return true;
+      }
+      if (o == null || getClass() != o.getClass()) {
+        return false;
+      }
+
+      Duration duration = (Duration) o;
+
+      return value == duration.value;
+
+    }
+
+    @Override
+    public int hashCode() {
+      return value;
+    }
+
+    @Override
+    public String toString() {
+      return "Duration{" + "value=" + value + '}';
+    }
+  }
+
+  protected BaseWindowedBolt() {
+    windowConfiguration = new com.twitter.heron.api.windowing.WindowingConfigs();
+  }
+
+  private BaseWindowedBolt withWindowLength(Count count) {
+    if (count.value <= 0) {
+      throw new IllegalArgumentException("Window length must be positive [" + count + "]");
+    }
+    windowConfiguration.setTopologyBoltsWindowLengthCount(count.value);
+    return this;
+  }
+
+  private BaseWindowedBolt withWindowLength(Duration duration) {
+    if (duration.value <= 0) {
+      throw new IllegalArgumentException("Window length must be positive [" + duration + "]");
+    }
+    windowConfiguration.setTopologyBoltsWindowLengthDurationMs(duration.value);
+    return this;
+  }
+
+  private BaseWindowedBolt withSlidingInterval(Count count) {
+    if (count.value <= 0) {
+      throw new IllegalArgumentException("Sliding interval must be positive [" + count + "]");
+    }
+    windowConfiguration.setTopologyBoltsSlidingIntervalCount(count.value);
+    return this;
+  }
+
+  private BaseWindowedBolt withSlidingInterval(Duration duration) {
+    if (duration.value <= 0) {
+      throw new IllegalArgumentException("Sliding interval must be positive [" + duration + "]");
+    }
+    windowConfiguration.setTopologyBoltsSlidingIntervalDurationMs(duration.value);
+    return this;
+  }
+
+  /**
+   * Tuple count based sliding window configuration.
+   *
+   * @param windowLength    the number of tuples in the window
+   * @param slidingInterval the number of tuples after which the window slides
+   */
+  public BaseWindowedBolt withWindow(Count windowLength, Count slidingInterval) {
+    return withWindowLength(windowLength).withSlidingInterval(slidingInterval);
+  }
+
+  /**
+   * Tuple count and time duration based sliding window configuration.
+   *
+   * @param windowLength    the number of tuples in the window
+   * @param slidingInterval the time duration after which the window slides
+   */
+  public BaseWindowedBolt withWindow(Count windowLength, Duration slidingInterval) {
+    return withWindowLength(windowLength).withSlidingInterval(slidingInterval);
+  }
+
+  /**
+   * Time duration and count based sliding window configuration.
+   *
+   * @param windowLength    the time duration of the window
+   * @param slidingInterval the number of tuples after which the window slides
+   */
+  public BaseWindowedBolt withWindow(Duration windowLength, Count slidingInterval) {
+    return withWindowLength(windowLength).withSlidingInterval(slidingInterval);
+  }
+
+  /**
+   * Time duration based sliding window configuration.
+   *
+   * @param windowLength    the time duration of the window
+   * @param slidingInterval the time duration after which the window slides
+   */
+  public BaseWindowedBolt withWindow(Duration windowLength, Duration slidingInterval) {
+    return withWindowLength(windowLength).withSlidingInterval(slidingInterval);
+  }
+
+  /**
+   * A tuple count based window that slides with every incoming tuple.
+   *
+   * @param windowLength the number of tuples in the window
+   */
+  public BaseWindowedBolt withWindow(Count windowLength) {
+    return withWindowLength(windowLength).withSlidingInterval(new Count(1));
+  }
+
+  /**
+   * A time duration based window that slides with every incoming tuple.
+   *
+   * @param windowLength the time duration of the window
+   */
+  public BaseWindowedBolt withWindow(Duration windowLength) {
+    return withWindowLength(windowLength).withSlidingInterval(new Count(1));
+  }
+
+  /**
+   * A count based tumbling window.
+   *
+   * @param count the number of tuples after which the window tumbles
+   */
+  public BaseWindowedBolt withTumblingWindow(Count count) {
+    return withWindowLength(count).withSlidingInterval(count);
+  }
+
+  /**
+   * A time duration based tumbling window.
+   *
+   * @param duration the time duration after which the window tumbles
+   */
+  public BaseWindowedBolt withTumblingWindow(Duration duration) {
+    return withWindowLength(duration).withSlidingInterval(duration);
+  }
+
+  /**
+   * Specify a field in the tuple that represents the timestamp as a long value. If this
+   * field is not present in the incoming tuple, an {@link IllegalArgumentException} will be thrown.
+   *
+   * @param fieldName the name of the field that contains the timestamp
+   */
+  public BaseWindowedBolt withTimestampField(String fieldName) {
+    return withTimestampExtractor(TupleFieldTimestampExtractor.of(fieldName));
+  }
+
+  /**
+   * Specify the timestamp extractor implementation.
+   *
+   * @param timestampExtractor the {@link TimestampExtractor} implementation
+   */
+  @SuppressWarnings("HiddenField")
+  public BaseWindowedBolt withTimestampExtractor(TimestampExtractor timestampExtractor) {
+    if (this.timestampExtractor != null) {
+      throw new IllegalArgumentException(
+          "Window is already configured with a timestamp extractor: " + timestampExtractor);
+    }
+
+    this.timestampExtractor = new com.twitter.heron.api.windowing.TimestampExtractor() {
+
+      @Override
+      public long extractTimestamp(Tuple tuple) {
+        return timestampExtractor.extractTimestamp(new TupleImpl(tuple));
+      }
+    };
+    return this;
+  }
+
+  @Override
+  public TimestampExtractor getTimestampExtractor() {
+    return (this.timestampExtractor == null) ? null : new TimestampExtractor() {
+
+      @Override
+      public long extractTimestamp(org.apache.storm.tuple.Tuple tuple) {
+
+        return timestampExtractor.extractTimestamp(new com.twitter.heron.api.tuple.Tuple() {
+
+          @Override
+          public int size() {
+            return tuple.size();
+          }
+
+          @Override
+          public int fieldIndex(String field) {
+            return tuple.fieldIndex(field);
+          }
+
+          @Override
+          public boolean contains(String field) {
+            return tuple.contains(field);
+          }
+
+          @Override
+          public Object getValue(int i) {
+            return tuple.getValue(i);
+          }
+
+          @Override
+          public String getString(int i) {
+            return tuple.getString(i);
+          }
+
+          @Override
+          public Integer getInteger(int i) {
+            return tuple.getInteger(i);
+          }
+
+          @Override
+          public Long getLong(int i) {
+            return tuple.getLong(i);
+          }
+
+          @Override
+          public Boolean getBoolean(int i) {
+            return tuple.getBoolean(i);
+          }
+
+          @Override
+          public Short getShort(int i) {
+            return tuple.getShort(i);
+          }
+
+          @Override
+          public Byte getByte(int i) {
+            return tuple.getByte(i);
+          }
+
+          @Override
+          public Double getDouble(int i) {
+            return tuple.getDouble(i);
+          }
+
+          @Override
+          public Float getFloat(int i) {
+            return tuple.getFloat(i);
+          }
+
+          @Override
+          public byte[] getBinary(int i) {
+            return tuple.getBinary(i);
+          }
+
+          @Override
+          public Object getValueByField(String field) {
+            return tuple.getValueByField(field);
+          }
+
+          @Override
+          public String getStringByField(String field) {
+            return tuple.getStringByField(field);
+          }
+
+          @Override
+          public Integer getIntegerByField(String field) {
+            return tuple.getIntegerByField(field);
+          }
+
+          @Override
+          public Long getLongByField(String field) {
+            return tuple.getLongByField(field);
+          }
+
+          @Override
+          public Boolean getBooleanByField(String field) {
+            return tuple.getBooleanByField(field);
+          }
+
+          @Override
+          public Short getShortByField(String field) {
+            return tuple.getShortByField(field);
+          }
+
+          @Override
+          public Byte getByteByField(String field) {
+            return tuple.getByteByField(field);
+          }
+
+          @Override
+          public Double getDoubleByField(String field) {
+            return tuple.getDoubleByField(field);
+          }
+
+          @Override
+          public Float getFloatByField(String field) {
+            return tuple.getFloatByField(field);
+          }
+
+          @Override
+          public byte[] getBinaryByField(String field) {
+            return tuple.getBinaryByField(field);
+          }
+
+          @Override
+          public List<Object> getValues() {
+            return tuple.getValues();
+          }
+
+          @Override
+          public Fields getFields() {
+            return new Fields(tuple.getFields().toList());
+          }
+
+          @Override
+          public List<Object> select(Fields selector) {
+            return tuple.select(new org.apache.storm.tuple.Fields(selector.toString()));
+          }
+
+          @Override
+          public TopologyAPI.StreamId getSourceGlobalStreamId() {
+            return TopologyAPI.StreamId.newBuilder().setId(tuple.getSourceStreamId())
+                .setComponentName(tuple.getSourceComponent()).build();
+          }
+
+          @Override
+          public String getSourceComponent() {
+            return tuple.getSourceComponent();
+          }
+
+          @Override
+          public int getSourceTask() {
+            return tuple.getSourceTask();
+          }
+
+          @Override
+          public String getSourceStreamId() {
+            return tuple.getSourceStreamId();
+          }
+
+          @Override
+          public void resetValues() {
+            tuple.resetValues();
+          }
+        });
+      }
+    };
+  }
+
+  /**
+   * Specify a stream id on which late tuples are going to be emitted.
+   * It must be defined on a per-component basis, and in conjunction with the
+   * {@link BaseWindowedBolt#withTimestampField}, otherwise {@link IllegalArgumentException} will
+   * be thrown.
+   *
+   * @param streamId the name of the stream used to emit late tuples on
+   */
+  public BaseWindowedBolt withLateTupleStream(String streamId) {
+    windowConfiguration.setTopologyBoltsLateTupleStream(streamId);
+    return this;
+  }
+
+
+  /**
+   * Specify the maximum time lag of the tuple timestamp in milliseconds. It means that the tuple
+   * timestamps
+   * cannot be out of order by more than this amount.
+   *
+   * @param duration the max lag duration
+   */
+  public BaseWindowedBolt withLag(Duration duration) {
+    windowConfiguration.setTopologyBoltsTupleTimestampMaxLagMs(duration.value);
+    return this;
+  }
+
+  /**
+   * Specify the watermark event generation interval. For tuple based timestamps, watermark events
+   * are used to track the progress of time
+   *
+   * @param interval the interval at which watermark events are generated
+   */
+  public BaseWindowedBolt withWatermarkInterval(Duration interval) {
+    windowConfiguration.setTopologyBoltsWatermarkEventIntervalMs(interval.value);
+    return this;
+  }
+
+  @Override
+  public void prepare(Map<String, Object> topoConf, TopologyContext context, OutputCollector
+      collector) {
+    // NOOP
+  }
+
+  @Override
+  public void cleanup() {
+    // NOOP
+  }
+
+  @Override
+  public void declareOutputFields(OutputFieldsDeclarer declarer) {
+    // NOOP
+  }
+
+  @Override
+  public Map<String, Object> getComponentConfiguration() {
+    return windowConfiguration;
+  }
+}
diff --git a/heron/storm/src/java/org/apache/storm/tuple/Tuple.java b/heron/storm/src/java/org/apache/storm/tuple/Tuple.java
index 65eb6c8..c238346 100644
--- a/heron/storm/src/java/org/apache/storm/tuple/Tuple.java
+++ b/heron/storm/src/java/org/apache/storm/tuple/Tuple.java
@@ -39,110 +39,190 @@
   int size();
 
   /**
-   * Returns the position of the specified field in this tuple.
-   */
-  int fieldIndex(String field);
-
-  /**
    * Returns true if this tuple contains the specified name of the field.
    */
   boolean contains(String field);
 
   /**
-   * Gets the field at position i in the tuple. Returns object since tuples are dynamically typed.
-   */
-  Object getValue(int i);
-
-  /**
-   * Returns the String at position i in the tuple. If that field is not a String,
-   * you will get a runtime error.
-   */
-  String getString(int i);
-
-  /**
-   * Returns the Integer at position i in the tuple. If that field is not an Integer,
-   * you will get a runtime error.
-   */
-  Integer getInteger(int i);
-
-  /**
-   * Returns the Long at position i in the tuple. If that field is not a Long,
-   * you will get a runtime error.
-   */
-  Long getLong(int i);
-
-  /**
-   * Returns the Boolean at position i in the tuple. If that field is not a Boolean,
-   * you will get a runtime error.
-   */
-  Boolean getBoolean(int i);
-
-  /**
-   * Returns the Short at position i in the tuple. If that field is not a Short,
-   * you will get a runtime error.
-   */
-  Short getShort(int i);
-
-  /**
-   * Returns the Byte at position i in the tuple. If that field is not a Byte,
-   * you will get a runtime error.
-   */
-  Byte getByte(int i);
-
-  /**
-   * Returns the Double at position i in the tuple. If that field is not a Double,
-   * you will get a runtime error.
-   */
-  Double getDouble(int i);
-
-  /**
-   * Returns the Float at position i in the tuple. If that field is not a Float,
-   * you will get a runtime error.
-   */
-  Float getFloat(int i);
-
-  /**
-   * Returns the byte array at position i in the tuple. If that field is not a byte array,
-   * you will get a runtime error.
-   */
-  byte[] getBinary(int i);
-
-
-  Object getValueByField(String field);
-
-  String getStringByField(String field);
-
-  Integer getIntegerByField(String field);
-
-  Long getLongByField(String field);
-
-  Boolean getBooleanByField(String field);
-
-  Short getShortByField(String field);
-
-  Byte getByteByField(String field);
-
-  Double getDoubleByField(String field);
-
-  Float getFloatByField(String field);
-
-  byte[] getBinaryByField(String field);
-
-  /**
-   * Gets all the values in this tuple.
-   */
-  List<Object> getValues();
-
-  /**
    * Gets the names of the fields in this tuple.
    */
   Fields getFields();
 
   /**
+   *  Returns the position of the specified field in this tuple.
+   *
+   * @throws IllegalArgumentException - if field does not exist
+   */
+  int fieldIndex(String field);
+
+  /**
    * Returns a subset of the tuple based on the fields selector.
    */
   List<Object> select(Fields selector);
 
+  /**
+   * Gets the field at position i in the tuple. Returns object since tuples are dynamically typed.
+   *
+   * @throws IndexOutOfBoundsException - if the index is out of range `(index < 0 || index >= size())`
+   */
+  Object getValue(int i);
+
+  /**
+   * Returns the String at position i in the tuple.
+   *
+   * @throws ClassCastException If that field is not a String
+   * @throws IndexOutOfBoundsException - if the index is out of range `(index < 0 || index >= size())`
+   */
+  String getString(int i);
+
+  /**
+   * Returns the Integer at position i in the tuple.
+   *
+   * @throws ClassCastException If that field is not a Integer
+   * @throws IndexOutOfBoundsException - if the index is out of range `(index < 0 || index >= size())`
+   */
+  Integer getInteger(int i);
+
+  /**
+   * Returns the Long at position i in the tuple.
+   *
+   * @throws ClassCastException If that field is not a Long
+   * @throws IndexOutOfBoundsException - if the index is out of range `(index < 0 || index >= size())`
+   */
+  Long getLong(int i);
+
+  /**
+   * Returns the Boolean at position i in the tuple.
+   *
+   * @throws ClassCastException If that field is not a Boolean
+   * @throws IndexOutOfBoundsException - if the index is out of range `(index < 0 || index >= size())`
+   */
+  Boolean getBoolean(int i);
+
+  /**
+   * Returns the Short at position i in the tuple.
+   *
+   * @throws ClassCastException If that field is not a Short
+   * @throws IndexOutOfBoundsException - if the index is out of range `(index < 0 || index >= size())`
+   */
+  Short getShort(int i);
+
+  /**
+   * Returns the Byte at position i in the tuple.
+   *
+   * @throws ClassCastException If that field is not a Byte
+   * @throws IndexOutOfBoundsException - if the index is out of range `(index < 0 || index >= size())`
+   */
+  Byte getByte(int i);
+
+  /**
+   * Returns the Double at position i in the tuple.
+   *
+   * @throws ClassCastException If that field is not a Double
+   * @throws IndexOutOfBoundsException - if the index is out of range `(index < 0 || index >= size())`
+   */
+  Double getDouble(int i);
+
+  /**
+   * Returns the Float at position i in the tuple.
+   *
+   * @throws ClassCastException If that field is not a Float
+   * @throws IndexOutOfBoundsException - if the index is out of range `(index < 0 || index >= size())`
+   */
+  Float getFloat(int i);
+
+  /**
+   * Returns the byte array at position i in the tuple.
+   *
+   * @throws ClassCastException If that field is not a byte array
+   * @throws IndexOutOfBoundsException - if the index is out of range `(index < 0 || index >= size())`
+   */
+  byte[] getBinary(int i);
+
+  /**
+   * Gets the field with a specific name. Returns object since tuples are dynamically typed.
+   *
+   * @throws IllegalArgumentException - if field does not exist
+   */
+  Object getValueByField(String field);
+
+  /**
+   * Gets the String field with a specific name.
+   *
+   * @throws ClassCastException If that field is not a String
+   * @throws IllegalArgumentException - if field does not exist
+   */
+  String getStringByField(String field);
+
+  /**
+   * Gets the Integer field with a specific name.
+   *
+   * @throws ClassCastException If that field is not an Integer
+   * @throws IllegalArgumentException - if field does not exist
+   */
+  Integer getIntegerByField(String field);
+
+  /**
+   * Gets the Long field with a specific name.
+   *
+   * @throws ClassCastException If that field is not a Long
+   * @throws IllegalArgumentException - if field does not exist
+   */
+  Long getLongByField(String field);
+
+  /**
+   * Gets the Boolean field with a specific name.
+   *
+   * @throws ClassCastException If that field is not a Boolean
+   * @throws IllegalArgumentException - if field does not exist
+   */
+  Boolean getBooleanByField(String field);
+
+  /**
+   * Gets the Short field with a specific name.
+   *
+   * @throws ClassCastException If that field is not a Short
+   * @throws IllegalArgumentException - if field does not exist
+   */
+  Short getShortByField(String field);
+
+  /**
+   * Gets the Byte field with a specific name.
+   *
+   * @throws ClassCastException If that field is not a Byte
+   * @throws IllegalArgumentException - if field does not exist
+   */
+  Byte getByteByField(String field);
+
+  /**
+   * Gets the Double field with a specific name.
+   *
+   * @throws ClassCastException If that field is not a Double
+   * @throws IllegalArgumentException - if field does not exist
+   */
+  Double getDoubleByField(String field);
+
+  /**
+   * Gets the Float field with a specific name.
+   *
+   * @throws ClassCastException If that field is not a Float
+   * @throws IllegalArgumentException - if field does not exist
+   */
+  Float getFloatByField(String field);
+
+  /**
+   * Gets the Byte array field with a specific name.
+   *
+   * @throws ClassCastException If that field is not a byte array
+   * @throws IllegalArgumentException - if field does not exist
+   */
+  byte[] getBinaryByField(String field);
+
+  /**
+   * Gets all the values in this tuple.
+   */
+  List<Object> getValues();
+
 
   /**
    * Returns the global stream id (component + stream) of this tuple.
diff --git a/heron/storm/src/java/org/apache/storm/tuple/TupleImpl.java b/heron/storm/src/java/org/apache/storm/tuple/TupleImpl.java
index 9c99768..30d4090 100644
--- a/heron/storm/src/java/org/apache/storm/tuple/TupleImpl.java
+++ b/heron/storm/src/java/org/apache/storm/tuple/TupleImpl.java
@@ -248,4 +248,20 @@
   public void resetValues() {
     delegate.resetValues();
   }
+
+  @Override
+  public String toString() {
+    return "source: " + getSourceComponent() + ":" + getSourceTask()
+        + ", stream: " + getSourceStreamId() + ", " + getValues().toString();
+  }
+
+  @Override
+  public boolean equals(Object other) {
+    return this == other;
+  }
+
+  @Override
+  public int hashCode() {
+    return System.identityHashCode(this);
+  }
 }
diff --git a/heron/storm/src/java/org/apache/storm/windowing/TimestampExtractor.java b/heron/storm/src/java/org/apache/storm/windowing/TimestampExtractor.java
new file mode 100644
index 0000000..14c6054
--- /dev/null
+++ b/heron/storm/src/java/org/apache/storm/windowing/TimestampExtractor.java
@@ -0,0 +1,36 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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 org.apache.storm.windowing;
+
+import java.io.Serializable;
+
+import org.apache.storm.tuple.Tuple;
+
+/**
+ * Interface to be implemented for extracting timestamp from a tuple.
+ */
+public interface TimestampExtractor extends Serializable {
+  /**
+   * Return the tuple timestamp indicating the time when the event happened.
+   *
+   * @param tuple the tuple
+   * @return the timestamp
+   */
+  long extractTimestamp(Tuple tuple);
+}
diff --git a/heron/storm/src/java/org/apache/storm/windowing/TupleWindow.java b/heron/storm/src/java/org/apache/storm/windowing/TupleWindow.java
new file mode 100644
index 0000000..0ed5810
--- /dev/null
+++ b/heron/storm/src/java/org/apache/storm/windowing/TupleWindow.java
@@ -0,0 +1,26 @@
+/**
+ * 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 org.apache.storm.windowing;
+
+import org.apache.storm.tuple.Tuple;
+
+/**
+ * A {@link Window} that contains {@link Tuple} objects.
+ */
+public interface TupleWindow extends Window<Tuple> {
+}
diff --git a/heron/storm/src/java/org/apache/storm/windowing/TupleWindowImpl.java b/heron/storm/src/java/org/apache/storm/windowing/TupleWindowImpl.java
new file mode 100644
index 0000000..1d2f9e4
--- /dev/null
+++ b/heron/storm/src/java/org/apache/storm/windowing/TupleWindowImpl.java
@@ -0,0 +1,67 @@
+/**
+ * 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 org.apache.storm.windowing;
+
+import java.util.LinkedList;
+import java.util.List;
+
+import org.apache.storm.tuple.Tuple;
+import org.apache.storm.tuple.TupleImpl;
+
+public class TupleWindowImpl implements TupleWindow {
+
+  private final com.twitter.heron.api.windowing.TupleWindow delegate;
+
+  public TupleWindowImpl(com.twitter.heron.api.windowing.TupleWindow tupleWindow) {
+    this.delegate = tupleWindow;
+  }
+
+  @Override
+  public List<Tuple> get() {
+    return convert(this.delegate.get());
+  }
+
+  @Override
+  public List<Tuple> getNew() {
+    return convert(this.delegate.getNew());
+  }
+
+  @Override
+  public List<Tuple> getExpired() {
+    return convert(this.delegate.getExpired());
+  }
+
+  @Override
+  public Long getEndTimestamp() {
+    return this.delegate.getEndTimestamp();
+  }
+
+  @Override
+  public Long getStartTimestamp() {
+    return this.delegate.getStartTimestamp();
+  }
+
+  private static List<Tuple> convert(List<com.twitter.heron.api.tuple.Tuple> tuples) {
+    List<Tuple> ret = new LinkedList<>();
+    for (com.twitter.heron.api.tuple.Tuple tuple : tuples) {
+      ret.add(new TupleImpl(tuple));
+    }
+    return ret;
+  }
+}
diff --git a/heron/storm/src/java/org/apache/storm/windowing/Window.java b/heron/storm/src/java/org/apache/storm/windowing/Window.java
new file mode 100644
index 0000000..5142ab8
--- /dev/null
+++ b/heron/storm/src/java/org/apache/storm/windowing/Window.java
@@ -0,0 +1,75 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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 org.apache.storm.windowing;
+
+import java.util.Iterator;
+import java.util.List;
+
+/**
+ * A view of events in a sliding window.
+ *
+ * @param <T> the type of event that this window contains. E.g. {@link org.apache.storm.tuple.Tuple}
+ */
+public interface Window<T> {
+  /**
+   * Gets the list of events in the window.
+   * <p>
+   *     <b>Note: </b> If the number of tuples in windows is huge, invoking {@code get} would
+   *                   load all the tuples into memory and may throw an OOM exception. Use
+   *                   windowing with persistence
+   * </p>
+   * @return the list of events in the window.
+   */
+  List<T> get();
+
+  /**
+   * Returns an iterator over the events in the window.
+   * @return an {@link Iterator} over the events in the current window.
+   */
+  default Iterator<T> getIter() {
+    throw new UnsupportedOperationException("Not implemented");
+  }
+
+  /**
+   * Get the list of newly added events in the window since the last time the window was generated.
+   * @return the list of newly added events in the window.
+   */
+  List<T> getNew();
+
+  /**
+   * Get the list of events expired from the window since the last time the window was generated.
+   * @return the list of events expired from the window.
+   */
+  List<T> getExpired();
+
+  /**
+   * If processing based on event time, returns the window end time based on watermark otherwise
+   * returns the window end time based on processing time.
+   *
+   * @return the window end timestamp
+   */
+  Long getEndTimestamp();
+
+  /**
+   * Returns the window start timestamp. Will return null if the window length is not based on
+   * time duration.
+   *
+   * @return the window start timestamp or null if the window length is not time based
+   */
+  Long getStartTimestamp();
+}
diff --git a/integration_test/src/__init__.py b/integration_test/src/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/integration_test/src/__init__.py
diff --git a/integration_test/src/java/BUILD b/integration_test/src/java/BUILD
index 71a0bd7..693d3c2 100644
--- a/integration_test/src/java/BUILD
+++ b/integration_test/src/java/BUILD
@@ -47,6 +47,7 @@
     deps = [
         "//heron/api/src/java:api-java",
         "//heron/storm/src/java:storm-compatibility-java",
+        "@com_googlecode_json_simple_json_simple//jar",
         "@commons_cli_commons_cli//jar",
         ":common",
         ":core"
@@ -75,6 +76,7 @@
         "//heron/api/src/java:api-java",
         "//heron/storm/src/java:storm-compatibility-java",
         "@commons_cli_commons_cli//jar",
+        "@com_googlecode_json_simple_json_simple//jar",
         ":common",
         ":core"
     ],
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/core/IntegrationTestBolt.java b/integration_test/src/java/com/twitter/heron/integration_test/core/IntegrationTestBolt.java
index dfaa034..19aec08 100644
--- a/integration_test/src/java/com/twitter/heron/integration_test/core/IntegrationTestBolt.java
+++ b/integration_test/src/java/com/twitter/heron/integration_test/core/IntegrationTestBolt.java
@@ -42,8 +42,12 @@
   private Tuple currentTupleProcessing = null;
   private OutputCollector collector;
 
-  public IntegrationTestBolt(IRichBolt delegate) {
+  //whether automatically acking should be done
+  private boolean ackAuto;
+
+  public IntegrationTestBolt(IRichBolt delegate, boolean ackAuto) {
     this.delegateBolt = delegate;
+    this.ackAuto = ackAuto;
   }
 
   @Override
@@ -61,7 +65,7 @@
                       OutputCollector outputCollector) {
     update(context);
     this.collector = new OutputCollector(new IntegrationTestBoltCollector(outputCollector));
-    this.delegateBolt.prepare(map, context, collector);
+    this.delegateBolt.prepare(map, new TestTopologyContext(context), collector);
   }
 
   private int calculateTerminalsToReceive(TopologyContext context) {
@@ -108,7 +112,9 @@
       currentTupleProcessing = tuple;
       delegateBolt.execute(tuple);
       // We ack only the tuples in user's logic
-      collector.ack(tuple);
+      if (this.ackAuto) {
+        collector.ack(tuple);
+      }
     }
   }
 
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/core/IntegrationTestSpout.java b/integration_test/src/java/com/twitter/heron/integration_test/core/IntegrationTestSpout.java
index 0fc4dd6..08aee93 100644
--- a/integration_test/src/java/com/twitter/heron/integration_test/core/IntegrationTestSpout.java
+++ b/integration_test/src/java/com/twitter/heron/integration_test/core/IntegrationTestSpout.java
@@ -96,7 +96,7 @@
     // Here the spoutOutputCollector should be a default one
     // to emit tuples without adding MessageId
     this.spoutOutputCollector = outputCollector;
-    delegateSpout.open(map, topologyContext,
+    delegateSpout.open(map, new TestTopologyContext(topologyContext),
         new SpoutOutputCollector(new IntegrationTestSpoutCollector(outputCollector)));
   }
 
@@ -153,6 +153,7 @@
   public void fail(Object messageId) {
     LOG.info("Received a fail with MessageId: " + messageId);
 
+    tuplesToAck--;
     if (!isTestMessageId(messageId)) {
       delegateSpout.fail(messageId);
     } else {
@@ -166,7 +167,8 @@
   }
 
   private static boolean isTestMessageId(Object messageId) {
-    return ((String) messageId).startsWith(Constants.INTEGRATION_TEST_MOCK_MESSAGE_ID);
+    return  (messageId instanceof String) && ((String) messageId)
+        .startsWith(Constants.INTEGRATION_TEST_MOCK_MESSAGE_ID);
   }
 
   protected boolean doneEmitting() {
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/core/TestTopologyBuilder.java b/integration_test/src/java/com/twitter/heron/integration_test/core/TestTopologyBuilder.java
index 7290506..e34d3d5 100644
--- a/integration_test/src/java/com/twitter/heron/integration_test/core/TestTopologyBuilder.java
+++ b/integration_test/src/java/com/twitter/heron/integration_test/core/TestTopologyBuilder.java
@@ -22,6 +22,8 @@
 import com.twitter.heron.api.Config;
 import com.twitter.heron.api.HeronTopology;
 import com.twitter.heron.api.bolt.IRichBolt;
+import com.twitter.heron.api.bolt.IWindowedBolt;
+import com.twitter.heron.api.bolt.WindowedBoltExecutor;
 import com.twitter.heron.api.generated.TopologyAPI;
 import com.twitter.heron.api.spout.IRichSpout;
 import com.twitter.heron.api.topology.BoltDeclarer;
@@ -68,7 +70,17 @@
 
   @Override
   public BoltDeclarer setBolt(String id, IRichBolt bolt, Number parallelismHint) {
-    return super.setBolt(id, new IntegrationTestBolt(bolt), parallelismHint);
+    return setBolt(id, bolt, parallelismHint, true);
+  }
+
+  public BoltDeclarer setBolt(String id, IRichBolt bolt, Number parallelismHint, boolean ackAuto) {
+    return super.setBolt(id, new IntegrationTestBolt(bolt, ackAuto), parallelismHint);
+  }
+
+  public BoltDeclarer setBolt(String id, IWindowedBolt bolt,
+                              Number parallelismHint, boolean ackAuto) throws
+      IllegalArgumentException {
+    return setBolt(id, new WindowedBoltExecutor(bolt), parallelismHint, ackAuto);
   }
 
   @Override
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/core/TestTopologyContext.java b/integration_test/src/java/com/twitter/heron/integration_test/core/TestTopologyContext.java
new file mode 100644
index 0000000..421506b
--- /dev/null
+++ b/integration_test/src/java/com/twitter/heron/integration_test/core/TestTopologyContext.java
@@ -0,0 +1,172 @@
+//  Copyright 2017 Twitter. All rights reserved.
+//
+//  Licensed 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.twitter.heron.integration_test.core;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+
+import com.twitter.heron.api.generated.TopologyAPI;
+import com.twitter.heron.api.hooks.ITaskHook;
+import com.twitter.heron.api.metric.CombinedMetric;
+import com.twitter.heron.api.metric.ICombiner;
+import com.twitter.heron.api.metric.IMetric;
+import com.twitter.heron.api.metric.IReducer;
+import com.twitter.heron.api.metric.ReducedMetric;
+import com.twitter.heron.api.topology.TopologyContext;
+import com.twitter.heron.api.tuple.Fields;
+
+public class TestTopologyContext implements TopologyContext {
+  private TopologyContext delegate;
+
+  public TestTopologyContext(TopologyContext topologyContext) {
+    this.delegate = topologyContext;
+  }
+
+  @Override
+  public int getThisTaskId() {
+    return this.delegate.getThisTaskId();
+  }
+
+  @Override
+  public String getThisComponentId() {
+    return this.delegate.getThisComponentId();
+  }
+
+  @Override
+  public Fields getThisOutputFields(String streamId) {
+    return this.delegate.getThisOutputFields(streamId);
+  }
+
+  @Override
+  public Set<String> getThisStreams() {
+    return this.delegate.getThisStreams();
+  }
+
+  @Override
+  public int getThisTaskIndex() {
+    return this.delegate.getThisTaskIndex();
+  }
+
+  /**
+   * remove INTEGRATION_TEST_CONTROL_STREAM_ID from topology context
+   */
+  @Override
+  public Map<TopologyAPI.StreamId, TopologyAPI.Grouping> getThisSources() {
+    Map<TopologyAPI.StreamId, TopologyAPI.Grouping> original = getSources(getThisComponentId());
+    Map<TopologyAPI.StreamId, TopologyAPI.Grouping> ret = new HashMap<>();
+    for (Map.Entry<TopologyAPI.StreamId, TopologyAPI.Grouping> entry : original.entrySet()) {
+      if (!entry.getKey().getId().equals(Constants.INTEGRATION_TEST_CONTROL_STREAM_ID)) {
+        ret.put(entry.getKey(), entry.getValue());
+      }
+    }
+    return ret;
+  }
+
+  @Override
+  public Map<String, Map<String, TopologyAPI.Grouping>> getThisTargets() {
+    return this.delegate.getThisTargets();
+  }
+
+  @Override
+  public void setTaskData(String name, Object data) {
+    this.delegate.setTaskData(name, data);
+  }
+
+  @Override
+  public Object getTaskData(String name) {
+    return this.delegate.getTaskData(name);
+  }
+
+  @Override
+  public void addTaskHook(ITaskHook hook) {
+    this.delegate.addTaskHook(hook);
+  }
+
+  @Override
+  public Collection<ITaskHook> getHooks() {
+    return this.delegate.getHooks();
+  }
+
+  @Override
+  public <T, U, V> ReducedMetric<T, U, V> registerMetric(String name, IReducer<T, U, V> reducer,
+                                                         int timeBucketSizeInSecs) {
+    return this.delegate.registerMetric(name, reducer, timeBucketSizeInSecs);
+  }
+
+  @Override
+  public <T> CombinedMetric<T> registerMetric(String name, ICombiner<T> combiner, int
+      timeBucketSizeInSecs) {
+    return this.delegate.registerMetric(name, combiner, timeBucketSizeInSecs);
+  }
+
+  @Override
+  public <T extends IMetric<U>, U> T registerMetric(String name, T metric, int
+      timeBucketSizeInSecs) {
+    return this.delegate.registerMetric(name, metric, timeBucketSizeInSecs);
+  }
+
+  @Override
+  public String getTopologyId() {
+    return this.delegate.getTopologyId();
+  }
+
+  @Override
+  public String getComponentId(int taskId) {
+    return this.delegate.getComponentId(taskId);
+  }
+
+  @Override
+  public Set<String> getComponentStreams(String componentId) {
+    return this.delegate.getComponentStreams(componentId);
+  }
+
+  @Override
+  public List<Integer> getComponentTasks(String componentId) {
+    return this.delegate.getComponentTasks(componentId);
+  }
+
+  @Override
+  public Fields getComponentOutputFields(String componentId, String streamId) {
+    return this.delegate.getComponentOutputFields(componentId, streamId);
+  }
+
+  @Override
+  public Map<TopologyAPI.StreamId, TopologyAPI.Grouping> getSources(String componentId) {
+    return this.delegate.getSources(componentId);
+  }
+
+  @Override
+  public Map<String, Map<String, TopologyAPI.Grouping>> getTargets(String componentId) {
+    return this.delegate.getTargets(componentId);
+  }
+
+  @Override
+  public Map<Integer, String> getTaskToComponent() {
+    return this.delegate.getTaskToComponent();
+  }
+
+  @Override
+  public Set<String> getComponentIds() {
+    return this.delegate.getComponentIds();
+  }
+
+  @Override
+  public int maxTopologyMessageTimeout() {
+    return this.delegate.maxTopologyMessageTimeout();
+  }
+}
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/AllGrouping.json b/integration_test/src/java/com/twitter/heron/integration_test/topology/AllGrouping.json
deleted file mode 100644
index 7e04cfc..0000000
--- a/integration_test/src/java/com/twitter/heron/integration_test/topology/AllGrouping.json
+++ /dev/null
@@ -1 +0,0 @@
-["A", "B", "A", "B", "A", "B", "A", "B", "A", "B", "A", "B", "A", "B", "A", "B", "A", "B", "A", "B"]
\ No newline at end of file
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/BoltDoubleEmitTuples.json b/integration_test/src/java/com/twitter/heron/integration_test/topology/BoltDoubleEmitTuples.json
deleted file mode 100644
index 7e04cfc..0000000
--- a/integration_test/src/java/com/twitter/heron/integration_test/topology/BoltDoubleEmitTuples.json
+++ /dev/null
@@ -1 +0,0 @@
-["A", "B", "A", "B", "A", "B", "A", "B", "A", "B", "A", "B", "A", "B", "A", "B", "A", "B", "A", "B"]
\ No newline at end of file
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/NonGrouping.json b/integration_test/src/java/com/twitter/heron/integration_test/topology/NonGrouping.json
deleted file mode 100644
index b822614..0000000
--- a/integration_test/src/java/com/twitter/heron/integration_test/topology/NonGrouping.json
+++ /dev/null
@@ -1 +0,0 @@
-["A", "B", "A", "B", "A", "B", "A", "B", "A", "B"]
\ No newline at end of file
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/OneSpoutMultiTasks.json b/integration_test/src/java/com/twitter/heron/integration_test/topology/OneSpoutMultiTasks.json
deleted file mode 100644
index a3cceb7..0000000
--- a/integration_test/src/java/com/twitter/heron/integration_test/topology/OneSpoutMultiTasks.json
+++ /dev/null
@@ -1 +0,0 @@
-["A", "B", "A", "B", "A", "B", "A", "B", "A", "B","A", "B", "A", "B", "A", "B", "A", "B", "A", "B","A", "B", "A", "B", "B", "A", "A", "A", "B", "B"]
\ No newline at end of file
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/ShuffleGrouping.json b/integration_test/src/java/com/twitter/heron/integration_test/topology/ShuffleGrouping.json
deleted file mode 100644
index b822614..0000000
--- a/integration_test/src/java/com/twitter/heron/integration_test/topology/ShuffleGrouping.json
+++ /dev/null
@@ -1 +0,0 @@
-["A", "B", "A", "B", "A", "B", "A", "B", "A", "B"]
\ No newline at end of file
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/AllGrouping.java b/integration_test/src/java/com/twitter/heron/integration_test/topology/all_grouping/AllGrouping.java
similarity index 100%
rename from integration_test/src/java/com/twitter/heron/integration_test/topology/AllGrouping.java
rename to integration_test/src/java/com/twitter/heron/integration_test/topology/all_grouping/AllGrouping.java
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/OneSpoutTwoBolts.json b/integration_test/src/java/com/twitter/heron/integration_test/topology/all_grouping/AllGroupingResults.json
similarity index 100%
rename from integration_test/src/java/com/twitter/heron/integration_test/topology/OneSpoutTwoBolts.json
rename to integration_test/src/java/com/twitter/heron/integration_test/topology/all_grouping/AllGroupingResults.json
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/BasicTopologyOneTask.java b/integration_test/src/java/com/twitter/heron/integration_test/topology/basic_topology_one_task/BasicTopologyOneTask.java
similarity index 100%
rename from integration_test/src/java/com/twitter/heron/integration_test/topology/BasicTopologyOneTask.java
rename to integration_test/src/java/com/twitter/heron/integration_test/topology/basic_topology_one_task/BasicTopologyOneTask.java
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/BasicTopologyOneTask.json b/integration_test/src/java/com/twitter/heron/integration_test/topology/basic_topology_one_task/BasicTopologyOneTaskResults.json
similarity index 100%
rename from integration_test/src/java/com/twitter/heron/integration_test/topology/BasicTopologyOneTask.json
rename to integration_test/src/java/com/twitter/heron/integration_test/topology/basic_topology_one_task/BasicTopologyOneTaskResults.json
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/BoltDoubleEmitTuples.java b/integration_test/src/java/com/twitter/heron/integration_test/topology/bolt_double_emit_tuples/BoltDoubleEmitTuples.java
similarity index 100%
rename from integration_test/src/java/com/twitter/heron/integration_test/topology/BoltDoubleEmitTuples.java
rename to integration_test/src/java/com/twitter/heron/integration_test/topology/bolt_double_emit_tuples/BoltDoubleEmitTuples.java
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/OneSpoutTwoBolts.json b/integration_test/src/java/com/twitter/heron/integration_test/topology/bolt_double_emit_tuples/BoltDoubleEmitTuplesResults.json
similarity index 100%
copy from integration_test/src/java/com/twitter/heron/integration_test/topology/OneSpoutTwoBolts.json
copy to integration_test/src/java/com/twitter/heron/integration_test/topology/bolt_double_emit_tuples/BoltDoubleEmitTuplesResults.json
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/FieldsGrouping.java b/integration_test/src/java/com/twitter/heron/integration_test/topology/fields_grouping/FieldsGrouping.java
similarity index 100%
rename from integration_test/src/java/com/twitter/heron/integration_test/topology/FieldsGrouping.java
rename to integration_test/src/java/com/twitter/heron/integration_test/topology/fields_grouping/FieldsGrouping.java
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/FieldsGrouping.json b/integration_test/src/java/com/twitter/heron/integration_test/topology/fields_grouping/FieldsGroupingResults.json
similarity index 100%
rename from integration_test/src/java/com/twitter/heron/integration_test/topology/FieldsGrouping.json
rename to integration_test/src/java/com/twitter/heron/integration_test/topology/fields_grouping/FieldsGroupingResults.json
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/GlobalGrouping.java b/integration_test/src/java/com/twitter/heron/integration_test/topology/global_grouping/GlobalGrouping.java
similarity index 100%
rename from integration_test/src/java/com/twitter/heron/integration_test/topology/GlobalGrouping.java
rename to integration_test/src/java/com/twitter/heron/integration_test/topology/global_grouping/GlobalGrouping.java
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/GlobalGrouping.json b/integration_test/src/java/com/twitter/heron/integration_test/topology/global_grouping/GlobalGroupingResults.json
similarity index 100%
rename from integration_test/src/java/com/twitter/heron/integration_test/topology/GlobalGrouping.json
rename to integration_test/src/java/com/twitter/heron/integration_test/topology/global_grouping/GlobalGroupingResults.json
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/MultiSpoutsMultiTasks.java b/integration_test/src/java/com/twitter/heron/integration_test/topology/multi_spouts_multi_tasks/MultiSpoutsMultiTasks.java
similarity index 100%
rename from integration_test/src/java/com/twitter/heron/integration_test/topology/MultiSpoutsMultiTasks.java
rename to integration_test/src/java/com/twitter/heron/integration_test/topology/multi_spouts_multi_tasks/MultiSpoutsMultiTasks.java
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/MultiSpoutsMultiTasks.json b/integration_test/src/java/com/twitter/heron/integration_test/topology/multi_spouts_multi_tasks/MultiSpoutsMultiTasksResults.json
similarity index 100%
rename from integration_test/src/java/com/twitter/heron/integration_test/topology/MultiSpoutsMultiTasks.json
rename to integration_test/src/java/com/twitter/heron/integration_test/topology/multi_spouts_multi_tasks/MultiSpoutsMultiTasksResults.json
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/NonGrouping.java b/integration_test/src/java/com/twitter/heron/integration_test/topology/non_grouping/NonGrouping.java
similarity index 100%
rename from integration_test/src/java/com/twitter/heron/integration_test/topology/NonGrouping.java
rename to integration_test/src/java/com/twitter/heron/integration_test/topology/non_grouping/NonGrouping.java
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/OneBoltMultiTasks.json b/integration_test/src/java/com/twitter/heron/integration_test/topology/non_grouping/NonGroupingResults.json
similarity index 100%
rename from integration_test/src/java/com/twitter/heron/integration_test/topology/OneBoltMultiTasks.json
rename to integration_test/src/java/com/twitter/heron/integration_test/topology/non_grouping/NonGroupingResults.json
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/OneBoltMultiTasks.java b/integration_test/src/java/com/twitter/heron/integration_test/topology/one_bolt_multi_tasks/OneBoltMultiTasks.java
similarity index 100%
rename from integration_test/src/java/com/twitter/heron/integration_test/topology/OneBoltMultiTasks.java
rename to integration_test/src/java/com/twitter/heron/integration_test/topology/one_bolt_multi_tasks/OneBoltMultiTasks.java
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/OneBoltMultiTasks.json b/integration_test/src/java/com/twitter/heron/integration_test/topology/one_bolt_multi_tasks/OneBoltMultiTasksResults.json
similarity index 100%
copy from integration_test/src/java/com/twitter/heron/integration_test/topology/OneBoltMultiTasks.json
copy to integration_test/src/java/com/twitter/heron/integration_test/topology/one_bolt_multi_tasks/OneBoltMultiTasksResults.json
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/OneSpoutBoltMultiTasks.java b/integration_test/src/java/com/twitter/heron/integration_test/topology/one_spout_bolt_multi_tasks/OneSpoutBoltMultiTasks.java
similarity index 100%
rename from integration_test/src/java/com/twitter/heron/integration_test/topology/OneSpoutBoltMultiTasks.java
rename to integration_test/src/java/com/twitter/heron/integration_test/topology/one_spout_bolt_multi_tasks/OneSpoutBoltMultiTasks.java
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/OneSpoutBoltMultiTasks.json b/integration_test/src/java/com/twitter/heron/integration_test/topology/one_spout_bolt_multi_tasks/OneSpoutBoltMultiTasksResults.json
similarity index 100%
copy from integration_test/src/java/com/twitter/heron/integration_test/topology/OneSpoutBoltMultiTasks.json
copy to integration_test/src/java/com/twitter/heron/integration_test/topology/one_spout_bolt_multi_tasks/OneSpoutBoltMultiTasksResults.json
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/OneSpoutMultiTasks.java b/integration_test/src/java/com/twitter/heron/integration_test/topology/one_spout_multi_tasks/OneSpoutMultiTasks.java
similarity index 100%
rename from integration_test/src/java/com/twitter/heron/integration_test/topology/OneSpoutMultiTasks.java
rename to integration_test/src/java/com/twitter/heron/integration_test/topology/one_spout_multi_tasks/OneSpoutMultiTasks.java
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/OneSpoutBoltMultiTasks.json b/integration_test/src/java/com/twitter/heron/integration_test/topology/one_spout_multi_tasks/OneSpoutMultiTasksResults.json
similarity index 100%
rename from integration_test/src/java/com/twitter/heron/integration_test/topology/OneSpoutBoltMultiTasks.json
rename to integration_test/src/java/com/twitter/heron/integration_test/topology/one_spout_multi_tasks/OneSpoutMultiTasksResults.json
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/OneSpoutTwoBolts.java b/integration_test/src/java/com/twitter/heron/integration_test/topology/one_spout_two_bolts/OneSpoutTwoBolts.java
similarity index 100%
rename from integration_test/src/java/com/twitter/heron/integration_test/topology/OneSpoutTwoBolts.java
rename to integration_test/src/java/com/twitter/heron/integration_test/topology/one_spout_two_bolts/OneSpoutTwoBolts.java
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/OneSpoutTwoBolts.json b/integration_test/src/java/com/twitter/heron/integration_test/topology/one_spout_two_bolts/OneSpoutTwoBoltsResults.json
similarity index 100%
copy from integration_test/src/java/com/twitter/heron/integration_test/topology/OneSpoutTwoBolts.json
copy to integration_test/src/java/com/twitter/heron/integration_test/topology/one_spout_two_bolts/OneSpoutTwoBoltsResults.json
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/CustomCheckBolt.java b/integration_test/src/java/com/twitter/heron/integration_test/topology/serialization/CustomCheckBolt.java
similarity index 100%
rename from integration_test/src/java/com/twitter/heron/integration_test/topology/CustomCheckBolt.java
rename to integration_test/src/java/com/twitter/heron/integration_test/topology/serialization/CustomCheckBolt.java
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/CustomObject.java b/integration_test/src/java/com/twitter/heron/integration_test/topology/serialization/CustomObject.java
similarity index 100%
rename from integration_test/src/java/com/twitter/heron/integration_test/topology/CustomObject.java
rename to integration_test/src/java/com/twitter/heron/integration_test/topology/serialization/CustomObject.java
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/CustomSpout.java b/integration_test/src/java/com/twitter/heron/integration_test/topology/serialization/CustomSpout.java
similarity index 100%
rename from integration_test/src/java/com/twitter/heron/integration_test/topology/CustomSpout.java
rename to integration_test/src/java/com/twitter/heron/integration_test/topology/serialization/CustomSpout.java
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/SerializationTopology.java b/integration_test/src/java/com/twitter/heron/integration_test/topology/serialization/SerializationTopology.java
similarity index 100%
rename from integration_test/src/java/com/twitter/heron/integration_test/topology/SerializationTopology.java
rename to integration_test/src/java/com/twitter/heron/integration_test/topology/serialization/SerializationTopology.java
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/SerializationTopology.json b/integration_test/src/java/com/twitter/heron/integration_test/topology/serialization/SerializationTopologyResults.json
similarity index 100%
rename from integration_test/src/java/com/twitter/heron/integration_test/topology/SerializationTopology.json
rename to integration_test/src/java/com/twitter/heron/integration_test/topology/serialization/SerializationTopologyResults.json
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/ShuffleGrouping.java b/integration_test/src/java/com/twitter/heron/integration_test/topology/shuffle_grouping/ShuffleGrouping.java
similarity index 100%
rename from integration_test/src/java/com/twitter/heron/integration_test/topology/ShuffleGrouping.java
rename to integration_test/src/java/com/twitter/heron/integration_test/topology/shuffle_grouping/ShuffleGrouping.java
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/OneBoltMultiTasks.json b/integration_test/src/java/com/twitter/heron/integration_test/topology/shuffle_grouping/ShuffleGroupingResults.json
similarity index 100%
copy from integration_test/src/java/com/twitter/heron/integration_test/topology/OneBoltMultiTasks.json
copy to integration_test/src/java/com/twitter/heron/integration_test/topology/shuffle_grouping/ShuffleGroupingResults.json
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/WindowTestBase.java b/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/WindowTestBase.java
new file mode 100644
index 0000000..8647bb1
--- /dev/null
+++ b/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/WindowTestBase.java
@@ -0,0 +1,260 @@
+//  Copyright 2017 Twitter. All rights reserved.
+//
+//  Licensed 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.twitter.heron.integration_test.topology.windowing;
+
+import java.net.MalformedURLException;
+import java.time.Duration;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.logging.Logger;
+
+import org.json.simple.JSONArray;
+import org.json.simple.JSONObject;
+
+import com.twitter.heron.api.bolt.BaseWindowedBolt;
+import com.twitter.heron.api.bolt.OutputCollector;
+import com.twitter.heron.api.spout.BaseRichSpout;
+import com.twitter.heron.api.spout.SpoutOutputCollector;
+import com.twitter.heron.api.topology.OutputFieldsDeclarer;
+import com.twitter.heron.api.topology.TopologyContext;
+import com.twitter.heron.api.tuple.Fields;
+import com.twitter.heron.api.tuple.Tuple;
+import com.twitter.heron.api.tuple.Values;
+import com.twitter.heron.api.utils.Utils;
+import com.twitter.heron.api.windowing.TupleWindow;
+import com.twitter.heron.integration_test.common.AbstractTestTopology;
+import com.twitter.heron.integration_test.core.TestTopologyBuilder;
+
+public class WindowTestBase extends AbstractTestTopology {
+
+  private static final String NUMBER_FIELD = "number";
+  private static final String STRING_FIELD = "numAsStr";
+  private static final String EVENT_TIME_FIELD = "eventTimeField";
+  private static final String DUMMY_FIELD = "dummy";
+
+  private BaseWindowedBolt.Count windowCountLength;
+  private BaseWindowedBolt.Count slideCountInterval;
+  private Duration windowDurationLength;
+  private Duration slideDurationInterval;
+  private Long sleepBetweenTuples;
+  private boolean useEventTime = false;
+  private Duration watermarkInterval;
+
+
+  public String getBoltName() {
+    return "VerificationBolt";
+  }
+
+  public String getSpoutName() {
+    return "IncrementingSpout";
+  }
+
+
+  public WindowTestBase(String[] args) throws MalformedURLException {
+    super(args);
+  }
+
+  public WindowTestBase withWindowLength(Duration duration) {
+    this.windowDurationLength = duration;
+    return this;
+  }
+
+  public WindowTestBase withWindowLength(BaseWindowedBolt.Count count) {
+    this.windowCountLength = count;
+    return this;
+  }
+
+  public WindowTestBase withSlidingInterval(Duration duration) {
+    this.slideDurationInterval = duration;
+    return this;
+  }
+
+  public WindowTestBase withSlidingInterval(BaseWindowedBolt.Count count) {
+    this.slideCountInterval = count;
+    return this;
+  }
+
+  public WindowTestBase withSleepBetweenTuples(long millis) {
+    this.sleepBetweenTuples = millis;
+    return this;
+  }
+
+  public WindowTestBase useEventTime() {
+    this.useEventTime = true;
+    return this;
+  }
+
+  @SuppressWarnings("HiddenField")
+  public WindowTestBase withWatermarkInterval(Duration watermarkInterval) {
+    this.watermarkInterval = watermarkInterval;
+    return this;
+  }
+
+  @Override
+  protected TestTopologyBuilder buildTopology(TestTopologyBuilder builder) {
+    builder.setSpout(getSpoutName(), new IncrementingSpout()
+        .withSleepBetweenTuples(sleepBetweenTuples), 1);
+
+    BaseWindowedBolt bolt = null;
+    if (this.windowCountLength != null) {
+      if (this.slideCountInterval != null) {
+        bolt = new VerificationBolt()
+            .withWindow(this.windowCountLength, this.slideCountInterval);
+      } else {
+        bolt = new VerificationBolt()
+            .withTumblingWindow(this.windowCountLength);
+      }
+    }
+
+    if (this.windowDurationLength != null) {
+      if (this.slideDurationInterval != null) {
+        bolt = new VerificationBolt()
+            .withWindow(this.windowDurationLength, this.slideDurationInterval);
+      } else {
+        bolt = new VerificationBolt()
+            .withTumblingWindow(this.windowDurationLength);
+      }
+    }
+
+    if (this.useEventTime) {
+      bolt.withTimestampField(EVENT_TIME_FIELD);
+    }
+
+    if (this.watermarkInterval != null) {
+      bolt.withWatermarkInterval(this.watermarkInterval);
+    }
+    builder.setBolt(getBoltName(), bolt, 1, false)
+        .shuffleGrouping(getSpoutName());
+    return builder;
+  }
+
+  public static class IncrementingSpout extends BaseRichSpout {
+    private static final Logger LOG = Logger.getLogger(IncrementingSpout.class.getName());
+    private static final long serialVersionUID = -6171170228097868632L;
+    private SpoutOutputCollector collector;
+    private static int currentNum;
+    private static Random rng = new Random();
+    private String componentId;
+    private Long sleepBetweenTuples = null;
+    // In millis
+    private long currentTime = 1504573536000L;
+
+    public IncrementingSpout withSleepBetweenTuples(long millis) {
+      this.sleepBetweenTuples = millis;
+      return this;
+    }
+
+    @Override
+    public void declareOutputFields(OutputFieldsDeclarer declarer) {
+      declarer.declare(new Fields(NUMBER_FIELD, STRING_FIELD, EVENT_TIME_FIELD));
+    }
+
+    @Override
+    @SuppressWarnings("HiddenField")
+    public void open(Map<String, Object> conf, TopologyContext context, SpoutOutputCollector
+        collector) {
+      componentId = context.getThisComponentId();
+      this.collector = collector;
+    }
+
+    @Override
+    public void nextTuple() {
+      if (sleepBetweenTuples == null) {
+        Utils.sleep(rng.nextInt(10));
+      } else {
+        Utils.sleep(sleepBetweenTuples);
+      }
+      currentNum++;
+      final String numAsStr = "str(" + currentNum + ")str";
+      final Values tuple = new Values(currentNum, numAsStr, currentTime += 1000);
+      LOG.info("Time = " + System.currentTimeMillis()
+          + " Component = " + componentId + " Tuple = " + tuple.toString());
+
+      collector.emit(tuple, currentNum);
+    }
+
+    @Override
+    public void ack(Object msgId) {
+      LOG.info("Received ACK for msgId : " + msgId);
+    }
+
+    @Override
+    public void fail(Object msgId) {
+      LOG.info("Received FAIL for msgId : " + msgId);
+    }
+  }
+
+  public static class VerificationBolt extends BaseWindowedBolt {
+    private static final long serialVersionUID = -6067634845003700125L;
+    private OutputCollector collector;
+    private String componentId;
+
+    private static int windowCount = 0;
+
+
+    private static final Logger LOG = Logger.getLogger(VerificationBolt.class.getName());
+
+
+    @Override
+    @SuppressWarnings("HiddenField")
+    public void prepare(Map<String, Object> topoConf, TopologyContext context, OutputCollector
+        collector) {
+      componentId = context.getThisComponentId();
+      this.collector = collector;
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    public void execute(TupleWindow inputWindow) {
+      windowCount++;
+      List<Tuple> tuplesInWindow = inputWindow.get();
+      List<Tuple> newTuples = inputWindow.getNew();
+      List<Tuple> expiredTuples = inputWindow.getExpired();
+      LOG.info("tuplesInWindow.size() = " + tuplesInWindow.size());
+      LOG.info("newTuples.size() = " + newTuples.size());
+      LOG.info("expiredTuples.size() = " + expiredTuples.size());
+
+      JSONObject jsonObject = new JSONObject();
+      JSONArray jsonArray = new JSONArray();
+      JSONObject tmp = new JSONObject();
+      tmp.put("tuplesInWindow", tuplesToListString(tuplesInWindow));
+      jsonArray.add(tmp);
+      tmp = new JSONObject();
+      tmp.put("newTuples", tuplesToListString(newTuples));
+      jsonArray.add(tmp);
+      tmp = new JSONObject();
+      tmp.put("expiredTuples", tuplesToListString(expiredTuples));
+      jsonArray.add(tmp);
+      jsonObject.put(windowCount, jsonArray);
+      LOG.info("Component = " + componentId + " Window Count = " + windowCount
+          + " tuplesInWindow = " + jsonArray.toJSONString());
+
+      collector.emit(new Values(jsonObject.toJSONString()));
+    }
+
+    public void declareOutputFields(OutputFieldsDeclarer declarer) {
+      declarer.declare(new Fields(DUMMY_FIELD));
+    }
+  }
+
+  public static List<String> tuplesToListString(List<Tuple> tuples) {
+    List<String> tmp = new LinkedList<>();
+    for (Tuple tuple : tuples) {
+      tmp.add(tuple.getValue(0).toString());
+    }
+    return tmp;
+  }
+}
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/count/SlidingCountWindowTest1.java b/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/count/SlidingCountWindowTest1.java
new file mode 100644
index 0000000..b77ec65
--- /dev/null
+++ b/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/count/SlidingCountWindowTest1.java
@@ -0,0 +1,34 @@
+//  Copyright 2017 Twitter. All rights reserved.
+//
+//  Licensed 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.twitter.heron.integration_test.topology.windowing.count;
+
+import java.net.MalformedURLException;
+
+import com.twitter.heron.api.bolt.BaseWindowedBolt;
+import com.twitter.heron.integration_test.topology.windowing.WindowTestBase;
+
+public class SlidingCountWindowTest1 extends WindowTestBase {
+
+  public SlidingCountWindowTest1(String[] args) throws MalformedURLException {
+    super(args);
+  }
+
+  public static void main(String[] args) throws Exception {
+    WindowTestBase topology = new SlidingCountWindowTest1(args)
+        .withSleepBetweenTuples(1000)
+        .withWindowLength(BaseWindowedBolt.Count.of(10))
+        .withSlidingInterval(BaseWindowedBolt.Count.of(5));
+    topology.submit();
+  }
+}
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/count/SlidingCountWindowTest1Results.json b/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/count/SlidingCountWindowTest1Results.json
new file mode 100644
index 0000000..e8680ad
--- /dev/null
+++ b/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/count/SlidingCountWindowTest1Results.json
@@ -0,0 +1,4 @@
+[
+  "{\"1\":[{\"tuplesInWindow\":[\"1\",\"2\",\"3\",\"4\",\"5\"]},{\"newTuples\":[\"1\",\"2\",\"3\",\"4\",\"5\"]},{\"expiredTuples\":[]}]}",
+  "{\"2\":[{\"tuplesInWindow\":[\"1\",\"2\",\"3\",\"4\",\"5\",\"6\",\"7\",\"8\",\"9\",\"10\"]},{\"newTuples\":[\"6\",\"7\",\"8\",\"9\",\"10\"]},{\"expiredTuples\":[]}]}"
+]
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/count/SlidingCountWindowTest2.java b/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/count/SlidingCountWindowTest2.java
new file mode 100644
index 0000000..b8bd7e6
--- /dev/null
+++ b/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/count/SlidingCountWindowTest2.java
@@ -0,0 +1,34 @@
+//  Copyright 2017 Twitter. All rights reserved.
+//
+//  Licensed 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.twitter.heron.integration_test.topology.windowing.count;
+
+import java.net.MalformedURLException;
+
+import com.twitter.heron.api.bolt.BaseWindowedBolt;
+import com.twitter.heron.integration_test.topology.windowing.WindowTestBase;
+
+public class SlidingCountWindowTest2 extends WindowTestBase {
+
+  public SlidingCountWindowTest2(String[] args) throws MalformedURLException {
+    super(args);
+  }
+
+  public static void main(String[] args) throws Exception {
+    WindowTestBase topology = new SlidingCountWindowTest2(args)
+        .withSleepBetweenTuples(1000)
+        .withWindowLength(BaseWindowedBolt.Count.of(2))
+        .withSlidingInterval(BaseWindowedBolt.Count.of(1));
+    topology.submit();
+  }
+}
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/count/SlidingCountWindowTest2Results.json b/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/count/SlidingCountWindowTest2Results.json
new file mode 100644
index 0000000..970469f
--- /dev/null
+++ b/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/count/SlidingCountWindowTest2Results.json
@@ -0,0 +1,12 @@
+[
+  "{\"1\":[{\"tuplesInWindow\":[\"1\"]},{\"newTuples\":[\"1\"]},{\"expiredTuples\":[]}]}",
+  "{\"2\":[{\"tuplesInWindow\":[\"1\",\"2\"]},{\"newTuples\":[\"2\"]},{\"expiredTuples\":[]}]}",
+  "{\"3\":[{\"tuplesInWindow\":[\"2\",\"3\"]},{\"newTuples\":[\"3\"]},{\"expiredTuples\":[\"1\"]}]}",
+  "{\"4\":[{\"tuplesInWindow\":[\"3\",\"4\"]},{\"newTuples\":[\"4\"]},{\"expiredTuples\":[\"2\"]}]}",
+  "{\"5\":[{\"tuplesInWindow\":[\"4\",\"5\"]},{\"newTuples\":[\"5\"]},{\"expiredTuples\":[\"3\"]}]}",
+  "{\"6\":[{\"tuplesInWindow\":[\"5\",\"6\"]},{\"newTuples\":[\"6\"]},{\"expiredTuples\":[\"4\"]}]}",
+  "{\"7\":[{\"tuplesInWindow\":[\"6\",\"7\"]},{\"newTuples\":[\"7\"]},{\"expiredTuples\":[\"5\"]}]}",
+  "{\"8\":[{\"tuplesInWindow\":[\"7\",\"8\"]},{\"newTuples\":[\"8\"]},{\"expiredTuples\":[\"6\"]}]}",
+  "{\"9\":[{\"tuplesInWindow\":[\"8\",\"9\"]},{\"newTuples\":[\"9\"]},{\"expiredTuples\":[\"7\"]}]}",
+  "{\"10\":[{\"tuplesInWindow\":[\"9\",\"10\"]},{\"newTuples\":[\"10\"]},{\"expiredTuples\":[\"8\"]}]}"
+]
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/count/SlidingCountWindowTest3.java b/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/count/SlidingCountWindowTest3.java
new file mode 100644
index 0000000..1080c88
--- /dev/null
+++ b/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/count/SlidingCountWindowTest3.java
@@ -0,0 +1,34 @@
+//  Copyright 2017 Twitter. All rights reserved.
+//
+//  Licensed 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.twitter.heron.integration_test.topology.windowing.count;
+
+import java.net.MalformedURLException;
+
+import com.twitter.heron.api.bolt.BaseWindowedBolt;
+import com.twitter.heron.integration_test.topology.windowing.WindowTestBase;
+
+public class SlidingCountWindowTest3 extends WindowTestBase {
+
+  public SlidingCountWindowTest3(String[] args) throws MalformedURLException {
+    super(args);
+  }
+
+  public static void main(String[] args) throws Exception {
+    WindowTestBase topology = new SlidingCountWindowTest3(args)
+        .withSleepBetweenTuples(1000)
+        .withWindowLength(BaseWindowedBolt.Count.of(7))
+        .withSlidingInterval(BaseWindowedBolt.Count.of(3));
+    topology.submit();
+  }
+}
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/count/SlidingCountWindowTest3Results.json b/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/count/SlidingCountWindowTest3Results.json
new file mode 100644
index 0000000..975f01c
--- /dev/null
+++ b/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/count/SlidingCountWindowTest3Results.json
@@ -0,0 +1,5 @@
+[
+  "{\"1\":[{\"tuplesInWindow\":[\"1\",\"2\",\"3\"]},{\"newTuples\":[\"1\",\"2\",\"3\"]},{\"expiredTuples\":[]}]}",
+  "{\"2\":[{\"tuplesInWindow\":[\"1\",\"2\",\"3\",\"4\",\"5\",\"6\"]},{\"newTuples\":[\"4\",\"5\",\"6\"]},{\"expiredTuples\":[]}]}",
+  "{\"3\":[{\"tuplesInWindow\":[\"3\",\"4\",\"5\",\"6\",\"7\",\"8\",\"9\"]},{\"newTuples\":[\"7\",\"8\",\"9\"]},{\"expiredTuples\":[\"1\",\"2\"]}]}"
+]
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/count/TumblingCountWindowTest1.java b/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/count/TumblingCountWindowTest1.java
new file mode 100644
index 0000000..3f9e612
--- /dev/null
+++ b/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/count/TumblingCountWindowTest1.java
@@ -0,0 +1,33 @@
+//  Copyright 2017 Twitter. All rights reserved.
+//
+//  Licensed 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.twitter.heron.integration_test.topology.windowing.count;
+
+import java.net.MalformedURLException;
+
+import com.twitter.heron.api.bolt.BaseWindowedBolt;
+import com.twitter.heron.integration_test.topology.windowing.WindowTestBase;
+
+public class TumblingCountWindowTest1 extends WindowTestBase {
+
+  public TumblingCountWindowTest1(String[] args) throws MalformedURLException {
+    super(args);
+  }
+
+  public static void main(String[] args) throws Exception {
+    WindowTestBase topology = new TumblingCountWindowTest1(args)
+        .withSleepBetweenTuples(1000)
+        .withWindowLength(BaseWindowedBolt.Count.of(10));
+    topology.submit();
+  }
+}
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/count/TumblingCountWindowTest1Results.json b/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/count/TumblingCountWindowTest1Results.json
new file mode 100644
index 0000000..2416022
--- /dev/null
+++ b/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/count/TumblingCountWindowTest1Results.json
@@ -0,0 +1,3 @@
+[
+  "{\"1\":[{\"tuplesInWindow\":[\"1\",\"2\",\"3\",\"4\",\"5\",\"6\",\"7\",\"8\",\"9\",\"10\"]},{\"newTuples\":[\"1\",\"2\",\"3\",\"4\",\"5\",\"6\",\"7\",\"8\",\"9\",\"10\"]},{\"expiredTuples\":[]}]}"
+]
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/count/TumblingCountWindowTest2.java b/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/count/TumblingCountWindowTest2.java
new file mode 100644
index 0000000..15c80d6
--- /dev/null
+++ b/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/count/TumblingCountWindowTest2.java
@@ -0,0 +1,33 @@
+//  Copyright 2017 Twitter. All rights reserved.
+//
+//  Licensed 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.twitter.heron.integration_test.topology.windowing.count;
+
+import java.net.MalformedURLException;
+
+import com.twitter.heron.api.bolt.BaseWindowedBolt;
+import com.twitter.heron.integration_test.topology.windowing.WindowTestBase;
+
+public class TumblingCountWindowTest2 extends WindowTestBase {
+
+  public TumblingCountWindowTest2(String[] args) throws MalformedURLException {
+    super(args);
+  }
+
+  public static void main(String[] args) throws Exception {
+    WindowTestBase topology = new TumblingCountWindowTest2(args)
+        .withSleepBetweenTuples(1000)
+        .withWindowLength(BaseWindowedBolt.Count.of(2));
+    topology.submit();
+  }
+}
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/count/TumblingCountWindowTest2Results.json b/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/count/TumblingCountWindowTest2Results.json
new file mode 100644
index 0000000..e258923
--- /dev/null
+++ b/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/count/TumblingCountWindowTest2Results.json
@@ -0,0 +1,7 @@
+[
+  "{\"1\":[{\"tuplesInWindow\":[\"1\",\"2\"]},{\"newTuples\":[\"1\",\"2\"]},{\"expiredTuples\":[]}]}",
+  "{\"2\":[{\"tuplesInWindow\":[\"3\",\"4\"]},{\"newTuples\":[\"3\",\"4\"]},{\"expiredTuples\":[\"1\",\"2\"]}]}",
+  "{\"3\":[{\"tuplesInWindow\":[\"5\",\"6\"]},{\"newTuples\":[\"5\",\"6\"]},{\"expiredTuples\":[\"3\",\"4\"]}]}",
+  "{\"4\":[{\"tuplesInWindow\":[\"7\",\"8\"]},{\"newTuples\":[\"7\",\"8\"]},{\"expiredTuples\":[\"5\",\"6\"]}]}",
+  "{\"5\":[{\"tuplesInWindow\":[\"9\",\"10\"]},{\"newTuples\":[\"9\",\"10\"]},{\"expiredTuples\":[\"7\",\"8\"]}]}"
+]
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/count/TumblingCountWindowTest3.java b/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/count/TumblingCountWindowTest3.java
new file mode 100644
index 0000000..227b1cc
--- /dev/null
+++ b/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/count/TumblingCountWindowTest3.java
@@ -0,0 +1,33 @@
+//  Copyright 2017 Twitter. All rights reserved.
+//
+//  Licensed 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.twitter.heron.integration_test.topology.windowing.count;
+
+import java.net.MalformedURLException;
+
+import com.twitter.heron.api.bolt.BaseWindowedBolt;
+import com.twitter.heron.integration_test.topology.windowing.WindowTestBase;
+
+public class TumblingCountWindowTest3 extends WindowTestBase {
+
+  public TumblingCountWindowTest3(String[] args) throws MalformedURLException {
+    super(args);
+  }
+
+  public static void main(String[] args) throws Exception {
+    WindowTestBase topology = new TumblingCountWindowTest3(args)
+        .withSleepBetweenTuples(1000)
+        .withWindowLength(BaseWindowedBolt.Count.of(7));
+    topology.submit();
+  }
+}
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/count/TumblingCountWindowTest3Results.json b/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/count/TumblingCountWindowTest3Results.json
new file mode 100644
index 0000000..8969a9a
--- /dev/null
+++ b/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/count/TumblingCountWindowTest3Results.json
@@ -0,0 +1,3 @@
+[
+  "{\"1\":[{\"tuplesInWindow\":[\"1\",\"2\",\"3\",\"4\",\"5\",\"6\",\"7\"]},{\"newTuples\":[\"1\",\"2\",\"3\",\"4\",\"5\",\"6\",\"7\"]},{\"expiredTuples\":[]}]}"
+]
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/time/SlidingTimeWindowTest1.java b/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/time/SlidingTimeWindowTest1.java
new file mode 100644
index 0000000..120fdea
--- /dev/null
+++ b/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/time/SlidingTimeWindowTest1.java
@@ -0,0 +1,32 @@
+//  Copyright 2017 Twitter. All rights reserved.
+//
+//  Licensed 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.twitter.heron.integration_test.topology.windowing.time;
+
+import java.net.MalformedURLException;
+import java.time.Duration;
+
+import com.twitter.heron.integration_test.topology.windowing.WindowTestBase;
+
+public class SlidingTimeWindowTest1 extends WindowTestBase {
+  public SlidingTimeWindowTest1(String[] args) throws MalformedURLException {
+    super(args);
+  }
+
+  public static void main(String[] args) throws Exception {
+    WindowTestBase topology = new SlidingTimeWindowTest1(args).withSleepBetweenTuples(1000)
+        .withWindowLength(Duration.ofMillis(1000))
+        .withSlidingInterval(Duration.ofMillis(1000));
+    topology.submit();
+  }
+}
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/time/SlidingTimeWindowTest1Results.json b/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/time/SlidingTimeWindowTest1Results.json
new file mode 100644
index 0000000..7af261c
--- /dev/null
+++ b/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/time/SlidingTimeWindowTest1Results.json
@@ -0,0 +1,12 @@
+[
+  "{\"1\":[{\"tuplesInWindow\":[\"1\"]},{\"newTuples\":[\"1\"]},{\"expiredTuples\":[]}]}",
+  "{\"10\":[{\"tuplesInWindow\":[\"10\"]},{\"newTuples\":[\"10\"]},{\"expiredTuples\":[\"9\"]}]}",
+  "{\"2\":[{\"tuplesInWindow\":[\"2\"]},{\"newTuples\":[\"2\"]},{\"expiredTuples\":[\"1\"]}]}",
+  "{\"3\":[{\"tuplesInWindow\":[\"3\"]},{\"newTuples\":[\"3\"]},{\"expiredTuples\":[\"2\"]}]}",
+  "{\"4\":[{\"tuplesInWindow\":[\"4\"]},{\"newTuples\":[\"4\"]},{\"expiredTuples\":[\"3\"]}]}",
+  "{\"5\":[{\"tuplesInWindow\":[\"5\"]},{\"newTuples\":[\"5\"]},{\"expiredTuples\":[\"4\"]}]}",
+  "{\"6\":[{\"tuplesInWindow\":[\"6\"]},{\"newTuples\":[\"6\"]},{\"expiredTuples\":[\"5\"]}]}",
+  "{\"7\":[{\"tuplesInWindow\":[\"7\"]},{\"newTuples\":[\"7\"]},{\"expiredTuples\":[\"6\"]}]}",
+  "{\"8\":[{\"tuplesInWindow\":[\"8\"]},{\"newTuples\":[\"8\"]},{\"expiredTuples\":[\"7\"]}]}",
+  "{\"9\":[{\"tuplesInWindow\":[\"9\"]},{\"newTuples\":[\"9\"]},{\"expiredTuples\":[\"8\"]}]}"
+]
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/watermark/SlidingWatermarkEventTimeWindowTest1.java b/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/watermark/SlidingWatermarkEventTimeWindowTest1.java
new file mode 100644
index 0000000..2077c3b
--- /dev/null
+++ b/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/watermark/SlidingWatermarkEventTimeWindowTest1.java
@@ -0,0 +1,36 @@
+//  Copyright 2017 Twitter. All rights reserved.
+//
+//  Licensed 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.twitter.heron.integration_test.topology.windowing.watermark;
+
+import java.net.MalformedURLException;
+import java.time.Duration;
+
+import com.twitter.heron.integration_test.topology.windowing.WindowTestBase;
+
+public class SlidingWatermarkEventTimeWindowTest1 extends WindowTestBase {
+
+  public SlidingWatermarkEventTimeWindowTest1(String[] args) throws MalformedURLException {
+    super(args);
+  }
+
+  public static void main(String[] args) throws Exception {
+    WindowTestBase topology = new SlidingWatermarkEventTimeWindowTest1(args)
+        .withSleepBetweenTuples(1000)
+        .useEventTime()
+        .withWatermarkInterval(Duration.ofMillis(1000))
+        .withWindowLength(Duration.ofMillis(1000))
+        .withSlidingInterval(Duration.ofMillis(1000));
+    topology.submit();
+  }
+}
diff --git a/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/watermark/SlidingWatermarkEventTimeWindowTest1Results.json b/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/watermark/SlidingWatermarkEventTimeWindowTest1Results.json
new file mode 100644
index 0000000..7af261c
--- /dev/null
+++ b/integration_test/src/java/com/twitter/heron/integration_test/topology/windowing/watermark/SlidingWatermarkEventTimeWindowTest1Results.json
@@ -0,0 +1,12 @@
+[
+  "{\"1\":[{\"tuplesInWindow\":[\"1\"]},{\"newTuples\":[\"1\"]},{\"expiredTuples\":[]}]}",
+  "{\"10\":[{\"tuplesInWindow\":[\"10\"]},{\"newTuples\":[\"10\"]},{\"expiredTuples\":[\"9\"]}]}",
+  "{\"2\":[{\"tuplesInWindow\":[\"2\"]},{\"newTuples\":[\"2\"]},{\"expiredTuples\":[\"1\"]}]}",
+  "{\"3\":[{\"tuplesInWindow\":[\"3\"]},{\"newTuples\":[\"3\"]},{\"expiredTuples\":[\"2\"]}]}",
+  "{\"4\":[{\"tuplesInWindow\":[\"4\"]},{\"newTuples\":[\"4\"]},{\"expiredTuples\":[\"3\"]}]}",
+  "{\"5\":[{\"tuplesInWindow\":[\"5\"]},{\"newTuples\":[\"5\"]},{\"expiredTuples\":[\"4\"]}]}",
+  "{\"6\":[{\"tuplesInWindow\":[\"6\"]},{\"newTuples\":[\"6\"]},{\"expiredTuples\":[\"5\"]}]}",
+  "{\"7\":[{\"tuplesInWindow\":[\"7\"]},{\"newTuples\":[\"7\"]},{\"expiredTuples\":[\"6\"]}]}",
+  "{\"8\":[{\"tuplesInWindow\":[\"8\"]},{\"newTuples\":[\"8\"]},{\"expiredTuples\":[\"7\"]}]}",
+  "{\"9\":[{\"tuplesInWindow\":[\"9\"]},{\"newTuples\":[\"9\"]},{\"expiredTuples\":[\"8\"]}]}"
+]
diff --git a/integration_test/src/python/__init__.py b/integration_test/src/python/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/integration_test/src/python/__init__.py
diff --git a/integration_test/src/python/integration_test/topology/__init__.py b/integration_test/src/python/integration_test/topology/__init__.py
index 53d0931..8b13789 100644
--- a/integration_test/src/python/integration_test/topology/__init__.py
+++ b/integration_test/src/python/integration_test/topology/__init__.py
@@ -1,19 +1 @@
-"""Integration test topologies"""
-__all__ = [
-    'all_grouping',
-    'basic_one_task'
-    'bolt_double_emit_tuples',
-    'fields_grouping',
-    'global_grouping',
-    'multi_spouts_multi_tasks',
-    'none_grouping',
-    'one_bolt_multi_tasks',
-    'one_spout_bolt_multi_tasks',
-    'one_spout_multi_tasks',
-    'one_spout_two_bolts',
-    'shuffle_grouping',
-]
 
-# import some core stuff
-import integration_test.src.python.integration_test.common as common
-import integration_test.src.python.integration_test.core as core
diff --git a/integration_test/src/python/integration_test/topology/all_grouping/__init__.py b/integration_test/src/python/integration_test/topology/all_grouping/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/integration_test/src/python/integration_test/topology/all_grouping/__init__.py
diff --git a/integration_test/src/python/integration_test/topology/all_grouping.py b/integration_test/src/python/integration_test/topology/all_grouping/all_grouping.py
similarity index 95%
rename from integration_test/src/python/integration_test/topology/all_grouping.py
rename to integration_test/src/python/integration_test/topology/all_grouping/all_grouping.py
index fa96200..02f569f 100644
--- a/integration_test/src/python/integration_test/topology/all_grouping.py
+++ b/integration_test/src/python/integration_test/topology/all_grouping/all_grouping.py
@@ -19,7 +19,8 @@
 from integration_test.src.python.integration_test.common.bolt import IdentityBolt
 from integration_test.src.python.integration_test.common.spout import ABSpout
 
-def all_grouping_buidler(topology_name, http_server_url):
+
+def all_grouping_builder(topology_name, http_server_url):
   """Integration test topology builder for all grouping"""
   builder = TestTopologyBuilder(topology_name, http_server_url)
   ab_spout = builder.add_spout("ab-spout", ABSpout, 1)
diff --git a/integration_test/src/python/integration_test/topology/all_grouping_result.json b/integration_test/src/python/integration_test/topology/all_grouping/all_grouping_result.json
similarity index 100%
rename from integration_test/src/python/integration_test/topology/all_grouping_result.json
rename to integration_test/src/python/integration_test/topology/all_grouping/all_grouping_result.json
diff --git a/integration_test/src/python/integration_test/topology/basic_one_task/__init__.py b/integration_test/src/python/integration_test/topology/basic_one_task/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/integration_test/src/python/integration_test/topology/basic_one_task/__init__.py
diff --git a/integration_test/src/python/integration_test/topology/basic_one_task.py b/integration_test/src/python/integration_test/topology/basic_one_task/basic_one_task.py
similarity index 100%
rename from integration_test/src/python/integration_test/topology/basic_one_task.py
rename to integration_test/src/python/integration_test/topology/basic_one_task/basic_one_task.py
diff --git a/integration_test/src/python/integration_test/topology/basic_one_task_result.json b/integration_test/src/python/integration_test/topology/basic_one_task/basic_one_task_result.json
similarity index 100%
rename from integration_test/src/python/integration_test/topology/basic_one_task_result.json
rename to integration_test/src/python/integration_test/topology/basic_one_task/basic_one_task_result.json
diff --git a/integration_test/src/python/integration_test/topology/bolt_double_emit_tuples/__init__.py b/integration_test/src/python/integration_test/topology/bolt_double_emit_tuples/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/integration_test/src/python/integration_test/topology/bolt_double_emit_tuples/__init__.py
diff --git a/integration_test/src/python/integration_test/topology/bolt_double_emit_tuples.py b/integration_test/src/python/integration_test/topology/bolt_double_emit_tuples/bolt_double_emit_tuples.py
similarity index 100%
rename from integration_test/src/python/integration_test/topology/bolt_double_emit_tuples.py
rename to integration_test/src/python/integration_test/topology/bolt_double_emit_tuples/bolt_double_emit_tuples.py
diff --git a/integration_test/src/python/integration_test/topology/bolt_double_emit_tuples_result.json b/integration_test/src/python/integration_test/topology/bolt_double_emit_tuples/bolt_double_emit_tuples_result.json
similarity index 100%
rename from integration_test/src/python/integration_test/topology/bolt_double_emit_tuples_result.json
rename to integration_test/src/python/integration_test/topology/bolt_double_emit_tuples/bolt_double_emit_tuples_result.json
diff --git a/integration_test/src/python/integration_test/topology/fields_grouping/__init__.py b/integration_test/src/python/integration_test/topology/fields_grouping/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/integration_test/src/python/integration_test/topology/fields_grouping/__init__.py
diff --git a/integration_test/src/python/integration_test/topology/fields_grouping.py b/integration_test/src/python/integration_test/topology/fields_grouping/fields_grouping.py
similarity index 100%
rename from integration_test/src/python/integration_test/topology/fields_grouping.py
rename to integration_test/src/python/integration_test/topology/fields_grouping/fields_grouping.py
diff --git a/integration_test/src/python/integration_test/topology/fields_grouping_result.json b/integration_test/src/python/integration_test/topology/fields_grouping/fields_grouping_result.json
similarity index 100%
rename from integration_test/src/python/integration_test/topology/fields_grouping_result.json
rename to integration_test/src/python/integration_test/topology/fields_grouping/fields_grouping_result.json
diff --git a/integration_test/src/python/integration_test/topology/global_grouping/__init__.py b/integration_test/src/python/integration_test/topology/global_grouping/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/integration_test/src/python/integration_test/topology/global_grouping/__init__.py
diff --git a/integration_test/src/python/integration_test/topology/global_grouping.py b/integration_test/src/python/integration_test/topology/global_grouping/global_grouping.py
similarity index 100%
rename from integration_test/src/python/integration_test/topology/global_grouping.py
rename to integration_test/src/python/integration_test/topology/global_grouping/global_grouping.py
diff --git a/integration_test/src/python/integration_test/topology/global_grouping_result.json b/integration_test/src/python/integration_test/topology/global_grouping/global_grouping_result.json
similarity index 100%
rename from integration_test/src/python/integration_test/topology/global_grouping_result.json
rename to integration_test/src/python/integration_test/topology/global_grouping/global_grouping_result.json
diff --git a/integration_test/src/python/integration_test/topology/multi_spouts_multi_tasks/__init__.py b/integration_test/src/python/integration_test/topology/multi_spouts_multi_tasks/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/integration_test/src/python/integration_test/topology/multi_spouts_multi_tasks/__init__.py
diff --git a/integration_test/src/python/integration_test/topology/multi_spouts_multi_tasks.py b/integration_test/src/python/integration_test/topology/multi_spouts_multi_tasks/multi_spouts_multi_tasks.py
similarity index 100%
rename from integration_test/src/python/integration_test/topology/multi_spouts_multi_tasks.py
rename to integration_test/src/python/integration_test/topology/multi_spouts_multi_tasks/multi_spouts_multi_tasks.py
diff --git a/integration_test/src/python/integration_test/topology/multi_spouts_multi_tasks_result.json b/integration_test/src/python/integration_test/topology/multi_spouts_multi_tasks/multi_spouts_multi_tasks_result.json
similarity index 100%
rename from integration_test/src/python/integration_test/topology/multi_spouts_multi_tasks_result.json
rename to integration_test/src/python/integration_test/topology/multi_spouts_multi_tasks/multi_spouts_multi_tasks_result.json
diff --git a/integration_test/src/python/integration_test/topology/none_grouping/__init__.py b/integration_test/src/python/integration_test/topology/none_grouping/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/integration_test/src/python/integration_test/topology/none_grouping/__init__.py
diff --git a/integration_test/src/python/integration_test/topology/none_grouping.py b/integration_test/src/python/integration_test/topology/none_grouping/none_grouping.py
similarity index 100%
rename from integration_test/src/python/integration_test/topology/none_grouping.py
rename to integration_test/src/python/integration_test/topology/none_grouping/none_grouping.py
diff --git a/integration_test/src/python/integration_test/topology/none_grouping_result.json b/integration_test/src/python/integration_test/topology/none_grouping/none_grouping_result.json
similarity index 100%
rename from integration_test/src/python/integration_test/topology/none_grouping_result.json
rename to integration_test/src/python/integration_test/topology/none_grouping/none_grouping_result.json
diff --git a/integration_test/src/python/integration_test/topology/one_bolt_multi_tasks/__init__.py b/integration_test/src/python/integration_test/topology/one_bolt_multi_tasks/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/integration_test/src/python/integration_test/topology/one_bolt_multi_tasks/__init__.py
diff --git a/integration_test/src/python/integration_test/topology/one_bolt_multi_tasks.py b/integration_test/src/python/integration_test/topology/one_bolt_multi_tasks/one_bolt_multi_tasks.py
similarity index 100%
rename from integration_test/src/python/integration_test/topology/one_bolt_multi_tasks.py
rename to integration_test/src/python/integration_test/topology/one_bolt_multi_tasks/one_bolt_multi_tasks.py
diff --git a/integration_test/src/python/integration_test/topology/one_bolt_multi_tasks_result.json b/integration_test/src/python/integration_test/topology/one_bolt_multi_tasks/one_bolt_multi_tasks_result.json
similarity index 100%
rename from integration_test/src/python/integration_test/topology/one_bolt_multi_tasks_result.json
rename to integration_test/src/python/integration_test/topology/one_bolt_multi_tasks/one_bolt_multi_tasks_result.json
diff --git a/integration_test/src/python/integration_test/topology/one_spout_bolt_multi_tasks/__init__.py b/integration_test/src/python/integration_test/topology/one_spout_bolt_multi_tasks/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/integration_test/src/python/integration_test/topology/one_spout_bolt_multi_tasks/__init__.py
diff --git a/integration_test/src/python/integration_test/topology/one_spout_bolt_multi_tasks.py b/integration_test/src/python/integration_test/topology/one_spout_bolt_multi_tasks/one_spout_bolt_multi_tasks.py
similarity index 100%
rename from integration_test/src/python/integration_test/topology/one_spout_bolt_multi_tasks.py
rename to integration_test/src/python/integration_test/topology/one_spout_bolt_multi_tasks/one_spout_bolt_multi_tasks.py
diff --git a/integration_test/src/python/integration_test/topology/one_spout_bolt_multi_tasks_result.json b/integration_test/src/python/integration_test/topology/one_spout_bolt_multi_tasks/one_spout_bolt_multi_tasks_result.json
similarity index 100%
rename from integration_test/src/python/integration_test/topology/one_spout_bolt_multi_tasks_result.json
rename to integration_test/src/python/integration_test/topology/one_spout_bolt_multi_tasks/one_spout_bolt_multi_tasks_result.json
diff --git a/integration_test/src/python/integration_test/topology/one_spout_multi_tasks/__init__.py b/integration_test/src/python/integration_test/topology/one_spout_multi_tasks/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/integration_test/src/python/integration_test/topology/one_spout_multi_tasks/__init__.py
diff --git a/integration_test/src/python/integration_test/topology/one_spout_multi_tasks.py b/integration_test/src/python/integration_test/topology/one_spout_multi_tasks/one_spout_multi_tasks.py
similarity index 100%
rename from integration_test/src/python/integration_test/topology/one_spout_multi_tasks.py
rename to integration_test/src/python/integration_test/topology/one_spout_multi_tasks/one_spout_multi_tasks.py
diff --git a/integration_test/src/python/integration_test/topology/one_spout_multi_tasks_result.json b/integration_test/src/python/integration_test/topology/one_spout_multi_tasks/one_spout_multi_tasks_result.json
similarity index 100%
rename from integration_test/src/python/integration_test/topology/one_spout_multi_tasks_result.json
rename to integration_test/src/python/integration_test/topology/one_spout_multi_tasks/one_spout_multi_tasks_result.json
diff --git a/integration_test/src/python/integration_test/topology/one_spout_two_bolts/__init__.py b/integration_test/src/python/integration_test/topology/one_spout_two_bolts/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/integration_test/src/python/integration_test/topology/one_spout_two_bolts/__init__.py
diff --git a/integration_test/src/python/integration_test/topology/one_spout_two_bolts.py b/integration_test/src/python/integration_test/topology/one_spout_two_bolts/one_spout_two_bolts.py
similarity index 100%
rename from integration_test/src/python/integration_test/topology/one_spout_two_bolts.py
rename to integration_test/src/python/integration_test/topology/one_spout_two_bolts/one_spout_two_bolts.py
diff --git a/integration_test/src/python/integration_test/topology/one_spout_two_bolts_result.json b/integration_test/src/python/integration_test/topology/one_spout_two_bolts/one_spout_two_bolts_result.json
similarity index 100%
rename from integration_test/src/python/integration_test/topology/one_spout_two_bolts_result.json
rename to integration_test/src/python/integration_test/topology/one_spout_two_bolts/one_spout_two_bolts_result.json
diff --git a/integration_test/src/python/integration_test/topology/shuffle_grouping/__init__.py b/integration_test/src/python/integration_test/topology/shuffle_grouping/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/integration_test/src/python/integration_test/topology/shuffle_grouping/__init__.py
diff --git a/integration_test/src/python/integration_test/topology/shuffle_grouping.py b/integration_test/src/python/integration_test/topology/shuffle_grouping/shuffle_grouping.py
similarity index 100%
rename from integration_test/src/python/integration_test/topology/shuffle_grouping.py
rename to integration_test/src/python/integration_test/topology/shuffle_grouping/shuffle_grouping.py
diff --git a/integration_test/src/python/integration_test/topology/shuffle_grouping_result.json b/integration_test/src/python/integration_test/topology/shuffle_grouping/shuffle_grouping_result.json
similarity index 100%
rename from integration_test/src/python/integration_test/topology/shuffle_grouping_result.json
rename to integration_test/src/python/integration_test/topology/shuffle_grouping/shuffle_grouping_result.json
diff --git a/integration_test/src/python/integration_test/topology/test_topology_main.py b/integration_test/src/python/integration_test/topology/test_topology_main.py
index 25a7964..08f587c 100644
--- a/integration_test/src/python/integration_test/topology/test_topology_main.py
+++ b/integration_test/src/python/integration_test/topology/test_topology_main.py
@@ -17,45 +17,45 @@
 import logging
 import sys
 
-from integration_test.src.python.integration_test.topology.all_grouping \
-  import all_grouping_buidler
+from integration_test.src.python.integration_test.topology.all_grouping.all_grouping \
+    import all_grouping_builder
 
-from integration_test.src.python.integration_test.topology.basic_one_task \
+from integration_test.src.python.integration_test.topology.basic_one_task.basic_one_task \
   import basic_one_task_builder
 
-from integration_test.src.python.integration_test.topology.bolt_double_emit_tuples \
-  import bolt_double_emit_tuples_builder
+from integration_test.src.python.integration_test.topology.bolt_double_emit_tuples\
+    .bolt_double_emit_tuples import bolt_double_emit_tuples_builder
 
-from integration_test.src.python.integration_test.topology.fields_grouping \
+from integration_test.src.python.integration_test.topology.fields_grouping.fields_grouping \
   import fields_grouping_builder
 
-from integration_test.src.python.integration_test.topology.global_grouping \
+from integration_test.src.python.integration_test.topology.global_grouping.global_grouping \
   import global_grouping_builder
 
-from integration_test.src.python.integration_test.topology.multi_spouts_multi_tasks \
-  import multi_spouts_multi_tasks_builder
+from integration_test.src.python.integration_test.topology.multi_spouts_multi_tasks\
+    .multi_spouts_multi_tasks import multi_spouts_multi_tasks_builder
 
-from integration_test.src.python.integration_test.topology.none_grouping \
+from integration_test.src.python.integration_test.topology.none_grouping.none_grouping \
   import none_grouping_builder
 
-from integration_test.src.python.integration_test.topology.one_bolt_multi_tasks \
-  import one_bolt_multi_tasks_builder
+from integration_test.src.python.integration_test.topology. one_bolt_multi_tasks\
+    .one_bolt_multi_tasks import one_bolt_multi_tasks_builder
 
-from integration_test.src.python.integration_test.topology.one_spout_bolt_multi_tasks \
-  import one_spout_bolt_multi_tasks_builder
+from integration_test.src.python.integration_test.topology.one_spout_bolt_multi_tasks\
+    .one_spout_bolt_multi_tasks import one_spout_bolt_multi_tasks_builder
 
-from integration_test.src.python.integration_test.topology.one_spout_multi_tasks \
-  import one_spout_multi_tasks_builder
+from integration_test.src.python.integration_test.topology.one_spout_multi_tasks\
+    .one_spout_multi_tasks import one_spout_multi_tasks_builder
 
-from integration_test.src.python.integration_test.topology.one_spout_two_bolts \
-  import one_spout_two_bolts_builder
+from integration_test.src.python.integration_test.topology.one_spout_two_bolts\
+    .one_spout_two_bolts  import one_spout_two_bolts_builder
 
-from integration_test.src.python.integration_test.topology.shuffle_grouping \
+from integration_test.src.python.integration_test.topology.shuffle_grouping.shuffle_grouping \
   import shuffle_grouping_builder
 
 TOPOLOGY_BUILDERS = {
     'Heron_IntegrationTest_BasicOneTask': basic_one_task_builder,
-    'Heron_IntegrationTest_AllGrouping': all_grouping_buidler,
+    'Heron_IntegrationTest_AllGrouping': all_grouping_builder,
     'Heron_IntegrationTest_NoneGrouping': none_grouping_builder,
     'Heron_IntegrationTest_OneBoltMultiTasks': one_bolt_multi_tasks_builder,
     'Heron_IntegrationTest_OneSpoutBoltMultiTasks': one_spout_bolt_multi_tasks_builder,
diff --git a/integration_test/src/python/test_runner/resources/test.json b/integration_test/src/python/test_runner/resources/test.json
index bed1d6b..8c1e15e 100644
--- a/integration_test/src/python/test_runner/resources/test.json
+++ b/integration_test/src/python/test_runner/resources/test.json
@@ -11,32 +11,32 @@
     {
       "topologyName" : "IntegrationTest_FieldsGrouping",
       "classPath"    : "fields_grouping.FieldsGrouping",
-      "expectedResultRelativePath" : "FieldsGrouping.json"
+      "expectedResultRelativePath" : "fields_grouping/FieldsGroupingResults.json"
     },
     {
       "topologyName" : "IntegrationTest_AllGrouping",
       "classPath"    : "all_grouping.AllGrouping",
-      "expectedResultRelativePath" : "AllGrouping.json"
+      "expectedResultRelativePath" : "all_grouping/AllGroupingResults.json"
     },
     {
       "topologyName" : "IntegrationTest_GlobalGrouping",
       "classPath"    : "global_grouping.GlobalGrouping",
-      "expectedResultRelativePath" : "GlobalGrouping.json"
+      "expectedResultRelativePath" : "global_grouping/GlobalGroupingResults.json"
     },
     {
       "topologyName" : "IntegrationTest_NonGrouping",
       "classPath"    : "non_grouping.NonGrouping",
-      "expectedResultRelativePath" : "NonGrouping.json"
+      "expectedResultRelativePath" : "non_grouping/NonGroupingResults.json"
     },
     {
       "topologyName" : "IntegrationTest_ShuffleGrouping",
       "classPath"    : "shuffle_grouping.ShuffleGrouping",
-      "expectedResultRelativePath" : "ShuffleGrouping.json"
+      "expectedResultRelativePath" : "shuffle_grouping/ShuffleGroupingResults.json"
     },
     {
       "topologyName" : "IntegrationTest_BasicTopologyOneTask",
       "classPath"    : "basic_topology_one_task.BasicTopologyOneTask",
-      "expectedResultRelativePath" : "BasicTopologyOneTask.json"
+      "expectedResultRelativePath" : "basic_topology_one_task/BasicTopologyOneTaskResults.json"
     },
     {
       "topologyName" : "IntegrationTest_BasicTopologyOneTaskScaleUp",
@@ -55,104 +55,144 @@
     {
       "topologyName" : "IntegrationTest_BoltDoubleEmitTuples",
       "classPath"    : "bolt_double_emit_tuples.BoltDoubleEmitTuples",
-      "expectedResultRelativePath" : "BoltDoubleEmitTuples.json"
+      "expectedResultRelativePath" : "bolt_double_emit_tuples/BoltDoubleEmitTuplesResults.json"
     },
     {
       "topologyName" : "IntegrationTest_MultiSpoutsMultiTasks",
       "classPath"    : "multi_spouts_multi_tasks.MultiSpoutsMultiTasks",
-      "expectedResultRelativePath" : "MultiSpoutsMultiTasks.json"
+      "expectedResultRelativePath" : "multi_spouts_multi_tasks/MultiSpoutsMultiTasksResults.json"
     },
     {
       "topologyName" : "IntegrationTest_OneBoltMultiTasks",
       "classPath"    : "one_bolt_multi_tasks.OneBoltMultiTasks",
-      "expectedResultRelativePath" : "OneBoltMultiTasks.json"
+      "expectedResultRelativePath" : "one_bolt_multi_tasks/OneBoltMultiTasksResults.json"
     },
     {
-      "topologyName" : "IntegrationTest_OneSpoutBoltMultiTasks",		
-      "classPath"    : "one_spout_bolt_multi_tasks.OneSpoutBoltMultiTasks",		
-      "expectedResultRelativePath" : "OneSpoutBoltMultiTasks.json"		
-    },		
+      "topologyName" : "IntegrationTest_OneSpoutBoltMultiTasks",
+      "classPath"    : "one_spout_bolt_multi_tasks.OneSpoutBoltMultiTasks",
+      "expectedResultRelativePath" : "one_spout_bolt_multi_tasks/OneSpoutBoltMultiTasksResults.json"
+    },
     {
       "topologyName" : "IntegrationTest_OneSpoutMultiTasks",
       "classPath"    : "one_spout_multi_tasks.OneSpoutMultiTasks",
-      "expectedResultRelativePath" : "OneSpoutMultiTasks.json"
+      "expectedResultRelativePath" : "one_spout_multi_tasks/OneSpoutMultiTasksResults.json"
     },
     {
       "topologyName" : "IntegrationTest_OneSpoutTwoBolts",
       "classPath"    : "one_spout_two_bolts.OneSpoutTwoBolts",
-      "expectedResultRelativePath" : "OneSpoutTwoBolts.json"
+      "expectedResultRelativePath" : "one_spout_two_bolts/OneSpoutTwoBoltsResults.json"
     },
     {
       "topologyName" : "IntegrationTest_CustomSerialization",
       "classPath"    : "serialization.SerializationTopology",
-      "expectedResultRelativePath" : "SerializationTopology.json"
+      "expectedResultRelativePath" : "serialization/SerializationTopologyResults.json"
+    },
+    {
+      "topologyName" : "IntegrationTest_SlidingCountWindow1",
+      "classPath"    : "windowing.count.SlidingCountWindowTest1",
+      "expectedResultRelativePath" : "windowing/count/SlidingCountWindowTest1Results.json"
+    },
+    {
+      "topologyName" : "IntegrationTest_SlidingCountWindow2",
+      "classPath"    : "windowing.count.SlidingCountWindowTest2",
+      "expectedResultRelativePath" : "windowing/count/SlidingCountWindowTest2Results.json"
+    },
+    {
+      "topologyName" : "IntegrationTest_SlidingCountWindow3",
+      "classPath"    : "windowing.count.SlidingCountWindowTest3",
+      "expectedResultRelativePath" : "windowing/count/SlidingCountWindowTest3Results.json"
+    },
+    {
+      "topologyName" : "IntegrationTest_TumblingCountWindow1",
+      "classPath"    : "windowing.count.TumblingCountWindowTest1",
+      "expectedResultRelativePath" : "windowing/count/TumblingCountWindowTest1Results.json"
+    },
+    {
+      "topologyName" : "IntegrationTest_TumblingCountWindow2",
+      "classPath"    : "windowing.count.TumblingCountWindowTest2",
+      "expectedResultRelativePath" : "windowing/count/TumblingCountWindowTest2Results.json"
+    },
+    {
+      "topologyName" : "IntegrationTest_TumblingCountWindow3",
+      "classPath"    : "windowing.count.TumblingCountWindowTest3",
+      "expectedResultRelativePath" : "windowing/count/TumblingCountWindowTest3Results.json"
+    },
+    {
+      "topologyName" : "IntegrationTest_SlidingTimeWindow1",
+      "classPath"    : "windowing.time.SlidingTimeWindowTest1",
+      "expectedResultRelativePath" : "windowing/time/SlidingTimeWindowTest1Results.json"
+    },
+    {
+      "topologyName" : "IntegrationTest_SlidingWatermarkEventTimeWindow1",
+      "classPath"    : "windowing.watermark.SlidingWatermarkEventTimeWindowTest1",
+      "expectedResultRelativePath" : "windowing/watermark/SlidingWatermarkEventTimeWindowTest1Results.json"
     }
   ],
   "pythonTopologies": [
     {
       "topologyName" : "Heron_IntegrationTest_AllGrouping",
       "classPath"    : "-",
-      "expectedResultRelativePath" : "all_grouping_result.json"
+      "expectedResultRelativePath" : "all_grouping/all_grouping_result.json"
     },
     {
       "topologyName" : "Heron_IntegrationTest_BasicOneTask",
       "classPath"    : "-",
-      "expectedResultRelativePath" : "basic_one_task_result.json"
+      "expectedResultRelativePath" : "basic_one_task/basic_one_task_result.json"
     },
     {
       "topologyName" : "Heron_IntegrationTest_NoneGrouping",
       "classPath"    : "-",
-      "expectedResultRelativePath" : "none_grouping_result.json"
+      "expectedResultRelativePath" : "none_grouping/none_grouping_result.json"
     },
     {
       "topologyName" : "Heron_IntegrationTest_OneBoltMultiTasks",
       "classPath"    : "-",
-      "expectedResultRelativePath" : "one_bolt_multi_tasks_result.json"
+      "expectedResultRelativePath" : "one_bolt_multi_tasks/one_bolt_multi_tasks_result.json"
     },
     {
       "topologyName" : "Heron_IntegrationTest_OneSpoutBoltMultiTasks",
       "classPath"    : "-",
-      "expectedResultRelativePath" : "one_spout_bolt_multi_tasks_result.json"
+      "expectedResultRelativePath" : "one_spout_bolt_multi_tasks/one_spout_bolt_multi_tasks_result.json"
     },
     {
       "topologyName" : "Heron_IntegrationTest_ShuffleGrouping",
       "classPath"    : "-",
-      "expectedResultRelativePath" : "shuffle_grouping_result.json"
+      "expectedResultRelativePath" : "shuffle_grouping/shuffle_grouping_result.json"
     },
     {
       "topologyName" : "Heron_IntegrationTest_OneSpoutTwoBolts",
       "classPath"    : "-",
-      "expectedResultRelativePath" : "one_spout_two_bolts_result.json"
+      "expectedResultRelativePath" : "one_spout_two_bolts/one_spout_two_bolts_result.json"
     },
     {
       "topologyName" : "Heron_IntegrationTest_OneSpoutTwoBolts",
       "classPath"    : "-",
-      "expectedResultRelativePath" : "one_spout_two_bolts_result.json"
+      "expectedResultRelativePath" : "one_spout_two_bolts/one_spout_two_bolts_result.json"
     },
     {
       "topologyName" : "Heron_IntegrationTest_OneSpoutMultiTasks",
       "classPath"    : "-",
-      "expectedResultRelativePath" : "one_spout_multi_tasks_result.json"
+      "expectedResultRelativePath" : "one_spout_multi_tasks/one_spout_multi_tasks_result.json"
     },
     {
       "topologyName" : "Heron_IntegrationTest_MultiSpoutsMultiTasks",
       "classPath"    : "-",
-      "expectedResultRelativePath" : "multi_spouts_multi_tasks_result.json"
+      "expectedResultRelativePath" : "multi_spouts_multi_tasks/multi_spouts_multi_tasks_result.json"
     },
     {
       "topologyName" : "Heron_IntegrationTest_FieldsGrouping",
       "classPath"    : "-",
-      "expectedResultRelativePath" : "fields_grouping_result.json"
+      "expectedResultRelativePath" : "fields_grouping/fields_grouping_result.json"
     },
     {
       "topologyName" : "Heron_IntegrationTest_BoltDoubleEmitTuples",
       "classPath"    : "-",
-      "expectedResultRelativePath" : "bolt_double_emit_tuples_result.json"
+      "expectedResultRelativePath" : "bolt_double_emit_tuples/bolt_double_emit_tuples_result.json"
     },
     {
       "topologyName" : "Heron_IntegrationTest_GlobalGrouping",
       "classPath"    : "-",
-      "expectedResultRelativePath" : "global_grouping_result.json"
+      "expectedResultRelativePath" : "global_grouping/global_grouping_result.json"
     }
   ]
 }
diff --git a/scripts/packages/BUILD b/scripts/packages/BUILD
index 18fdfbe..fafc619 100644
--- a/scripts/packages/BUILD
+++ b/scripts/packages/BUILD
@@ -311,6 +311,7 @@
     srcs = [
         "//integration_test/src/java:test-data-files",
     ],
+    strip_prefix = '/integration_test/src/java/com/twitter/heron/integration_test/topology/'
 )
 
 pkg_tar(
@@ -319,6 +320,7 @@
     srcs = [
         "//integration_test/src/python/integration_test/topology:test-data-files",
     ],
+    strip_prefix = '/integration_test/src/python/integration_test/topology/'
 )
 
 pkg_tar(