序

本文主要研究一下storm的OpaquePartitionedTridentSpoutExecutor

TridentTopology.newStream

storm-core-1.2.2-sources.jar!/org/apache/storm/trident/TridentTopology.java

public Stream newStream(String txId, IOpaquePartitionedTridentSpout spout) {        return newStream(txId, new OpaquePartitionedTridentSpoutExecutor(spout));    }

• TridentTopology.newStream方法，对于IOpaquePartitionedTridentSpout类型的spout会使用OpaquePartitionedTridentSpoutExecutor来包装；而KafkaTridentSpoutOpaque则实现了IOpaquePartitionedTridentSpout接口

TridentTopologyBuilder.buildTopology

storm-core-1.2.2-sources.jar!/org/apache/storm/trident/topology/TridentTopologyBuilder.java

public StormTopology buildTopology(Map
    
      masterCoordResources) {        TopologyBuilder builder = new TopologyBuilder();        Map
     
       batchIdsForSpouts = fleshOutStreamBatchIds(false);        Map
      
        batchIdsForBolts = fleshOutStreamBatchIds(true);        Map
       
        > batchesToCommitIds = new HashMap<>();        Map
        
         > batchesToSpouts = new HashMap<>();                for(String id: _spouts.keySet()) {            TransactionalSpoutComponent c = _spouts.get(id);            if(c.spout instanceof IRichSpout) {                                //TODO: wrap this to set the stream name                builder.setSpout(id, (IRichSpout) c.spout, c.parallelism);            } else {                String batchGroup = c.batchGroupId;                if(!batchesToCommitIds.containsKey(batchGroup)) {                    batchesToCommitIds.put(batchGroup, new ArrayList
         
          ()); } batchesToCommitIds.get(batchGroup).add(c.commitStateId); if(!batchesToSpouts.containsKey(batchGroup)) { batchesToSpouts.put(batchGroup, new ArrayList
          
           ()); } batchesToSpouts.get(batchGroup).add((ITridentSpout) c.spout); BoltDeclarer scd = builder.setBolt(spoutCoordinator(id), new TridentSpoutCoordinator(c.commitStateId, (ITridentSpout) c.spout)) .globalGrouping(masterCoordinator(c.batchGroupId), MasterBatchCoordinator.BATCH_STREAM_ID) .globalGrouping(masterCoordinator(c.batchGroupId), MasterBatchCoordinator.SUCCESS_STREAM_ID); for(Map
           
             m: c.componentConfs) { scd.addConfigurations(m); } Map
            
              specs = new HashMap(); specs.put(c.batchGroupId, new CoordSpec()); BoltDeclarer bd = builder.setBolt(id, new TridentBoltExecutor( new TridentSpoutExecutor( c.commitStateId, c.streamName, ((ITridentSpout) c.spout)), batchIdsForSpouts, specs), c.parallelism); bd.allGrouping(spoutCoordinator(id), MasterBatchCoordinator.BATCH_STREAM_ID); bd.allGrouping(masterCoordinator(batchGroup), MasterBatchCoordinator.SUCCESS_STREAM_ID); if(c.spout instanceof ICommitterTridentSpout) { bd.allGrouping(masterCoordinator(batchGroup), MasterBatchCoordinator.COMMIT_STREAM_ID); } for(Map
             
               m: c.componentConfs) { bd.addConfigurations(m); } } } //...... return builder.createTopology(); }
             
            
           
          
         
        
       
      
     
    

• TridentTopologyBuilder.buildTopology会将IOpaquePartitionedTridentSpout(OpaquePartitionedTridentSpoutExecutor)使用TridentSpoutExecutor包装，然后再使用TridentBoltExecutor包装为bolt

OpaquePartitionedTridentSpoutExecutor

storm-core-1.2.2-sources.jar!/org/apache/storm/trident/spout/OpaquePartitionedTridentSpoutExecutor.java

public class OpaquePartitionedTridentSpoutExecutor implements ICommitterTridentSpout

• OpaquePartitionedTridentSpoutExecutor实现了ICommitterTridentSpout，这里getCoordinator返回的是ITridentSpout.BatchCoordinator，getEmitter返回的是ICommitterTridentSpout.Emitter

ITridentSpout.BatchCoordinator

storm-core-1.2.2-sources.jar!/org/apache/storm/trident/spout/OpaquePartitionedTridentSpoutExecutor.java

public class Coordinator implements ITridentSpout.BatchCoordinator

• 包装了spout的_coordinator，它的类型IOpaquePartitionedTridentSpout.Coordinator，这里仅仅是多了debug日志

ICommitterTridentSpout.Emitter

storm-core-1.2.2-sources.jar!/org/apache/storm/trident/spout/OpaquePartitionedTridentSpoutExecutor.java

public class Emitter implements ICommitterTridentSpout.Emitter {                IOpaquePartitionedTridentSpout.Emitter
    
      _emitter;        TransactionalState _state;        TreeMap
     
      > _cachedMetas = new TreeMap<>();        Map
      
        _partitionStates = new HashMap<>();        int _index;        int _numTasks;        public Emitter(String txStateId, Map conf, TopologyContext context) {            _emitter = _spout.getEmitter(conf, context);            _index = context.getThisTaskIndex();            _numTasks = context.getComponentTasks(context.getThisComponentId()).size();            _state = TransactionalState.newUserState(conf, txStateId);            LOG.debug("Created {}", this);        }        Object _savedCoordinatorMeta = null;        boolean _changedMeta = false;        @Override        public void emitBatch(TransactionAttempt tx, Object coordinatorMeta, TridentCollector collector) {            LOG.debug("Emitting Batch. [transaction = {}], [coordinatorMeta = {}], [collector = {}], [{}]",                    tx, coordinatorMeta, collector, this);            if(_savedCoordinatorMeta==null || !_savedCoordinatorMeta.equals(coordinatorMeta)) {                _partitionStates.clear();                final List
       
         taskPartitions = _emitter.getPartitionsForTask(_index, _numTasks, coordinatorMeta);                for (ISpoutPartition partition : taskPartitions) {                    _partitionStates.put(partition.getId(), new EmitterPartitionState(new RotatingTransactionalState(_state, partition.getId()), partition));                }                // refresh all partitions for backwards compatibility with old spout                _emitter.refreshPartitions(_emitter.getOrderedPartitions(coordinatorMeta));                _savedCoordinatorMeta = coordinatorMeta;                _changedMeta = true;            }            Map
        
          metas = new HashMap<>();            _cachedMetas.put(tx.getTransactionId(), metas);            Entry
         
          > entry = _cachedMetas.lowerEntry(tx.getTransactionId()); Map
          
            prevCached; if(entry!=null) { prevCached = entry.getValue(); } else { prevCached = new HashMap<>(); } for(Entry
           
             e: _partitionStates.entrySet()) { String id = e.getKey(); EmitterPartitionState s = e.getValue(); s.rotatingState.removeState(tx.getTransactionId()); Object lastMeta = prevCached.get(id); if(lastMeta==null) lastMeta = s.rotatingState.getLastState(); Object meta = _emitter.emitPartitionBatch(tx, collector, s.partition, lastMeta); metas.put(id, meta); } LOG.debug("Emitted Batch. [transaction = {}], [coordinatorMeta = {}], [collector = {}], [{}]", tx, coordinatorMeta, collector, this); } @Override public void success(TransactionAttempt tx) { for(EmitterPartitionState state: _partitionStates.values()) { state.rotatingState.cleanupBefore(tx.getTransactionId()); } LOG.debug("Success transaction {}. [{}]", tx, this); } @Override public void commit(TransactionAttempt attempt) { LOG.debug("Committing transaction {}. [{}]", attempt, this); // this code here handles a case where a previous commit failed, and the partitions // changed since the last commit. This clears out any state for the removed partitions // for this txid. // we make sure only a single task ever does this. we're also guaranteed that // it's impossible for there to be another writer to the directory for that partition // because only a single commit can be happening at once. this is because in order for // another attempt of the batch to commit, the batch phase must have succeeded in between. // hence, all tasks for the prior commit must have finished committing (whether successfully or not) if(_changedMeta && _index==0) { Set
            
              validIds = new HashSet<>(); for(ISpoutPartition p: _emitter.getOrderedPartitions(_savedCoordinatorMeta)) { validIds.add(p.getId()); } for(String existingPartition: _state.list("")) { if(!validIds.contains(existingPartition)) { RotatingTransactionalState s = new RotatingTransactionalState(_state, existingPartition); s.removeState(attempt.getTransactionId()); } } _changedMeta = false; } Long txid = attempt.getTransactionId(); Map
             
               metas = _cachedMetas.remove(txid); for(Entry
              
                entry: metas.entrySet()) { _partitionStates.get(entry.getKey()).rotatingState.overrideState(txid, entry.getValue()); } LOG.debug("Exiting commit method for transaction {}. [{}]", attempt, this); } @Override public void close() { LOG.debug("Closing"); _emitter.close(); LOG.debug("Closed"); } @Override public String toString() { return "Emitter{" + ", _state=" + _state + ", _cachedMetas=" + _cachedMetas + ", _partitionStates=" + _partitionStates + ", _index=" + _index + ", _numTasks=" + _numTasks + ", _savedCoordinatorMeta=" + _savedCoordinatorMeta + ", _changedMeta=" + _changedMeta + '}'; } } static class EmitterPartitionState { public RotatingTransactionalState rotatingState; public ISpoutPartition partition; public EmitterPartitionState(RotatingTransactionalState s, ISpoutPartition p) { rotatingState = s; partition = p; } }
              
             
            
           
          
         
        
       
      
     
    

• 这里对spout的IOpaquePartitionedTridentSpout.Emitter进行了封装，_partitionStates使用了EmitterPartitionState
• emitBatch方法首先计算_partitionStates，然后计算prevCached，最后调用_emitter.emitPartitionBatch(tx, collector, s.partition, lastMeta)
• success方法调用state.rotatingState.cleanupBefore(tx.getTransactionId())，清空该txid之前的状态信息；commit方法主要是更新_partitionStates

KafkaTridentSpoutOpaque

storm-kafka-client-1.2.2-sources.jar!/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutOpaque.java

public class KafkaTridentSpoutOpaque
    
      implements IOpaquePartitionedTridentSpout
     
      
     
    

• KafkaTridentSpoutOpaque的getCoordinator返回的是KafkaTridentSpoutOpaqueCoordinator；getEmitter返回的是KafkaTridentSpoutEmitter

KafkaTridentSpoutOpaqueCoordinator

storm-kafka-client-1.2.2-sources.jar!/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutOpaqueCoordinator.java

public class KafkaTridentSpoutOpaqueCoordinator
    
      implements IOpaquePartitionedTridentSpout.Coordinator
     
      
     
    

• 这里的isReady始终返回true，getPartitionsForBatch方法主要是将kafkaManager.getTopicPartitions()信息转换为map结构

KafkaTridentSpoutEmitter

storm-kafka-client-1.2.2-sources.jar!/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutEmitter.java

public class KafkaTridentSpoutEmitter
    
      implements IOpaquePartitionedTridentSpout.Emitter<        List
     
    

• 这里的refreshSubscriptionTimer的interval取的是kafkaManager.getKafkaSpoutConfig().getPartitionRefreshPeriodMs()，默认是2000
• emitPartitionBatch方法没调用一次都会判断refreshSubscriptionTimer.isExpiredResetOnTrue()，如果时间到了，就会调用kafkaManager.getKafkaSpoutConfig().getSubscription().refreshAssignment()刷新assignment
• emitPartitionBatch方法主要是找到与该batch关联的partition，停止从其他parition拉取消息，然后根据firstPollOffsetStrategy以及lastBatchMeta信息，调用kafkaConsumer的seek相关方法seek到指定位置
• 之后就是用kafkaConsumer.poll(pollTimeoutMs)拉取数据，然后emitTuples；emitTuples方法会是用translator转换数据，然后调用collector.emit发射出去
• refreshPartitions方法目前仅仅是trace下日志；getOrderedPartitions方法先将allPartitionInfo的数据从map结构反序列化回来，然后转换为KafkaTridentSpoutTopicPartition返回；getPartitionsForTask方法主要是通过kafkaConsumer.assignment()的信息转换为KafkaTridentSpoutTopicPartition返回

小结

• storm-kafka-client提供了KafkaTridentSpoutOpaque这个spout作为trident的kafka spout(旧版的为OpaqueTridentKafkaSpout，在storm-kafka类库中)，它实现了IOpaquePartitionedTridentSpout接口
• TridentTopology.newStream方法，对于IOpaquePartitionedTridentSpout类型的spout会使用OpaquePartitionedTridentSpoutExecutor来包装；TridentTopologyBuilder.buildTopology会将IOpaquePartitionedTridentSpout(OpaquePartitionedTridentSpoutExecutor)先使用TridentSpoutExecutor包装，然后再使用TridentBoltExecutor包装为bolt
• OpaquePartitionedTridentSpoutExecutor的getCoordinator返回的是ITridentSpout.BatchCoordinator，getEmitter返回的是ICommitterTridentSpout.Emitter；他们分别对KafkaTridentSpoutOpaque这个原始spout返回的KafkaTridentSpoutOpaqueCoordinator以及KafkaTridentSpoutEmitter进行包装再处理；其中对coordinator加了debug日志，对emitter则主要多了对EmitterPartitionState的存取

doc