storm-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From kabh...@apache.org
Subject [01/12] storm git commit: [STORM-1961] Stream api for storm core use cases
Date Tue, 07 Feb 2017 01:28:14 GMT
Repository: storm
Updated Branches:
  refs/heads/master a7cdfefd7 -> b5f02d4e4


http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/topology/StatefulWindowedBoltExecutor.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/topology/StatefulWindowedBoltExecutor.java b/storm-core/src/jvm/org/apache/storm/topology/StatefulWindowedBoltExecutor.java
index aff46e6..2fc6f8d 100644
--- a/storm-core/src/jvm/org/apache/storm/topology/StatefulWindowedBoltExecutor.java
+++ b/storm-core/src/jvm/org/apache/storm/topology/StatefulWindowedBoltExecutor.java
@@ -207,14 +207,14 @@ public class StatefulWindowedBoltExecutor<T extends State> extends WindowedBoltE
             }
 
             @Override
-            public void onActivation(List<Tuple> events, List<Tuple> newEvents, List<Tuple> expired) {
+            public void onActivation(List<Tuple> events, List<Tuple> newEvents, List<Tuple> expired, Long timestamp) {
                 if (isRecovering()) {
                     String msg = String.format("Unexpected activation with events %s, newEvents %s, expired %s in recovering state. " +
                                                        "recoveryStates %s ", events, newEvents, expired, recoveryStates);
                     LOG.error(msg);
                     throw new IllegalStateException(msg);
                 } else {
-                    parentListener.onActivation(events, newEvents, expired);
+                    parentListener.onActivation(events, newEvents, expired, timestamp);
                     updateWindowState(expired, newEvents);
                 }
             }

http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/topology/WindowedBoltExecutor.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/topology/WindowedBoltExecutor.java b/storm-core/src/jvm/org/apache/storm/topology/WindowedBoltExecutor.java
index fd98274..b592e0b 100644
--- a/storm-core/src/jvm/org/apache/storm/topology/WindowedBoltExecutor.java
+++ b/storm-core/src/jvm/org/apache/storm/topology/WindowedBoltExecutor.java
@@ -327,9 +327,9 @@ public class WindowedBoltExecutor implements IRichBolt {
             }
 
             @Override
-            public void onActivation(List<Tuple> tuples, List<Tuple> newTuples, List<Tuple> expiredTuples) {
+            public void onActivation(List<Tuple> tuples, List<Tuple> newTuples, List<Tuple> expiredTuples, Long timestamp) {
                 windowedOutputCollector.setContext(tuples);
-                bolt.execute(new TupleWindowImpl(tuples, newTuples, expiredTuples));
+                bolt.execute(new TupleWindowImpl(tuples, newTuples, expiredTuples, timestamp));
             }
         };
     }

http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/topology/base/BaseWindowedBolt.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/topology/base/BaseWindowedBolt.java b/storm-core/src/jvm/org/apache/storm/topology/base/BaseWindowedBolt.java
index 2e2af0d..ba97c26 100644
--- a/storm-core/src/jvm/org/apache/storm/topology/base/BaseWindowedBolt.java
+++ b/storm-core/src/jvm/org/apache/storm/topology/base/BaseWindowedBolt.java
@@ -27,6 +27,7 @@ import org.apache.storm.windowing.TimestampExtractor;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.Serializable;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
@@ -40,7 +41,7 @@ public abstract class BaseWindowedBolt implements IWindowedBolt {
     /**
      * Holds a count value for count based windows and sliding intervals.
      */
-    public static class Count {
+    public static class Count implements Serializable {
         public final int value;
 
         public Count(int value) {
@@ -58,6 +59,22 @@ public abstract class BaseWindowedBolt implements IWindowedBolt {
         }
 
         @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 +
@@ -68,7 +85,7 @@ public abstract class BaseWindowedBolt implements IWindowedBolt {
     /**
      * Holds a Time duration for time based windows and sliding intervals.
      */
-    public static class Duration {
+    public static class Duration implements Serializable {
         public final int value;
 
         public Duration(int value, TimeUnit timeUnit) {
@@ -126,6 +143,22 @@ public abstract class BaseWindowedBolt implements IWindowedBolt {
         }
 
         @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 +

http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/trident/windowing/AbstractTridentWindowManager.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/trident/windowing/AbstractTridentWindowManager.java b/storm-core/src/jvm/org/apache/storm/trident/windowing/AbstractTridentWindowManager.java
index f93527a..a8fbb41 100644
--- a/storm-core/src/jvm/org/apache/storm/trident/windowing/AbstractTridentWindowManager.java
+++ b/storm-core/src/jvm/org/apache/storm/trident/windowing/AbstractTridentWindowManager.java
@@ -121,7 +121,7 @@ public abstract class AbstractTridentWindowManager<T> implements ITridentWindowM
         }
 
         @Override
-        public void onActivation(List<T> events, List<T> newEvents, List<T> expired) {
+        public void onActivation(List<T> events, List<T> newEvents, List<T> expired, Long timestamp) {
             LOG.debug("onActivation is invoked with events size: [{}]", events.size());
             // trigger occurred, create an aggregation and keep them in store
             int currentTriggerId = triggerId.incrementAndGet();

http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/windowing/CountEvictionPolicy.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/windowing/CountEvictionPolicy.java b/storm-core/src/jvm/org/apache/storm/windowing/CountEvictionPolicy.java
index fb12202..6a9a4f8 100644
--- a/storm-core/src/jvm/org/apache/storm/windowing/CountEvictionPolicy.java
+++ b/storm-core/src/jvm/org/apache/storm/windowing/CountEvictionPolicy.java
@@ -28,6 +28,7 @@ import java.util.concurrent.atomic.AtomicLong;
 public class CountEvictionPolicy<T> implements EvictionPolicy<T> {
     protected final int threshold;
     protected final AtomicLong currentCount;
+    private EvictionContext context;
 
     public CountEvictionPolicy(int count) {
         this.threshold = count;
@@ -62,7 +63,12 @@ public class CountEvictionPolicy<T> implements EvictionPolicy<T> {
 
     @Override
     public void setContext(EvictionContext context) {
-        // NOOP
+        this.context = context;
+    }
+
+    @Override
+    public EvictionContext getContext() {
+        return context;
     }
 
     @Override

http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/windowing/EvictionPolicy.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/windowing/EvictionPolicy.java b/storm-core/src/jvm/org/apache/storm/windowing/EvictionPolicy.java
index 05e4d93..774d0a3 100644
--- a/storm-core/src/jvm/org/apache/storm/windowing/EvictionPolicy.java
+++ b/storm-core/src/jvm/org/apache/storm/windowing/EvictionPolicy.java
@@ -72,4 +72,11 @@ public interface EvictionPolicy<T> {
      */
     void setContext(EvictionContext context);
 
+    /**
+     * Returns the current context that is part of this eviction policy
+     *
+     * @return the eviction context
+     */
+    EvictionContext getContext();
+
 }

http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/windowing/TimeEvictionPolicy.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/windowing/TimeEvictionPolicy.java b/storm-core/src/jvm/org/apache/storm/windowing/TimeEvictionPolicy.java
index e646207..802e6bb 100644
--- a/storm-core/src/jvm/org/apache/storm/windowing/TimeEvictionPolicy.java
+++ b/storm-core/src/jvm/org/apache/storm/windowing/TimeEvictionPolicy.java
@@ -22,11 +22,7 @@ package org.apache.storm.windowing;
  */
 public class TimeEvictionPolicy<T> implements EvictionPolicy<T> {
     private final int windowLength;
-    /**
-     * The reference time in millis for window calculations and
-     * expiring events. If not set it will default to System.currentTimeMillis()
-     */
-    protected Long referenceTime;
+    protected EvictionContext evictionContext;
 
     /**
      * Constructs a TimeEvictionPolicy that evicts events older
@@ -43,7 +39,7 @@ public class TimeEvictionPolicy<T> implements EvictionPolicy<T> {
      */
     @Override
     public Action evict(Event<T> event) {
-        long now = referenceTime == null ? System.currentTimeMillis() : referenceTime;
+        long now = evictionContext == null ? System.currentTimeMillis() : evictionContext.getReferenceTime();
         long diff = now - event.getTimestamp();
         if (diff >= windowLength) {
             return Action.EXPIRE;
@@ -58,14 +54,19 @@ public class TimeEvictionPolicy<T> implements EvictionPolicy<T> {
 
     @Override
     public void setContext(EvictionContext context) {
-        referenceTime = context.getReferenceTime();
+        this.evictionContext = context;
+    }
+
+    @Override
+    public EvictionContext getContext() {
+        return evictionContext;
     }
 
     @Override
     public String toString() {
         return "TimeEvictionPolicy{" +
                 "windowLength=" + windowLength +
-                ", referenceTime=" + referenceTime +
+                ", evictionContext=" + evictionContext +
                 '}';
     }
 }

http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/windowing/TupleWindowImpl.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/windowing/TupleWindowImpl.java b/storm-core/src/jvm/org/apache/storm/windowing/TupleWindowImpl.java
index 2560d25..1e8b022 100644
--- a/storm-core/src/jvm/org/apache/storm/windowing/TupleWindowImpl.java
+++ b/storm-core/src/jvm/org/apache/storm/windowing/TupleWindowImpl.java
@@ -28,11 +28,17 @@ public class TupleWindowImpl implements TupleWindow {
     private final List<Tuple> tuples;
     private final List<Tuple> newTuples;
     private final List<Tuple> expiredTuples;
+    private final Long timestamp;
 
     public TupleWindowImpl(List<Tuple> tuples, List<Tuple> newTuples, List<Tuple> expiredTuples) {
+        this(tuples, newTuples, expiredTuples, null);
+    }
+
+    public TupleWindowImpl(List<Tuple> tuples, List<Tuple> newTuples, List<Tuple> expiredTuples, Long timestamp) {
         this.tuples = tuples;
         this.newTuples = newTuples;
         this.expiredTuples = expiredTuples;
+        this.timestamp = timestamp;
     }
 
     @Override
@@ -51,6 +57,11 @@ public class TupleWindowImpl implements TupleWindow {
     }
 
     @Override
+    public Long getTimestamp() {
+        return timestamp != null ? timestamp : System.currentTimeMillis();
+    }
+
+    @Override
     public String toString() {
         return "TupleWindowImpl{" +
                 "tuples=" + tuples +

http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/windowing/WatermarkCountEvictionPolicy.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/windowing/WatermarkCountEvictionPolicy.java b/storm-core/src/jvm/org/apache/storm/windowing/WatermarkCountEvictionPolicy.java
index 74240bb..7304366 100644
--- a/storm-core/src/jvm/org/apache/storm/windowing/WatermarkCountEvictionPolicy.java
+++ b/storm-core/src/jvm/org/apache/storm/windowing/WatermarkCountEvictionPolicy.java
@@ -24,11 +24,6 @@ package org.apache.storm.windowing;
  * @param <T> the type of event tracked by this policy.
  */
 public class WatermarkCountEvictionPolicy<T> extends CountEvictionPolicy<T> {
-    /*
-     * The reference time in millis for window calculations and
-     * expiring events. If not set it will default to System.currentTimeMillis()
-     */
-    private long referenceTime;
     private long processed = 0L;
 
     public WatermarkCountEvictionPolicy(int count) {
@@ -38,7 +33,7 @@ public class WatermarkCountEvictionPolicy<T> extends CountEvictionPolicy<T> {
     @Override
     public Action evict(Event<T> event) {
         Action action;
-        if (event.getTimestamp() <= referenceTime && processed < currentCount.get()) {
+        if (event.getTimestamp() <= super.getContext().getReferenceTime() && processed < currentCount.get()) {
             action = super.evict(event);
             if (action == Action.PROCESS) {
                 ++processed;
@@ -56,7 +51,7 @@ public class WatermarkCountEvictionPolicy<T> extends CountEvictionPolicy<T> {
 
     @Override
     public void setContext(EvictionContext context) {
-        referenceTime = context.getReferenceTime();
+        super.setContext(context);
         if (context.getCurrentCount() != null) {
             currentCount.set(context.getCurrentCount());
         } else {
@@ -68,7 +63,6 @@ public class WatermarkCountEvictionPolicy<T> extends CountEvictionPolicy<T> {
     @Override
     public String toString() {
         return "WatermarkCountEvictionPolicy{" +
-                "referenceTime=" + referenceTime +
                 "} " + super.toString();
     }
 }

http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/windowing/WatermarkTimeEvictionPolicy.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/windowing/WatermarkTimeEvictionPolicy.java b/storm-core/src/jvm/org/apache/storm/windowing/WatermarkTimeEvictionPolicy.java
index 53361d2..e5ecba4 100644
--- a/storm-core/src/jvm/org/apache/storm/windowing/WatermarkTimeEvictionPolicy.java
+++ b/storm-core/src/jvm/org/apache/storm/windowing/WatermarkTimeEvictionPolicy.java
@@ -45,7 +45,6 @@ public class WatermarkTimeEvictionPolicy<T> extends TimeEvictionPolicy<T> {
      */
     public WatermarkTimeEvictionPolicy(int windowLength, int lag) {
         super(windowLength);
-        referenceTime = 0L;
         this.lag = lag;
     }
 
@@ -58,7 +57,8 @@ public class WatermarkTimeEvictionPolicy<T> extends TimeEvictionPolicy<T> {
      */
     @Override
     public Action evict(Event<T> event) {
-        long diff = referenceTime - event.getTimestamp();
+        long referenceTime = evictionContext.getReferenceTime() != null ? evictionContext.getReferenceTime() : 0L;
+        long diff =  referenceTime - event.getTimestamp();
         if (diff < -lag) {
             return Action.STOP;
         } else if (diff < 0) {

http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/windowing/Window.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/windowing/Window.java b/storm-core/src/jvm/org/apache/storm/windowing/Window.java
index 8382448..9a62eef 100644
--- a/storm-core/src/jvm/org/apache/storm/windowing/Window.java
+++ b/storm-core/src/jvm/org/apache/storm/windowing/Window.java
@@ -45,4 +45,11 @@ public interface Window<T> {
      * @return the list of events expired from the window.
      */
     List<T> getExpired();
+
+    /**
+     * If processing based on event time, returns the watermark time otherwise the current timestamp.
+     *
+     * @return the window timestamp
+     */
+    Long getTimestamp();
 }

http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/windowing/WindowLifecycleListener.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/windowing/WindowLifecycleListener.java b/storm-core/src/jvm/org/apache/storm/windowing/WindowLifecycleListener.java
index 03c0213..ea2c997 100644
--- a/storm-core/src/jvm/org/apache/storm/windowing/WindowLifecycleListener.java
+++ b/storm-core/src/jvm/org/apache/storm/windowing/WindowLifecycleListener.java
@@ -37,6 +37,7 @@ public interface WindowLifecycleListener<T> {
      * @param events the list of current events in the window.
      * @param newEvents the newly added events since last activation.
      * @param expired the expired events since last activation.
+     * @param referenceTime the reference (event or processing) time that resulted in activation
      */
-    void onActivation(List<T> events, List<T> newEvents, List<T> expired);
+    void onActivation(List<T> events, List<T> newEvents, List<T> expired, Long referenceTime);
 }

http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/src/jvm/org/apache/storm/windowing/WindowManager.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/windowing/WindowManager.java b/storm-core/src/jvm/org/apache/storm/windowing/WindowManager.java
index 792509e..8021ba8 100644
--- a/storm-core/src/jvm/org/apache/storm/windowing/WindowManager.java
+++ b/storm-core/src/jvm/org/apache/storm/windowing/WindowManager.java
@@ -142,7 +142,7 @@ public class WindowManager<T> implements TriggerHandler {
         if (!events.isEmpty()) {
             prevWindowEvents.addAll(windowEvents);
             LOG.debug("invoking windowLifecycleListener onActivation, [{}] events in window.", events.size());
-            windowLifecycleListener.onActivation(events, newEvents, expired);
+            windowLifecycleListener.onActivation(events, newEvents, expired, evictionPolicy.getContext().getReferenceTime());
         } else {
             LOG.debug("No events in the window, skipping onActivation");
         }

http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/test/jvm/org/apache/storm/streams/ProcessorBoltTest.java
----------------------------------------------------------------------
diff --git a/storm-core/test/jvm/org/apache/storm/streams/ProcessorBoltTest.java b/storm-core/test/jvm/org/apache/storm/streams/ProcessorBoltTest.java
new file mode 100644
index 0000000..e9d5127
--- /dev/null
+++ b/storm-core/test/jvm/org/apache/storm/streams/ProcessorBoltTest.java
@@ -0,0 +1,165 @@
+/**
+ * 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.streams;
+
+import com.google.common.collect.Multimap;
+import org.apache.storm.streams.operations.aggregators.Sum;
+import org.apache.storm.streams.processors.AggregateProcessor;
+import org.apache.storm.streams.processors.FilterProcessor;
+import org.apache.storm.streams.processors.Processor;
+import org.apache.storm.task.OutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Tuple;
+import org.apache.storm.tuple.Values;
+import org.jgrapht.DirectedGraph;
+import org.jgrapht.graph.DefaultDirectedGraph;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Set;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Unit tests for {@link ProcessorBolt}
+ */
+public class ProcessorBoltTest {
+    TopologyContext mockTopologyContext;
+    OutputCollector mockOutputCollector;
+    ProcessorBolt bolt;
+    Tuple mockTuple1;
+    Tuple mockTuple2;
+    Tuple mockTuple3;
+    Tuple punctuation;
+    Multimap<String, ProcessorNode> mockStreamToProcessors;
+    DirectedGraph<Node, Edge> graph;
+
+    @Before
+    public void setUp() throws Exception {
+        mockTopologyContext = Mockito.mock(TopologyContext.class);
+        mockOutputCollector = Mockito.mock(OutputCollector.class);
+        mockTuple1 = Mockito.mock(Tuple.class);
+        mockTuple2 = Mockito.mock(Tuple.class);
+        mockTuple3 = Mockito.mock(Tuple.class);
+        setUpMockTuples(mockTuple1, mockTuple2, mockTuple3);
+        punctuation = Mockito.mock(Tuple.class);
+        setUpPunctuation(punctuation);
+        mockStreamToProcessors = Mockito.mock(Multimap.class);
+        graph = new DefaultDirectedGraph(new StreamsEdgeFactory());
+
+    }
+
+    @Test
+    public void testEmitAndAck() throws Exception {
+        setUpProcessorBolt(new FilterProcessor<Integer>(x -> true));
+        bolt.execute(mockTuple1);
+        ArgumentCaptor<Collection> anchor = ArgumentCaptor.forClass(Collection.class);
+        ArgumentCaptor<Values> values = ArgumentCaptor.forClass(Values.class);
+        ArgumentCaptor<String> os = ArgumentCaptor.forClass(String.class);
+        Mockito.verify(mockOutputCollector).emit(os.capture(), anchor.capture(), values.capture());
+        assertEquals("outputstream", os.getValue());
+        assertArrayEquals(new Object[]{mockTuple1}, anchor.getValue().toArray());
+        assertEquals(new Values(100), values.getValue());
+        Mockito.verify(mockOutputCollector, Mockito.times(1)).ack(mockTuple1);
+    }
+
+    @Test
+    public void testAggResultAndAck() throws Exception {
+        setUpProcessorBolt(new AggregateProcessor<>(new Sum()), Collections.singleton("inputstream"), true, null);
+        bolt.execute(mockTuple2);
+        bolt.execute(mockTuple3);
+        bolt.execute(punctuation);
+        ArgumentCaptor<Collection> anchor = ArgumentCaptor.forClass(Collection.class);
+        ArgumentCaptor<Values> values = ArgumentCaptor.forClass(Values.class);
+        ArgumentCaptor<String> os = ArgumentCaptor.forClass(String.class);
+        Mockito.verify(mockOutputCollector, Mockito.times(2)).emit(os.capture(), anchor.capture(), values.capture());
+        assertArrayEquals(new Object[]{mockTuple2, mockTuple3, punctuation}, anchor.getAllValues().get(0).toArray());
+        assertArrayEquals(new Object[]{mockTuple2, mockTuple3, punctuation}, anchor.getAllValues().get(1).toArray());
+        assertArrayEquals(new Object[]{new Values(200L), new Values("__punctuation")}, values.getAllValues().toArray());
+        assertArrayEquals(new Object[]{"outputstream", "outputstream"}, os.getAllValues().toArray());
+        Mockito.verify(mockOutputCollector).ack(mockTuple2);
+        Mockito.verify(mockOutputCollector).ack(mockTuple3);
+        Mockito.verify(mockOutputCollector).ack(punctuation);
+    }
+
+    @Test
+    public void testEmitTs() throws Exception {
+        Tuple tupleWithTs = Mockito.mock(Tuple.class);
+        setUpMockTuples(tupleWithTs);
+        Mockito.when(tupleWithTs.getLongByField("ts")).thenReturn(12345L);
+        setUpProcessorBolt(new FilterProcessor(x -> true), "ts");
+        bolt.execute(tupleWithTs);
+        ArgumentCaptor<Collection> anchor = ArgumentCaptor.forClass(Collection.class);
+        ArgumentCaptor<Values> values = ArgumentCaptor.forClass(Values.class);
+        ArgumentCaptor<String> os = ArgumentCaptor.forClass(String.class);
+        Mockito.verify(mockOutputCollector).emit(os.capture(), anchor.capture(), values.capture());
+        assertEquals("outputstream", os.getValue());
+        assertArrayEquals(new Object[]{tupleWithTs}, anchor.getValue().toArray());
+        assertEquals(new Values(100, 12345L), values.getValue());
+        Mockito.verify(mockOutputCollector, Mockito.times(1)).ack(tupleWithTs);
+    }
+
+    private void setUpProcessorBolt(Processor<?> processor) {
+        setUpProcessorBolt(processor, Collections.emptySet(), false, null);
+    }
+
+    private void setUpProcessorBolt(Processor<?> processor, String tsFieldName) {
+        setUpProcessorBolt(processor, Collections.emptySet(), false, tsFieldName);
+    }
+
+    private void setUpProcessorBolt(Processor<?> processor,
+                                    Set<String> windowedParentStreams,
+                                    boolean isWindowed,
+                                    String tsFieldName) {
+        ProcessorNode node = new ProcessorNode(processor, "outputstream", new Fields("value"));
+        node.setWindowedParentStreams(windowedParentStreams);
+        node.setWindowed(isWindowed);
+        Mockito.when(mockStreamToProcessors.get(Mockito.anyString())).thenReturn(Collections.singletonList(node));
+        graph.addVertex(node);
+        bolt = new ProcessorBolt("bolt1", graph, Collections.singletonList(node));
+        if (tsFieldName != null && !tsFieldName.isEmpty()) {
+            bolt.setTimestampField(tsFieldName);
+        }
+        bolt.setStreamToInitialProcessors(mockStreamToProcessors);
+        bolt.prepare(new HashMap<>(), mockTopologyContext, mockOutputCollector);
+    }
+
+    private void setUpMockTuples(Tuple... tuples) {
+        for (Tuple tuple : tuples) {
+            Mockito.when(tuple.size()).thenReturn(1);
+            Mockito.when(tuple.getValue(0)).thenReturn(100);
+            Mockito.when(tuple.getSourceComponent()).thenReturn("bolt0");
+            Mockito.when(tuple.getSourceStreamId()).thenReturn("inputstream");
+        }
+    }
+
+    private void setUpPunctuation(Tuple punctuation) {
+        Mockito.when(punctuation.size()).thenReturn(1);
+        Mockito.when(punctuation.getValue(0)).thenReturn(WindowNode.PUNCTUATION);
+        Mockito.when(punctuation.getSourceComponent()).thenReturn("bolt0");
+        Mockito.when(punctuation.getSourceStreamId()).thenReturn("inputstream");
+    }
+}

http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/test/jvm/org/apache/storm/streams/StatefulProcessorBoltTest.java
----------------------------------------------------------------------
diff --git a/storm-core/test/jvm/org/apache/storm/streams/StatefulProcessorBoltTest.java b/storm-core/test/jvm/org/apache/storm/streams/StatefulProcessorBoltTest.java
new file mode 100644
index 0000000..dbc7e27
--- /dev/null
+++ b/storm-core/test/jvm/org/apache/storm/streams/StatefulProcessorBoltTest.java
@@ -0,0 +1,100 @@
+/**
+ * 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.streams;
+
+import com.google.common.collect.Multimap;
+import org.apache.storm.state.KeyValueState;
+import org.apache.storm.streams.operations.aggregators.Count;
+import org.apache.storm.streams.processors.Processor;
+import org.apache.storm.streams.processors.UpdateStateByKeyProcessor;
+import org.apache.storm.task.OutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Tuple;
+import org.apache.storm.tuple.Values;
+import org.jgrapht.DirectedGraph;
+import org.jgrapht.graph.DefaultDirectedGraph;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Unit tests for {@link StatefulProcessorBolt}
+ */
+public class StatefulProcessorBoltTest {
+    TopologyContext mockTopologyContext;
+    OutputCollector mockOutputCollector;
+    StatefulProcessorBolt<String, Long> bolt;
+    Tuple mockTuple1;
+    DirectedGraph<Node, Edge> graph;
+    Multimap<String, ProcessorNode> mockStreamToProcessors;
+    KeyValueState<String, Long> mockKeyValueState;
+
+    @Before
+    public void setUp() throws Exception {
+        mockTopologyContext = Mockito.mock(TopologyContext.class);
+        mockOutputCollector = Mockito.mock(OutputCollector.class);
+        mockTuple1 = Mockito.mock(Tuple.class);
+        mockStreamToProcessors = Mockito.mock(Multimap.class);
+        mockKeyValueState = Mockito.mock(KeyValueState.class);
+        setUpMockTuples(mockTuple1);
+    }
+
+    @Test
+    public void testEmitAndAck() throws Exception {
+        setUpStatefulProcessorBolt(new UpdateStateByKeyProcessor<>(new Count<>()));
+        bolt.execute(mockTuple1);
+        ArgumentCaptor<Collection> anchor = ArgumentCaptor.forClass(Collection.class);
+        ArgumentCaptor<Values> values = ArgumentCaptor.forClass(Values.class);
+        ArgumentCaptor<String> os = ArgumentCaptor.forClass(String.class);
+        Mockito.verify(mockOutputCollector).emit(os.capture(), anchor.capture(), values.capture());
+        assertEquals("outputstream", os.getValue());
+        assertArrayEquals(new Object[]{mockTuple1}, anchor.getValue().toArray());
+        assertEquals(new Values("k", 1L), values.getValue());
+        Mockito.verify(mockOutputCollector, Mockito.times(1)).ack(mockTuple1);
+        Mockito.verify(mockKeyValueState, Mockito.times(1)).put("k", 1L );
+    }
+
+    private void setUpStatefulProcessorBolt(Processor<?> processor) {
+        ProcessorNode node = new ProcessorNode(processor, "outputstream", new Fields("value"));
+        Mockito.when(mockStreamToProcessors.get(Mockito.anyString())).thenReturn(Collections.singletonList(node));
+        graph = new DefaultDirectedGraph(new StreamsEdgeFactory());
+        graph.addVertex(node);
+        bolt = new StatefulProcessorBolt<>("bolt1", graph, Collections.singletonList(node));
+        bolt.setStreamToInitialProcessors(mockStreamToProcessors);
+        bolt.prepare(new HashMap<>(), mockTopologyContext, mockOutputCollector);
+        bolt.initState(mockKeyValueState);
+    }
+
+    private void setUpMockTuples(Tuple... tuples) {
+        for (Tuple tuple : tuples) {
+            Mockito.when(tuple.size()).thenReturn(1);
+            Mockito.when(tuple.getValue(0)).thenReturn(Pair.of("k", "v"));
+            Mockito.when(tuple.getSourceComponent()).thenReturn("bolt0");
+            Mockito.when(tuple.getSourceStreamId()).thenReturn("inputstream");
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/test/jvm/org/apache/storm/streams/StreamBuilderTest.java
----------------------------------------------------------------------
diff --git a/storm-core/test/jvm/org/apache/storm/streams/StreamBuilderTest.java b/storm-core/test/jvm/org/apache/storm/streams/StreamBuilderTest.java
new file mode 100644
index 0000000..1498ae4
--- /dev/null
+++ b/storm-core/test/jvm/org/apache/storm/streams/StreamBuilderTest.java
@@ -0,0 +1,219 @@
+/**
+ * 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.streams;
+
+import org.apache.storm.generated.Bolt;
+import org.apache.storm.generated.GlobalStreamId;
+import org.apache.storm.generated.Grouping;
+import org.apache.storm.generated.NullStruct;
+import org.apache.storm.generated.SpoutSpec;
+import org.apache.storm.generated.StormTopology;
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.storm.streams.operations.aggregators.Count;
+import org.apache.storm.streams.operations.mappers.PairValueMapper;
+import org.apache.storm.streams.operations.mappers.ValueMapper;
+import org.apache.storm.streams.processors.BranchProcessor;
+import org.apache.storm.task.OutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.IRichBolt;
+import org.apache.storm.topology.IRichSpout;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.topology.base.BaseRichBolt;
+import org.apache.storm.topology.base.BaseRichSpout;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Tuple;
+import org.apache.storm.utils.Utils;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Unit tests for {@link StreamBuilder}
+ */
+public class StreamBuilderTest {
+    StreamBuilder streamBuilder;
+
+    @Before
+    public void setUp() throws Exception {
+        streamBuilder = new StreamBuilder();
+        UniqueIdGen.getInstance().reset();
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testSpoutNoDefaultStream() throws Exception {
+        Stream<Tuple> stream = streamBuilder.newStream(newSpout("test"));
+        stream.filter(x -> true);
+        streamBuilder.build();
+    }
+
+    @Test
+    public void testSpoutToBolt() throws Exception {
+        Stream<Tuple> stream = streamBuilder.newStream(newSpout(Utils.DEFAULT_STREAM_ID));
+        stream.to(newBolt());
+        StormTopology topology = streamBuilder.build();
+        assertEquals(1, topology.get_spouts_size());
+        assertEquals(1, topology.get_bolts_size());
+        String spoutId = topology.get_spouts().keySet().iterator().next();
+        Map<GlobalStreamId, Grouping> expected = new HashMap<>();
+        expected.put(new GlobalStreamId(spoutId, "default"), Grouping.shuffle(new NullStruct()));
+        assertEquals(expected, topology.get_bolts().values().iterator().next().get_common().get_inputs());
+    }
+
+    @Test
+    public void testBranch() throws Exception {
+        Stream<Tuple> stream = streamBuilder.newStream(newSpout(Utils.DEFAULT_STREAM_ID));
+        Stream<Tuple>[] streams = stream.branch(x -> true);
+        StormTopology topology = streamBuilder.build();
+        assertEquals(1, topology.get_spouts_size());
+        assertEquals(1, topology.get_bolts_size());
+        Map<GlobalStreamId, Grouping> expected = new HashMap<>();
+        String spoutId = topology.get_spouts().keySet().iterator().next();
+        expected.put(new GlobalStreamId(spoutId, "default"), Grouping.shuffle(new NullStruct()));
+        assertEquals(expected, topology.get_bolts().values().iterator().next().get_common().get_inputs());
+        assertEquals(1, streams.length);
+        assertEquals(1, streams[0].node.getOutputStreams().size());
+        String parentStream = streams[0].node.getOutputStreams().iterator().next() + "-branch";
+        assertEquals(1, streams[0].node.getParents(parentStream).size());
+        Node processorNdoe = streams[0].node.getParents(parentStream).iterator().next();
+        assertTrue(processorNdoe instanceof ProcessorNode);
+        assertTrue(((ProcessorNode) processorNdoe).getProcessor() instanceof BranchProcessor);
+        assertTrue(processorNdoe.getParents("default").iterator().next() instanceof SpoutNode);
+    }
+
+    @Test
+    public void testJoin() throws Exception {
+        Stream<Integer> stream = streamBuilder.newStream(newSpout(Utils.DEFAULT_STREAM_ID), new ValueMapper<>(0));
+        Stream<Integer>[] streams = stream.branch(x -> x % 2 == 0, x-> x % 3 == 0);
+        PairStream<Integer, Integer> s1 = streams[0].mapToPair(x -> Pair.of(x, 1));
+        PairStream<Integer, Integer> s2 = streams[1].mapToPair(x -> Pair.of(x, 1));
+        PairStream<Integer, Pair<Integer, Integer>> sj = s1.join(s2);
+        assertEquals(Collections.singleton(s1.node), sj.node.getParents(s1.stream));
+        assertEquals(Collections.singleton(s2.node), sj.node.getParents(s2.stream));
+    }
+
+    @Test
+    public void testGroupBy() throws Exception {
+        PairStream<String, String> stream = streamBuilder.newStream(newSpout(Utils.DEFAULT_STREAM_ID), new PairValueMapper<>(0, 1));
+
+        stream.groupByKey().aggregateByKey(new Count<>());
+
+        StormTopology topology = streamBuilder.build();
+        assertEquals(2, topology.get_bolts_size());
+        Bolt bolt1 = topology.get_bolts().get("bolt1");
+        Bolt bolt2 = topology.get_bolts().get("bolt2");
+        assertEquals(Grouping.shuffle(new NullStruct()), bolt1.get_common().get_inputs().values().iterator().next());
+        assertEquals(Grouping.fields(Collections.singletonList("key")), bolt2.get_common().get_inputs().values().iterator().next());
+    }
+
+    @Test
+    public void testGlobalAggregate() throws Exception {
+        Stream<String> stream = streamBuilder.newStream(newSpout(Utils.DEFAULT_STREAM_ID), new ValueMapper<>(0));
+
+        stream.aggregate(new Count<>());
+
+        StormTopology topology = streamBuilder.build();
+        assertEquals(2, topology.get_bolts_size());
+        Bolt bolt1 = topology.get_bolts().get("bolt1");
+        Bolt bolt2 = topology.get_bolts().get("bolt2");
+        String spoutId = topology.get_spouts().keySet().iterator().next();
+        Map<GlobalStreamId, Grouping> expected1 = new HashMap<>();
+        expected1.put(new GlobalStreamId(spoutId, "default"), Grouping.shuffle(new NullStruct()));
+        Map<GlobalStreamId, Grouping> expected2 = new HashMap<>();
+        expected2.put(new GlobalStreamId("bolt1", "s1"), Grouping.fields(Collections.emptyList()));
+        assertEquals(expected1, bolt1.get_common().get_inputs());
+        assertEquals(expected2, bolt2.get_common().get_inputs());
+    }
+
+    @Test
+    public void testRepartition() throws Exception {
+        Stream<String> stream = streamBuilder.newStream(newSpout(Utils.DEFAULT_STREAM_ID), new ValueMapper<>(0));
+        stream.repartition(3).filter(x -> true).repartition(2).filter(x -> true).aggregate(new Count<>());
+        StormTopology topology = streamBuilder.build();
+        assertEquals(1, topology.get_spouts_size());
+        SpoutSpec spout = topology.get_spouts().get("spout1");
+        assertEquals(4, topology.get_bolts_size());
+        Bolt bolt1 = topology.get_bolts().get("bolt1");
+        Bolt bolt2 = topology.get_bolts().get("bolt2");
+        Bolt bolt3 = topology.get_bolts().get("bolt3");
+        Bolt bolt4 = topology.get_bolts().get("bolt4");
+        assertEquals(1, spout.get_common().get_parallelism_hint());
+        assertEquals(1, bolt1.get_common().get_parallelism_hint());
+        assertEquals(3, bolt2.get_common().get_parallelism_hint());
+        assertEquals(2, bolt3.get_common().get_parallelism_hint());
+        assertEquals(2, bolt4.get_common().get_parallelism_hint());
+    }
+
+    @Test
+    public void testBranchAndJoin() throws Exception {
+        TopologyContext mockContext = Mockito.mock(TopologyContext.class);
+        OutputCollector mockCollector = Mockito.mock(OutputCollector.class);
+        Stream<Integer> stream = streamBuilder.newStream(newSpout(Utils.DEFAULT_STREAM_ID), new ValueMapper<>(0));
+        Stream<Integer>[] streams = stream.branch(x -> x % 2 == 0, x -> x % 2 == 1);
+        PairStream<Integer, Pair<Integer, Integer>> joined = streams[0].mapToPair(x -> Pair.of(x, 1)).join(streams[1].mapToPair(x -> Pair.of(x, 1)));
+        assertTrue(joined.getNode() instanceof ProcessorNode);
+        StormTopology topology = streamBuilder.build();
+        assertEquals(1, topology.get_bolts_size());
+    }
+
+    private static IRichSpout newSpout(final String os) {
+        return new BaseRichSpout() {
+
+            @Override
+            public void declareOutputFields(OutputFieldsDeclarer declarer) {
+                declarer.declareStream(os, new Fields("value"));
+            }
+
+            @Override
+            public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
+
+            }
+
+            @Override
+            public void nextTuple() {
+
+            }
+        };
+    }
+
+    private static IRichBolt newBolt() {
+        return new BaseRichBolt() {
+
+            @Override
+            public void declareOutputFields(OutputFieldsDeclarer declarer) {
+
+            }
+
+            @Override
+            public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
+
+            }
+
+            @Override
+            public void execute(Tuple input) {
+
+            }
+        };
+    }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/test/jvm/org/apache/storm/streams/WindowedProcessorBoltTest.java
----------------------------------------------------------------------
diff --git a/storm-core/test/jvm/org/apache/storm/streams/WindowedProcessorBoltTest.java b/storm-core/test/jvm/org/apache/storm/streams/WindowedProcessorBoltTest.java
new file mode 100644
index 0000000..7428e3f
--- /dev/null
+++ b/storm-core/test/jvm/org/apache/storm/streams/WindowedProcessorBoltTest.java
@@ -0,0 +1,110 @@
+/**
+ * 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.streams;
+
+import com.google.common.collect.Multimap;
+import org.apache.storm.streams.operations.aggregators.Count;
+import org.apache.storm.streams.processors.AggregateProcessor;
+import org.apache.storm.streams.processors.Processor;
+import org.apache.storm.streams.windowing.TumblingWindows;
+import org.apache.storm.streams.windowing.Window;
+import org.apache.storm.task.OutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.base.BaseWindowedBolt;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Tuple;
+import org.apache.storm.tuple.Values;
+import org.apache.storm.windowing.TupleWindow;
+import org.jgrapht.DirectedGraph;
+import org.jgrapht.graph.DefaultDirectedGraph;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Unit test for {@link WindowedProcessorBolt}
+ */
+public class WindowedProcessorBoltTest {
+    TopologyContext mockTopologyContext;
+    OutputCollector mockOutputCollector;
+    WindowedProcessorBolt bolt;
+    Tuple mockTuple1;
+    Tuple mockTuple2;
+    Tuple mockTuple3;
+    DirectedGraph<Node, Edge> graph;
+    Multimap<String, ProcessorNode> mockStreamToProcessors;
+
+    @Before
+    public void setUp() throws Exception {
+        mockTopologyContext = Mockito.mock(TopologyContext.class);
+        mockOutputCollector = Mockito.mock(OutputCollector.class);
+        mockTuple1 = Mockito.mock(Tuple.class);
+        mockTuple2 = Mockito.mock(Tuple.class);
+        mockTuple3 = Mockito.mock(Tuple.class);
+        setUpMockTuples(mockTuple1, mockTuple2, mockTuple3);
+        mockStreamToProcessors = Mockito.mock(Multimap.class);
+    }
+
+    @Test
+    public void testEmit() throws Exception {
+        Window<?, ?> window = TumblingWindows.of(BaseWindowedBolt.Count.of(2));
+        setUpWindowedProcessorBolt(new AggregateProcessor<>(new Count<>()), window);
+        bolt.execute(getMockTupleWindow(mockTuple1, mockTuple2, mockTuple3));
+        ArgumentCaptor<Values> values = ArgumentCaptor.forClass(Values.class);
+        ArgumentCaptor<String> os = ArgumentCaptor.forClass(String.class);
+        Mockito.verify(mockOutputCollector, Mockito.times(2)).emit(os.capture(), values.capture());
+        assertEquals("outputstream", os.getAllValues().get(0));
+        assertEquals(new Values(3L), values.getAllValues().get(0));
+        assertEquals("outputstream", os.getAllValues().get(1));
+        assertEquals(new Values(WindowNode.PUNCTUATION), values.getAllValues().get(1));
+    }
+
+    private void setUpWindowedProcessorBolt(Processor<?> processor, Window<?, ?> window) {
+        ProcessorNode node = new ProcessorNode(processor, "outputstream", new Fields("value"));
+        node.setWindowed(true);
+        Mockito.when(mockStreamToProcessors.get(Mockito.anyString())).thenReturn(Collections.singletonList(node));
+        Mockito.when(mockStreamToProcessors.keySet()).thenReturn(Collections.singleton("inputstream"));
+        graph = new DefaultDirectedGraph<>(new StreamsEdgeFactory());
+        graph.addVertex(node);
+        bolt = new WindowedProcessorBolt("bolt1", graph, Collections.singletonList(node), window);
+        bolt.setStreamToInitialProcessors(mockStreamToProcessors);
+        bolt.prepare(new HashMap<>(), mockTopologyContext, mockOutputCollector);
+    }
+
+    private void setUpMockTuples(Tuple... tuples) {
+        for (Tuple tuple : tuples) {
+            Mockito.when(tuple.size()).thenReturn(1);
+            Mockito.when(tuple.getValue(0)).thenReturn(100);
+            Mockito.when(tuple.getSourceComponent()).thenReturn("bolt0");
+            Mockito.when(tuple.getSourceStreamId()).thenReturn("inputstream");
+        }
+    }
+
+    private TupleWindow getMockTupleWindow(Tuple... tuples) {
+        TupleWindow tupleWindow = Mockito.mock(TupleWindow.class);
+        Mockito.when(tupleWindow.get()).thenReturn(Arrays.asList(tuples));
+        return tupleWindow;
+    }
+}

http://git-wip-us.apache.org/repos/asf/storm/blob/e251573d/storm-core/test/jvm/org/apache/storm/windowing/WindowManagerTest.java
----------------------------------------------------------------------
diff --git a/storm-core/test/jvm/org/apache/storm/windowing/WindowManagerTest.java b/storm-core/test/jvm/org/apache/storm/windowing/WindowManagerTest.java
index 6645566..6c170c6 100644
--- a/storm-core/test/jvm/org/apache/storm/windowing/WindowManagerTest.java
+++ b/storm-core/test/jvm/org/apache/storm/windowing/WindowManagerTest.java
@@ -60,7 +60,7 @@ public class WindowManagerTest {
         }
 
         @Override
-        public void onActivation(List<Integer> events, List<Integer> newEvents, List<Integer> expired) {
+        public void onActivation(List<Integer> events, List<Integer> newEvents, List<Integer> expired, Long timestamp) {
             onActivationEvents = events;
             allOnActivationEvents.add(events);
             onActivationNewEvents = newEvents;


Mime
View raw message