<commit_before>package backtype.storm.transactional;

import backtype.storm.coordination.IBatchBolt;
import backtype.storm.coordination.BatchBoltExecutor;
import backtype.storm.Config;
import backtype.storm.Constants;
import backtype.storm.coordination.CoordinatedBolt;
import backtype.storm.coordination.CoordinatedBolt.IdStreamSpec;
import backtype.storm.coordination.CoordinatedBolt.SourceArgs;
import backtype.storm.generated.GlobalStreamId;
import backtype.storm.generated.Grouping;
import backtype.storm.generated.StormTopology;
import backtype.storm.grouping.CustomStreamGrouping;
import backtype.storm.topology.BaseConfigurationDeclarer;
import backtype.storm.topology.BasicBoltExecutor;
import backtype.storm.topology.BoltDeclarer;
import backtype.storm.topology.IBasicBolt;
import backtype.storm.topology.IRichBolt;
import backtype.storm.topology.InputDeclarer;
import backtype.storm.topology.SpoutDeclarer;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.transactional.partitioned.IPartitionedTransactionalSpout;
import backtype.storm.transactional.partitioned.PartitionedTransactionalSpoutExecutor;
import backtype.storm.tuple.Fields;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * TODO: check to see if there are two topologies active with the same transactional id 
 * essentially want to implement a file lock on top of zk (use ephemeral nodes?)
 * or just use the topology name?
 * 
 */

public class TransactionalTopologyBuilder {
    String _id;
    String _spoutId;
    ITransactionalSpout _spout;
    Map<String, Component> _bolts = new HashMap<String, Component>();
    Integer _spoutParallelism;
    List<Map> _spoutConfs = new ArrayList();
    
    // id is used to store the state of this transactionalspout in zookeeper
    // it would be very dangerous to have 2 topologies active with the same id in the same cluster    
    public TransactionalTopologyBuilder(String id, String spoutId, ITransactionalSpout spout, Integer spoutParallelism) {
        _id = id;
        _spoutId = spoutId;
        _spout = spout;
        _spoutParallelism = spoutParallelism;
    }
    
    public TransactionalTopologyBuilder(String id, String spoutId, IPartitionedTransactionalSpout spout, Integer spoutParallelism) {
        this(id, spoutId, new PartitionedTransactionalSpoutExecutor(spout), spoutParallelism);
    }

    public TransactionalTopologyBuilder(String id, String spoutId, ITransactionalSpout spout) {
        this(id, spoutId, spout, null);
    }
    
    public TransactionalTopologyBuilder(String id, String spoutId, IPartitionedTransactionalSpout spout) {
        this(id, spoutId, spout, null);
    }    
    
    public SpoutDeclarer getSpoutDeclarer() {
        return new SpoutDeclarerImpl();
    }
    
    public BoltDeclarer setBolt(String id, IBatchBolt bolt) {
        return setBolt(id, bolt, null);
    }
    
    public BoltDeclarer setBolt(String id, IBatchBolt bolt, Integer parallelism) {
        return setBolt(id, new BatchBoltExecutor(bolt), parallelism, bolt instanceof ICommitter);
    }

    public BoltDeclarer setCommitterBolt(String id, IBatchBolt bolt) {
        return setCommitterBolt(id, bolt, null);
    }
    
    public BoltDeclarer setCommitterBolt(String id, IBatchBolt bolt, Integer parallelism) {
        return setBolt(id, new BatchBoltExecutor(bolt), parallelism, true);
    }      
    
    public BoltDeclarer setBolt(String id, IBasicBolt bolt) {
        return setBolt(id, bolt, null);
    }    
    
    public BoltDeclarer setBolt(String id, IBasicBolt bolt, Integer parallelism) {
        return setBolt(id, new BasicBoltExecutor(bolt), parallelism, false);
    }
    
    private BoltDeclarer setBolt(String id, IRichBolt bolt, Integer parallelism, boolean committer) {
        Component component = new Component(bolt, parallelism, committer);
        _bolts.put(id, component);
        return new BoltDeclarerImpl(component);
    }
    
    public TopologyBuilder buildTopologyBuilder() {
        String coordinator = _spoutId + "/coordinator";
        TopologyBuilder builder = new TopologyBuilder();
        SpoutDeclarer declarer = builder.setSpout(coordinator, new TransactionalSpoutCoordinator(_spout));
        for(Map conf: _spoutConfs) {
            declarer.addConfigurations(conf);
        }
        declarer.addConfiguration(Config.TOPOLOGY_TRANSACTIONAL_ID, _id);

        builder.setBolt(_spoutId,
                        new CoordinatedBolt(new TransactionalSpoutBatchExecutor(_spout),
                                             null,
                                             null),
                        _spoutParallelism)
                .allGrouping(coordinator, TransactionalSpoutCoordinator.TRANSACTION_BATCH_STREAM_ID)
                .addConfiguration(Config.TOPOLOGY_TRANSACTIONAL_ID, _id);
        for(String id: _bolts.keySet()) {
            Component component = _bolts.get(id);
            Map<String, SourceArgs> coordinatedArgs = new HashMap<String, SourceArgs>();
            for(String c: componentBoltSubscriptions(component)) {
                coordinatedArgs.put(c, SourceArgs.all());
            }
            
            IdStreamSpec idSpec = null;
            if(component.committer) {
                idSpec = IdStreamSpec.makeDetectSpec(coordinator, TransactionalSpoutCoordinator.TRANSACTION_COMMIT_STREAM_ID);          
            }
            BoltDeclarer input = builder.setBolt(id,
                                                  new CoordinatedBolt(component.bolt,
                                                                      coordinatedArgs,
                                                                      idSpec),
                                                  component.parallelism);
            for(Map conf: component.componentConfs) {
                input.addConfigurations(conf);
            }
            for(String c: componentBoltSubscriptions(component)) {
                input.directGrouping(c, Constants.COORDINATED_STREAM_ID);
            }
            for(InputDeclaration d: component.declarations) {
                d.declare(input);
            }
            if(component.committer) {
                input.allGrouping(coordinator, TransactionalSpoutCoordinator.TRANSACTION_COMMIT_STREAM_ID);                
            }
        }
        return builder;
    }
    
    public StormTopology buildTopology() {
        return buildTopologyBuilder().createTopology();
    }
    
    private Set<String> componentBoltSubscriptions(Component component) {
        Set<String> ret = new HashSet<String>();
        for(InputDeclaration d: component.declarations) {
            ret.add(d.getComponent());
        }
        return ret;
    }

    private static class Component {
        public IRichBolt bolt;
        public int parallelism;
        public List<InputDeclaration> declarations = new ArrayList<InputDeclaration>();
        public List<Map> componentConfs = new ArrayList<Map>();
        public boolean committer;
        
        public Component(IRichBolt bolt, int parallelism, boolean committer) {
            this.bolt = bolt;
            this.parallelism = parallelism;
            this.committer = committer;
        }<commit_msg>allow null parallelism for components<commit_after>package backtype.storm.transactional;

import backtype.storm.coordination.IBatchBolt;
import backtype.storm.coordination.BatchBoltExecutor;
import backtype.storm.Config;
import backtype.storm.Constants;
import backtype.storm.coordination.CoordinatedBolt;
import backtype.storm.coordination.CoordinatedBolt.IdStreamSpec;
import backtype.storm.coordination.CoordinatedBolt.SourceArgs;
import backtype.storm.generated.GlobalStreamId;
import backtype.storm.generated.Grouping;
import backtype.storm.generated.StormTopology;
import backtype.storm.grouping.CustomStreamGrouping;
import backtype.storm.topology.BaseConfigurationDeclarer;
import backtype.storm.topology.BasicBoltExecutor;
import backtype.storm.topology.BoltDeclarer;
import backtype.storm.topology.IBasicBolt;
import backtype.storm.topology.IRichBolt;
import backtype.storm.topology.InputDeclarer;
import backtype.storm.topology.SpoutDeclarer;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.transactional.partitioned.IPartitionedTransactionalSpout;
import backtype.storm.transactional.partitioned.PartitionedTransactionalSpoutExecutor;
import backtype.storm.tuple.Fields;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * TODO: check to see if there are two topologies active with the same transactional id 
 * essentially want to implement a file lock on top of zk (use ephemeral nodes?)
 * or just use the topology name?
 * 
 */

public class TransactionalTopologyBuilder {
    String _id;
    String _spoutId;
    ITransactionalSpout _spout;
    Map<String, Component> _bolts = new HashMap<String, Component>();
    Integer _spoutParallelism;
    List<Map> _spoutConfs = new ArrayList();
    
    // id is used to store the state of this transactionalspout in zookeeper
    // it would be very dangerous to have 2 topologies active with the same id in the same cluster    
    public TransactionalTopologyBuilder(String id, String spoutId, ITransactionalSpout spout, Integer spoutParallelism) {
        _id = id;
        _spoutId = spoutId;
        _spout = spout;
        _spoutParallelism = spoutParallelism;
    }
    
    public TransactionalTopologyBuilder(String id, String spoutId, IPartitionedTransactionalSpout spout, Integer spoutParallelism) {
        this(id, spoutId, new PartitionedTransactionalSpoutExecutor(spout), spoutParallelism);
    }

    public TransactionalTopologyBuilder(String id, String spoutId, ITransactionalSpout spout) {
        this(id, spoutId, spout, null);
    }
    
    public TransactionalTopologyBuilder(String id, String spoutId, IPartitionedTransactionalSpout spout) {
        this(id, spoutId, spout, null);
    }    
    
    public SpoutDeclarer getSpoutDeclarer() {
        return new SpoutDeclarerImpl();
    }
    
    public BoltDeclarer setBolt(String id, IBatchBolt bolt) {
        return setBolt(id, bolt, null);
    }
    
    public BoltDeclarer setBolt(String id, IBatchBolt bolt, Integer parallelism) {
        return setBolt(id, new BatchBoltExecutor(bolt), parallelism, bolt instanceof ICommitter);
    }

    public BoltDeclarer setCommitterBolt(String id, IBatchBolt bolt) {
        return setCommitterBolt(id, bolt, null);
    }
    
    public BoltDeclarer setCommitterBolt(String id, IBatchBolt bolt, Integer parallelism) {
        return setBolt(id, new BatchBoltExecutor(bolt), parallelism, true);
    }      
    
    public BoltDeclarer setBolt(String id, IBasicBolt bolt) {
        return setBolt(id, bolt, null);
    }    
    
    public BoltDeclarer setBolt(String id, IBasicBolt bolt, Integer parallelism) {
        return setBolt(id, new BasicBoltExecutor(bolt), parallelism, false);
    }
    
    private BoltDeclarer setBolt(String id, IRichBolt bolt, Integer parallelism, boolean committer) {
        Component component = new Component(bolt, parallelism, committer);
        _bolts.put(id, component);
        return new BoltDeclarerImpl(component);
    }
    
    public TopologyBuilder buildTopologyBuilder() {
        String coordinator = _spoutId + "/coordinator";
        TopologyBuilder builder = new TopologyBuilder();
        SpoutDeclarer declarer = builder.setSpout(coordinator, new TransactionalSpoutCoordinator(_spout));
        for(Map conf: _spoutConfs) {
            declarer.addConfigurations(conf);
        }
        declarer.addConfiguration(Config.TOPOLOGY_TRANSACTIONAL_ID, _id);

        builder.setBolt(_spoutId,
                        new CoordinatedBolt(new TransactionalSpoutBatchExecutor(_spout),
                                             null,
                                             null),
                        _spoutParallelism)
                .allGrouping(coordinator, TransactionalSpoutCoordinator.TRANSACTION_BATCH_STREAM_ID)
                .addConfiguration(Config.TOPOLOGY_TRANSACTIONAL_ID, _id);
        for(String id: _bolts.keySet()) {
            Component component = _bolts.get(id);
            Map<String, SourceArgs> coordinatedArgs = new HashMap<String, SourceArgs>();
            for(String c: componentBoltSubscriptions(component)) {
                coordinatedArgs.put(c, SourceArgs.all());
            }
            
            IdStreamSpec idSpec = null;
            if(component.committer) {
                idSpec = IdStreamSpec.makeDetectSpec(coordinator, TransactionalSpoutCoordinator.TRANSACTION_COMMIT_STREAM_ID);          
            }
            BoltDeclarer input = builder.setBolt(id,
                                                  new CoordinatedBolt(component.bolt,
                                                                      coordinatedArgs,
                                                                      idSpec),
                                                  component.parallelism);
            for(Map conf: component.componentConfs) {
                input.addConfigurations(conf);
            }
            for(String c: componentBoltSubscriptions(component)) {
                input.directGrouping(c, Constants.COORDINATED_STREAM_ID);
            }
            for(InputDeclaration d: component.declarations) {
                d.declare(input);
            }
            if(component.committer) {
                input.allGrouping(coordinator, TransactionalSpoutCoordinator.TRANSACTION_COMMIT_STREAM_ID);                
            }
        }
        return builder;
    }
    
    public StormTopology buildTopology() {
        return buildTopologyBuilder().createTopology();
    }
    
    private Set<String> componentBoltSubscriptions(Component component) {
        Set<String> ret = new HashSet<String>();
        for(InputDeclaration d: component.declarations) {
            ret.add(d.getComponent());
        }
        return ret;
    }

    private static class Component {
        public IRichBolt bolt;
        public Integer parallelism;
        public List<InputDeclaration> declarations = new ArrayList<InputDeclaration>();
        public List<Map> componentConfs = new ArrayList<Map>();
        public boolean committer;
        
        public Component(IRichBolt bolt, Integer parallelism, boolean committer) {
            this.bolt = bolt;
            this.parallelism = parallelism;
            this.committer = committer;
        }
{'fraction_non_alphanumeric': 0.0745721271393643, 'fraction_numerical': 0.00013583265417006248, 'mean_word_length': 2.6529397171917637, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 27, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 28, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5452950', 'n_tokens_mistral': 4020, 'n_tokens_neox': 3934, 'n_words': 876}