Package org.apache.storm

Class Testing

  • public class Testing
extends Object
    A utility that helps with testing topologies, Bolts and Spouts.

    • Field Detail

      • TEST_TIMEOUT_MS

        public static final int TEST_TIMEOUT_MS
        The default amount of wall time should be spent waiting for specific conditions to happen. Default is 10 seconds unless the environment variable STORM_TEST_TIMEOUT_MS is set.

    • Constructor Detail

      • Testing

        public Testing()

    • Method Detail

      • whileTimeout

        public static void whileTimeout​(Testing.Condition condition,
                                Runnable body)
        Continue to execute body repeatedly until condition is true or TEST_TIMEOUT_MS has passed.
        Parameters:
        condition - what we are waiting for
        body - what to run in the loop
        Throws:
        AssertionError - if the loop timed out.

      • whileTimeout

        public static void whileTimeout​(long timeoutMs,
                                Testing.Condition condition,
                                Runnable body)
        Continue to execute body repeatedly until condition is true or TEST_TIMEOUT_MS has passed.
        Parameters:
        timeoutMs - the number of ms to wait before timing out.
        condition - what we are waiting for
        body - what to run in the loop
        Throws:
        AssertionError - if the loop timed out.

      • isEvery

        public static <T> boolean isEvery​(Collection<T> data,
                                  Predicate<T> pred)
        Convenience method for data.stream.allMatch(pred).

      • withSimulatedTime

        @Deprecated
public static void withSimulatedTime​(Runnable code)
        Deprecated.
        use ``` try (Time.SimulatedTime time = new Time.SimulatedTime()) { ... } ```
        Run with simulated time.
        Parameters:
        code - what to run

      • withLocalCluster

        @Deprecated
public static void withLocalCluster​(TestJob code)
        Deprecated.
        use ``` try (LocalCluster cluster = new LocalCluster()) { ... } ```
        Run with a local cluster.
        Parameters:
        code - what to run

      • withLocalCluster

        @Deprecated
public static void withLocalCluster​(MkClusterParam param,
                                    TestJob code)
        Deprecated.
        use ``` try (LocalCluster cluster = new LocalCluster.Builder()....build()) { ... } ```
        Run with a local cluster.
        Parameters:
        param - configs to set in the cluster
        code - what to run

      • getLocalCluster

        @Deprecated
public static ILocalCluster getLocalCluster​(Map<String,​Object> clusterConf)
        Deprecated.
        use ``` try (LocalCluster cluster = new LocalCluster.Builder()....build()) { ... } ```
        Run with a local cluster.
        Parameters:
        clusterConf - some configs to set in the cluster

      • withSimulatedTimeLocalCluster

        @Deprecated
public static void withSimulatedTimeLocalCluster​(TestJob code)
        Deprecated.
        use ``` try (LocalCluster cluster = new LocalCluster.Builder().withSimulatedTime().build()) { ... } ```
        Run with a local cluster.
        Parameters:
        code - what to run

      • withSimulatedTimeLocalCluster

        @Deprecated
public static void withSimulatedTimeLocalCluster​(MkClusterParam param,
                                                 TestJob code)
        Deprecated.
        use ``` try (LocalCluster cluster = new LocalCluster.Builder().withSimulatedTime()....build()) { ... } ```
        Run with a local cluster.
        Parameters:
        param - configs to set in the cluster
        code - what to run

      • withTrackedCluster

        @Deprecated
public static void withTrackedCluster​(TestJob code)
        Deprecated.
        use ``` try (LocalCluster cluster = new LocalCluster.Builder().withTracked().build()) { ... } ```
        Run with a local cluster.
        Parameters:
        code - what to run

      • withTrackedCluster

        @Deprecated
public static void withTrackedCluster​(MkClusterParam param,
                                      TestJob code)
        Deprecated.
        use ``` try (LocalCluster cluster = new LocalCluster.Builder().withTracked()....build()) { ... } ```
        Run with a local tracked cluster.
        Parameters:
        param - configs to set in the cluster
        code - what to run

      • globalAmt

        @Deprecated
public static int globalAmt​(String id,
                            String key)
        Deprecated.
        In a tracked topology some metrics are tracked. This provides a way to get those metrics. This is intended mostly for internal testing.
        Parameters:
        id - the id of the tracked cluster
        key - the name of the metric to get.
        Returns:
        the metric

      • captureTopology

        public static Testing.CapturedTopology<StormTopology> captureTopology​(StormTopology topology)
        Rewrites a topology so that all the tuples flowing through it are captured.
        Parameters:
        topology - the topology to rewrite
        Returns:
        the modified topology and a new Bolt that can retrieve the captured tuples.

      • completeTopology

        public static Map<String,​List<FixedTuple>> completeTopology​(ILocalCluster cluster,
                                                                  StormTopology topology)
                                                           throws InterruptedException,
                                                                  org.apache.storm.thrift.TException
        Run a topology to completion capturing all of the messages that are emitted. This only works when all of the spouts are instances of CompletableSpout.
        Parameters:
        cluster - the cluster to submit the topology to
        topology - the topology itself
        Returns:
        a map of the component to the list of tuples it emitted
        Throws:
        org.apache.storm.thrift.TException - on any error from nimbus
        InterruptedException

      • completeTopology

        public static Map<String,​List<FixedTuple>> completeTopology​(ILocalCluster cluster,
                                                                  StormTopology topology,
                                                                  CompleteTopologyParam param)
                                                           throws org.apache.storm.thrift.TException,
                                                                  InterruptedException
        Run a topology to completion capturing all of the messages that are emitted. This only works when all of the spouts are instances of CompletableSpout or are overwritten by MockedSources in param
        Parameters:
        cluster - the cluster to submit the topology to
        topology - the topology itself
        param - parameters to describe how to complete a topology
        Returns:
        a map of the component to the list of tuples it emitted
        Throws:
        org.apache.storm.thrift.TException - on any error from nimbus.
        InterruptedException

      • readTuples

        public static List<List<Object>> readTuples​(Map<String,​List<FixedTuple>> results,
                                            String componentId)
        Get all of the tuples from a given component on the default stream.
        Parameters:
        results - the results of running a completed topology
        componentId - the id of the component to look at
        Returns:
        a list of the tuple values.

      • readTuples

        public static List<List<Object>> readTuples​(Map<String,​List<FixedTuple>> results,
                                            String componentId,
                                            String streamId)
        Get all of the tuples from a given component on a given stream.
        Parameters:
        results - the results of running a completed topology
        componentId - the id of the component to look at
        streamId - the id of the stream to look for.
        Returns:
        a list of the tuple values.

      • trackedWait

        public static void trackedWait​(TrackedTopology topo)
        Simulated time wait for a tracked topology. This is intended for internal testing.

      • trackedWait

        public static void trackedWait​(TrackedTopology topo,
                               Integer amt)
        Simulated time wait for a tracked topology. This is intended for internal testing.

      • trackedWait

        public static void trackedWait​(TrackedTopology topo,
                               Integer amt,
                               Integer timeoutMs)
        Simulated time wait for a tracked topology. This is intended for internal testing.

      • multiset

        public static <T> Map<T,​Integer> multiset​(Collection<T> c)
        Count how many times each element appears in the Collection.
        Parameters:
        c - a collection of values
        Returns:
        a map of the unique values in c to the count of those values.

      • multiseteq

        public static <T> boolean multiseteq​(Collection<T> a,
                                     Collection<T> b)
        Check if two collections are equivalent ignoring the order of elements.

      • testTuple

        public static Tuple testTuple​(List<Object> values)
        Create a Tuple for use with testing.
        Parameters:
        values - the values to appear in the tuple

      • testTuple

        public static Tuple testTuple​(List<Object> values,
                              MkTupleParam param)
        Create a Tuple for use with testing.
        Parameters:
        values - the values to appear in the tuple
        param - parametrs describing more details about the tuple