import storm.trident.topology.state.RotatingTransactionalState; //导入依赖的package包/类
@Override
public void emitBatch(final TransactionAttempt tx, final Object coordinatorMeta,
        final TridentCollector collector) {
    if(_savedCoordinatorMeta == null || !_savedCoordinatorMeta.equals(coordinatorMeta)) {
        List<ISpoutPartition> partitions = _emitter.getOrderedPartitions(coordinatorMeta);
        _partitionStates.clear();
        List<ISpoutPartition> myPartitions = new ArrayList();
        for(int i=_index; i < partitions.size(); i+=_numTasks) {
            ISpoutPartition p = partitions.get(i);
            String id = p.getId();
            myPartitions.add(p);
            _partitionStates.put(id, new EmitterPartitionState(new RotatingTransactionalState(_state, id), p));
        }
        _emitter.refreshPartitions(myPartitions);
        _savedCoordinatorMeta = coordinatorMeta;
    }
    for(EmitterPartitionState s: _partitionStates.values()) {
        RotatingTransactionalState state = s.rotatingState;
        final ISpoutPartition partition = s.partition;
        Object meta = state.getStateOrCreate(tx.getTransactionId(),
                new RotatingTransactionalState.StateInitializer() {
            @Override
            public Object init(long txid, Object lastState) {
                return _emitter.emitPartitionBatchNew(tx, collector, partition, lastState);
            }
        });
        // it's null if one of:
        //   a) a later transaction batch was emitted before this, so we should skip this batch
        //   b) if didn't exist and was created (in which case the StateInitializer was invoked and 
        //      it was emitted
        if(meta!=null) {
            _emitter.emitPartitionBatch(tx, collector, partition, meta);
        }
    }            
}
 

import storm.trident.topology.state.RotatingTransactionalState; //导入依赖的package包/类
@Override
public void commit(TransactionAttempt attempt) {
    // 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<String> validIds = new HashSet<String>();
        for(ISpoutPartition p: (List<ISpoutPartition>) _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<String, Object> metas = _cachedMetas.remove(txid);
    for(String partitionId: metas.keySet()) {
        Object meta = metas.get(partitionId);
        _partitionStates.get(partitionId).rotatingState.overrideState(txid, meta);
    }
}
 

import storm.trident.topology.state.RotatingTransactionalState; //导入依赖的package包/类
@Override
public void emitBatch(final TransactionAttempt tx, final Object coordinatorMeta, final TridentCollector collector) {
    if (_savedCoordinatorMeta == null || !_savedCoordinatorMeta.equals(coordinatorMeta)) {
        List<ISpoutPartition> partitions = _emitter.getOrderedPartitions(coordinatorMeta);
        _partitionStates.clear();
        List<ISpoutPartition> myPartitions = new ArrayList();
        for (int i = _index; i < partitions.size(); i += _numTasks) {
            ISpoutPartition p = partitions.get(i);
            String id = p.getId();
            myPartitions.add(p);
            _partitionStates.put(id, new EmitterPartitionState(new RotatingTransactionalState(_state, id), p));
        }
        _emitter.refreshPartitions(myPartitions);
        _savedCoordinatorMeta = coordinatorMeta;
    }
    for (EmitterPartitionState s : _partitionStates.values()) {
        RotatingTransactionalState state = s.rotatingState;
        final ISpoutPartition partition = s.partition;
        Object meta = state.getStateOrCreate(tx.getTransactionId(), new RotatingTransactionalState.StateInitializer() {
            @Override
            public Object init(long txid, Object lastState) {
                return _emitter.emitPartitionBatchNew(tx, collector, partition, lastState);
            }
        });
        // it's null if one of:
        // a) a later transaction batch was emitted before this, so we should skip this batch
        // b) if didn't exist and was created (in which case the StateInitializer was invoked and
        // it was emitted
        if (meta != null) {
            _emitter.emitPartitionBatch(tx, collector, partition, meta);
        }
    }
}
 

import storm.trident.topology.state.RotatingTransactionalState; //导入依赖的package包/类
@Override
public void commit(TransactionAttempt attempt) {
    // 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<String> validIds = new HashSet<String>();
        for (ISpoutPartition p : (List<ISpoutPartition>) _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<String, Object> metas = _cachedMetas.remove(txid);
    for (String partitionId : metas.keySet()) {
        Object meta = metas.get(partitionId);
        _partitionStates.get(partitionId).rotatingState.overrideState(txid, meta);
    }
}
 

import storm.trident.topology.state.RotatingTransactionalState; //导入依赖的package包/类
@Override
public void emitBatch(final TransactionAttempt tx, final Integer coordinatorMeta,
        final TridentCollector collector) {
    if(_savedCoordinatorMeta == null || !_savedCoordinatorMeta.equals(coordinatorMeta)) {
        List<ISpoutPartition> partitions = _emitter.getOrderedPartitions(coordinatorMeta);
        _partitionStates.clear();
        List<ISpoutPartition> myPartitions = new ArrayList<>();
        for(int i=_index; i < partitions.size(); i+=_numTasks) {
            ISpoutPartition p = partitions.get(i);
            String id = p.getId();
            myPartitions.add(p);
            _partitionStates.put(id, new EmitterPartitionState(new RotatingTransactionalState(_state, id), p));
        }
        _emitter.refreshPartitions(myPartitions);
        _savedCoordinatorMeta = coordinatorMeta;
    }
    for(EmitterPartitionState s: _partitionStates.values()) {
        RotatingTransactionalState state = s.rotatingState;
        final ISpoutPartition partition = s.partition;
        Object meta = state.getStateOrCreate(tx.getTransactionId(),
                new RotatingTransactionalState.StateInitializer() {
            @Override
            public Object init(long txid, Object lastState) {
                return _emitter.emitPartitionBatchNew(tx, collector, partition, lastState);
            }
        });
        // it's null if one of:
        // a) a later transaction batch was emitted before this, so we should skip this batch
        // b) if didn't exist and was created (in which case the StateInitializer was invoked and
        // it was emitted
        if (meta != null) {
            _emitter.emitPartitionBatch(tx, collector, partition, meta);
        }
    }
}
 

import storm.trident.topology.state.RotatingTransactionalState; //导入依赖的package包/类
@Override
public void commit(TransactionAttempt attempt) {
    // 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<String> 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<String, Object> metas = _cachedMetas.remove(txid);
    for(Entry<String, Object> entry: metas.entrySet()) {
        _partitionStates.get(entry.getKey()).rotatingState.overrideState(txid, entry.getValue());
    }
}
 

import storm.trident.topology.state.RotatingTransactionalState; //导入依赖的package包/类
public EmitterPartitionState(RotatingTransactionalState s, ISpoutPartition p) {
    rotatingState = s;
    partition = p;
}
 

import storm.trident.topology.state.RotatingTransactionalState; //导入依赖的package包/类
@Override
public void prepare(Map conf, TopologyContext context) {
    _coord = _spout.getCoordinator(_id, conf, context);
    _underlyingState = TransactionalState.newCoordinatorState(conf, _id);
    _state = new RotatingTransactionalState(_underlyingState, META_DIR);
}