import storm.trident.topology.TransactionAttempt; //导入依赖的package包/类
private Map emitNewPartitionBatch(TransactionAttempt attempt, TridentCollector collector, Partition partition, Map lastMeta) {
    try {
        return failFastEmitNewPartitionBatch(attempt, collector, partition, lastMeta);
    } catch (FailedFetchException e) {
        e.printStackTrace();
    }

    Map ret = new HashMap();

    if (lastMeta != null && partition != null) {
        ret.put("offset", lastMeta.get("nextOffset"));
        ret.put("nextOffset", lastMeta.get("nextOffset"));
        ret.put("partition", partition.partition);
        ret.put("broker", ImmutableMap.of("host", partition.host.host, "port", partition.host.port));
        ret.put("topic", _config.topic);
        ret.put("topology", ImmutableMap.of("name", _topologyName, "id", _topologyInstanceId));
    }

    return ret;
}
 

import storm.trident.topology.TransactionAttempt; //导入依赖的package包/类
@Override
public void finishBatch(ProcessorContext processorContext) {
    _collector.setContext(processorContext);
    Object batchId = processorContext.batchId;
    // since this processor type is a committer, this occurs in the commit phase
    List<TridentTuple> buffer = (List) processorContext.state[_context.getStateIndex()];
    
    // don't update unless there are tuples
    // this helps out with things like global partition persist, where multiple tasks may still
    // exist for this processor. Only want the global one to do anything
    // this is also a helpful optimization that state implementations don't need to manually do
    if(buffer.size() > 0) {
        Long txid = null;
        // this is to support things like persisting off of drpc stream, which is inherently unreliable
        // and won't have a tx attempt
        if(batchId instanceof TransactionAttempt) {
            txid = ((TransactionAttempt) batchId).getTransactionId();
        }
        _state.beginCommit(txid);
        _updater.updateState(_state, buffer, _collector);
        _state.commit(txid);
    }
}
 

import storm.trident.topology.TransactionAttempt; //导入依赖的package包/类
@Override
public void execute(BatchInfo info, Tuple input) {
    // there won't be a BatchInfo for the success stream
    TransactionAttempt attempt = (TransactionAttempt) input.getValue(0);
    if(input.getSourceStreamId().equals(MasterBatchCoordinator.COMMIT_STREAM_ID)) {
        if(attempt.equals(_activeBatches.get(attempt.getTransactionId()))) {
            ((ICommitterTridentSpout.Emitter) _emitter).commit(attempt);
            _activeBatches.remove(attempt.getTransactionId());
        } else {
             throw new FailedException("Received commit for different transaction attempt");
        }
    } else if(input.getSourceStreamId().equals(MasterBatchCoordinator.SUCCESS_STREAM_ID)) {
        // valid to delete before what's been committed since 
        // those batches will never be accessed again
        _activeBatches.headMap(attempt.getTransactionId()).clear();
        _emitter.success(attempt);
    } else {            
        _collector.setBatch(info.batchId);
        _emitter.emitBatch(attempt, input.getValue(1), _collector);
        _activeBatches.put(attempt.getTransactionId(), attempt);
    }
}
 

import storm.trident.topology.TransactionAttempt; //导入依赖的package包/类
@Override
public void execute(Tuple tuple, BasicOutputCollector collector) {
    TransactionAttempt attempt = (TransactionAttempt) tuple.getValue(0);

    if(tuple.getSourceStreamId().equals(MasterBatchCoordinator.SUCCESS_STREAM_ID)) {
        _state.cleanupBefore(attempt.getTransactionId());
        _coord.success(attempt.getTransactionId());
    } else if (tuple.getSourceStreamId().equals(MasterBatchCoordinator.COMMIT_STREAM_ID)) { 
    	// Do nothing.
    } else {
        long txid = attempt.getTransactionId();
        Object prevMeta = _state.getPreviousState(txid);
        Object meta = _coord.initializeTransaction(txid, prevMeta, _state.getState(txid));
        _state.overrideState(txid, meta);
        collector.emit(MasterBatchCoordinator.BATCH_STREAM_ID, new Values(attempt, meta));
    }
            
}
 

import storm.trident.topology.TransactionAttempt; //导入依赖的package包/类
/**
    * This attempt committed successfully, so all state for this commit and before can be safely cleaned up.
    * In case of batch replay scenario in Opaque Transactional State Management, this method will get invoked
    * (irrespective of thrown FailedException for a trxnId). This is because as per design of Opaque 
    * Transactional State Management, the same data isn't tied to a partuclar trxnId.
    * The same failed data can be transmitted via another trxnId.
    */

@Override
public void success(TransactionAttempt trxnAttempt) {
	// TODO Auto-generated method stub
	String curThreadName = Thread.currentThread().getName();
	try {
			throw new Exception("success: throwing intentional Exception - check code flow");
	} catch (Exception e) {
		// TODO: handle exception
		e.printStackTrace();
	}
	long trxnId = trxnAttempt.getTransactionId();
	
	CommonUtil.logMessage(logger, curThreadName, "success: entered for trxnId = %d", trxnId);
	//successfulTransactions.incrementAndGet();
	//Util.logMessage(logger, curThreadName, "success: exiting with updated succesfulTrxns = %d", successfulTransactions.get());

}
 

import storm.trident.topology.TransactionAttempt; //导入依赖的package包/类
/**
    * Emit a batch for the specified transaction attempt and metadata for the transaction. The metadata
    * was created by the Coordinator in the initializeTranaction method. This method must always emit
    * the same batch of tuples across all tasks for the same transaction id.
    * 
    * emitBatch will keep getting called repeatedly. Each invocation to emitBatch will be done in a new trxn.
    * emitBatch is internally invoked by Storm repeatedly, like a infinite while loop.Each iteration of the
    * loop is a trxn.Trxn Ids are auto-generated.
    * 
    */

@Override
public void emitBatch(TransactionAttempt trxnAttempt, Object coordinatorMeta,
		TridentCollector collector) {
	// TODO Auto-generated method stub
	String curThreadName = Thread.currentThread().getName();
	long trxnId = trxnAttempt.getTransactionId();
	for (int i = 0; i < BATCH_SIZE; i++) {
          List<Object> events = new ArrayList<Object>();
          
          String uid = UUID.randomUUID().toString();
          int field1 = rand.nextInt(1);//possible values - 0
          int field2 = rand.nextInt(1);//possible values - 0
          
          TupleDS tupleDs = new TupleDS(uid, uid, field1, field2);
          
          events.add(tupleDs);
          collector.emit(events);
  		  CommonUtil.logMessage(logger, curThreadName, "emitBatch: emitting tuple for trxnId = %d with uid = %s, field1 = %d, field2 = %d",  trxnId, uid, field1, field2);

	}
	
}
 

import storm.trident.topology.TransactionAttempt; //导入依赖的package包/类
@Override
public void execute(BatchInfo info, Tuple input) {
    // there won't be a BatchInfo for the success stream
    TransactionAttempt attempt = (TransactionAttempt) input.getValue(0);
    if (input.getSourceStreamId().equals(MasterBatchCoordinator.COMMIT_STREAM_ID)) {
        if (attempt.equals(_activeBatches.get(attempt.getTransactionId()))) {
            ((ICommitterTridentSpout.Emitter) _emitter).commit(attempt);
            _activeBatches.remove(attempt.getTransactionId());
        } else {
            throw new FailedException("Received commit for different transaction attempt");
        }
    } else if (input.getSourceStreamId().equals(MasterBatchCoordinator.SUCCESS_STREAM_ID)) {
        // valid to delete before what's been committed since
        // those batches will never be accessed again
        _activeBatches.headMap(attempt.getTransactionId()).clear();
        _emitter.success(attempt);
    } else {
        _collector.setBatch(info.batchId);
        _emitter.emitBatch(attempt, input.getValue(1), _collector);
        _activeBatches.put(attempt.getTransactionId(), attempt);
    }
}
 

import storm.trident.topology.TransactionAttempt; //导入依赖的package包/类
@Override
public void execute(Tuple tuple, BasicOutputCollector collector) {
    TransactionAttempt attempt = (TransactionAttempt) tuple.getValue(0);

    if (tuple.getSourceStreamId().equals(MasterBatchCoordinator.SUCCESS_STREAM_ID)) {
        _state.cleanupBefore(attempt.getTransactionId());
        _coord.success(attempt.getTransactionId());
    } else {
        long txid = attempt.getTransactionId();
        Object prevMeta = _state.getPreviousState(txid);
        Object meta = _coord.initializeTransaction(txid, prevMeta, _state.getState(txid));
        _state.overrideState(txid, meta);
        collector.emit(MasterBatchCoordinator.BATCH_STREAM_ID, new Values(attempt, meta));
    }

}
 

import storm.trident.topology.TransactionAttempt; //导入依赖的package包/类
/**
 * re-emit the batch described by the meta data provided
 *
 * @param attempt
 * @param collector
 * @param partition
 * @param meta
 */
private void reEmitPartitionBatch(TransactionAttempt attempt, TridentCollector collector, Partition partition, Map meta) {
    LOG.info("re-emitting batch, attempt " + attempt);
    String instanceId = (String) meta.get("instanceId");
    if (!_config.forceFromStart || instanceId.equals(_topologyInstanceId)) {
        SimpleConsumer consumer = _connections.register(partition);
        long offset = (Long) meta.get("offset");
        long nextOffset = (Long) meta.get("nextOffset");
        ByteBufferMessageSet msgs = null;
        try {
            msgs = fetchMessages(consumer, partition, offset);
        } catch (TopicOffsetOutOfRangeException e) {
            LOG.warn("OffsetOutOfRange during reEmitPartitionBatch, the transaction can not be replayed." +
                    "Returning empty messages");
        }

        if (msgs != null) {
            for (MessageAndOffset msg : msgs) {
                if (offset == nextOffset) {
                    break;
                }
                if (offset > nextOffset) {
                    throw new RuntimeException("Error when re-emitting batch. overshot the end offset");
                }
                emit(collector, msg.message());
                offset = msg.nextOffset();
            }
        }
    }
}
 

import storm.trident.topology.TransactionAttempt; //导入依赖的package包/类
public IOpaquePartitionedTridentSpout.Emitter<GlobalPartitionInformation, Partition, Map> asOpaqueEmitter() {

        return new IOpaquePartitionedTridentSpout.Emitter<GlobalPartitionInformation, Partition, Map>() {

            /**
             * Emit a batch of tuples for a partition/transaction.
             *
             * Return the metadata describing this batch that will be used as lastPartitionMeta
             * for defining the parameters of the next batch.
             */
            @Override
            public Map emitPartitionBatch(TransactionAttempt transactionAttempt, TridentCollector tridentCollector, Partition partition, Map map) {
                return emitNewPartitionBatch(transactionAttempt, tridentCollector, partition, map);
            }

            @Override
            public void refreshPartitions(List<Partition> partitions) {
                refresh(partitions);
            }

            @Override
            public List<Partition> getOrderedPartitions(GlobalPartitionInformation partitionInformation) {
                return orderPartitions(partitionInformation);
            }

            @Override
            public void close() {
                clear();
            }
        };
    }
 

import storm.trident.topology.TransactionAttempt; //导入依赖的package包/类
@Override
public void emitBatch(TransactionAttempt tx, Object coordinatorMeta, TridentCollector collector) {
    long txid = tx.getTransactionId();
    
    long now = System.currentTimeMillis();
    if(now - lastRotate > rotateTime) {
        Map<Long, List<Object>> failed = idsMap.rotate();
        for(Long id: failed.keySet()) {
            // this isn't right... it's not in the map anymore
            fail(id);
        }
        lastRotate = now;
    }
    
    if(idsMap.containsKey(txid)) {
        fail(txid);
    }
    
    _collector.reset(collector);
    if(!prepared) {
        _spout.open(_conf, _context, new SpoutOutputCollector(_collector));
        prepared = true;
    }
    for(int i=0; i<_maxBatchSize; i++) {
        _spout.nextTuple();
        if(_collector.numEmitted < i) {
            break;
        }
    }
    idsMap.put(txid, _collector.ids);

}
 

import storm.trident.topology.TransactionAttempt; //导入依赖的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.TransactionAttempt; //导入依赖的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.TransactionAttempt; //导入依赖的package包/类
@Override
public void emitBatch(TransactionAttempt tx, Map<Integer, List<List<Object>>> coordinatorMeta, TridentCollector collector) {
    List<List<Object>> tuples = coordinatorMeta.get(_index);
    if(tuples!=null) {
        for(List<Object> t: tuples) {
            collector.emit(t);
        }
    }
}
 

import storm.trident.topology.TransactionAttempt; //导入依赖的package包/类
@Override
public void emitBatch(TransactionAttempt tx, Object coordinatorMeta, TridentCollector collector) {
    long txid = tx.getTransactionId();
    
    long now = System.currentTimeMillis();
    if(now - lastRotate > rotateTime) {
        Map<Long, List<Object>> failed = idsMap.rotate();
        for(Long id: failed.keySet()) {
            //TODO: this isn't right... it's not in the map anymore
            fail(id);
        }
        lastRotate = now;
    }
    
    if(idsMap.containsKey(txid)) {
        fail(txid);
    }
    
    _collector.reset(collector);
    if(!prepared) {
        _spout.open(_conf, _context, new SpoutOutputCollector(_collector));
        prepared = true;
    }
    for(int i=0; i<_maxBatchSize; i++) {
        _spout.nextTuple();
        if(_collector.numEmitted < i) {
            break;
        }
    }
    idsMap.put(txid, _collector.ids);

}
 

import storm.trident.topology.TransactionAttempt; //导入依赖的package包/类
@Override
public void emitBatch(TransactionAttempt tx, Long coordinatorMeta,
                      TridentCollector collector) {
    for (int i = 0; i < 100; i++) {
        List<Object> sale = new ArrayList<Object>();
        sale.add(states[generator.nextInt(4)]);
        sale.add(products[generator.nextInt(4)]);
        sale.add(generator.nextInt(100));
        collector.emit(sale);
    }
}
 

import storm.trident.topology.TransactionAttempt; //导入依赖的package包/类
@Override
public void emitBatch(TransactionAttempt tx, Long coordinatorMeta,
                      TridentCollector collector) {
    for (int i = 0; i < 100; i++) {
        List<Object> message = new ArrayList<Object>();
        message.add(Integer.toString(i));
        collector.emit(message);
    }
}
 

import storm.trident.topology.TransactionAttempt; //导入依赖的package包/类
/**
    * This attempt committed successfully, so all state for this commit and before can be safely cleaned up.
    */

@Override
public void success(TransactionAttempt trxnAttempt) {
	// TODO Auto-generated method stub
	long trxnId = trxnAttempt.getTransactionId();
	String curThreadName = Thread.currentThread().getName();
	CommonUtil.logMessage(logger, curThreadName, "success: entered for trxnId = %d", trxnId);
	//successfulTransactions.incrementAndGet();
	//Util.logMessage(logger, curThreadName, "success: exiting with updated succesfulTrxns = %d", successfulTransactions.get());

}
 

import storm.trident.topology.TransactionAttempt; //导入依赖的package包/类
/**
    * Emit a batch for the specified transaction attempt and metadata for the transaction. The metadata
    * was created by the Coordinator in the initializeTranaction method. This method must always emit
    * the same batch of tuples across all tasks for the same transaction id - Transactional Spout feature.
    * 
    * emitBatch will keep getting called repeatedly. Each invocation to emitBatch will be done in a new trxn.
    * emitBatch is internally invoked by Storm repeatedly, like a infinite while loop.Each iteration of the
    * loop is a trxn.Trxn Ids are auto-generated.
    * 
    */

@Override
public void emitBatch(TransactionAttempt trxnAttempt, String coordinatorMeta,
		TridentCollector collector) {
	// TODO Auto-generated method stub
	
	String curThreadName = Thread.currentThread().getName();
	long trxnId = trxnAttempt.getTransactionId();
	CommonUtil.logMessage(logger, curThreadName, "emitBatch: entered with latestIndex = %d",  latestIndex);


	CommonUtil.logMessage(logger, curThreadName, "emitBatch: entered with trxnId = %d",  trxnId);
	/**
	 *  2014-11-28 21:07:16 b.s.util [ERROR] Async loop died!
              java.lang.RuntimeException: java.lang.ArrayIndexOutOfBoundsException: 498
           2014-11-28 21:07:16 b.s.util [INFO] Halting process: ("Worker died")
           014-11-28 21:13:43 b.s.d.worker [INFO] Launching worker for   trident-eg5-2-1417188174 on 223733d0-502e-4bfd-a074-ca39e7d5c130:6703 with  id e4c3658f-881b-493b-8304-65bc01ddf18a and conf {"dev.zookeeper.path"  "/tmp/dev-storm-zookeeper".......................

	 */
	if(latestIndex < 498)//To stop java.lang.ArrayIndexOutOfBoundsException , and to prevent worker from dying and restarting.
	{
		String randomPhrase = sentences[latestIndex];
		latestIndex++;
	    
		CommonUtil.logMessage(logger, curThreadName, "emitBatch: randomPhrase = %s",  randomPhrase);
				
		List<Object> randomPhraseList = new ArrayList<>();
		randomPhraseList.add(randomPhrase);
		
		collector.emit(randomPhraseList);
	}
	
		
}
 

import storm.trident.topology.TransactionAttempt; //导入依赖的package包/类
/**
    * This attempt committed successfully, so all state for this commit and before can be safely cleaned up.
    * In case of Transactional State Management, 
    * success will NOT be called when a batch is explicitly failed - by throwing FailedException.
    */

@Override
public void success(TransactionAttempt trxnAttempt) {
	// TODO Auto-generated method stub
	/**
	 * 	at bharati.binita.storm.trident.eg6.RandomPhraseEmitter.success(RandomPhraseEmitter.java:150)
		at storm.trident.spout.TridentSpoutExecutor.execute(TridentSpoutExecutor.java:79)
		at storm.trident.topology.TridentBoltExecutor.execute(TridentBoltExecutor.java:333)
		at backtype.storm.daemon.executor$fn__5641$tuple_action_fn__5643.invoke(executor.clj:631)
		at backtype.storm.daemon.executor$mk_task_receiver$fn__5564.invoke(executor.clj:399)
		at backtype.storm.disruptor$clojure_handler$reify__745.onEvent(disruptor.clj:58)
		at backtype.storm.utils.DisruptorQueue.consumeBatchToCursor(DisruptorQueue.java:125)
		at backtype.storm.utils.DisruptorQueue.consumeBatchWhenAvailable(DisruptorQueue.java:99)
		at backtype.storm.disruptor$consume_batch_when_available.invoke(disruptor.clj:80)
	 */
	try {
		throw new Exception("success: throwing intentional Exception - check code flow");
	} catch (Exception e) {
		// TODO: handle exception
		e.printStackTrace();
	}
	long trxnId = trxnAttempt.getTransactionId();
	String curThreadName = Thread.currentThread().getName();
	CommonUtil.logMessage(logger, curThreadName, "success: entered for trxnId = %d", trxnId);
	//successfulTransactions.incrementAndGet();
	//Util.logMessage(logger, curThreadName, "success: exiting with updated succesfulTrxns = %d", successfulTransactions.get());

}