Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/2334#discussion_r140039574
--- Diff: storm-server/src/test/java/org/apache/storm/TickTupleTest.java ---
@@ -18,85 +18,113 @@
package org.apache.storm;
+import java.util.Map;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.storm.ILocalCluster.ILocalTopology;
import org.apache.storm.generated.StormTopology;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
-import org.apache.storm.topology.IRichSpout;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.TopologyBuilder;
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.Assert;
+import org.apache.storm.utils.Time;
+import org.apache.storm.utils.TupleUtils;
import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
-import java.util.Map;
+import static org.junit.Assert.*;
public class TickTupleTest {
+ private final static Logger LOG =
LoggerFactory.getLogger(TickTupleTest.class);
+ private static LinkedBlockingQueue<Long> tickTupleTimes = new
LinkedBlockingQueue<>();
+ private static AtomicReference<Tuple> nonTickTuple = new
AtomicReference<>(null);
@Test
public void testTickTupleWorksWithSystemBolt() throws Exception {
- ILocalCluster cluster = null;
- try {
- cluster = new
LocalCluster.Builder().withSimulatedTime().withNimbusDaemon(true).build();
+ try (ILocalCluster cluster = new
LocalCluster.Builder().withSimulatedTime().withNimbusDaemon(true).build()){
StormTopology topology = createNoOpTopology();
Config topoConf = new Config();
- topoConf.putAll(Utils.readDefaultConfig());
- topoConf.put("storm.cluster.mode", "local");
topoConf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, 1);
- cluster.submitTopology("test", topoConf, topology);
- cluster.advanceClusterTime(2);
- Assert.assertTrue("Test is passed", true);
- } finally {
- cluster.close();
+ try (ILocalTopology topo = cluster.submitTopology("test",
topoConf, topology)) {
+ //Give the cluster some time to come up
+ long time = 0;
+ while (tickTupleTimes.size() <= 0) {
+ assert time <= 100_000 : "took over " + time + " ms of
simulated time to get a message back...";
+ cluster.advanceClusterTime(10);
+ time += 10_000;
+ }
+ tickTupleTimes.clear();
+ cluster.advanceClusterTime(1);
+ time += 1000;
--- End diff --
I think this can be put in a loop now, the following lines don't differ
anymore.
---
