You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by pt...@apache.org on 2015/11/05 21:40:58 UTC
[19/60] [abbrv] [partial] storm git commit: Release 2.0.4-SNAPSHOT
http://git-wip-us.apache.org/repos/asf/storm/blob/e935da91/jstorm-client/src/main/java/storm/trident/planner/processor/StateQueryProcessor.java
----------------------------------------------------------------------
diff --git a/jstorm-client/src/main/java/storm/trident/planner/processor/StateQueryProcessor.java b/jstorm-client/src/main/java/storm/trident/planner/processor/StateQueryProcessor.java
deleted file mode 100644
index 878fb6c..0000000
--- a/jstorm-client/src/main/java/storm/trident/planner/processor/StateQueryProcessor.java
+++ /dev/null
@@ -1,89 +0,0 @@
-package storm.trident.planner.processor;
-
-import backtype.storm.task.TopologyContext;
-import backtype.storm.tuple.Fields;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import storm.trident.operation.TridentOperationContext;
-import storm.trident.planner.ProcessorContext;
-import storm.trident.planner.TridentProcessor;
-import storm.trident.state.QueryFunction;
-import storm.trident.state.State;
-import storm.trident.tuple.TridentTuple;
-import storm.trident.tuple.TridentTuple.Factory;
-import storm.trident.tuple.TridentTupleView.ProjectionFactory;
-
-
-public class StateQueryProcessor implements TridentProcessor {
- QueryFunction _function;
- State _state;
- String _stateId;
- TridentContext _context;
- Fields _inputFields;
- ProjectionFactory _projection;
- AppendCollector _collector;
-
- public StateQueryProcessor(String stateId, Fields inputFields, QueryFunction function) {
- _stateId = stateId;
- _function = function;
- _inputFields = inputFields;
- }
-
- @Override
- public void prepare(Map conf, TopologyContext context, TridentContext tridentContext) {
- List<Factory> parents = tridentContext.getParentTupleFactories();
- if(parents.size()!=1) {
- throw new RuntimeException("State query operation can only have one parent");
- }
- _context = tridentContext;
- _state = (State) context.getTaskData(_stateId);
- _projection = new ProjectionFactory(parents.get(0), _inputFields);
- _collector = new AppendCollector(tridentContext);
- _function.prepare(conf, new TridentOperationContext(context, _projection));
- }
-
- @Override
- public void cleanup() {
- _function.cleanup();
- }
-
- @Override
- public void startBatch(ProcessorContext processorContext) {
- processorContext.state[_context.getStateIndex()] = new BatchState();
- }
-
- @Override
- public void execute(ProcessorContext processorContext, String streamId, TridentTuple tuple) {
- BatchState state = (BatchState) processorContext.state[_context.getStateIndex()];
- state.tuples.add(tuple);
- state.args.add(_projection.create(tuple));
- }
-
- @Override
- public void finishBatch(ProcessorContext processorContext) {
- BatchState state = (BatchState) processorContext.state[_context.getStateIndex()];
- if(!state.tuples.isEmpty()) {
- List<Object> results = _function.batchRetrieve(_state, state.args);
- if(results.size()!=state.tuples.size()) {
- throw new RuntimeException("Results size is different than argument size: " + results.size() + " vs " + state.tuples.size());
- }
- for(int i=0; i<state.tuples.size(); i++) {
- TridentTuple tuple = state.tuples.get(i);
- Object result = results.get(i);
- _collector.setContext(processorContext, tuple);
- _function.execute(_projection.create(tuple), result, _collector);
- }
- }
- }
-
- private static class BatchState {
- public List<TridentTuple> tuples = new ArrayList<TridentTuple>();
- public List<TridentTuple> args = new ArrayList<TridentTuple>();
- }
-
- @Override
- public Factory getOutputFactory() {
- return _collector.getOutputFactory();
- }
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e935da91/jstorm-client/src/main/java/storm/trident/planner/processor/TridentContext.java
----------------------------------------------------------------------
diff --git a/jstorm-client/src/main/java/storm/trident/planner/processor/TridentContext.java b/jstorm-client/src/main/java/storm/trident/planner/processor/TridentContext.java
deleted file mode 100644
index ae39768..0000000
--- a/jstorm-client/src/main/java/storm/trident/planner/processor/TridentContext.java
+++ /dev/null
@@ -1,59 +0,0 @@
-package storm.trident.planner.processor;
-
-import backtype.storm.coordination.BatchOutputCollector;
-import backtype.storm.tuple.Fields;
-import java.util.List;
-import storm.trident.planner.TupleReceiver;
-import storm.trident.tuple.TridentTuple.Factory;
-
-
-public class TridentContext {
- Fields selfFields;
- List<Factory> parentFactories;
- List<String> parentStreams;
- List<TupleReceiver> receivers;
- String outStreamId;
- int stateIndex;
- BatchOutputCollector collector;
-
- public TridentContext(Fields selfFields, List<Factory> parentFactories,
- List<String> parentStreams, List<TupleReceiver> receivers,
- String outStreamId, int stateIndex, BatchOutputCollector collector) {
- this.selfFields = selfFields;
- this.parentFactories = parentFactories;
- this.parentStreams = parentStreams;
- this.receivers = receivers;
- this.outStreamId = outStreamId;
- this.stateIndex = stateIndex;
- this.collector = collector;
- }
-
- public List<Factory> getParentTupleFactories() {
- return parentFactories;
- }
-
- public Fields getSelfOutputFields() {
- return selfFields;
- }
-
- public List<String> getParentStreams() {
- return parentStreams;
- }
-
- public List<TupleReceiver> getReceivers() {
- return receivers;
- }
-
- public String getOutStreamId() {
- return outStreamId;
- }
-
- public int getStateIndex() {
- return stateIndex;
- }
-
- //for reporting errors
- public BatchOutputCollector getDelegateCollector() {
- return collector;
- }
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e935da91/jstorm-client/src/main/java/storm/trident/spout/BatchSpoutExecutor.java
----------------------------------------------------------------------
diff --git a/jstorm-client/src/main/java/storm/trident/spout/BatchSpoutExecutor.java b/jstorm-client/src/main/java/storm/trident/spout/BatchSpoutExecutor.java
deleted file mode 100644
index 4e5fc99..0000000
--- a/jstorm-client/src/main/java/storm/trident/spout/BatchSpoutExecutor.java
+++ /dev/null
@@ -1,75 +0,0 @@
-package storm.trident.spout;
-
-import backtype.storm.task.TopologyContext;
-import storm.trident.topology.TransactionAttempt;
-import backtype.storm.tuple.Fields;
-import java.util.Map;
-import storm.trident.operation.TridentCollector;
-
-public class BatchSpoutExecutor implements ITridentSpout {
- public static class EmptyCoordinator implements BatchCoordinator {
- @Override
- public Object initializeTransaction(long txid, Object prevMetadata, Object currMetadata) {
- return null;
- }
-
- @Override
- public void close() {
- }
-
- @Override
- public void success(long txid) {
- }
-
- @Override
- public boolean isReady(long txid) {
- return true;
- }
- }
-
- public class BatchSpoutEmitter implements Emitter {
-
- @Override
- public void emitBatch(TransactionAttempt tx, Object coordinatorMeta, TridentCollector collector) {
- _spout.emitBatch(tx.getTransactionId(), collector);
- }
-
- @Override
- public void success(TransactionAttempt tx) {
- _spout.ack(tx.getTransactionId());
- }
-
- @Override
- public void close() {
- _spout.close();
- }
- }
-
- IBatchSpout _spout;
-
- public BatchSpoutExecutor(IBatchSpout spout) {
- _spout = spout;
- }
-
- @Override
- public BatchCoordinator getCoordinator(String txStateId, Map conf, TopologyContext context) {
- return new EmptyCoordinator();
- }
-
- @Override
- public Emitter getEmitter(String txStateId, Map conf, TopologyContext context) {
- _spout.open(conf, context);
- return new BatchSpoutEmitter();
- }
-
- @Override
- public Map getComponentConfiguration() {
- return _spout.getComponentConfiguration();
- }
-
- @Override
- public Fields getOutputFields() {
- return _spout.getOutputFields();
- }
-
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e935da91/jstorm-client/src/main/java/storm/trident/spout/IBatchID.java
----------------------------------------------------------------------
diff --git a/jstorm-client/src/main/java/storm/trident/spout/IBatchID.java b/jstorm-client/src/main/java/storm/trident/spout/IBatchID.java
deleted file mode 100644
index e41d8e2..0000000
--- a/jstorm-client/src/main/java/storm/trident/spout/IBatchID.java
+++ /dev/null
@@ -1,7 +0,0 @@
-package storm.trident.spout;
-
-
-public interface IBatchID {
- Object getId();
- int getAttemptId();
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e935da91/jstorm-client/src/main/java/storm/trident/spout/IBatchSpout.java
----------------------------------------------------------------------
diff --git a/jstorm-client/src/main/java/storm/trident/spout/IBatchSpout.java b/jstorm-client/src/main/java/storm/trident/spout/IBatchSpout.java
deleted file mode 100644
index 8034026..0000000
--- a/jstorm-client/src/main/java/storm/trident/spout/IBatchSpout.java
+++ /dev/null
@@ -1,16 +0,0 @@
-package storm.trident.spout;
-
-import backtype.storm.task.TopologyContext;
-import backtype.storm.tuple.Fields;
-import java.io.Serializable;
-import java.util.Map;
-import storm.trident.operation.TridentCollector;
-
-public interface IBatchSpout extends Serializable {
- void open(Map conf, TopologyContext context);
- void emitBatch(long batchId, TridentCollector collector);
- void ack(long batchId);
- void close();
- Map getComponentConfiguration();
- Fields getOutputFields();
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e935da91/jstorm-client/src/main/java/storm/trident/spout/ICommitterTridentSpout.java
----------------------------------------------------------------------
diff --git a/jstorm-client/src/main/java/storm/trident/spout/ICommitterTridentSpout.java b/jstorm-client/src/main/java/storm/trident/spout/ICommitterTridentSpout.java
deleted file mode 100644
index e163c77..0000000
--- a/jstorm-client/src/main/java/storm/trident/spout/ICommitterTridentSpout.java
+++ /dev/null
@@ -1,14 +0,0 @@
-package storm.trident.spout;
-
-import backtype.storm.task.TopologyContext;
-import storm.trident.topology.TransactionAttempt;
-import java.util.Map;
-
-public interface ICommitterTridentSpout<X> extends ITridentSpout<X> {
- public interface Emitter extends ITridentSpout.Emitter {
- void commit(TransactionAttempt attempt);
- }
-
- @Override
- public Emitter getEmitter(String txStateId, Map conf, TopologyContext context);
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/storm/blob/e935da91/jstorm-client/src/main/java/storm/trident/spout/IOpaquePartitionedTridentSpout.java
----------------------------------------------------------------------
diff --git a/jstorm-client/src/main/java/storm/trident/spout/IOpaquePartitionedTridentSpout.java b/jstorm-client/src/main/java/storm/trident/spout/IOpaquePartitionedTridentSpout.java
deleted file mode 100644
index 7a0e254..0000000
--- a/jstorm-client/src/main/java/storm/trident/spout/IOpaquePartitionedTridentSpout.java
+++ /dev/null
@@ -1,45 +0,0 @@
-package storm.trident.spout;
-
-import backtype.storm.task.TopologyContext;
-import backtype.storm.tuple.Fields;
-import java.io.Serializable;
-import java.util.List;
-import java.util.Map;
-import storm.trident.operation.TridentCollector;
-import storm.trident.topology.TransactionAttempt;
-
-/**
- * This defines a transactional spout which does *not* necessarily
- * replay the same batch every time it emits a batch for a transaction id.
- *
- */
-public interface IOpaquePartitionedTridentSpout<Partitions, Partition extends ISpoutPartition, M> extends Serializable {
- public interface Coordinator<Partitions> {
- boolean isReady(long txid);
- Partitions getPartitionsForBatch();
- void close();
- }
-
- public interface Emitter<Partitions, Partition extends ISpoutPartition, M> {
- /**
- * 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.
- */
- M emitPartitionBatch(TransactionAttempt tx, TridentCollector collector, Partition partition, M lastPartitionMeta);
-
- /**
- * This method is called when this task is responsible for a new set of partitions. Should be used
- * to manage things like connections to brokers.
- */
- void refreshPartitions(List<Partition> partitionResponsibilities);
- List<Partition> getOrderedPartitions(Partitions allPartitionInfo);
- void close();
- }
-
- Emitter<Partitions, Partition, M> getEmitter(Map conf, TopologyContext context);
- Coordinator getCoordinator(Map conf, TopologyContext context);
- Map getComponentConfiguration();
- Fields getOutputFields();
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/storm/blob/e935da91/jstorm-client/src/main/java/storm/trident/spout/IPartitionedTridentSpout.java
----------------------------------------------------------------------
diff --git a/jstorm-client/src/main/java/storm/trident/spout/IPartitionedTridentSpout.java b/jstorm-client/src/main/java/storm/trident/spout/IPartitionedTridentSpout.java
deleted file mode 100644
index c27bdc9..0000000
--- a/jstorm-client/src/main/java/storm/trident/spout/IPartitionedTridentSpout.java
+++ /dev/null
@@ -1,60 +0,0 @@
-package storm.trident.spout;
-
-import backtype.storm.task.TopologyContext;
-import backtype.storm.tuple.Fields;
-import java.io.Serializable;
-import java.util.List;
-import java.util.Map;
-import storm.trident.operation.TridentCollector;
-import storm.trident.topology.TransactionAttempt;
-
-/**
- * This interface defines a transactional spout that reads its tuples from a partitioned set of
- * brokers. It automates the storing of metadata for each partition to ensure that the same batch
- * is always emitted for the same transaction id. The partition metadata is stored in Zookeeper.
- */
-public interface IPartitionedTridentSpout<Partitions, Partition extends ISpoutPartition, T> extends Serializable {
- public interface Coordinator<Partitions> {
- /**
- * Return the partitions currently in the source of data. The idea is
- * is that if a new partition is added and a prior transaction is replayed, it doesn't
- * emit tuples for the new partition because it knows what partitions were in
- * that transaction.
- */
- Partitions getPartitionsForBatch();
-
- boolean isReady(long txid);
-
- void close();
- }
-
- public interface Emitter<Partitions, Partition extends ISpoutPartition, X> {
-
- List<Partition> getOrderedPartitions(Partitions allPartitionInfo);
-
- /**
- * Emit a batch of tuples for a partition/transaction that's never been emitted before.
- * Return the metadata that can be used to reconstruct this partition/batch in the future.
- */
- X emitPartitionBatchNew(TransactionAttempt tx, TridentCollector collector, Partition partition, X lastPartitionMeta);
-
- /**
- * This method is called when this task is responsible for a new set of partitions. Should be used
- * to manage things like connections to brokers.
- */
- void refreshPartitions(List<Partition> partitionResponsibilities);
-
- /**
- * Emit a batch of tuples for a partition/transaction that has been emitted before, using
- * the metadata created when it was first emitted.
- */
- void emitPartitionBatch(TransactionAttempt tx, TridentCollector collector, Partition partition, X partitionMeta);
- void close();
- }
-
- Coordinator<Partitions> getCoordinator(Map conf, TopologyContext context);
- Emitter<Partitions, Partition, T> getEmitter(Map conf, TopologyContext context);
-
- Map getComponentConfiguration();
- Fields getOutputFields();
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e935da91/jstorm-client/src/main/java/storm/trident/spout/ISpoutPartition.java
----------------------------------------------------------------------
diff --git a/jstorm-client/src/main/java/storm/trident/spout/ISpoutPartition.java b/jstorm-client/src/main/java/storm/trident/spout/ISpoutPartition.java
deleted file mode 100644
index 38a561f..0000000
--- a/jstorm-client/src/main/java/storm/trident/spout/ISpoutPartition.java
+++ /dev/null
@@ -1,8 +0,0 @@
-package storm.trident.spout;
-
-public interface ISpoutPartition {
- /**
- * This is used as a Zookeeper node path for storing metadata.
- */
- String getId();
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e935da91/jstorm-client/src/main/java/storm/trident/spout/ITridentSpout.java
----------------------------------------------------------------------
diff --git a/jstorm-client/src/main/java/storm/trident/spout/ITridentSpout.java b/jstorm-client/src/main/java/storm/trident/spout/ITridentSpout.java
deleted file mode 100644
index 37d8c11..0000000
--- a/jstorm-client/src/main/java/storm/trident/spout/ITridentSpout.java
+++ /dev/null
@@ -1,77 +0,0 @@
-package storm.trident.spout;
-
-import backtype.storm.task.TopologyContext;
-import storm.trident.topology.TransactionAttempt;
-import backtype.storm.tuple.Fields;
-import java.io.Serializable;
-import java.util.Map;
-import storm.trident.operation.TridentCollector;
-
-
-public interface ITridentSpout<T> extends Serializable {
- public interface BatchCoordinator<X> {
- /**
- * Create metadata for this particular transaction id which has never
- * been emitted before. The metadata should contain whatever is necessary
- * to be able to replay the exact batch for the transaction at a later point.
- *
- * The metadata is stored in Zookeeper.
- *
- * Storm uses the Kryo serializations configured in the component configuration
- * for this spout to serialize and deserialize the metadata.
- *
- * @param txid The id of the transaction.
- * @param prevMetadata The metadata of the previous transaction
- * @param currMetadata The metadata for this transaction the last time it was initialized.
- * null if this is the first attempt
- * @return the metadata for this new transaction
- */
- X initializeTransaction(long txid, X prevMetadata, X currMetadata);
-
- void success(long txid);
-
- boolean isReady(long txid);
-
- /**
- * Release any resources from this coordinator.
- */
- void close();
- }
-
- public interface Emitter<X> {
- /**
- * 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.
- *
- */
- void emitBatch(TransactionAttempt tx, X coordinatorMeta, TridentCollector collector);
-
- /**
- * This attempt committed successfully, so all state for this commit and before can be safely cleaned up.
- */
- void success(TransactionAttempt tx);
-
- /**
- * Release any resources held by this emitter.
- */
- void close();
- }
-
- /**
- * The coordinator for a TransactionalSpout runs in a single thread and indicates when batches
- * of tuples should be emitted and when transactions should commit. The Coordinator that you provide
- * in a TransactionalSpout provides metadata for each transaction so that the transactions can be replayed.
- */
- BatchCoordinator<T> getCoordinator(String txStateId, Map conf, TopologyContext context);
-
- /**
- * The emitter for a TransactionalSpout runs as many tasks across the cluster. Emitters are responsible for
- * emitting batches of tuples for a transaction and must ensure that the same batch of tuples is always
- * emitted for the same transaction id.
- */
- Emitter<T> getEmitter(String txStateId, Map conf, TopologyContext context);
-
- Map getComponentConfiguration();
- Fields getOutputFields();
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e935da91/jstorm-client/src/main/java/storm/trident/spout/OpaquePartitionedTridentSpoutExecutor.java
----------------------------------------------------------------------
diff --git a/jstorm-client/src/main/java/storm/trident/spout/OpaquePartitionedTridentSpoutExecutor.java b/jstorm-client/src/main/java/storm/trident/spout/OpaquePartitionedTridentSpoutExecutor.java
deleted file mode 100644
index 0e02205..0000000
--- a/jstorm-client/src/main/java/storm/trident/spout/OpaquePartitionedTridentSpoutExecutor.java
+++ /dev/null
@@ -1,184 +0,0 @@
-package storm.trident.spout;
-
-
-import backtype.storm.task.TopologyContext;
-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.Map.Entry;
-import java.util.Set;
-import java.util.TreeMap;
-import storm.trident.operation.TridentCollector;
-import storm.trident.topology.state.RotatingTransactionalState;
-import storm.trident.topology.state.TransactionalState;
-import storm.trident.topology.TransactionAttempt;
-
-
-public class OpaquePartitionedTridentSpoutExecutor implements ICommitterTridentSpout<Object> {
- IOpaquePartitionedTridentSpout _spout;
-
- public class Coordinator implements ITridentSpout.BatchCoordinator<Object> {
- IOpaquePartitionedTridentSpout.Coordinator _coordinator;
-
- public Coordinator(Map conf, TopologyContext context) {
- _coordinator = _spout.getCoordinator(conf, context);
- }
-
- @Override
- public Object initializeTransaction(long txid, Object prevMetadata, Object currMetadata) {
- return _coordinator.getPartitionsForBatch();
- }
-
- @Override
- public void close() {
- _coordinator.close();
- }
-
- @Override
- public void success(long txid) {
- }
-
- @Override
- public boolean isReady(long txid) {
- return _coordinator.isReady(txid);
- }
- }
-
- static class EmitterPartitionState {
- public RotatingTransactionalState rotatingState;
- public ISpoutPartition partition;
-
- public EmitterPartitionState(RotatingTransactionalState s, ISpoutPartition p) {
- rotatingState = s;
- partition = p;
- }
- }
-
- public class Emitter implements ICommitterTridentSpout.Emitter {
- IOpaquePartitionedTridentSpout.Emitter _emitter;
- TransactionalState _state;
- TreeMap<Long, Map<String, Object>> _cachedMetas = new TreeMap<Long, Map<String, Object>>();
- Map<String, EmitterPartitionState> _partitionStates = new HashMap<String, EmitterPartitionState>();
- 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);
- }
-
- Object _savedCoordinatorMeta = null;
- boolean _changedMeta = false;
-
- @Override
- public void emitBatch(TransactionAttempt tx, Object coordinatorMeta, 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;
- _changedMeta = true;
- }
- Map<String, Object> metas = new HashMap<String, Object>();
- _cachedMetas.put(tx.getTransactionId(), metas);
-
- Entry<Long, Map<String, Object>> entry = _cachedMetas.lowerEntry(tx.getTransactionId());
- Map<String, Object> prevCached;
- if(entry!=null) {
- prevCached = entry.getValue();
- } else {
- prevCached = new HashMap<String, Object>();
- }
-
- for(String id: _partitionStates.keySet()) {
- EmitterPartitionState s = _partitionStates.get(id);
- 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);
- }
- }
-
- @Override
- public void success(TransactionAttempt tx) {
- for(EmitterPartitionState state: _partitionStates.values()) {
- state.rotatingState.cleanupBefore(tx.getTransactionId());
- }
- }
-
- @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);
- }
- }
-
- @Override
- public void close() {
- _emitter.close();
- }
- }
-
- public OpaquePartitionedTridentSpoutExecutor(IOpaquePartitionedTridentSpout spout) {
- _spout = spout;
- }
-
- @Override
- public ITridentSpout.BatchCoordinator<Object> getCoordinator(String txStateId, Map conf, TopologyContext context) {
- return new Coordinator(conf, context);
- }
-
- @Override
- public ICommitterTridentSpout.Emitter getEmitter(String txStateId, Map conf, TopologyContext context) {
- return new Emitter(txStateId, conf, context);
- }
-
- @Override
- public Fields getOutputFields() {
- return _spout.getOutputFields();
- }
-
- @Override
- public Map<String, Object> getComponentConfiguration() {
- return _spout.getComponentConfiguration();
- }
-
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e935da91/jstorm-client/src/main/java/storm/trident/spout/PartitionedTridentSpoutExecutor.java
----------------------------------------------------------------------
diff --git a/jstorm-client/src/main/java/storm/trident/spout/PartitionedTridentSpoutExecutor.java b/jstorm-client/src/main/java/storm/trident/spout/PartitionedTridentSpoutExecutor.java
deleted file mode 100644
index 484d275..0000000
--- a/jstorm-client/src/main/java/storm/trident/spout/PartitionedTridentSpoutExecutor.java
+++ /dev/null
@@ -1,154 +0,0 @@
-package storm.trident.spout;
-
-import backtype.storm.task.TopologyContext;
-import backtype.storm.tuple.Fields;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import storm.trident.operation.TridentCollector;
-import storm.trident.topology.TransactionAttempt;
-import storm.trident.topology.state.RotatingTransactionalState;
-import storm.trident.topology.state.TransactionalState;
-
-
-public class PartitionedTridentSpoutExecutor implements ITridentSpout<Integer> {
- IPartitionedTridentSpout _spout;
-
- public PartitionedTridentSpoutExecutor(IPartitionedTridentSpout spout) {
- _spout = spout;
- }
-
- public IPartitionedTridentSpout getPartitionedSpout() {
- return _spout;
- }
-
- class Coordinator implements ITridentSpout.BatchCoordinator<Object> {
- private IPartitionedTridentSpout.Coordinator _coordinator;
-
- public Coordinator(Map conf, TopologyContext context) {
- _coordinator = _spout.getCoordinator(conf, context);
- }
-
- @Override
- public Object initializeTransaction(long txid, Object prevMetadata, Object currMetadata) {
- if(currMetadata!=null) {
- return currMetadata;
- } else {
- return _coordinator.getPartitionsForBatch();
- }
- }
-
-
- @Override
- public void close() {
- _coordinator.close();
- }
-
- @Override
- public void success(long txid) {
- }
-
- @Override
- public boolean isReady(long txid) {
- return _coordinator.isReady(txid);
- }
- }
-
- static class EmitterPartitionState {
- public RotatingTransactionalState rotatingState;
- public ISpoutPartition partition;
-
- public EmitterPartitionState(RotatingTransactionalState s, ISpoutPartition p) {
- rotatingState = s;
- partition = p;
- }
- }
-
- class Emitter implements ITridentSpout.Emitter<Object> {
- private IPartitionedTridentSpout.Emitter _emitter;
- private TransactionalState _state;
- private Map<String, EmitterPartitionState> _partitionStates = new HashMap<String, EmitterPartitionState>();
- private int _index;
- private int _numTasks;
-
- public Emitter(String txStateId, Map conf, TopologyContext context) {
- _emitter = _spout.getEmitter(conf, context);
- _state = TransactionalState.newUserState(conf, txStateId);
- _index = context.getThisTaskIndex();
- _numTasks = context.getComponentTasks(context.getThisComponentId()).size();
- }
-
- Object _savedCoordinatorMeta = null;
-
-
- @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);
- }
- }
- }
-
- @Override
- public void success(TransactionAttempt tx) {
- for(EmitterPartitionState state: _partitionStates.values()) {
- state.rotatingState.cleanupBefore(tx.getTransactionId());
- }
- }
-
- @Override
- public void close() {
- _state.close();
- _emitter.close();
- }
- }
-
- @Override
- public ITridentSpout.BatchCoordinator getCoordinator(String txStateId, Map conf, TopologyContext context) {
- return new Coordinator(conf, context);
- }
-
- @Override
- public ITridentSpout.Emitter getEmitter(String txStateId, Map conf, TopologyContext context) {
- return new Emitter(txStateId, conf, context);
- }
-
- @Override
- public Map<String, Object> getComponentConfiguration() {
- return _spout.getComponentConfiguration();
- }
-
- @Override
- public Fields getOutputFields() {
- return _spout.getOutputFields();
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/storm/blob/e935da91/jstorm-client/src/main/java/storm/trident/spout/RichSpoutBatchExecutor.java
----------------------------------------------------------------------
diff --git a/jstorm-client/src/main/java/storm/trident/spout/RichSpoutBatchExecutor.java b/jstorm-client/src/main/java/storm/trident/spout/RichSpoutBatchExecutor.java
deleted file mode 100644
index f6a5a59..0000000
--- a/jstorm-client/src/main/java/storm/trident/spout/RichSpoutBatchExecutor.java
+++ /dev/null
@@ -1,181 +0,0 @@
-package storm.trident.spout;
-
-import backtype.storm.Config;
-import backtype.storm.spout.ISpoutOutputCollector;
-import backtype.storm.spout.SpoutOutputCollector;
-import backtype.storm.task.TopologyContext;
-import backtype.storm.topology.IRichSpout;
-import backtype.storm.tuple.Fields;
-import backtype.storm.utils.RotatingMap;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import storm.trident.operation.TridentCollector;
-import storm.trident.topology.TransactionAttempt;
-import storm.trident.util.TridentUtils;
-
-public class RichSpoutBatchExecutor implements ITridentSpout {
- public static final String MAX_BATCH_SIZE_CONF = "topology.spout.max.batch.size";
-
- IRichSpout _spout;
-
- public RichSpoutBatchExecutor(IRichSpout spout) {
- _spout = spout;
- }
-
- @Override
- public Map getComponentConfiguration() {
- return _spout.getComponentConfiguration();
- }
-
- @Override
- public Fields getOutputFields() {
- return TridentUtils.getSingleOutputStreamFields(_spout);
-
- }
-
- @Override
- public BatchCoordinator getCoordinator(String txStateId, Map conf, TopologyContext context) {
- return new RichSpoutCoordinator();
- }
-
- @Override
- public Emitter getEmitter(String txStateId, Map conf, TopologyContext context) {
- return new RichSpoutEmitter(conf, context);
- }
-
- class RichSpoutEmitter implements ITridentSpout.Emitter<Object> {
- int _maxBatchSize;
- boolean prepared = false;
- CaptureCollector _collector;
- RotatingMap<Long, List<Object>> idsMap;
- Map _conf;
- TopologyContext _context;
- long lastRotate = System.currentTimeMillis();
- long rotateTime;
-
- public RichSpoutEmitter(Map conf, TopologyContext context) {
- _conf = conf;
- _context = context;
- Number batchSize = (Number) conf.get(MAX_BATCH_SIZE_CONF);
- if(batchSize==null) batchSize = 1000;
- _maxBatchSize = batchSize.intValue();
- _collector = new CaptureCollector();
- idsMap = new RotatingMap(3);
- rotateTime = 1000L * ((Number)conf.get(Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS)).intValue();
- }
-
- @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);
-
- }
-
- @Override
- public void success(TransactionAttempt tx) {
- ack(tx.getTransactionId());
- }
-
- private void ack(long batchId) {
- List<Object> ids = (List<Object>) idsMap.remove(batchId);
- if(ids!=null) {
- for(Object id: ids) {
- _spout.ack(id);
- }
- }
- }
-
- private void fail(long batchId) {
- List<Object> ids = (List<Object>) idsMap.remove(batchId);
- if(ids!=null) {
- for(Object id: ids) {
- _spout.fail(id);
- }
- }
- }
-
- @Override
- public void close() {
- }
-
- }
-
- class RichSpoutCoordinator implements ITridentSpout.BatchCoordinator {
- @Override
- public Object initializeTransaction(long txid, Object prevMetadata, Object currMetadata) {
- return null;
- }
-
- @Override
- public void success(long txid) {
- }
-
- @Override
- public boolean isReady(long txid) {
- return true;
- }
-
- @Override
- public void close() {
- }
- }
-
- static class CaptureCollector implements ISpoutOutputCollector {
-
- TridentCollector _collector;
- public List<Object> ids;
- public int numEmitted;
-
- public void reset(TridentCollector c) {
- _collector = c;
- ids = new ArrayList<Object>();
- }
-
- @Override
- public void reportError(Throwable t) {
- _collector.reportError(t);
- }
-
- @Override
- public List<Integer> emit(String stream, List<Object> values, Object id) {
- if(id!=null) ids.add(id);
- numEmitted++;
- _collector.emit(values);
- return null;
- }
-
- @Override
- public void emitDirect(int task, String stream, List<Object> values, Object id) {
- throw new UnsupportedOperationException("Trident does not support direct streams");
- }
-
- }
-
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e935da91/jstorm-client/src/main/java/storm/trident/spout/RichSpoutBatchId.java
----------------------------------------------------------------------
diff --git a/jstorm-client/src/main/java/storm/trident/spout/RichSpoutBatchId.java b/jstorm-client/src/main/java/storm/trident/spout/RichSpoutBatchId.java
deleted file mode 100644
index 1340d21..0000000
--- a/jstorm-client/src/main/java/storm/trident/spout/RichSpoutBatchId.java
+++ /dev/null
@@ -1,32 +0,0 @@
-package storm.trident.spout;
-
-public class RichSpoutBatchId implements IBatchID {
- long _id;
-
- public RichSpoutBatchId(long id) {
- _id = id;
- }
-
- @Override
- public Object getId() {
- // this is to distinguish from TransactionAttempt
- return this;
- }
-
- @Override
- public int getAttemptId() {
- return 0; // each drpc request is always a single attempt
- }
-
- @Override
- public int hashCode() {
- return ((Long) _id).hashCode();
- }
-
- @Override
- public boolean equals(Object o) {
- if(!(o instanceof RichSpoutBatchId)) return false;
- RichSpoutBatchId other = (RichSpoutBatchId) o;
- return _id == other._id;
- }
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e935da91/jstorm-client/src/main/java/storm/trident/spout/RichSpoutBatchIdSerializer.java
----------------------------------------------------------------------
diff --git a/jstorm-client/src/main/java/storm/trident/spout/RichSpoutBatchIdSerializer.java b/jstorm-client/src/main/java/storm/trident/spout/RichSpoutBatchIdSerializer.java
deleted file mode 100644
index d544fa7..0000000
--- a/jstorm-client/src/main/java/storm/trident/spout/RichSpoutBatchIdSerializer.java
+++ /dev/null
@@ -1,21 +0,0 @@
-package storm.trident.spout;
-
-import com.esotericsoftware.kryo.Kryo;
-import com.esotericsoftware.kryo.Serializer;
-import com.esotericsoftware.kryo.io.Input;
-import com.esotericsoftware.kryo.io.Output;
-
-
-public class RichSpoutBatchIdSerializer extends Serializer<RichSpoutBatchId> {
-
- @Override
- public void write(Kryo kryo, Output output, RichSpoutBatchId id) {
- output.writeLong(id._id);
- }
-
- @Override
- public RichSpoutBatchId read(Kryo kryo, Input input, Class type) {
- long l = input.readLong();
- return new RichSpoutBatchId(l);
- }
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e935da91/jstorm-client/src/main/java/storm/trident/spout/RichSpoutBatchTriggerer.java
----------------------------------------------------------------------
diff --git a/jstorm-client/src/main/java/storm/trident/spout/RichSpoutBatchTriggerer.java b/jstorm-client/src/main/java/storm/trident/spout/RichSpoutBatchTriggerer.java
deleted file mode 100644
index 39878a1..0000000
--- a/jstorm-client/src/main/java/storm/trident/spout/RichSpoutBatchTriggerer.java
+++ /dev/null
@@ -1,161 +0,0 @@
-package storm.trident.spout;
-
-import backtype.storm.Config;
-import backtype.storm.generated.Grouping;
-import backtype.storm.spout.ISpoutOutputCollector;
-import backtype.storm.spout.SpoutOutputCollector;
-import backtype.storm.task.TopologyContext;
-import backtype.storm.topology.IRichSpout;
-import backtype.storm.topology.OutputFieldsDeclarer;
-import backtype.storm.tuple.Fields;
-import backtype.storm.tuple.Values;
-import backtype.storm.utils.Utils;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Random;
-import java.util.Set;
-import storm.trident.topology.TridentBoltExecutor;
-import storm.trident.tuple.ConsList;
-import storm.trident.util.TridentUtils;
-
-
-public class RichSpoutBatchTriggerer implements IRichSpout {
-
- String _stream;
- IRichSpout _delegate;
- List<Integer> _outputTasks;
- Random _rand;
- String _coordStream;
-
- public RichSpoutBatchTriggerer(IRichSpout delegate, String streamName, String batchGroup) {
- _delegate = delegate;
- _stream = streamName;
- _coordStream = TridentBoltExecutor.COORD_STREAM(batchGroup);
- }
-
- @Override
- public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
- _delegate.open(conf, context, new SpoutOutputCollector(new StreamOverrideCollector(collector)));
- _outputTasks = new ArrayList<Integer>();
- for(String component: Utils.get(context.getThisTargets(),
- _coordStream,
- new HashMap<String, Grouping>()).keySet()) {
- _outputTasks.addAll(context.getComponentTasks(component));
- }
- _rand = new Random(Utils.secureRandomLong());
- }
-
- @Override
- public void close() {
- _delegate.close();
- }
-
- @Override
- public void activate() {
- _delegate.activate();
- }
-
- @Override
- public void deactivate() {
- _delegate.deactivate();
- }
-
- @Override
- public void nextTuple() {
- _delegate.nextTuple();
- }
-
- @Override
- public void ack(Object msgId) {
- Long batchId = _msgIdToBatchId.remove((Long) msgId);
- FinishCondition cond = _finishConditions.get(batchId);
- if(cond!=null) {
- cond.vals.remove((Long) msgId);
- if(cond.vals.isEmpty()) {
- _finishConditions.remove(batchId);
- _delegate.ack(cond.msgId);
- }
- }
- }
-
- @Override
- public void fail(Object msgId) {
- Long batchId = _msgIdToBatchId.remove((Long) msgId);
- FinishCondition cond = _finishConditions.remove(batchId);
- if(cond!=null) {
- _delegate.fail(cond.msgId);
- }
- }
-
- @Override
- public void declareOutputFields(OutputFieldsDeclarer declarer) {
- Fields outFields = TridentUtils.getSingleOutputStreamFields(_delegate);
- outFields = TridentUtils.fieldsConcat(new Fields("$id$"), outFields);
- declarer.declareStream(_stream, outFields);
- // try to find a way to merge this code with what's already done in TridentBoltExecutor
- declarer.declareStream(_coordStream, true, new Fields("id", "count"));
- }
-
- @Override
- public Map<String, Object> getComponentConfiguration() {
- Map<String, Object> conf = _delegate.getComponentConfiguration();
- if(conf==null) conf = new HashMap();
- else conf = new HashMap(conf);
- Config.registerSerialization(conf, RichSpoutBatchId.class, RichSpoutBatchIdSerializer.class);
- return conf;
- }
-
- static class FinishCondition {
- Set<Long> vals = new HashSet<Long>();
- Object msgId;
- }
-
- Map<Long, Long> _msgIdToBatchId = new HashMap();
-
- Map<Long, FinishCondition> _finishConditions = new HashMap();
-
- class StreamOverrideCollector implements ISpoutOutputCollector {
-
- SpoutOutputCollector _collector;
-
- public StreamOverrideCollector(SpoutOutputCollector collector) {
- _collector = collector;
- }
-
- @Override
- public List<Integer> emit(String ignore, List<Object> values, Object msgId) {
- long batchIdVal = _rand.nextLong();
- Object batchId = new RichSpoutBatchId(batchIdVal);
- FinishCondition finish = new FinishCondition();
- finish.msgId = msgId;
- List<Integer> tasks = _collector.emit(_stream, new ConsList(batchId, values));
- Set<Integer> outTasksSet = new HashSet<Integer>(tasks);
- for(Integer t: _outputTasks) {
- int count = 0;
- if(outTasksSet.contains(t)) {
- count = 1;
- }
- long r = _rand.nextLong();
- _collector.emitDirect(t, _coordStream, new Values(batchId, count), r);
- finish.vals.add(r);
- _msgIdToBatchId.put(r, batchIdVal);
- }
- _finishConditions.put(batchIdVal, finish);
- return tasks;
- }
-
- @Override
- public void emitDirect(int task, String ignore, List<Object> values, Object msgId) {
- throw new RuntimeException("Trident does not support direct emits from spouts");
- }
-
- @Override
- public void reportError(Throwable t) {
- _collector.reportError(t);
- }
-
- }
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e935da91/jstorm-client/src/main/java/storm/trident/spout/TridentSpoutCoordinator.java
----------------------------------------------------------------------
diff --git a/jstorm-client/src/main/java/storm/trident/spout/TridentSpoutCoordinator.java b/jstorm-client/src/main/java/storm/trident/spout/TridentSpoutCoordinator.java
deleted file mode 100644
index 7095f36..0000000
--- a/jstorm-client/src/main/java/storm/trident/spout/TridentSpoutCoordinator.java
+++ /dev/null
@@ -1,78 +0,0 @@
-package storm.trident.spout;
-
-import backtype.storm.Config;
-import backtype.storm.task.TopologyContext;
-import backtype.storm.topology.BasicOutputCollector;
-import backtype.storm.topology.IBasicBolt;
-import backtype.storm.topology.OutputFieldsDeclarer;
-import storm.trident.topology.TransactionAttempt;
-import backtype.storm.tuple.Fields;
-import backtype.storm.tuple.Tuple;
-import backtype.storm.tuple.Values;
-import java.util.Map;
-import org.apache.log4j.Logger;
-import storm.trident.topology.MasterBatchCoordinator;
-import storm.trident.topology.state.RotatingTransactionalState;
-import storm.trident.topology.state.TransactionalState;
-
-
-public class TridentSpoutCoordinator implements IBasicBolt {
- public static final Logger LOG = Logger.getLogger(TridentSpoutCoordinator.class);
- private static final String META_DIR = "meta";
-
- ITridentSpout _spout;
- ITridentSpout.BatchCoordinator _coord;
- RotatingTransactionalState _state;
- TransactionalState _underlyingState;
- String _id;
-
-
- public TridentSpoutCoordinator(String id, ITridentSpout spout) {
- _spout = spout;
- _id = id;
- }
-
- @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);
- }
-
- @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));
- }
-
- }
-
- @Override
- public void cleanup() {
- _coord.close();
- _underlyingState.close();
- }
-
- @Override
- public void declareOutputFields(OutputFieldsDeclarer declarer) {
- declarer.declareStream(MasterBatchCoordinator.BATCH_STREAM_ID, new Fields("tx", "metadata"));
- }
-
- @Override
- public Map<String, Object> getComponentConfiguration() {
- Config ret = new Config();
- ret.setMaxTaskParallelism(1);
- return ret;
- }
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e935da91/jstorm-client/src/main/java/storm/trident/spout/TridentSpoutExecutor.java
----------------------------------------------------------------------
diff --git a/jstorm-client/src/main/java/storm/trident/spout/TridentSpoutExecutor.java b/jstorm-client/src/main/java/storm/trident/spout/TridentSpoutExecutor.java
deleted file mode 100644
index 10b40ae..0000000
--- a/jstorm-client/src/main/java/storm/trident/spout/TridentSpoutExecutor.java
+++ /dev/null
@@ -1,120 +0,0 @@
-package storm.trident.spout;
-
-import backtype.storm.coordination.BatchOutputCollector;
-import backtype.storm.task.TopologyContext;
-import backtype.storm.topology.FailedException;
-import backtype.storm.topology.OutputFieldsDeclarer;
-import storm.trident.topology.TransactionAttempt;
-import backtype.storm.tuple.Fields;
-import backtype.storm.tuple.Tuple;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import java.util.TreeMap;
-import org.apache.log4j.Logger;
-import storm.trident.operation.TridentCollector;
-import storm.trident.topology.BatchInfo;
-import storm.trident.topology.ITridentBatchBolt;
-import storm.trident.topology.MasterBatchCoordinator;
-import storm.trident.tuple.ConsList;
-
-public class TridentSpoutExecutor implements ITridentBatchBolt {
- public static String ID_FIELD = "$tx";
-
- public static Logger LOG = Logger.getLogger(TridentSpoutExecutor.class);
-
- AddIdCollector _collector;
- ITridentSpout _spout;
- ITridentSpout.Emitter _emitter;
- String _streamName;
- String _txStateId;
-
- TreeMap<Long, TransactionAttempt> _activeBatches = new TreeMap<Long, TransactionAttempt>();
-
- public TridentSpoutExecutor(String txStateId, String streamName, ITridentSpout spout) {
- _txStateId = txStateId;
- _spout = spout;
- _streamName = streamName;
- }
-
- @Override
- public void prepare(Map conf, TopologyContext context, BatchOutputCollector collector) {
- _emitter = _spout.getEmitter(_txStateId, conf, context);
- _collector = new AddIdCollector(_streamName, collector);
- }
-
- @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);
- }
- }
-
- @Override
- public void cleanup() {
- _emitter.close();
- }
-
- @Override
- public void declareOutputFields(OutputFieldsDeclarer declarer) {
- List<String> fields = new ArrayList(_spout.getOutputFields().toList());
- fields.add(0, ID_FIELD);
- declarer.declareStream(_streamName, new Fields(fields));
- }
-
- @Override
- public Map<String, Object> getComponentConfiguration() {
- return _spout.getComponentConfiguration();
- }
-
- @Override
- public void finishBatch(BatchInfo batchInfo) {
- }
-
- @Override
- public Object initBatchState(String batchGroup, Object batchId) {
- return null;
- }
-
- private static class AddIdCollector implements TridentCollector {
- BatchOutputCollector _delegate;
- Object _id;
- String _stream;
-
- public AddIdCollector(String stream, BatchOutputCollector c) {
- _delegate = c;
- _stream = stream;
- }
-
-
- public void setBatch(Object id) {
- _id = id;
- }
-
- @Override
- public void emit(List<Object> values) {
- _delegate.emit(_stream, new ConsList(_id, values));
- }
-
- @Override
- public void reportError(Throwable t) {
- _delegate.reportError(t);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e935da91/jstorm-client/src/main/java/storm/trident/state/BaseQueryFunction.java
----------------------------------------------------------------------
diff --git a/jstorm-client/src/main/java/storm/trident/state/BaseQueryFunction.java b/jstorm-client/src/main/java/storm/trident/state/BaseQueryFunction.java
deleted file mode 100644
index 49be20e..0000000
--- a/jstorm-client/src/main/java/storm/trident/state/BaseQueryFunction.java
+++ /dev/null
@@ -1,8 +0,0 @@
-package storm.trident.state;
-
-import storm.trident.operation.BaseOperation;
-
-
-public abstract class BaseQueryFunction<S extends State, T> extends BaseOperation implements QueryFunction<S, T> {
-
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e935da91/jstorm-client/src/main/java/storm/trident/state/BaseStateUpdater.java
----------------------------------------------------------------------
diff --git a/jstorm-client/src/main/java/storm/trident/state/BaseStateUpdater.java b/jstorm-client/src/main/java/storm/trident/state/BaseStateUpdater.java
deleted file mode 100644
index 292c59f..0000000
--- a/jstorm-client/src/main/java/storm/trident/state/BaseStateUpdater.java
+++ /dev/null
@@ -1,8 +0,0 @@
-package storm.trident.state;
-
-import storm.trident.operation.BaseOperation;
-
-
-public abstract class BaseStateUpdater<S extends State> extends BaseOperation implements StateUpdater<S> {
-
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e935da91/jstorm-client/src/main/java/storm/trident/state/CombinerValueUpdater.java
----------------------------------------------------------------------
diff --git a/jstorm-client/src/main/java/storm/trident/state/CombinerValueUpdater.java b/jstorm-client/src/main/java/storm/trident/state/CombinerValueUpdater.java
deleted file mode 100644
index ea2248f..0000000
--- a/jstorm-client/src/main/java/storm/trident/state/CombinerValueUpdater.java
+++ /dev/null
@@ -1,19 +0,0 @@
-package storm.trident.state;
-
-import storm.trident.operation.CombinerAggregator;
-
-public class CombinerValueUpdater implements ValueUpdater<Object> {
- Object arg;
- CombinerAggregator agg;
-
- public CombinerValueUpdater(CombinerAggregator agg, Object arg) {
- this.agg = agg;
- this.arg = arg;
- }
-
- @Override
- public Object update(Object stored) {
- if(stored==null) return arg;
- else return agg.combine(stored, arg);
- }
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e935da91/jstorm-client/src/main/java/storm/trident/state/ITupleCollection.java
----------------------------------------------------------------------
diff --git a/jstorm-client/src/main/java/storm/trident/state/ITupleCollection.java b/jstorm-client/src/main/java/storm/trident/state/ITupleCollection.java
deleted file mode 100644
index 23c1253..0000000
--- a/jstorm-client/src/main/java/storm/trident/state/ITupleCollection.java
+++ /dev/null
@@ -1,9 +0,0 @@
-package storm.trident.state;
-
-import java.util.Iterator;
-import java.util.List;
-
-/* Container of a collection of tuples */
-public interface ITupleCollection {
- public Iterator<List<Object>> getTuples();
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e935da91/jstorm-client/src/main/java/storm/trident/state/JSONNonTransactionalSerializer.java
----------------------------------------------------------------------
diff --git a/jstorm-client/src/main/java/storm/trident/state/JSONNonTransactionalSerializer.java b/jstorm-client/src/main/java/storm/trident/state/JSONNonTransactionalSerializer.java
deleted file mode 100644
index a289ddb..0000000
--- a/jstorm-client/src/main/java/storm/trident/state/JSONNonTransactionalSerializer.java
+++ /dev/null
@@ -1,28 +0,0 @@
-package storm.trident.state;
-
-import java.io.UnsupportedEncodingException;
-
-import backtype.storm.utils.Utils;
-
-
-public class JSONNonTransactionalSerializer implements Serializer {
-
- @Override
- public byte[] serialize(Object obj) {
- try {
- return Utils.to_json(obj).getBytes("UTF-8");
- } catch (UnsupportedEncodingException e) {
- throw new RuntimeException(e);
- }
- }
-
- @Override
- public Object deserialize(byte[] b) {
- try {
- return Utils.from_json(new String(b, "UTF-8"));
- } catch (UnsupportedEncodingException e) {
- throw new RuntimeException(e);
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e935da91/jstorm-client/src/main/java/storm/trident/state/JSONOpaqueSerializer.java
----------------------------------------------------------------------
diff --git a/jstorm-client/src/main/java/storm/trident/state/JSONOpaqueSerializer.java b/jstorm-client/src/main/java/storm/trident/state/JSONOpaqueSerializer.java
deleted file mode 100644
index a96d457..0000000
--- a/jstorm-client/src/main/java/storm/trident/state/JSONOpaqueSerializer.java
+++ /dev/null
@@ -1,36 +0,0 @@
-package storm.trident.state;
-
-import java.io.UnsupportedEncodingException;
-import java.util.ArrayList;
-import java.util.List;
-
-import backtype.storm.utils.Utils;
-
-
-public class JSONOpaqueSerializer implements Serializer<OpaqueValue> {
-
- @Override
- public byte[] serialize(OpaqueValue obj) {
- List toSer = new ArrayList(3);
- toSer.add(obj.currTxid);
- toSer.add(obj.curr);
- toSer.add(obj.prev);
- try {
- return Utils.to_json(toSer).getBytes("UTF-8");
- } catch (UnsupportedEncodingException e) {
- throw new RuntimeException(e);
- }
- }
-
- @Override
- public OpaqueValue deserialize(byte[] b) {
- try {
- String s = new String(b, "UTF-8");
- List deser = (List) Utils.from_json(s);
- return new OpaqueValue((Long) deser.get(0), deser.get(1), deser.get(2));
- } catch (UnsupportedEncodingException e) {
- throw new RuntimeException(e);
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e935da91/jstorm-client/src/main/java/storm/trident/state/JSONTransactionalSerializer.java
----------------------------------------------------------------------
diff --git a/jstorm-client/src/main/java/storm/trident/state/JSONTransactionalSerializer.java b/jstorm-client/src/main/java/storm/trident/state/JSONTransactionalSerializer.java
deleted file mode 100644
index a0df3af..0000000
--- a/jstorm-client/src/main/java/storm/trident/state/JSONTransactionalSerializer.java
+++ /dev/null
@@ -1,34 +0,0 @@
-package storm.trident.state;
-
-import java.io.UnsupportedEncodingException;
-import java.util.ArrayList;
-import java.util.List;
-
-import backtype.storm.utils.Utils;
-
-
-public class JSONTransactionalSerializer implements Serializer<TransactionalValue> {
- @Override
- public byte[] serialize(TransactionalValue obj) {
- List toSer = new ArrayList(2);
- toSer.add(obj.getTxid());
- toSer.add(obj.getVal());
- try {
- return Utils.to_json(toSer).getBytes("UTF-8");
- } catch (UnsupportedEncodingException e) {
- throw new RuntimeException(e);
- }
- }
-
- @Override
- public TransactionalValue deserialize(byte[] b) {
- try {
- String s = new String(b, "UTF-8");
- List deser = (List) Utils.from_json(s);
- return new TransactionalValue((Long) deser.get(0), deser.get(1));
- } catch (UnsupportedEncodingException e) {
- throw new RuntimeException(e);
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e935da91/jstorm-client/src/main/java/storm/trident/state/OpaqueValue.java
----------------------------------------------------------------------
diff --git a/jstorm-client/src/main/java/storm/trident/state/OpaqueValue.java b/jstorm-client/src/main/java/storm/trident/state/OpaqueValue.java
deleted file mode 100644
index 14a39d4..0000000
--- a/jstorm-client/src/main/java/storm/trident/state/OpaqueValue.java
+++ /dev/null
@@ -1,58 +0,0 @@
-package storm.trident.state;
-
-import org.apache.commons.lang.builder.ToStringBuilder;
-
-public class OpaqueValue<T> {
- Long currTxid;
- T prev;
- T curr;
-
- public OpaqueValue(Long currTxid, T val, T prev) {
- this.curr = val;
- this.currTxid = currTxid;
- this.prev = prev;
- }
-
- public OpaqueValue(Long currTxid, T val) {
- this(currTxid, val, null);
- }
-
- public OpaqueValue<T> update(Long batchTxid, T newVal) {
- T prev;
- if(batchTxid==null || (this.currTxid < batchTxid)) {
- prev = this.curr;
- } else if(batchTxid.equals(this.currTxid)){
- prev = this.prev;
- } else {
- throw new RuntimeException("Current batch (" + batchTxid + ") is behind state's batch: " + this.toString());
- }
- return new OpaqueValue<T>(batchTxid, newVal, prev);
- }
-
- public T get(Long batchTxid) {
- if(batchTxid==null || (this.currTxid < batchTxid)) {
- return curr;
- } else if(batchTxid.equals(this.currTxid)){
- return prev;
- } else {
- throw new RuntimeException("Current batch (" + batchTxid + ") is behind state's batch: " + this.toString());
- }
- }
-
- public T getCurr() {
- return curr;
- }
-
- public Long getCurrTxid() {
- return currTxid;
- }
-
- public T getPrev() {
- return prev;
- }
-
- @Override
- public String toString() {
- return ToStringBuilder.reflectionToString(this);
- }
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e935da91/jstorm-client/src/main/java/storm/trident/state/QueryFunction.java
----------------------------------------------------------------------
diff --git a/jstorm-client/src/main/java/storm/trident/state/QueryFunction.java b/jstorm-client/src/main/java/storm/trident/state/QueryFunction.java
deleted file mode 100644
index 38eb41f..0000000
--- a/jstorm-client/src/main/java/storm/trident/state/QueryFunction.java
+++ /dev/null
@@ -1,11 +0,0 @@
-package storm.trident.state;
-
-import java.util.List;
-import storm.trident.operation.EachOperation;
-import storm.trident.operation.TridentCollector;
-import storm.trident.tuple.TridentTuple;
-
-public interface QueryFunction<S extends State, T> extends EachOperation {
- List<T> batchRetrieve(S state, List<TridentTuple> args);
- void execute(TridentTuple tuple, T result, TridentCollector collector);
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e935da91/jstorm-client/src/main/java/storm/trident/state/ReadOnlyState.java
----------------------------------------------------------------------
diff --git a/jstorm-client/src/main/java/storm/trident/state/ReadOnlyState.java b/jstorm-client/src/main/java/storm/trident/state/ReadOnlyState.java
deleted file mode 100644
index f8c62f0..0000000
--- a/jstorm-client/src/main/java/storm/trident/state/ReadOnlyState.java
+++ /dev/null
@@ -1,14 +0,0 @@
-package storm.trident.state;
-
-public class ReadOnlyState implements State {
-
- @Override
- public void beginCommit(Long txid) {
- throw new UnsupportedOperationException("This state is read-only and does not support updates");
- }
-
- @Override
- public void commit(Long txid) {
- throw new UnsupportedOperationException("This state is read-only and does not support updates");
- }
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e935da91/jstorm-client/src/main/java/storm/trident/state/ReducerValueUpdater.java
----------------------------------------------------------------------
diff --git a/jstorm-client/src/main/java/storm/trident/state/ReducerValueUpdater.java b/jstorm-client/src/main/java/storm/trident/state/ReducerValueUpdater.java
deleted file mode 100644
index 2ba3aec..0000000
--- a/jstorm-client/src/main/java/storm/trident/state/ReducerValueUpdater.java
+++ /dev/null
@@ -1,24 +0,0 @@
-package storm.trident.state;
-
-import java.util.List;
-import storm.trident.operation.ReducerAggregator;
-import storm.trident.tuple.TridentTuple;
-
-public class ReducerValueUpdater implements ValueUpdater<Object> {
- List<TridentTuple> tuples;
- ReducerAggregator agg;
-
- public ReducerValueUpdater(ReducerAggregator agg, List<TridentTuple> tuples) {
- this.agg = agg;
- this.tuples = tuples;
- }
-
- @Override
- public Object update(Object stored) {
- Object ret = (stored == null) ? this.agg.init() : stored;
- for(TridentTuple t: tuples) {
- ret = this.agg.reduce(ret, t);
- }
- return ret;
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/storm/blob/e935da91/jstorm-client/src/main/java/storm/trident/state/Serializer.java
----------------------------------------------------------------------
diff --git a/jstorm-client/src/main/java/storm/trident/state/Serializer.java b/jstorm-client/src/main/java/storm/trident/state/Serializer.java
deleted file mode 100644
index 9f91a38..0000000
--- a/jstorm-client/src/main/java/storm/trident/state/Serializer.java
+++ /dev/null
@@ -1,9 +0,0 @@
-package storm.trident.state;
-
-import java.io.Serializable;
-
-
-public interface Serializer<T> extends Serializable {
- byte[] serialize(T obj);
- T deserialize(byte[] b);
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e935da91/jstorm-client/src/main/java/storm/trident/state/State.java
----------------------------------------------------------------------
diff --git a/jstorm-client/src/main/java/storm/trident/state/State.java b/jstorm-client/src/main/java/storm/trident/state/State.java
deleted file mode 100644
index 93f7255..0000000
--- a/jstorm-client/src/main/java/storm/trident/state/State.java
+++ /dev/null
@@ -1,22 +0,0 @@
-package storm.trident.state;
-
-/**
- * There's 3 different kinds of state:
- *
- * 1. non-transactional: ignores commits, updates are permanent. no rollback. a cassandra incrementing state would be like this
- * 2. repeat-transactional: idempotent as long as all batches for a txid are identical
- * 3. opaque-transactional: the most general kind of state. updates are always done
- * based on the previous version of the value if the current commit = latest stored commit
- * Idempotent even if the batch for a txid can change.
- *
- * repeat transactional is idempotent for transactional spouts
- * opaque transactional is idempotent for opaque or transactional spouts
- *
- * Trident should log warnings when state is idempotent but updates will not be idempotent
- * because of spout
- */
-// retrieving is encapsulated in Retrieval interface
-public interface State {
- void beginCommit(Long txid); // can be null for things like partitionPersist occuring off a DRPC stream
- void commit(Long txid);
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e935da91/jstorm-client/src/main/java/storm/trident/state/StateFactory.java
----------------------------------------------------------------------
diff --git a/jstorm-client/src/main/java/storm/trident/state/StateFactory.java b/jstorm-client/src/main/java/storm/trident/state/StateFactory.java
deleted file mode 100644
index a77321b..0000000
--- a/jstorm-client/src/main/java/storm/trident/state/StateFactory.java
+++ /dev/null
@@ -1,9 +0,0 @@
-package storm.trident.state;
-
-import backtype.storm.task.IMetricsContext;
-import java.io.Serializable;
-import java.util.Map;
-
-public interface StateFactory extends Serializable {
- State makeState(Map conf, IMetricsContext metrics, int partitionIndex, int numPartitions);
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e935da91/jstorm-client/src/main/java/storm/trident/state/StateSpec.java
----------------------------------------------------------------------
diff --git a/jstorm-client/src/main/java/storm/trident/state/StateSpec.java b/jstorm-client/src/main/java/storm/trident/state/StateSpec.java
deleted file mode 100644
index 569311e..0000000
--- a/jstorm-client/src/main/java/storm/trident/state/StateSpec.java
+++ /dev/null
@@ -1,13 +0,0 @@
-package storm.trident.state;
-
-import java.io.Serializable;
-
-
-public class StateSpec implements Serializable {
- public StateFactory stateFactory;
- public Integer requiredNumPartitions = null;
-
- public StateSpec(StateFactory stateFactory) {
- this.stateFactory = stateFactory;
- }
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e935da91/jstorm-client/src/main/java/storm/trident/state/StateType.java
----------------------------------------------------------------------
diff --git a/jstorm-client/src/main/java/storm/trident/state/StateType.java b/jstorm-client/src/main/java/storm/trident/state/StateType.java
deleted file mode 100644
index f77ec9d..0000000
--- a/jstorm-client/src/main/java/storm/trident/state/StateType.java
+++ /dev/null
@@ -1,8 +0,0 @@
-package storm.trident.state;
-
-
-public enum StateType {
- NON_TRANSACTIONAL,
- TRANSACTIONAL,
- OPAQUE
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e935da91/jstorm-client/src/main/java/storm/trident/state/StateUpdater.java
----------------------------------------------------------------------
diff --git a/jstorm-client/src/main/java/storm/trident/state/StateUpdater.java b/jstorm-client/src/main/java/storm/trident/state/StateUpdater.java
deleted file mode 100644
index 7a1f19c..0000000
--- a/jstorm-client/src/main/java/storm/trident/state/StateUpdater.java
+++ /dev/null
@@ -1,16 +0,0 @@
-package storm.trident.state;
-
-import java.util.List;
-import storm.trident.operation.Operation;
-import storm.trident.operation.TridentCollector;
-import storm.trident.tuple.TridentTuple;
-
-
-public interface StateUpdater<S extends State> extends Operation {
- // maybe it needs a start phase (where it can do a retrieval, an update phase, and then a finish phase...?
- // shouldn't really be a one-at-a-time interface, since we have all the tuples already?
- // TOOD: used for the new values stream
- // the list is needed to be able to get reduceragg and combineragg persistentaggregate
- // for grouped streams working efficiently
- void updateState(S state, List<TridentTuple> tuples, TridentCollector collector);
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e935da91/jstorm-client/src/main/java/storm/trident/state/TransactionalValue.java
----------------------------------------------------------------------
diff --git a/jstorm-client/src/main/java/storm/trident/state/TransactionalValue.java b/jstorm-client/src/main/java/storm/trident/state/TransactionalValue.java
deleted file mode 100644
index 933a0ea..0000000
--- a/jstorm-client/src/main/java/storm/trident/state/TransactionalValue.java
+++ /dev/null
@@ -1,27 +0,0 @@
-package storm.trident.state;
-
-import org.apache.commons.lang.builder.ToStringBuilder;
-
-
-public class TransactionalValue<T> {
- T val;
- Long txid;
-
- public TransactionalValue(Long txid, T val) {
- this.val = val;
- this.txid = txid;
- }
-
- public T getVal() {
- return val;
- }
-
- public Long getTxid() {
- return txid;
- }
-
- @Override
- public String toString() {
- return ToStringBuilder.reflectionToString(this);
- }
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e935da91/jstorm-client/src/main/java/storm/trident/state/ValueUpdater.java
----------------------------------------------------------------------
diff --git a/jstorm-client/src/main/java/storm/trident/state/ValueUpdater.java b/jstorm-client/src/main/java/storm/trident/state/ValueUpdater.java
deleted file mode 100644
index 466a921..0000000
--- a/jstorm-client/src/main/java/storm/trident/state/ValueUpdater.java
+++ /dev/null
@@ -1,6 +0,0 @@
-package storm.trident.state;
-
-
-public interface ValueUpdater<T> {
- T update(T stored);
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e935da91/jstorm-client/src/main/java/storm/trident/state/map/CachedBatchReadsMap.java
----------------------------------------------------------------------
diff --git a/jstorm-client/src/main/java/storm/trident/state/map/CachedBatchReadsMap.java b/jstorm-client/src/main/java/storm/trident/state/map/CachedBatchReadsMap.java
deleted file mode 100644
index 7d8c442..0000000
--- a/jstorm-client/src/main/java/storm/trident/state/map/CachedBatchReadsMap.java
+++ /dev/null
@@ -1,63 +0,0 @@
-package storm.trident.state.map;
-
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import storm.trident.state.ValueUpdater;
-
-
-public class CachedBatchReadsMap<T> {
- public static class RetVal<T> {
- public boolean cached;
- public T val;
-
- public RetVal(T v, boolean c) {
- val = v;
- cached = c;
- }
- }
-
- Map<List<Object>, T> _cached = new HashMap<List<Object>, T>();
-
- public IBackingMap<T> _delegate;
-
- public CachedBatchReadsMap(IBackingMap<T> delegate) {
- _delegate = delegate;
- }
-
- public void reset() {
- _cached.clear();
- }
-
- public List<RetVal<T>> multiGet(List<List<Object>> keys) {
- // TODO: can optimize further by only querying backing map for keys not in the cache
- List<T> vals = _delegate.multiGet(keys);
- List<RetVal<T>> ret = new ArrayList(vals.size());
- for(int i=0; i<keys.size(); i++) {
- List<Object> key = keys.get(i);
- if(_cached.containsKey(key)) {
- ret.add(new RetVal(_cached.get(key), true));
- } else {
- ret.add(new RetVal(vals.get(i), false));
- }
- }
- return ret;
- }
-
- public void multiPut(List<List<Object>> keys, List<T> vals) {
- _delegate.multiPut(keys, vals);
- cache(keys, vals);
- }
-
- private void cache(List<List<Object>> keys, List<T> vals) {
- for(int i=0; i<keys.size(); i++) {
- List<Object> key = keys.get(i);
- T val = vals.get(i);
- _cached.put(key, val);
- }
- }
-
-
-
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e935da91/jstorm-client/src/main/java/storm/trident/state/map/CachedMap.java
----------------------------------------------------------------------
diff --git a/jstorm-client/src/main/java/storm/trident/state/map/CachedMap.java b/jstorm-client/src/main/java/storm/trident/state/map/CachedMap.java
deleted file mode 100644
index f7ad646..0000000
--- a/jstorm-client/src/main/java/storm/trident/state/map/CachedMap.java
+++ /dev/null
@@ -1,62 +0,0 @@
-package storm.trident.state.map;
-
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import storm.trident.util.LRUMap;
-
-/**
- * Useful to layer over a map that communicates with a database. you generally layer opaque map over this over your database store
- * @author nathan
- * @param <T>
- */
-public class CachedMap<T> implements IBackingMap<T> {
- LRUMap<List<Object>, T> _cache;
- IBackingMap<T> _delegate;
-
- public CachedMap(IBackingMap<T> delegate, int cacheSize) {
- _cache = new LRUMap<List<Object>, T>(cacheSize);
- _delegate = delegate;
- }
-
- @Override
- public List<T> multiGet(List<List<Object>> keys) {
- Map<List<Object>, T> results = new HashMap<List<Object>, T>();
- List<List<Object>> toGet = new ArrayList<List<Object>>();
- for(List<Object> key: keys) {
- if(_cache.containsKey(key)) {
- results.put(key, _cache.get(key));
- } else {
- toGet.add(key);
- }
- }
-
- List<T> fetchedVals = _delegate.multiGet(toGet);
- for(int i=0; i<toGet.size(); i++) {
- List<Object> key = toGet.get(i);
- T val = fetchedVals.get(i);
- _cache.put(key, val);
- results.put(key, val);
- }
-
- List<T> ret = new ArrayList<T>(keys.size());
- for(List<Object> key: keys) {
- ret.add(results.get(key));
- }
- return ret;
- }
-
- @Override
- public void multiPut(List<List<Object>> keys, List<T> values) {
- cache(keys, values);
- _delegate.multiPut(keys, values);
- }
-
- private void cache(List<List<Object>> keys, List<T> values) {
- for(int i=0; i<keys.size(); i++) {
- _cache.put(keys.get(i), values.get(i));
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e935da91/jstorm-client/src/main/java/storm/trident/state/map/IBackingMap.java
----------------------------------------------------------------------
diff --git a/jstorm-client/src/main/java/storm/trident/state/map/IBackingMap.java b/jstorm-client/src/main/java/storm/trident/state/map/IBackingMap.java
deleted file mode 100644
index c1fdc27..0000000
--- a/jstorm-client/src/main/java/storm/trident/state/map/IBackingMap.java
+++ /dev/null
@@ -1,9 +0,0 @@
-package storm.trident.state.map;
-
-import java.util.List;
-
-
-public interface IBackingMap<T> {
- List<T> multiGet(List<List<Object>> keys);
- void multiPut(List<List<Object>> keys, List<T> vals);
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e935da91/jstorm-client/src/main/java/storm/trident/state/map/MapCombinerAggStateUpdater.java
----------------------------------------------------------------------
diff --git a/jstorm-client/src/main/java/storm/trident/state/map/MapCombinerAggStateUpdater.java b/jstorm-client/src/main/java/storm/trident/state/map/MapCombinerAggStateUpdater.java
deleted file mode 100644
index 338c9ac..0000000
--- a/jstorm-client/src/main/java/storm/trident/state/map/MapCombinerAggStateUpdater.java
+++ /dev/null
@@ -1,66 +0,0 @@
-package storm.trident.state.map;
-
-import backtype.storm.tuple.Fields;
-import backtype.storm.tuple.Values;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import storm.trident.operation.CombinerAggregator;
-import storm.trident.operation.TridentCollector;
-import storm.trident.operation.TridentOperationContext;
-import storm.trident.state.CombinerValueUpdater;
-import storm.trident.state.StateUpdater;
-import storm.trident.state.ValueUpdater;
-import storm.trident.tuple.ComboList;
-import storm.trident.tuple.TridentTuple;
-import storm.trident.tuple.TridentTupleView.ProjectionFactory;
-
-public class MapCombinerAggStateUpdater implements StateUpdater<MapState> {
- CombinerAggregator _agg;
- Fields _groupFields;
- Fields _inputFields;
- ProjectionFactory _groupFactory;
- ProjectionFactory _inputFactory;
- ComboList.Factory _factory;
-
-
- public MapCombinerAggStateUpdater(CombinerAggregator agg, Fields groupFields, Fields inputFields) {
- _agg = agg;
- _groupFields = groupFields;
- _inputFields = inputFields;
- if(inputFields.size()!=1) {
- throw new IllegalArgumentException("Combiner aggs only take a single field as input. Got this instead: " + inputFields.toString());
- }
- _factory = new ComboList.Factory(groupFields.size(), inputFields.size());
- }
-
-
- @Override
- public void updateState(MapState map, List<TridentTuple> tuples, TridentCollector collector) {
- List<List<Object>> groups = new ArrayList<List<Object>>(tuples.size());
- List<ValueUpdater> updaters = new ArrayList<ValueUpdater>(tuples.size());
-
- for(TridentTuple t: tuples) {
- groups.add(_groupFactory.create(t));
- updaters.add(new CombinerValueUpdater(_agg,_inputFactory.create(t).getValue(0)));
- }
- List<Object> newVals = map.multiUpdate(groups, updaters);
-
- for(int i=0; i<tuples.size(); i++) {
- List<Object> key = groups.get(i);
- Object result = newVals.get(i);
- collector.emit(_factory.create(new List[] {key, new Values(result) }));
- }
- }
-
- @Override
- public void prepare(Map conf, TridentOperationContext context) {
- _groupFactory = context.makeProjectionFactory(_groupFields);
- _inputFactory = context.makeProjectionFactory(_inputFields);
- }
-
- @Override
- public void cleanup() {
- }
-
-}