/*
 * Copyright 2018 The Apache Software Foundation.
 *
 * 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 org.apache.storm.integration;

import static org.apache.storm.AssertLoop.assertAcked;
import static org.apache.storm.AssertLoop.assertFailed;
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.Matchers.contains;
import static org.hamcrest.Matchers.containsInAnyOrder;
import static org.junit.Assert.assertThat;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Consumer;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.Testing;
import org.apache.storm.Thrift;
import org.apache.storm.Thrift.BoltDetails;
import org.apache.storm.Thrift.SpoutDetails;
import org.apache.storm.generated.GlobalStreamId;
import org.apache.storm.generated.Grouping;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.generated.StormTopology;
import org.apache.storm.generated.SubmitOptions;
import org.apache.storm.generated.TopologyInitialStatus;
import org.apache.storm.hooks.BaseTaskHook;
import org.apache.storm.hooks.info.BoltAckInfo;
import org.apache.storm.hooks.info.BoltExecuteInfo;
import org.apache.storm.hooks.info.BoltFailInfo;
import org.apache.storm.hooks.info.EmitInfo;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.testing.AckFailMapTracker;
import org.apache.storm.testing.CompleteTopologyParam;
import org.apache.storm.testing.FeederSpout;
import org.apache.storm.testing.FixedTuple;
import org.apache.storm.testing.IntegrationTest;
import org.apache.storm.testing.MockedSources;
import org.apache.storm.testing.TestAggregatesCounter;
import org.apache.storm.testing.TestConfBolt;
import org.apache.storm.testing.TestGlobalCount;
import org.apache.storm.testing.TestPlannerSpout;
import org.apache.storm.testing.TestWordCounter;
import org.apache.storm.testing.TestWordSpout;
import org.apache.storm.testing.TrackedTopology;
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.tuple.Values;
import org.apache.storm.utils.Utils;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;

@IntegrationTest
public class TopologyIntegrationTest {

    @ParameterizedTest
    @ValueSource(strings = {"true", "false"})
    public void testBasicTopology(boolean useLocalMessaging) throws Exception {
        try (LocalCluster cluster = new LocalCluster.Builder()
            .withSimulatedTime()
            .withSupervisors(4)
            .withDaemonConf(Collections.singletonMap(Config.STORM_LOCAL_MODE_ZMQ, !useLocalMessaging))
            .build()) {
            Map<String, SpoutDetails> spoutMap = Collections.singletonMap("1", Thrift.prepareSpoutDetails(new TestWordSpout(true), 3));
            Map<String, BoltDetails> boltMap = new HashMap<>();
            boltMap.put("2",
                Thrift.prepareBoltDetails(
                    Collections.singletonMap(
                        Utils.getGlobalStreamId("1", null),
                        Thrift.prepareFieldsGrouping(Collections.singletonList("word"))),
                    new TestWordCounter(), 4));
            boltMap.put("3",
                Thrift.prepareBoltDetails(
                    Collections.singletonMap(
                        Utils.getGlobalStreamId("1", null),
                        Thrift.prepareGlobalGrouping()),
                    new TestGlobalCount()));
            boltMap.put("4",
                Thrift.prepareBoltDetails(
                    Collections.singletonMap(
                        Utils.getGlobalStreamId("2", null),
                        Thrift.prepareGlobalGrouping()),
                    new TestAggregatesCounter()));
            StormTopology topology = Thrift.buildTopology(spoutMap, boltMap);

            Map<String, Object> stormConf = new HashMap<>();
            stormConf.put(Config.TOPOLOGY_WORKERS, 2);
            stormConf.put(Config.TOPOLOGY_TESTING_ALWAYS_TRY_SERIALIZE, true);

            List<FixedTuple> testTuples = Arrays.asList("nathan", "bob", "joey", "nathan").stream()
                .map(value -> new FixedTuple(new Values(value)))
                .collect(Collectors.toList());

            MockedSources mockedSources = new MockedSources(Collections.singletonMap("1", testTuples));

            CompleteTopologyParam completeTopologyParams = new CompleteTopologyParam();
            completeTopologyParams.setMockedSources(mockedSources);
            completeTopologyParams.setStormConf(stormConf);

            Map<String, List<FixedTuple>> results = Testing.completeTopology(cluster, topology, completeTopologyParams);

            assertThat(Testing.readTuples(results, "1"), containsInAnyOrder(
                new Values("nathan"),
                new Values("nathan"),
                new Values("bob"),
                new Values("joey")));
            assertThat(Testing.readTuples(results, "2"), containsInAnyOrder(
                new Values("nathan", 1),
                new Values("nathan", 2),
                new Values("bob", 1),
                new Values("joey", 1)
            ));
            assertThat(Testing.readTuples(results, "3"), contains(
                new Values(1),
                new Values(2),
                new Values(3),
                new Values(4)
            ));
            assertThat(Testing.readTuples(results, "4"), contains(
                new Values(1),
                new Values(2),
                new Values(3),
                new Values(4)
            ));
        }
    }

    private static class EmitTaskIdBolt extends BaseRichBolt {

        private int taskIndex;
        private OutputCollector collector;

        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
            declarer.declare(new Fields("tid"));
        }

        @Override
        public void prepare(Map<String, Object> topoConf, TopologyContext context, OutputCollector collector) {
            this.collector = collector;
            this.taskIndex = context.getThisTaskIndex();
        }

        @Override
        public void execute(Tuple input) {
            collector.emit(input, new Values(taskIndex));
            collector.ack(input);
        }

    }

    @Test
    public void testMultiTasksPerCluster() throws Exception {
        try (LocalCluster cluster = new LocalCluster.Builder()
            .withSimulatedTime()
            .withSupervisors(4)
            .build()) {
            Map<String, SpoutDetails> spoutMap = Collections.singletonMap("1", Thrift.prepareSpoutDetails(new TestWordSpout(true)));
            Map<String, BoltDetails> boltMap = new HashMap<>();
            boltMap.put("2",
                Thrift.prepareBoltDetails(
                    Collections.singletonMap(
                        Utils.getGlobalStreamId("1", null),
                        Thrift.prepareAllGrouping()),
                    new EmitTaskIdBolt(), 3, Collections.singletonMap(Config.TOPOLOGY_TASKS, 6)));
            StormTopology topology = Thrift.buildTopology(spoutMap, boltMap);

            MockedSources mockedSources = new MockedSources(Collections.singletonMap("1", Collections.singletonList(new FixedTuple(new Values("a")))));

            CompleteTopologyParam completeTopologyParams = new CompleteTopologyParam();
            completeTopologyParams.setMockedSources(mockedSources);

            Map<String, List<FixedTuple>> results = Testing.completeTopology(cluster, topology, completeTopologyParams);

            assertThat(Testing.readTuples(results, "2"), containsInAnyOrder(
                new Values(0),
                new Values(1),
                new Values(2),
                new Values(3),
                new Values(4),
                new Values(5)
            ));
        }
    }

    @Test
    public void testTimeout() throws Exception {
        try (LocalCluster cluster = new LocalCluster.Builder()
            .withSimulatedTime()
            .withSupervisors(4)
            .withDaemonConf(Collections.singletonMap(Config.TOPOLOGY_ENABLE_MESSAGE_TIMEOUTS, true))
            .build()) {
            FeederSpout feeder = new FeederSpout(new Fields("field1"));
            AckFailMapTracker tracker = new AckFailMapTracker();
            feeder.setAckFailDelegate(tracker);
            Map<String, SpoutDetails> spoutMap = Collections.singletonMap("1", Thrift.prepareSpoutDetails(feeder));
            Map<String, BoltDetails> boltMap = new HashMap<>();
            boltMap.put("2",
                Thrift.prepareBoltDetails(
                    Collections.singletonMap(
                        Utils.getGlobalStreamId("1", null),
                        Thrift.prepareGlobalGrouping()),
                    new AckEveryOtherBolt()));
            StormTopology topology = Thrift.buildTopology(spoutMap, boltMap);

            cluster.submitTopology("timeout-tester", Collections.singletonMap(Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS, 10), topology);

            cluster.advanceClusterTime(11);
            feeder.feed(new Values("a"), 1);
            feeder.feed(new Values("b"), 2);
            feeder.feed(new Values("c"), 3);
            cluster.advanceClusterTime(9);
            assertAcked(tracker, 1, 3);
            assertThat(tracker.isFailed(2), is(false));
            cluster.advanceClusterTime(12);
            assertFailed(tracker, 2);
        }
    }

    private static class ResetTimeoutBolt extends BaseRichBolt {

        private int tupleCounter = 1;
        private Tuple firstTuple = null;
        private OutputCollector collector;

        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
        }

        @Override
        public void prepare(Map<String, Object> topoConf, TopologyContext context, OutputCollector collector) {
            this.collector = collector;
        }

        @Override
        public void execute(Tuple input) {
            if (tupleCounter == 1) {
                firstTuple = input;
            } else if (tupleCounter == 2) {
                collector.resetTimeout(firstTuple);
            } else if (tupleCounter == 5) {
                collector.ack(firstTuple);
                collector.ack(input);
            } else {
                collector.resetTimeout(firstTuple);
                collector.ack(input);
            }
            tupleCounter++;
        }
    }

    @Test
    public void testResetTimeout() throws Exception {
        try (LocalCluster cluster = new LocalCluster.Builder()
            .withSimulatedTime()
            .withDaemonConf(Collections.singletonMap(Config.TOPOLOGY_ENABLE_MESSAGE_TIMEOUTS, true))
            .build()) {
            FeederSpout feeder = new FeederSpout(new Fields("field1"));
            AckFailMapTracker tracker = new AckFailMapTracker();
            feeder.setAckFailDelegate(tracker);
            Map<String, SpoutDetails> spoutMap = Collections.singletonMap("1", Thrift.prepareSpoutDetails(feeder));
            Map<String, BoltDetails> boltMap = new HashMap<>();
            boltMap.put("2",
                Thrift.prepareBoltDetails(
                    Collections.singletonMap(
                        Utils.getGlobalStreamId("1", null),
                        Thrift.prepareGlobalGrouping()),
                    new ResetTimeoutBolt()));
            StormTopology topology = Thrift.buildTopology(spoutMap, boltMap);

            cluster.submitTopology("reset-timeout-tester", Collections.singletonMap(Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS, 10), topology);

            //The first tuple wil be used to check timeout reset
            feeder.feed(new Values("a"), 1);
            //The second tuple is used to wait for the spout to rotate its pending map
            feeder.feed(new Values("b"), 2);
            cluster.advanceClusterTime(9);
            //The other tuples are used to reset the first tuple's timeout,
            //and to wait for the message to get through to the spout (acks use the same path as timeout resets)
            feeder.feed(new Values("c"), 3);
            assertAcked(tracker, 3);
            cluster.advanceClusterTime(9);
            feeder.feed(new Values("d"), 4);
            assertAcked(tracker, 4);
            cluster.advanceClusterTime(2);
            //The time is now twice the message timeout, the second tuple should expire since it was not acked
            //Waiting for this also ensures that the first tuple gets failed if reset-timeout doesn't work
            assertFailed(tracker, 2);
            //Put in a tuple to cause the first tuple to be acked
            feeder.feed(new Values("e"), 5);
            assertAcked(tracker, 5);
            //The first tuple should be acked, and should not have failed
            assertThat(tracker.isFailed(1), is(false));
            assertAcked(tracker, 1);
        }
    }

    private StormTopology mkValidateTopology() {
        Map<String, SpoutDetails> spoutMap = Collections.singletonMap("1", Thrift.prepareSpoutDetails(new TestWordSpout(true), 3));
        Map<String, BoltDetails> boltMap = Collections.singletonMap("2",
            Thrift.prepareBoltDetails(
                Collections.singletonMap(
                    Utils.getGlobalStreamId("1", null),
                    Thrift.prepareFieldsGrouping(Collections.singletonList("word"))),
                new TestWordCounter(), 4));
        return Thrift.buildTopology(spoutMap, boltMap);
    }

    private StormTopology mkInvalidateTopology1() {
        Map<String, SpoutDetails> spoutMap = Collections.singletonMap("1", Thrift.prepareSpoutDetails(new TestWordSpout(true), 3));
        Map<String, BoltDetails> boltMap = Collections.singletonMap("2",
            Thrift.prepareBoltDetails(
                Collections.singletonMap(
                    Utils.getGlobalStreamId("3", null),
                    Thrift.prepareFieldsGrouping(Collections.singletonList("word"))),
                new TestWordCounter(), 4));
        return Thrift.buildTopology(spoutMap, boltMap);
    }

    private StormTopology mkInvalidateTopology2() {
        Map<String, SpoutDetails> spoutMap = Collections.singletonMap("1", Thrift.prepareSpoutDetails(new TestWordSpout(true), 3));
        Map<String, BoltDetails> boltMap = Collections.singletonMap("2",
            Thrift.prepareBoltDetails(
                Collections.singletonMap(
                    Utils.getGlobalStreamId("1", null),
                    Thrift.prepareFieldsGrouping(Collections.singletonList("non-exists-field"))),
                new TestWordCounter(), 4));
        return Thrift.buildTopology(spoutMap, boltMap);
    }

    private StormTopology mkInvalidateTopology3() {
        Map<String, SpoutDetails> spoutMap = Collections.singletonMap("1", Thrift.prepareSpoutDetails(new TestWordSpout(true), 3));
        Map<String, BoltDetails> boltMap = Collections.singletonMap("2",
            Thrift.prepareBoltDetails(
                Collections.singletonMap(
                    Utils.getGlobalStreamId("1", "non-exists-stream"),
                    Thrift.prepareFieldsGrouping(Collections.singletonList("word"))),
                new TestWordCounter(), 4));
        return Thrift.buildTopology(spoutMap, boltMap);
    }

    private boolean tryCompleteWordCountTopology(LocalCluster cluster, StormTopology topology) throws Exception {
        try {
            List<FixedTuple> testTuples = Arrays.asList("nathan", "bob", "joey", "nathan").stream()
                .map(value -> new FixedTuple(new Values(value)))
                .collect(Collectors.toList());
            MockedSources mockedSources = new MockedSources(Collections.singletonMap("1", testTuples));
            CompleteTopologyParam completeTopologyParam = new CompleteTopologyParam();
            completeTopologyParam.setMockedSources(mockedSources);
            completeTopologyParam.setStormConf(Collections.singletonMap(Config.TOPOLOGY_WORKERS, 2));
            Testing.completeTopology(cluster, topology, completeTopologyParam);
            return false;
        } catch (InvalidTopologyException e) {
            return true;
        }
    }

    @Test
    public void testValidateTopologystructure() throws Exception {
        try (LocalCluster cluster = new LocalCluster.Builder()
            .withSimulatedTime()
            .withDaemonConf(Collections.singletonMap(Config.TOPOLOGY_ENABLE_MESSAGE_TIMEOUTS, true))
            .build()) {
            assertThat(tryCompleteWordCountTopology(cluster, mkValidateTopology()), is(false));
            assertThat(tryCompleteWordCountTopology(cluster, mkInvalidateTopology1()), is(true));
            assertThat(tryCompleteWordCountTopology(cluster, mkInvalidateTopology2()), is(true));
            assertThat(tryCompleteWordCountTopology(cluster, mkInvalidateTopology3()), is(true));
        }
    }

    @Test
    public void testSystemStream() throws Exception {
        //this test works because mocking a spout splits up the tuples evenly among the tasks
        try (LocalCluster cluster = new LocalCluster.Builder()
            .withSimulatedTime()
            .build()) {
            Map<String, SpoutDetails> spoutMap = Collections.singletonMap("1", Thrift.prepareSpoutDetails(new TestWordSpout(true), 3));

            Map<GlobalStreamId, Grouping> boltInputs = new HashMap<>();
            boltInputs.put(Utils.getGlobalStreamId("1", null), Thrift.prepareFieldsGrouping(Collections.singletonList("word")));
            boltInputs.put(Utils.getGlobalStreamId("1", "__system"), Thrift.prepareGlobalGrouping());
            Map<String, BoltDetails> boltMap = Collections.singletonMap("2",
                Thrift.prepareBoltDetails(
                    boltInputs,
                    new IdentityBolt(), 1));
            StormTopology topology = Thrift.buildTopology(spoutMap, boltMap);

            Map<String, Object> stormConf = new HashMap<>();
            stormConf.put(Config.TOPOLOGY_WORKERS, 2);

            List<FixedTuple> testTuples = Arrays.asList("a", "b", "c").stream()
                .map(value -> new FixedTuple(new Values(value)))
                .collect(Collectors.toList());

            MockedSources mockedSources = new MockedSources(Collections.singletonMap("1", testTuples));

            CompleteTopologyParam completeTopologyParams = new CompleteTopologyParam();
            completeTopologyParams.setMockedSources(mockedSources);
            completeTopologyParams.setStormConf(stormConf);

            Map<String, List<FixedTuple>> results = Testing.completeTopology(cluster, topology, completeTopologyParams);

            assertThat(Testing.readTuples(results, "2"), containsInAnyOrder(
                new Values("a"),
                new Values("b"),
                new Values("c")
            ));
        }
    }

    private static class SpoutAndChecker {

        private final FeederSpout spout;
        private final Consumer<Integer> checker;

        public SpoutAndChecker(FeederSpout spout, Consumer<Integer> checker) {
            this.spout = spout;
            this.checker = checker;
        }
    }

    private static class BranchingBolt extends BaseRichBolt {

        private final int branches;
        private OutputCollector collector;

        public BranchingBolt(int branches) {
            this.branches = branches;
        }

        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
            declarer.declare(new Fields("num"));
        }

        @Override
        public void prepare(Map<String, Object> topoConf, TopologyContext context, OutputCollector collector) {
            this.collector = collector;
        }

        @Override
        public void execute(Tuple input) {
            IntStream.range(0, branches)
                .forEach(i -> collector.emit(input, new Values(i)));
            collector.ack(input);
        }
    }

    private static class AckBolt extends BaseRichBolt {

        private OutputCollector collector;

        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
        }

        @Override
        public void prepare(Map<String, Object> topoConf, TopologyContext context, OutputCollector collector) {
            this.collector = collector;
        }

        @Override
        public void execute(Tuple input) {
            collector.ack(input);
        }
    }

    @Test
    public void testAcking() throws Exception {
        try (LocalCluster cluster = new LocalCluster.Builder()
            .withSimulatedTime()
            .withTracked()
            .build()) {
            AckTrackingFeeder feeder1 = new AckTrackingFeeder("num");
            AckTrackingFeeder feeder2 = new AckTrackingFeeder("num");
            AckTrackingFeeder feeder3 = new AckTrackingFeeder("num");

            Map<String, SpoutDetails> spoutMap = new HashMap<>();
            spoutMap.put("1", Thrift.prepareSpoutDetails(feeder1.getSpout()));
            spoutMap.put("2", Thrift.prepareSpoutDetails(feeder2.getSpout()));
            spoutMap.put("3", Thrift.prepareSpoutDetails(feeder3.getSpout()));

            Map<String, BoltDetails> boltMap = new HashMap<>();
            boltMap.put("4", Thrift.prepareBoltDetails(Collections.singletonMap(Utils.getGlobalStreamId("1", null), Thrift.prepareShuffleGrouping()), new BranchingBolt(2)));
            boltMap.put("5", Thrift.prepareBoltDetails(Collections.singletonMap(Utils.getGlobalStreamId("2", null), Thrift.prepareShuffleGrouping()), new BranchingBolt(4)));
            boltMap.put("6", Thrift.prepareBoltDetails(Collections.singletonMap(Utils.getGlobalStreamId("3", null), Thrift.prepareShuffleGrouping()), new BranchingBolt(1)));

            Map<GlobalStreamId, Grouping> aggregatorInputs = new HashMap<>();
            aggregatorInputs.put(Utils.getGlobalStreamId("4", null), Thrift.prepareShuffleGrouping());
            aggregatorInputs.put(Utils.getGlobalStreamId("5", null), Thrift.prepareShuffleGrouping());
            aggregatorInputs.put(Utils.getGlobalStreamId("6", null), Thrift.prepareShuffleGrouping());
            boltMap.put("7", Thrift.prepareBoltDetails(aggregatorInputs, new AggBolt(3)));

            boltMap.put("8", Thrift.prepareBoltDetails(Collections.singletonMap(Utils.getGlobalStreamId("7", null), Thrift.prepareShuffleGrouping()), new BranchingBolt(2)));
            boltMap.put("9", Thrift.prepareBoltDetails(Collections.singletonMap(Utils.getGlobalStreamId("8", null), Thrift.prepareShuffleGrouping()), new AckBolt()));

            TrackedTopology tracked = new TrackedTopology(Thrift.buildTopology(spoutMap, boltMap), cluster);;

            cluster.submitTopology("acking-test1", Collections.emptyMap(), tracked);

            cluster.advanceClusterTime(11);
            feeder1.feed(new Values(1));
            Testing.trackedWait(tracked, 1);
            feeder1.assertNumAcks(0);
            feeder2.feed(new Values(1));
            Testing.trackedWait(tracked, 1);
            feeder1.assertNumAcks(1);
            feeder2.assertNumAcks(1);
            feeder1.feed(new Values(1));
            Testing.trackedWait(tracked, 1);
            feeder1.assertNumAcks(0);
            feeder1.feed(new Values(1));
            Testing.trackedWait(tracked, 1);
            feeder1.assertNumAcks(1);
            feeder3.feed(new Values(1));
            Testing.trackedWait(tracked, 1);
            feeder1.assertNumAcks(0);
            feeder3.assertNumAcks(0);
            feeder2.feed(new Values(1));
            Testing.trackedWait(tracked, 1);
            feeder1.feed(new Values(1));
            feeder2.feed(new Values(1));
            feeder3.feed(new Values(1));
        }
    }

    @Test
    public void testAckBranching() throws Exception {
        try (LocalCluster cluster = new LocalCluster.Builder()
            .withSimulatedTime()
            .withTracked()
            .build()) {
            AckTrackingFeeder feeder = new AckTrackingFeeder("num");

            Map<String, SpoutDetails> spoutMap = new HashMap<>();
            spoutMap.put("1", Thrift.prepareSpoutDetails(feeder.getSpout()));

            Map<String, BoltDetails> boltMap = new HashMap<>();
            boltMap.put("2", Thrift.prepareBoltDetails(Collections.singletonMap(Utils.getGlobalStreamId("1", null), Thrift.prepareShuffleGrouping()), new IdentityBolt()));
            boltMap.put("3", Thrift.prepareBoltDetails(Collections.singletonMap(Utils.getGlobalStreamId("1", null), Thrift.prepareShuffleGrouping()), new IdentityBolt()));

            Map<GlobalStreamId, Grouping> aggregatorInputs = new HashMap<>();
            aggregatorInputs.put(Utils.getGlobalStreamId("2", null), Thrift.prepareShuffleGrouping());
            aggregatorInputs.put(Utils.getGlobalStreamId("3", null), Thrift.prepareShuffleGrouping());
            boltMap.put("4", Thrift.prepareBoltDetails(aggregatorInputs, new AggBolt(4)));

            TrackedTopology tracked = new TrackedTopology(Thrift.buildTopology(spoutMap, boltMap), cluster);;

            cluster.submitTopology("test-acking2", Collections.emptyMap(), tracked);

            cluster.advanceClusterTime(11);
            feeder.feed(new Values(1));
            Testing.trackedWait(tracked, 1);
            feeder.assertNumAcks(0);
            feeder.feed(new Values(1));
            Testing.trackedWait(tracked, 1);
            feeder.assertNumAcks(2);
        }
    }

    private static class DupAnchorBolt extends BaseRichBolt {

        private OutputCollector collector;

        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
            declarer.declare(new Fields("num"));
        }

        @Override
        public void prepare(Map<String, Object> topoConf, TopologyContext context, OutputCollector collector) {
            this.collector = collector;
        }

        @Override
        public void execute(Tuple input) {
            ArrayList<Tuple> anchors = new ArrayList<>();
            anchors.add(input);
            anchors.add(input);
            collector.emit(anchors, new Values(1));
            collector.ack(input);
        }
    }

    private static boolean boltPrepared = false;

    private static class PrepareTrackedBolt extends BaseRichBolt {

        private OutputCollector collector;

        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
        }

        @Override
        public void prepare(Map<String, Object> topoConf, TopologyContext context, OutputCollector collector) {
            this.collector = collector;
        }

        @Override
        public void execute(Tuple input) {
            boltPrepared = true;
            collector.ack(input);
        }
    }

    private static boolean spoutOpened = false;

    private static class OpenTrackedSpout extends BaseRichSpout {

        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
            declarer.declare(new Fields("val"));
        }

        @Override
        public void open(Map<String, Object> conf, TopologyContext context, SpoutOutputCollector collector) {
        }

        @Override
        public void nextTuple() {
            spoutOpened = true;
        }

    }

    @Test
    public void testSubmitInactiveTopology() throws Exception {
        try (LocalCluster cluster = new LocalCluster.Builder()
            .withSimulatedTime()
            .withDaemonConf(Collections.singletonMap(Config.TOPOLOGY_ENABLE_MESSAGE_TIMEOUTS, true))
            .build()) {
            FeederSpout feeder = new FeederSpout(new Fields("field1"));
            AckFailMapTracker tracker = new AckFailMapTracker();
            feeder.setAckFailDelegate(tracker);

            Map<String, SpoutDetails> spoutMap = new HashMap<>();
            spoutMap.put("1", Thrift.prepareSpoutDetails(feeder));
            spoutMap.put("2", Thrift.prepareSpoutDetails(new OpenTrackedSpout()));

            Map<String, BoltDetails> boltMap = new HashMap<>();
            boltMap.put("3", Thrift.prepareBoltDetails(Collections.singletonMap(Utils.getGlobalStreamId("1", null), Thrift.prepareGlobalGrouping()), new PrepareTrackedBolt()));

            boltPrepared = false;
            spoutOpened = false;

            StormTopology topology = Thrift.buildTopology(spoutMap, boltMap);

            cluster.submitTopologyWithOpts("test", Collections.singletonMap(Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS, 10), topology, new SubmitOptions(TopologyInitialStatus.INACTIVE));

            cluster.advanceClusterTime(11);
            feeder.feed(new Values("a"), 1);
            cluster.advanceClusterTime(9);
            assertThat(boltPrepared, is(false));
            assertThat(spoutOpened, is(false));
            cluster.getNimbus().activate("test");

            cluster.advanceClusterTime(12);
            assertAcked(tracker, 1);
            assertThat(boltPrepared, is(true));
            assertThat(spoutOpened, is(true));
        }
    }

    @Test
    public void testAckingSelfAnchor() throws Exception {
        try (LocalCluster cluster = new LocalCluster.Builder()
            .withSimulatedTime()
            .withTracked()
            .build()) {
            AckTrackingFeeder feeder = new AckTrackingFeeder("num");

            Map<String, SpoutDetails> spoutMap = new HashMap<>();
            spoutMap.put("1", Thrift.prepareSpoutDetails(feeder.getSpout()));

            Map<String, BoltDetails> boltMap = new HashMap<>();
            boltMap.put("2", Thrift.prepareBoltDetails(Collections.singletonMap(Utils.getGlobalStreamId("1", null), Thrift.prepareShuffleGrouping()), new DupAnchorBolt()));
            boltMap.put("3", Thrift.prepareBoltDetails(Collections.singletonMap(Utils.getGlobalStreamId("2", null), Thrift.prepareShuffleGrouping()), new AckBolt()));

            TrackedTopology tracked = new TrackedTopology(Thrift.buildTopology(spoutMap, boltMap), cluster);;

            cluster.submitTopology("test", Collections.emptyMap(), tracked);

            cluster.advanceClusterTime(11);
            feeder.feed(new Values(1));
            Testing.trackedWait(tracked, 1);
            feeder.assertNumAcks(1);
            feeder.feed(new Values(1));
            feeder.feed(new Values(1));
            feeder.feed(new Values(1));
            Testing.trackedWait(tracked, 3);
            feeder.assertNumAcks(3);
        }
    }

    private Map<Object, Object> listToMap(List<Object> list) {
        assertThat(list.size() % 2, is(0));
        Map<Object, Object> res = new HashMap<>();
        for (int i = 0; i < list.size(); i += 2) {
            res.put(list.get(i), list.get(i + 1));
        }
        return res;
    }

    @Test
    public void testKryoDecoratorsConfig() throws Exception {
        Map<String, Object> daemonConf = new HashMap<>();
        daemonConf.put(Config.TOPOLOGY_SKIP_MISSING_KRYO_REGISTRATIONS, true);
        daemonConf.put(Config.TOPOLOGY_KRYO_DECORATORS, "this-is-overridden");
        try (LocalCluster cluster = new LocalCluster.Builder()
            .withSimulatedTime()
            .withDaemonConf(daemonConf)
            .build()) {
            TopologyBuilder topologyBuilder = new TopologyBuilder();
            topologyBuilder.setSpout("1", new TestPlannerSpout(new Fields("conf")));
            topologyBuilder.setBolt("2", new TestConfBolt(Collections.singletonMap(Config.TOPOLOGY_KRYO_DECORATORS, Arrays.asList("one", "two"))))
                .shuffleGrouping("1");

            List<FixedTuple> testTuples = Arrays.asList(new Values(Config.TOPOLOGY_KRYO_DECORATORS)).stream()
                .map(value -> new FixedTuple(value))
                .collect(Collectors.toList());

            MockedSources mockedSources = new MockedSources(Collections.singletonMap("1", testTuples));

            CompleteTopologyParam completeTopologyParams = new CompleteTopologyParam();
            completeTopologyParams.setMockedSources(mockedSources);
            completeTopologyParams.setStormConf(Collections.singletonMap(Config.TOPOLOGY_KRYO_DECORATORS, Arrays.asList("one", "three")));

            Map<String, List<FixedTuple>> results = Testing.completeTopology(cluster, topologyBuilder.createTopology(), completeTopologyParams);

            List<Object> concatValues = Testing.readTuples(results, "2").stream()
                .flatMap(values -> values.stream())
                .collect(Collectors.toList());
            assertThat(concatValues.get(0), is(Config.TOPOLOGY_KRYO_DECORATORS));
            assertThat(concatValues.get(1), is(Arrays.asList("one", "two", "three")));
        }
    }

    @Test
    public void testComponentSpecificConfig() throws Exception {
        Map<String, Object> daemonConf = new HashMap<>();
        daemonConf.put(Config.TOPOLOGY_SKIP_MISSING_KRYO_REGISTRATIONS, true);
        try (LocalCluster cluster = new LocalCluster.Builder()
            .withSimulatedTime()
            .withDaemonConf(daemonConf)
            .build()) {
            TopologyBuilder topologyBuilder = new TopologyBuilder();
            topologyBuilder.setSpout("1", new TestPlannerSpout(new Fields("conf")));
            Map<String, Object> componentConf = new HashMap<>();
            componentConf.put("fake.config", 123);
            componentConf.put(Config.TOPOLOGY_MAX_TASK_PARALLELISM, 20);
            componentConf.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 30);
            componentConf.put(Config.TOPOLOGY_KRYO_REGISTER, Arrays.asList(Collections.singletonMap("fake.type", "bad.serializer"), Collections.singletonMap("fake.type2", "a.serializer")));
            topologyBuilder.setBolt("2", new TestConfBolt(componentConf))
                .shuffleGrouping("1")
                .setMaxTaskParallelism(2)
                .addConfiguration("fake.config2", 987);

            List<FixedTuple> testTuples = Arrays.asList("fake.config", Config.TOPOLOGY_MAX_TASK_PARALLELISM, Config.TOPOLOGY_MAX_SPOUT_PENDING, "fake.config2", Config.TOPOLOGY_KRYO_REGISTER).stream()
                .map(value -> new FixedTuple(new Values(value)))
                .collect(Collectors.toList());
            Map<String, String> kryoRegister = new HashMap<>();
            kryoRegister.put("fake.type", "good.serializer");
            kryoRegister.put("fake.type3", "a.serializer3");
            Map<String, Object> stormConf = new HashMap<>();
            stormConf.put(Config.TOPOLOGY_KRYO_REGISTER, Arrays.asList(kryoRegister));

            MockedSources mockedSources = new MockedSources(Collections.singletonMap("1", testTuples));

            CompleteTopologyParam completeTopologyParams = new CompleteTopologyParam();
            completeTopologyParams.setMockedSources(mockedSources);
            completeTopologyParams.setStormConf(stormConf);

            Map<String, List<FixedTuple>> results = Testing.completeTopology(cluster, topologyBuilder.createTopology(), completeTopologyParams);

            Map<String, Object> expectedValues = new HashMap<>();
            expectedValues.put("fake.config", 123L);
            expectedValues.put("fake.config2", 987L);
            expectedValues.put(Config.TOPOLOGY_MAX_TASK_PARALLELISM, 2L);
            expectedValues.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 30L);
            Map<String, String> expectedKryoRegister = new HashMap<>();
            expectedKryoRegister.putAll(kryoRegister);
            expectedKryoRegister.put("fake.type2", "a.serializer");
            expectedValues.put(Config.TOPOLOGY_KRYO_REGISTER, expectedKryoRegister);
            List<Object> concatValues = Testing.readTuples(results, "2").stream()
                .flatMap(values -> values.stream())
                .collect(Collectors.toList());
            assertThat(listToMap(concatValues), is(expectedValues));
        }
    }
    
    private static class HooksBolt extends BaseRichBolt {

        private int acked = 0;
        private int failed = 0;
        private int executed = 0;
        private int emitted = 0;
        private OutputCollector collector;

        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
            declarer.declare(new Fields("emit", "ack", "fail", "executed"));
        }

        @Override
        public void prepare(Map<String, Object> topoConf, TopologyContext context, OutputCollector collector) {
            this.collector = collector;
            context.addTaskHook(new BaseTaskHook() {
                @Override
                public void boltExecute(BoltExecuteInfo info) {
                    executed++;
                }

                @Override
                public void boltFail(BoltFailInfo info) {
                    failed++;
                }

                @Override
                public void boltAck(BoltAckInfo info) {
                    acked++;
                }

                @Override
                public void emit(EmitInfo info) {
                    emitted++;
                }
                
            });
        }

        @Override
        public void execute(Tuple input) {
            collector.emit(new Values(emitted, acked, failed, executed));
            if (acked - failed == 0) {
                collector.ack(input);
            } else {
                collector.fail(input);
            }
        }
    }
    
    @Test
    public void testHooks() throws Exception {
        try (LocalCluster cluster = new LocalCluster.Builder()
            .withSimulatedTime()
            .build()) {
            Map<String, SpoutDetails> spoutMap = Collections.singletonMap("1", Thrift.prepareSpoutDetails(new TestPlannerSpout(new Fields("conf"))));

            Map<String, BoltDetails> boltMap = Collections.singletonMap("2",
                Thrift.prepareBoltDetails(
                    Collections.singletonMap(Utils.getGlobalStreamId("1", null), Thrift.prepareShuffleGrouping()),
                    new HooksBolt()));
            StormTopology topology = Thrift.buildTopology(spoutMap, boltMap);

            List<FixedTuple> testTuples = Arrays.asList(1, 1, 1, 1).stream()
                .map(value -> new FixedTuple(new Values(value)))
                .collect(Collectors.toList());

            MockedSources mockedSources = new MockedSources(Collections.singletonMap("1", testTuples));

            CompleteTopologyParam completeTopologyParams = new CompleteTopologyParam();
            completeTopologyParams.setMockedSources(mockedSources);

            Map<String, List<FixedTuple>> results = Testing.completeTopology(cluster, topology, completeTopologyParams);
            
            List<List<Object>> expectedTuples = Arrays.asList(
                Arrays.asList(0, 0, 0, 0),
                Arrays.asList(2, 1, 0, 1),
                Arrays.asList(4, 1, 1, 2),
                Arrays.asList(6, 2, 1, 3));

            assertThat(Testing.readTuples(results, "2"), is(expectedTuples));
        }
    }

}