Запуск топологии Trident в модульном тесте с отслеживаемой топологией Storm

Question

Как запустить тест JUnit для топологии Trident, чтобы позволить кортежам проходить через топологию во время тестирования и проверки выходных данных на каждом этапе? Я пытался запустить в рамках среды тестирования Storm, но он не дает возможности проверки и последовательного выполнения Trident.

Вот пример топологии с некоторыми встроенными комментариями, где у меня больше всего проблем.

import static org.junit.Assert.assertEquals;
import java.util.Arrays;
import java.util.List;
import org.junit.Test;
import storm.trident.TridentState;
import storm.trident.TridentTopology;
import storm.trident.operation.builtin.Count;
import storm.trident.testing.MemoryMapState;
import storm.trident.testing.Split;
import backtype.storm.Config;
import backtype.storm.ILocalCluster;
import backtype.storm.Testing;
import backtype.storm.testing.FeederSpout;
import backtype.storm.testing.TestJob;
import backtype.storm.testing.TrackedTopology;
import backtype.storm.tuple.Fields;
import backtype.storm.utils.Utils;

public class WordCountTopologyTest {

    @Test
    public void testWordCountTopology() throws Exception {
        Testing.withTrackedCluster(new WordCountTestJob());
    }

    public class WordCountTestJob implements TestJob {

        @Override
        public void run(ILocalCluster cluster) throws Exception {

            // Create the test topology to submit
            TridentTopology termCountTopology = new TridentTopology();

            FeederSpout feeder = new FeederSpout(new Fields("text", "author"));

            TridentState tridentState = termCountTopology.newStream("inputSpout", feeder)
                    .each(new Fields("text"), new Split(), new Fields("word"))
                   .groupBy(new Fields("word"))
                   .name("counter-output")
                   .persistentAggregate(new MemoryMapState.Factory(), new Count(), new Fields("count"))                
                   .parallelismHint(6);

            TrackedTopology tracked = Testing.mkTrackedTopology(cluster, termCountTopology.build());

            // Feed some random data into the topology
            feeder.feed(Arrays.asList("Nearly all men can stand adversity, but if you want to test a man's character, give him power.", "Abraham Lincoln"));
            feeder.feed(Arrays.asList("No man has a good enough memory to be a successful liar.", "Abraham Lincoln"));
            feeder.feed(Arrays.asList("Either write something worth reading or do something worth writing.", "Benjamin Franklin"));
            feeder.feed(Arrays.asList("Well done is better than well said.", "Benjamin Franklin"));

            cluster.submitTopology("word-count-testing", new Config(), tracked.getTopology());

            // (!!) Runs, but bad to sleep for any time when may run faster or slower on other systems
            // Utils.sleep(5000);

            // (!!) Fails with 5000ms Topology timeout
            // Testing.trackedWait(tracked, 3);

            /*
             * (!!) Always 0. Trident creates the streams and bolts internally with
             * different names, so how can we read them to verify?
             */
            List outputTuples = Testing.readTuples(tracked, "counter-output");
            assertEquals(0, outputTuples.size());
        }
    }
}

Помимо этого, я попытался написать свой собственный BaseFilter, чтобы пометить конец потока, в котором хранятся все кортежи, но, похоже, должен быть лучший способ. Кроме того, это не решает проблему запуска топологии контролируемым образом. Это то, что поддерживает Trident?

Answer 1

Используйте класс FeederBatchSpout (для Trident) вместо FeederSpout. FeederBatchSpout блокирует сам по себе, нет необходимости использовать Testing.trackedWait() или что-то подобное.

Источник: https://groups.google.com/forum/