You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by bo...@apache.org on 2018/07/30 18:45:15 UTC
[2/4] storm git commit: STORM-3156: Remove the transactional topology
API
http://git-wip-us.apache.org/repos/asf/storm/blob/da0bc6a7/storm-client/src/jvm/org/apache/storm/transactional/TransactionAttempt.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/transactional/TransactionAttempt.java b/storm-client/src/jvm/org/apache/storm/transactional/TransactionAttempt.java
index a192c8f..683e414 100644
--- a/storm-client/src/jvm/org/apache/storm/transactional/TransactionAttempt.java
+++ b/storm-client/src/jvm/org/apache/storm/transactional/TransactionAttempt.java
@@ -14,6 +14,10 @@ package org.apache.storm.transactional;
import java.math.BigInteger;
+/**
+ * This is dead code. It is retained to avoid breaking Kryo registration order.
+ */
+@Deprecated
public class TransactionAttempt {
BigInteger _txid;
long _attemptId;
http://git-wip-us.apache.org/repos/asf/storm/blob/da0bc6a7/storm-client/src/jvm/org/apache/storm/transactional/TransactionalSpoutBatchExecutor.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/transactional/TransactionalSpoutBatchExecutor.java b/storm-client/src/jvm/org/apache/storm/transactional/TransactionalSpoutBatchExecutor.java
deleted file mode 100644
index add1d64..0000000
--- a/storm-client/src/jvm/org/apache/storm/transactional/TransactionalSpoutBatchExecutor.java
+++ /dev/null
@@ -1,91 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership. The ASF licenses this file to you under the Apache License, Version
- * 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions
- * and limitations under the License.
- */
-
-package org.apache.storm.transactional;
-
-import java.math.BigInteger;
-import java.util.Map;
-import java.util.TreeMap;
-import org.apache.storm.coordination.BatchOutputCollectorImpl;
-import org.apache.storm.task.OutputCollector;
-import org.apache.storm.task.TopologyContext;
-import org.apache.storm.topology.FailedException;
-import org.apache.storm.topology.IRichBolt;
-import org.apache.storm.topology.OutputFieldsDeclarer;
-import org.apache.storm.tuple.Tuple;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class TransactionalSpoutBatchExecutor implements IRichBolt {
- public static final Logger LOG = LoggerFactory.getLogger(TransactionalSpoutBatchExecutor.class);
-
- BatchOutputCollectorImpl _collector;
- ITransactionalSpout _spout;
- ITransactionalSpout.Emitter _emitter;
-
- TreeMap<BigInteger, TransactionAttempt> _activeTransactions = new TreeMap<>();
-
- public TransactionalSpoutBatchExecutor(ITransactionalSpout spout) {
- _spout = spout;
- }
-
- @Override
- public void prepare(Map<String, Object> conf, TopologyContext context, OutputCollector collector) {
- _collector = new BatchOutputCollectorImpl(collector);
- _emitter = _spout.getEmitter(conf, context);
- }
-
- @Override
- public void execute(Tuple input) {
- TransactionAttempt attempt = (TransactionAttempt) input.getValue(0);
- try {
- if (input.getSourceStreamId().equals(TransactionalSpoutCoordinator.TRANSACTION_COMMIT_STREAM_ID)) {
- if (attempt.equals(_activeTransactions.get(attempt.getTransactionId()))) {
- ((ICommitterTransactionalSpout.Emitter) _emitter).commit(attempt);
- _activeTransactions.remove(attempt.getTransactionId());
- _collector.ack(input);
- } else {
- _collector.fail(input);
- }
- } else {
- _emitter.emitBatch(attempt, input.getValue(1), _collector);
- _activeTransactions.put(attempt.getTransactionId(), attempt);
- _collector.ack(input);
- BigInteger committed = (BigInteger) input.getValue(2);
- if (committed != null) {
- // valid to delete before what's been committed since
- // those batches will never be accessed again
- _activeTransactions.headMap(committed).clear();
- _emitter.cleanupBefore(committed);
- }
- }
- } catch (FailedException e) {
- LOG.warn("Failed to emit batch for transaction", e);
- _collector.fail(input);
- }
- }
-
- @Override
- public void cleanup() {
- _emitter.close();
- }
-
- @Override
- public void declareOutputFields(OutputFieldsDeclarer declarer) {
- _spout.declareOutputFields(declarer);
- }
-
- @Override
- public Map<String, Object> getComponentConfiguration() {
- return _spout.getComponentConfiguration();
- }
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/da0bc6a7/storm-client/src/jvm/org/apache/storm/transactional/TransactionalSpoutCoordinator.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/transactional/TransactionalSpoutCoordinator.java b/storm-client/src/jvm/org/apache/storm/transactional/TransactionalSpoutCoordinator.java
deleted file mode 100644
index 55c570a..0000000
--- a/storm-client/src/jvm/org/apache/storm/transactional/TransactionalSpoutCoordinator.java
+++ /dev/null
@@ -1,209 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership. The ASF licenses this file to you under the Apache License, Version
- * 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions
- * and limitations under the License.
- */
-
-package org.apache.storm.transactional;
-
-import java.math.BigInteger;
-import java.util.Map;
-import java.util.Random;
-import java.util.TreeMap;
-import org.apache.storm.Config;
-import org.apache.storm.spout.SpoutOutputCollector;
-import org.apache.storm.task.TopologyContext;
-import org.apache.storm.topology.FailedException;
-import org.apache.storm.topology.OutputFieldsDeclarer;
-import org.apache.storm.topology.base.BaseRichSpout;
-import org.apache.storm.transactional.state.RotatingTransactionalState;
-import org.apache.storm.transactional.state.TransactionalState;
-import org.apache.storm.tuple.Fields;
-import org.apache.storm.tuple.Values;
-import org.apache.storm.utils.ObjectReader;
-import org.apache.storm.utils.Utils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class TransactionalSpoutCoordinator extends BaseRichSpout {
- public static final Logger LOG = LoggerFactory.getLogger(TransactionalSpoutCoordinator.class);
-
- public static final BigInteger INIT_TXID = BigInteger.ONE;
-
-
- public static final String TRANSACTION_BATCH_STREAM_ID = TransactionalSpoutCoordinator.class.getName() + "/batch";
- public static final String TRANSACTION_COMMIT_STREAM_ID = TransactionalSpoutCoordinator.class.getName() + "/commit";
-
- private static final String CURRENT_TX = "currtx";
- private static final String META_DIR = "meta";
- TreeMap<BigInteger, TransactionStatus> _activeTx = new TreeMap<>();
- BigInteger _currTransaction;
- int _maxTransactionActive;
- StateInitializer _initializer;
- private ITransactionalSpout _spout;
- private ITransactionalSpout.Coordinator _coordinator;
- private TransactionalState _state;
- private RotatingTransactionalState _coordinatorState;
- private SpoutOutputCollector _collector;
- private Random _rand;
-
-
- public TransactionalSpoutCoordinator(ITransactionalSpout spout) {
- _spout = spout;
- }
-
- public ITransactionalSpout getSpout() {
- return _spout;
- }
-
- @Override
- public void open(Map<String, Object> conf, TopologyContext context, SpoutOutputCollector collector) {
- _rand = new Random(Utils.secureRandomLong());
- _state = TransactionalState
- .newCoordinatorState(conf, (String) conf.get(Config.TOPOLOGY_TRANSACTIONAL_ID), _spout.getComponentConfiguration());
- _coordinatorState = new RotatingTransactionalState(_state, META_DIR, true);
- _collector = collector;
- _coordinator = _spout.getCoordinator(conf, context);
- _currTransaction = getStoredCurrTransaction(_state);
- _maxTransactionActive = ObjectReader.getInt(conf.get(Config.TOPOLOGY_MAX_SPOUT_PENDING), 1);
- _initializer = new StateInitializer();
- }
-
- @Override
- public void close() {
- _state.close();
- }
-
- @Override
- public void nextTuple() {
- sync();
- }
-
- @Override
- public void ack(Object msgId) {
- TransactionAttempt tx = (TransactionAttempt) msgId;
- TransactionStatus status = _activeTx.get(tx.getTransactionId());
- if (status != null && tx.equals(status.attempt)) {
- if (status.status == AttemptStatus.PROCESSING) {
- status.status = AttemptStatus.PROCESSED;
- } else if (status.status == AttemptStatus.COMMITTING) {
- _activeTx.remove(tx.getTransactionId());
- _coordinatorState.cleanupBefore(tx.getTransactionId());
- _currTransaction = nextTransactionId(tx.getTransactionId());
- _state.setData(CURRENT_TX, _currTransaction);
- }
- sync();
- }
- }
-
- @Override
- public void fail(Object msgId) {
- TransactionAttempt tx = (TransactionAttempt) msgId;
- TransactionStatus stored = _activeTx.remove(tx.getTransactionId());
- if (stored != null && tx.equals(stored.attempt)) {
- _activeTx.tailMap(tx.getTransactionId()).clear();
- sync();
- }
- }
-
- @Override
- public void declareOutputFields(OutputFieldsDeclarer declarer) {
- // in partitioned example, in case an emitter task receives a later transaction than it's emitted so far,
- // when it sees the earlier txid it should know to emit nothing
- declarer.declareStream(TRANSACTION_BATCH_STREAM_ID, new Fields("tx", "tx-meta", "committed-txid"));
- declarer.declareStream(TRANSACTION_COMMIT_STREAM_ID, new Fields("tx"));
- }
-
- private void sync() {
- // note that sometimes the tuples active may be less than max_spout_pending, e.g.
- // max_spout_pending = 3
- // tx 1, 2, 3 active, tx 2 is acked. there won't be a commit for tx 2 (because tx 1 isn't committed yet),
- // and there won't be a batch for tx 4 because there's max_spout_pending tx active
- TransactionStatus maybeCommit = _activeTx.get(_currTransaction);
- if (maybeCommit != null && maybeCommit.status == AttemptStatus.PROCESSED) {
- maybeCommit.status = AttemptStatus.COMMITTING;
- _collector.emit(TRANSACTION_COMMIT_STREAM_ID, new Values(maybeCommit.attempt), maybeCommit.attempt);
- }
-
- try {
- if (_activeTx.size() < _maxTransactionActive) {
- BigInteger curr = _currTransaction;
- for (int i = 0; i < _maxTransactionActive; i++) {
- if ((_coordinatorState.hasCache(curr) || _coordinator.isReady())
- && !_activeTx.containsKey(curr)) {
- TransactionAttempt attempt = new TransactionAttempt(curr, _rand.nextLong());
- Object state = _coordinatorState.getState(curr, _initializer);
- _activeTx.put(curr, new TransactionStatus(attempt));
- _collector.emit(TRANSACTION_BATCH_STREAM_ID, new Values(attempt, state, previousTransactionId(_currTransaction)),
- attempt);
- }
- curr = nextTransactionId(curr);
- }
- }
- } catch (FailedException e) {
- LOG.warn("Failed to get metadata for a transaction", e);
- }
- }
-
- @Override
- public Map<String, Object> getComponentConfiguration() {
- Config ret = new Config();
- ret.setMaxTaskParallelism(1);
- return ret;
- }
-
- private BigInteger nextTransactionId(BigInteger id) {
- return id.add(BigInteger.ONE);
- }
-
- private BigInteger previousTransactionId(BigInteger id) {
- if (id.equals(INIT_TXID)) {
- return null;
- } else {
- return id.subtract(BigInteger.ONE);
- }
- }
-
- private BigInteger getStoredCurrTransaction(TransactionalState state) {
- BigInteger ret = (BigInteger) state.getData(CURRENT_TX);
- if (ret == null) {
- return INIT_TXID;
- } else {
- return ret;
- }
- }
-
- private static enum AttemptStatus {
- PROCESSING,
- PROCESSED,
- COMMITTING
- }
-
- private static class TransactionStatus {
- TransactionAttempt attempt;
- AttemptStatus status;
-
- public TransactionStatus(TransactionAttempt attempt) {
- this.attempt = attempt;
- this.status = AttemptStatus.PROCESSING;
- }
-
- @Override
- public String toString() {
- return attempt.toString() + " <" + status.toString() + ">";
- }
- }
-
- private class StateInitializer implements RotatingTransactionalState.StateInitializer {
- @Override
- public Object init(BigInteger txid, Object lastState) {
- return _coordinator.initializeTransaction(txid, lastState);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/da0bc6a7/storm-client/src/jvm/org/apache/storm/transactional/TransactionalTopologyBuilder.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/transactional/TransactionalTopologyBuilder.java b/storm-client/src/jvm/org/apache/storm/transactional/TransactionalTopologyBuilder.java
deleted file mode 100644
index 9e850df..0000000
--- a/storm-client/src/jvm/org/apache/storm/transactional/TransactionalTopologyBuilder.java
+++ /dev/null
@@ -1,562 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership. The ASF licenses this file to you under the Apache License, Version
- * 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions
- * and limitations under the License.
- */
-
-package org.apache.storm.transactional;
-
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import org.apache.storm.Config;
-import org.apache.storm.Constants;
-import org.apache.storm.coordination.BatchBoltExecutor;
-import org.apache.storm.coordination.CoordinatedBolt;
-import org.apache.storm.coordination.CoordinatedBolt.IdStreamSpec;
-import org.apache.storm.coordination.CoordinatedBolt.SourceArgs;
-import org.apache.storm.coordination.IBatchBolt;
-import org.apache.storm.generated.GlobalStreamId;
-import org.apache.storm.generated.Grouping;
-import org.apache.storm.generated.SharedMemory;
-import org.apache.storm.generated.StormTopology;
-import org.apache.storm.grouping.CustomStreamGrouping;
-import org.apache.storm.grouping.PartialKeyGrouping;
-import org.apache.storm.topology.BaseConfigurationDeclarer;
-import org.apache.storm.topology.BasicBoltExecutor;
-import org.apache.storm.topology.BoltDeclarer;
-import org.apache.storm.topology.IBasicBolt;
-import org.apache.storm.topology.IRichBolt;
-import org.apache.storm.topology.InputDeclarer;
-import org.apache.storm.topology.SpoutDeclarer;
-import org.apache.storm.topology.TopologyBuilder;
-import org.apache.storm.transactional.partitioned.IOpaquePartitionedTransactionalSpout;
-import org.apache.storm.transactional.partitioned.IPartitionedTransactionalSpout;
-import org.apache.storm.transactional.partitioned.OpaquePartitionedTransactionalSpoutExecutor;
-import org.apache.storm.transactional.partitioned.PartitionedTransactionalSpoutExecutor;
-import org.apache.storm.tuple.Fields;
-
-/**
- * Trident subsumes the functionality provided by transactional topologies, so this class is deprecated.
- */
-@Deprecated
-public class TransactionalTopologyBuilder {
- final String id;
- final String spoutId;
- final ITransactionalSpout spout;
- final Map<String, Component> bolts = new HashMap<>();
- final Integer spoutParallelism;
- final Map<String, Object> spoutConf = new HashMap<>();
- final Set<SharedMemory> spoutSharedMemory = new HashSet<>();
-
- // id is used to store the state of this transactionalspout in zookeeper
- // it would be very dangerous to have 2 topologies active with the same id in the same cluster
- public TransactionalTopologyBuilder(String id, String spoutId, ITransactionalSpout spout, Number spoutParallelism) {
- this.id = id;
- this.spoutId = spoutId;
- this.spout = spout;
- this.spoutParallelism = (spoutParallelism == null) ? null : spoutParallelism.intValue();
- }
-
- public TransactionalTopologyBuilder(String id, String spoutId, ITransactionalSpout spout) {
- this(id, spoutId, spout, null);
- }
-
- public TransactionalTopologyBuilder(String id, String spoutId, IPartitionedTransactionalSpout spout, Number spoutParallelism) {
- this(id, spoutId, new PartitionedTransactionalSpoutExecutor(spout), spoutParallelism);
- }
-
- public TransactionalTopologyBuilder(String id, String spoutId, IPartitionedTransactionalSpout spout) {
- this(id, spoutId, spout, null);
- }
-
- public TransactionalTopologyBuilder(String id, String spoutId, IOpaquePartitionedTransactionalSpout spout, Number spoutParallelism) {
- this(id, spoutId, new OpaquePartitionedTransactionalSpoutExecutor(spout), spoutParallelism);
- }
-
- public TransactionalTopologyBuilder(String id, String spoutId, IOpaquePartitionedTransactionalSpout spout) {
- this(id, spoutId, spout, null);
- }
-
- public SpoutDeclarer getSpoutDeclarer() {
- return new SpoutDeclarerImpl();
- }
-
- public BoltDeclarer setBolt(String id, IBatchBolt bolt) {
- return setBolt(id, bolt, null);
- }
-
- public BoltDeclarer setBolt(String id, IBatchBolt bolt, Number parallelism) {
- return setBolt(id, new BatchBoltExecutor(bolt), parallelism, bolt instanceof ICommitter);
- }
-
- public BoltDeclarer setCommitterBolt(String id, IBatchBolt bolt) {
- return setCommitterBolt(id, bolt, null);
- }
-
- public BoltDeclarer setCommitterBolt(String id, IBatchBolt bolt, Number parallelism) {
- return setBolt(id, new BatchBoltExecutor(bolt), parallelism, true);
- }
-
- public BoltDeclarer setBolt(String id, IBasicBolt bolt) {
- return setBolt(id, bolt, null);
- }
-
- public BoltDeclarer setBolt(String id, IBasicBolt bolt, Number parallelism) {
- return setBolt(id, new BasicBoltExecutor(bolt), parallelism, false);
- }
-
- private BoltDeclarer setBolt(String id, IRichBolt bolt, Number parallelism, boolean committer) {
- Integer p = null;
- if (parallelism != null) {
- p = parallelism.intValue();
- }
- Component component = new Component(bolt, p, committer);
- bolts.put(id, component);
- return new BoltDeclarerImpl(component);
- }
-
- public TopologyBuilder buildTopologyBuilder() {
- String coordinator = spoutId + "/coordinator";
- TopologyBuilder builder = new TopologyBuilder();
- SpoutDeclarer declarer = builder.setSpout(coordinator, new TransactionalSpoutCoordinator(spout));
- for (SharedMemory request : spoutSharedMemory) {
- declarer.addSharedMemory(request);
- }
- if (!spoutConf.isEmpty()) {
- declarer.addConfigurations(spoutConf);
- }
- declarer.addConfiguration(Config.TOPOLOGY_TRANSACTIONAL_ID, id);
-
- BoltDeclarer emitterDeclarer =
- builder.setBolt(spoutId,
- new CoordinatedBolt(new TransactionalSpoutBatchExecutor(spout),
- null,
- null),
- spoutParallelism)
- .allGrouping(coordinator, TransactionalSpoutCoordinator.TRANSACTION_BATCH_STREAM_ID)
- .addConfiguration(Config.TOPOLOGY_TRANSACTIONAL_ID, id);
- if (spout instanceof ICommitterTransactionalSpout) {
- emitterDeclarer.allGrouping(coordinator, TransactionalSpoutCoordinator.TRANSACTION_COMMIT_STREAM_ID);
- }
- for (String id : bolts.keySet()) {
- Component component = bolts.get(id);
- Map<String, SourceArgs> coordinatedArgs = new HashMap<String, SourceArgs>();
- for (String c : componentBoltSubscriptions(component)) {
- coordinatedArgs.put(c, SourceArgs.all());
- }
-
- IdStreamSpec idSpec = null;
- if (component.committer) {
- idSpec = IdStreamSpec.makeDetectSpec(coordinator, TransactionalSpoutCoordinator.TRANSACTION_COMMIT_STREAM_ID);
- }
- BoltDeclarer input = builder.setBolt(id,
- new CoordinatedBolt(component.bolt,
- coordinatedArgs,
- idSpec),
- component.parallelism);
- for (SharedMemory request : component.sharedMemory) {
- input.addSharedMemory(request);
- }
- if (!component.componentConf.isEmpty()) {
- input.addConfigurations(component.componentConf);
- }
- for (String c : componentBoltSubscriptions(component)) {
- input.directGrouping(c, Constants.COORDINATED_STREAM_ID);
- }
- for (InputDeclaration d : component.declarations) {
- d.declare(input);
- }
- if (component.committer) {
- input.allGrouping(coordinator, TransactionalSpoutCoordinator.TRANSACTION_COMMIT_STREAM_ID);
- }
- }
- return builder;
- }
-
- public StormTopology buildTopology() {
- return buildTopologyBuilder().createTopology();
- }
-
- private Set<String> componentBoltSubscriptions(Component component) {
- Set<String> ret = new HashSet<String>();
- for (InputDeclaration d : component.declarations) {
- ret.add(d.getComponent());
- }
- return ret;
- }
-
- private static interface InputDeclaration {
- void declare(InputDeclarer declarer);
-
- String getComponent();
- }
-
- private static class Component {
- public final IRichBolt bolt;
- public final Integer parallelism;
- public final List<InputDeclaration> declarations = new ArrayList<>();
- public final Map<String, Object> componentConf = new HashMap<>();
- public final boolean committer;
- public final Set<SharedMemory> sharedMemory = new HashSet<>();
-
- public Component(IRichBolt bolt, Integer parallelism, boolean committer) {
- this.bolt = bolt;
- this.parallelism = parallelism;
- this.committer = committer;
- }
- }
-
- private static class BoltDeclarerImpl extends BaseConfigurationDeclarer<BoltDeclarer> implements BoltDeclarer {
- Component component;
-
- public BoltDeclarerImpl(Component component) {
- this.component = component;
- }
-
- @Override
- public BoltDeclarer fieldsGrouping(final String component, final Fields fields) {
- addDeclaration(new InputDeclaration() {
- @Override
- public void declare(InputDeclarer declarer) {
- declarer.fieldsGrouping(component, fields);
- }
-
- @Override
- public String getComponent() {
- return component;
- }
- });
- return this;
- }
-
- @Override
- public BoltDeclarer fieldsGrouping(final String component, final String streamId, final Fields fields) {
- addDeclaration(new InputDeclaration() {
- @Override
- public void declare(InputDeclarer declarer) {
- declarer.fieldsGrouping(component, streamId, fields);
- }
-
- @Override
- public String getComponent() {
- return component;
- }
- });
- return this;
- }
-
- @Override
- public BoltDeclarer globalGrouping(final String component) {
- addDeclaration(new InputDeclaration() {
- @Override
- public void declare(InputDeclarer declarer) {
- declarer.globalGrouping(component);
- }
-
- @Override
- public String getComponent() {
- return component;
- }
- });
- return this;
- }
-
- @Override
- public BoltDeclarer globalGrouping(final String component, final String streamId) {
- addDeclaration(new InputDeclaration() {
- @Override
- public void declare(InputDeclarer declarer) {
- declarer.globalGrouping(component, streamId);
- }
-
- @Override
- public String getComponent() {
- return component;
- }
- });
- return this;
- }
-
- @Override
- public BoltDeclarer shuffleGrouping(final String component) {
- addDeclaration(new InputDeclaration() {
- @Override
- public void declare(InputDeclarer declarer) {
- declarer.shuffleGrouping(component);
- }
-
- @Override
- public String getComponent() {
- return component;
- }
- });
- return this;
- }
-
- @Override
- public BoltDeclarer shuffleGrouping(final String component, final String streamId) {
- addDeclaration(new InputDeclaration() {
- @Override
- public void declare(InputDeclarer declarer) {
- declarer.shuffleGrouping(component, streamId);
- }
-
- @Override
- public String getComponent() {
- return component;
- }
- });
- return this;
- }
-
- @Override
- public BoltDeclarer localOrShuffleGrouping(final String component) {
- addDeclaration(new InputDeclaration() {
- @Override
- public void declare(InputDeclarer declarer) {
- declarer.localOrShuffleGrouping(component);
- }
-
- @Override
- public String getComponent() {
- return component;
- }
- });
- return this;
- }
-
- @Override
- public BoltDeclarer localOrShuffleGrouping(final String component, final String streamId) {
- addDeclaration(new InputDeclaration() {
- @Override
- public void declare(InputDeclarer declarer) {
- declarer.localOrShuffleGrouping(component, streamId);
- }
-
- @Override
- public String getComponent() {
- return component;
- }
- });
- return this;
- }
-
- @Override
- public BoltDeclarer noneGrouping(final String component) {
- addDeclaration(new InputDeclaration() {
- @Override
- public void declare(InputDeclarer declarer) {
- declarer.noneGrouping(component);
- }
-
- @Override
- public String getComponent() {
- return component;
- }
- });
- return this;
- }
-
- @Override
- public BoltDeclarer noneGrouping(final String component, final String streamId) {
- addDeclaration(new InputDeclaration() {
- @Override
- public void declare(InputDeclarer declarer) {
- declarer.noneGrouping(component, streamId);
- }
-
- @Override
- public String getComponent() {
- return component;
- }
- });
- return this;
- }
-
- @Override
- public BoltDeclarer allGrouping(final String component) {
- addDeclaration(new InputDeclaration() {
- @Override
- public void declare(InputDeclarer declarer) {
- declarer.allGrouping(component);
- }
-
- @Override
- public String getComponent() {
- return component;
- }
- });
- return this;
- }
-
- @Override
- public BoltDeclarer allGrouping(final String component, final String streamId) {
- addDeclaration(new InputDeclaration() {
- @Override
- public void declare(InputDeclarer declarer) {
- declarer.allGrouping(component, streamId);
- }
-
- @Override
- public String getComponent() {
- return component;
- }
- });
- return this;
- }
-
- @Override
- public BoltDeclarer directGrouping(final String component) {
- addDeclaration(new InputDeclaration() {
- @Override
- public void declare(InputDeclarer declarer) {
- declarer.directGrouping(component);
- }
-
- @Override
- public String getComponent() {
- return component;
- }
- });
- return this;
- }
-
- @Override
- public BoltDeclarer directGrouping(final String component, final String streamId) {
- addDeclaration(new InputDeclaration() {
- @Override
- public void declare(InputDeclarer declarer) {
- declarer.directGrouping(component, streamId);
- }
-
- @Override
- public String getComponent() {
- return component;
- }
- });
- return this;
- }
-
- @Override
- public BoltDeclarer partialKeyGrouping(String componentId, Fields fields) {
- return customGrouping(componentId, new PartialKeyGrouping(fields));
- }
-
- @Override
- public BoltDeclarer partialKeyGrouping(String componentId, String streamId, Fields fields) {
- return customGrouping(componentId, streamId, new PartialKeyGrouping(fields));
- }
-
- @Override
- public BoltDeclarer customGrouping(final String component, final CustomStreamGrouping grouping) {
- addDeclaration(new InputDeclaration() {
- @Override
- public void declare(InputDeclarer declarer) {
- declarer.customGrouping(component, grouping);
- }
-
- @Override
- public String getComponent() {
- return component;
- }
- });
- return this;
- }
-
- @Override
- public BoltDeclarer customGrouping(final String component, final String streamId, final CustomStreamGrouping grouping) {
- addDeclaration(new InputDeclaration() {
- @Override
- public void declare(InputDeclarer declarer) {
- declarer.customGrouping(component, streamId, grouping);
- }
-
- @Override
- public String getComponent() {
- return component;
- }
- });
- return this;
- }
-
- @Override
- public BoltDeclarer grouping(final GlobalStreamId stream, final Grouping grouping) {
- addDeclaration(new InputDeclaration() {
- @Override
- public void declare(InputDeclarer declarer) {
- declarer.grouping(stream, grouping);
- }
-
- @Override
- public String getComponent() {
- return stream.get_componentId();
- }
- });
- return this;
- }
-
- private void addDeclaration(InputDeclaration declaration) {
- component.declarations.add(declaration);
- }
-
- @Override
- public BoltDeclarer addConfigurations(Map<String, Object> conf) {
- if (conf != null) {
- getComponentConfiguration().putAll(conf);
- }
- return this;
- }
-
- /**
- * return the current component configuration.
- *
- * @return the current configuration.
- */
- @Override
- public Map<String, Object> getComponentConfiguration() {
- return component.componentConf;
- }
-
- @Override
- public BoltDeclarer addSharedMemory(SharedMemory request) {
- component.sharedMemory.add(request);
- return this;
- }
- }
-
- private class SpoutDeclarerImpl extends BaseConfigurationDeclarer<SpoutDeclarer> implements SpoutDeclarer {
- @Override
- public SpoutDeclarer addConfigurations(Map<String, Object> conf) {
- if (conf != null) {
- spoutConf.putAll(conf);
- }
- return this;
- }
-
- /**
- * return the current component configuration.
- *
- * @return the current configuration.
- */
- @Override
- public Map<String, Object> getComponentConfiguration() {
- return spoutConf;
- }
-
- @Override
- public SpoutDeclarer addSharedMemory(SharedMemory request) {
- spoutSharedMemory.add(request);
- return this;
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/da0bc6a7/storm-client/src/jvm/org/apache/storm/transactional/partitioned/IOpaquePartitionedTransactionalSpout.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/transactional/partitioned/IOpaquePartitionedTransactionalSpout.java b/storm-client/src/jvm/org/apache/storm/transactional/partitioned/IOpaquePartitionedTransactionalSpout.java
deleted file mode 100644
index 763b6a0..0000000
--- a/storm-client/src/jvm/org/apache/storm/transactional/partitioned/IOpaquePartitionedTransactionalSpout.java
+++ /dev/null
@@ -1,52 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership. The ASF licenses this file to you under the Apache License, Version
- * 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions
- * and limitations under the License.
- */
-
-package org.apache.storm.transactional.partitioned;
-
-import java.util.Map;
-import org.apache.storm.coordination.BatchOutputCollector;
-import org.apache.storm.task.TopologyContext;
-import org.apache.storm.topology.IComponent;
-import org.apache.storm.transactional.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 IOpaquePartitionedTransactionalSpout<T> extends IComponent {
- Emitter<T> getEmitter(Map<String, Object> conf, TopologyContext context);
-
- Coordinator getCoordinator(Map<String, Object> conf, TopologyContext context);
-
- public interface Coordinator {
- /**
- * Returns true if its ok to emit start a new transaction, false otherwise (will skip this transaction).
- *
- * You should sleep here if you want a delay between asking for the next transaction (this will be called repeatedly in a loop).
- */
- boolean isReady();
-
- void close();
- }
-
- public interface Emitter<X> {
- /**
- * 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.
- */
- X emitPartitionBatch(TransactionAttempt tx, BatchOutputCollector collector, int partition, X lastPartitionMeta);
-
- int numPartitions();
-
- void close();
- }
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/da0bc6a7/storm-client/src/jvm/org/apache/storm/transactional/partitioned/IPartitionedTransactionalSpout.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/transactional/partitioned/IPartitionedTransactionalSpout.java b/storm-client/src/jvm/org/apache/storm/transactional/partitioned/IPartitionedTransactionalSpout.java
deleted file mode 100644
index ed86d01..0000000
--- a/storm-client/src/jvm/org/apache/storm/transactional/partitioned/IPartitionedTransactionalSpout.java
+++ /dev/null
@@ -1,64 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership. The ASF licenses this file to you under the Apache License, Version
- * 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions
- * and limitations under the License.
- */
-
-package org.apache.storm.transactional.partitioned;
-
-import java.util.Map;
-import org.apache.storm.coordination.BatchOutputCollector;
-import org.apache.storm.task.TopologyContext;
-import org.apache.storm.topology.IComponent;
-import org.apache.storm.transactional.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 IPartitionedTransactionalSpout<T> extends IComponent {
- Coordinator getCoordinator(Map<String, Object> conf, TopologyContext context);
-
- Emitter<T> getEmitter(Map<String, Object> conf, TopologyContext context);
-
- public interface Coordinator {
- /**
- * Return the number of 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 how many partitions were in that
- * transaction.
- */
- int numPartitions();
-
- /**
- * Returns true if its ok to emit start a new transaction, false otherwise (will skip this transaction).
- *
- * You should sleep here if you want a delay between asking for the next transaction (this will be called repeatedly in a loop).
- */
- boolean isReady();
-
- void close();
- }
-
- public interface Emitter<X> {
- /**
- * 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, BatchOutputCollector collector, int partition, X lastPartitionMeta);
-
- /**
- * 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, BatchOutputCollector collector, int partition, X partitionMeta);
-
- void close();
- }
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/da0bc6a7/storm-client/src/jvm/org/apache/storm/transactional/partitioned/OpaquePartitionedTransactionalSpoutExecutor.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/transactional/partitioned/OpaquePartitionedTransactionalSpoutExecutor.java b/storm-client/src/jvm/org/apache/storm/transactional/partitioned/OpaquePartitionedTransactionalSpoutExecutor.java
deleted file mode 100644
index 26f87d0..0000000
--- a/storm-client/src/jvm/org/apache/storm/transactional/partitioned/OpaquePartitionedTransactionalSpoutExecutor.java
+++ /dev/null
@@ -1,158 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership. The ASF licenses this file to you under the Apache License, Version
- * 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions
- * and limitations under the License.
- */
-
-package org.apache.storm.transactional.partitioned;
-
-import java.math.BigInteger;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.TreeMap;
-import org.apache.storm.Config;
-import org.apache.storm.coordination.BatchOutputCollector;
-import org.apache.storm.task.TopologyContext;
-import org.apache.storm.topology.OutputFieldsDeclarer;
-import org.apache.storm.transactional.ICommitterTransactionalSpout;
-import org.apache.storm.transactional.ITransactionalSpout;
-import org.apache.storm.transactional.TransactionAttempt;
-import org.apache.storm.transactional.state.RotatingTransactionalState;
-import org.apache.storm.transactional.state.TransactionalState;
-
-
-public class OpaquePartitionedTransactionalSpoutExecutor implements ICommitterTransactionalSpout<Object> {
- IOpaquePartitionedTransactionalSpout _spout;
-
- public OpaquePartitionedTransactionalSpoutExecutor(IOpaquePartitionedTransactionalSpout spout) {
- _spout = spout;
- }
-
- @Override
- public ITransactionalSpout.Coordinator<Object> getCoordinator(Map<String, Object> conf, TopologyContext context) {
- return new Coordinator(conf, context);
- }
-
- @Override
- public ICommitterTransactionalSpout.Emitter getEmitter(Map<String, Object> conf, TopologyContext context) {
- return new Emitter(conf, context);
- }
-
- @Override
- public void declareOutputFields(OutputFieldsDeclarer declarer) {
- _spout.declareOutputFields(declarer);
- }
-
- @Override
- public Map<String, Object> getComponentConfiguration() {
- return _spout.getComponentConfiguration();
- }
-
- public class Coordinator implements ITransactionalSpout.Coordinator<Object> {
- IOpaquePartitionedTransactionalSpout.Coordinator _coordinator;
-
- public Coordinator(Map<String, Object> conf, TopologyContext context) {
- _coordinator = _spout.getCoordinator(conf, context);
- }
-
- @Override
- public Object initializeTransaction(BigInteger txid, Object prevMetadata) {
- return null;
- }
-
- @Override
- public boolean isReady() {
- return _coordinator.isReady();
- }
-
- @Override
- public void close() {
- _coordinator.close();
- }
- }
-
- public class Emitter implements ICommitterTransactionalSpout.Emitter {
- IOpaquePartitionedTransactionalSpout.Emitter _emitter;
- TransactionalState _state;
- TreeMap<BigInteger, Map<Integer, Object>> _cachedMetas = new TreeMap<>();
- Map<Integer, RotatingTransactionalState> _partitionStates = new HashMap<>();
- int _index;
- int _numTasks;
-
- public Emitter(Map<String, Object> conf, TopologyContext context) {
- _emitter = _spout.getEmitter(conf, context);
- _index = context.getThisTaskIndex();
- _numTasks = context.getComponentTasks(context.getThisComponentId()).size();
- _state =
- TransactionalState.newUserState(conf, (String) conf.get(Config.TOPOLOGY_TRANSACTIONAL_ID), getComponentConfiguration());
- List<String> existingPartitions = _state.list("");
- for (String p : existingPartitions) {
- int partition = Integer.parseInt(p);
- if ((partition - _index) % _numTasks == 0) {
- _partitionStates.put(partition, new RotatingTransactionalState(_state, p));
- }
- }
- }
-
- @Override
- public void emitBatch(TransactionAttempt tx, Object coordinatorMeta, BatchOutputCollector collector) {
- Map<Integer, Object> metas = new HashMap<>();
- _cachedMetas.put(tx.getTransactionId(), metas);
- int partitions = _emitter.numPartitions();
- Entry<BigInteger, Map<Integer, Object>> entry = _cachedMetas.lowerEntry(tx.getTransactionId());
- Map<Integer, Object> prevCached;
- if (entry != null) {
- prevCached = entry.getValue();
- } else {
- prevCached = new HashMap<>();
- }
-
- for (int i = _index; i < partitions; i += _numTasks) {
- RotatingTransactionalState state = _partitionStates.get(i);
- if (state == null) {
- state = new RotatingTransactionalState(_state, "" + i);
- _partitionStates.put(i, state);
- }
- state.removeState(tx.getTransactionId());
- Object lastMeta = prevCached.get(i);
- if (lastMeta == null) {
- lastMeta = state.getLastState();
- }
- Object meta = _emitter.emitPartitionBatch(tx, collector, i, lastMeta);
- metas.put(i, meta);
- }
- }
-
- @Override
- public void cleanupBefore(BigInteger txid) {
- for (RotatingTransactionalState state : _partitionStates.values()) {
- state.cleanupBefore(txid);
- }
- }
-
- @Override
- public void commit(TransactionAttempt attempt) {
- BigInteger txid = attempt.getTransactionId();
- Map<Integer, Object> metas = _cachedMetas.remove(txid);
- for (Entry<Integer, Object> entry : metas.entrySet()) {
- Integer partition = entry.getKey();
- Object meta = entry.getValue();
- _partitionStates.get(partition).overrideState(txid, meta);
- }
- }
-
- @Override
- public void close() {
- _emitter.close();
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/da0bc6a7/storm-client/src/jvm/org/apache/storm/transactional/partitioned/PartitionedTransactionalSpoutExecutor.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/transactional/partitioned/PartitionedTransactionalSpoutExecutor.java b/storm-client/src/jvm/org/apache/storm/transactional/partitioned/PartitionedTransactionalSpoutExecutor.java
deleted file mode 100644
index 898ffbf..0000000
--- a/storm-client/src/jvm/org/apache/storm/transactional/partitioned/PartitionedTransactionalSpoutExecutor.java
+++ /dev/null
@@ -1,138 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership. The ASF licenses this file to you under the Apache License, Version
- * 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions
- * and limitations under the License.
- */
-
-package org.apache.storm.transactional.partitioned;
-
-import java.math.BigInteger;
-import java.util.HashMap;
-import java.util.Map;
-import org.apache.storm.Config;
-import org.apache.storm.coordination.BatchOutputCollector;
-import org.apache.storm.task.TopologyContext;
-import org.apache.storm.topology.OutputFieldsDeclarer;
-import org.apache.storm.transactional.ITransactionalSpout;
-import org.apache.storm.transactional.TransactionAttempt;
-import org.apache.storm.transactional.state.RotatingTransactionalState;
-import org.apache.storm.transactional.state.TransactionalState;
-
-
-public class PartitionedTransactionalSpoutExecutor implements ITransactionalSpout<Integer> {
- IPartitionedTransactionalSpout _spout;
-
- public PartitionedTransactionalSpoutExecutor(IPartitionedTransactionalSpout spout) {
- _spout = spout;
- }
-
- public IPartitionedTransactionalSpout getPartitionedSpout() {
- return _spout;
- }
-
- @Override
- public ITransactionalSpout.Coordinator getCoordinator(Map<String, Object> conf, TopologyContext context) {
- return new Coordinator(conf, context);
- }
-
- @Override
- public ITransactionalSpout.Emitter getEmitter(Map<String, Object> conf, TopologyContext context) {
- return new Emitter(conf, context);
- }
-
- @Override
- public void declareOutputFields(OutputFieldsDeclarer declarer) {
- _spout.declareOutputFields(declarer);
- }
-
- @Override
- public Map<String, Object> getComponentConfiguration() {
- return _spout.getComponentConfiguration();
- }
-
- class Coordinator implements ITransactionalSpout.Coordinator<Integer> {
- private IPartitionedTransactionalSpout.Coordinator _coordinator;
-
- public Coordinator(Map<String, Object> conf, TopologyContext context) {
- _coordinator = _spout.getCoordinator(conf, context);
- }
-
- @Override
- public Integer initializeTransaction(BigInteger txid, Integer prevMetadata) {
- return _coordinator.numPartitions();
- }
-
- @Override
- public boolean isReady() {
- return _coordinator.isReady();
- }
-
- @Override
- public void close() {
- _coordinator.close();
- }
- }
-
- class Emitter implements ITransactionalSpout.Emitter<Integer> {
- private IPartitionedTransactionalSpout.Emitter _emitter;
- private TransactionalState _state;
- private Map<Integer, RotatingTransactionalState> _partitionStates = new HashMap<>();
- private int _index;
- private int _numTasks;
-
- public Emitter(Map<String, Object> conf, TopologyContext context) {
- _emitter = _spout.getEmitter(conf, context);
- _state =
- TransactionalState.newUserState(conf, (String) conf.get(Config.TOPOLOGY_TRANSACTIONAL_ID), getComponentConfiguration());
- _index = context.getThisTaskIndex();
- _numTasks = context.getComponentTasks(context.getThisComponentId()).size();
- }
-
- @Override
- public void emitBatch(final TransactionAttempt tx, final Integer partitions,
- final BatchOutputCollector collector) {
- for (int i = _index; i < partitions; i += _numTasks) {
- if (!_partitionStates.containsKey(i)) {
- _partitionStates.put(i, new RotatingTransactionalState(_state, "" + i));
- }
- RotatingTransactionalState state = _partitionStates.get(i);
- final int partition = i;
- Object meta = state.getStateOrCreate(tx.getTransactionId(),
- new RotatingTransactionalState.StateInitializer() {
- @Override
- public Object init(BigInteger 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 cleanupBefore(BigInteger txid) {
- for (RotatingTransactionalState state : _partitionStates.values()) {
- state.cleanupBefore(txid);
- }
- }
-
- @Override
- public void close() {
- _state.close();
- _emitter.close();
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/da0bc6a7/storm-client/src/jvm/org/apache/storm/transactional/state/RotatingTransactionalState.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/transactional/state/RotatingTransactionalState.java b/storm-client/src/jvm/org/apache/storm/transactional/state/RotatingTransactionalState.java
deleted file mode 100644
index e551a00..0000000
--- a/storm-client/src/jvm/org/apache/storm/transactional/state/RotatingTransactionalState.java
+++ /dev/null
@@ -1,149 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership. The ASF licenses this file to you under the Apache License, Version
- * 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions
- * and limitations under the License.
- */
-
-package org.apache.storm.transactional.state;
-
-import java.math.BigInteger;
-import java.util.HashSet;
-import java.util.List;
-import java.util.SortedMap;
-import java.util.TreeMap;
-import org.apache.storm.transactional.TransactionalSpoutCoordinator;
-
-/**
- * A map from txid to a value. Automatically deletes txids that have been committed.
- */
-public class RotatingTransactionalState {
- private TransactionalState _state;
- private String _subdir;
- private boolean _strictOrder;
- private TreeMap<BigInteger, Object> _curr = new TreeMap<BigInteger, Object>();
-
- public RotatingTransactionalState(TransactionalState state, String subdir, boolean strictOrder) {
- _state = state;
- _subdir = subdir;
- _strictOrder = strictOrder;
- state.mkdir(subdir);
- sync();
- }
-
- public RotatingTransactionalState(TransactionalState state, String subdir) {
- this(state, subdir, false);
- }
-
- public Object getLastState() {
- if (_curr.isEmpty()) {
- return null;
- } else {
- return _curr.lastEntry().getValue();
- }
- }
-
- public void overrideState(BigInteger txid, Object state) {
- _state.setData(txPath(txid), state);
- _curr.put(txid, state);
- }
-
- public void removeState(BigInteger txid) {
- if (_curr.containsKey(txid)) {
- _curr.remove(txid);
- _state.delete(txPath(txid));
- }
- }
-
- public Object getState(BigInteger txid, StateInitializer init) {
- if (!_curr.containsKey(txid)) {
- SortedMap<BigInteger, Object> prevMap = _curr.headMap(txid);
- SortedMap<BigInteger, Object> afterMap = _curr.tailMap(txid);
-
- BigInteger prev = null;
- if (!prevMap.isEmpty()) {
- prev = prevMap.lastKey();
- }
-
- if (_strictOrder) {
- if (prev == null && !txid.equals(TransactionalSpoutCoordinator.INIT_TXID)) {
- throw new IllegalStateException("Trying to initialize transaction for which there should be a previous state");
- }
- if (prev != null && !prev.equals(txid.subtract(BigInteger.ONE))) {
- throw new IllegalStateException("Expecting previous txid state to be the previous transaction");
- }
- if (!afterMap.isEmpty()) {
- throw new IllegalStateException(
- "Expecting tx state to be initialized in strict order but there are txids after that have state");
- }
- }
-
-
- Object data;
- if (afterMap.isEmpty()) {
- Object prevData;
- if (prev != null) {
- prevData = _curr.get(prev);
- } else {
- prevData = null;
- }
- data = init.init(txid, prevData);
- } else {
- data = null;
- }
- _curr.put(txid, data);
- _state.setData(txPath(txid), data);
- }
- return _curr.get(txid);
- }
-
- public boolean hasCache(BigInteger txid) {
- return _curr.containsKey(txid);
- }
-
- /**
- * Returns null if it was created, the value otherwise.
- */
- public Object getStateOrCreate(BigInteger txid, StateInitializer init) {
- if (_curr.containsKey(txid)) {
- return _curr.get(txid);
- } else {
- getState(txid, init);
- return null;
- }
- }
-
- public void cleanupBefore(BigInteger txid) {
- SortedMap<BigInteger, Object> toDelete = _curr.headMap(txid);
- for (BigInteger tx : new HashSet<BigInteger>(toDelete.keySet())) {
- _curr.remove(tx);
- _state.delete(txPath(tx));
- }
- }
-
- private void sync() {
- List<String> txids = _state.list(_subdir);
- for (String txid_s : txids) {
- Object data = _state.getData(txPath(txid_s));
- _curr.put(new BigInteger(txid_s), data);
- }
- }
-
- private String txPath(BigInteger tx) {
- return txPath(tx.toString());
- }
-
- private String txPath(String tx) {
- return _subdir + "/" + tx;
- }
-
- public static interface StateInitializer {
- Object init(BigInteger txid, Object lastState);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/da0bc6a7/storm-client/src/jvm/org/apache/storm/transactional/state/TestTransactionalState.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/transactional/state/TestTransactionalState.java b/storm-client/src/jvm/org/apache/storm/transactional/state/TestTransactionalState.java
deleted file mode 100644
index af0bbf5..0000000
--- a/storm-client/src/jvm/org/apache/storm/transactional/state/TestTransactionalState.java
+++ /dev/null
@@ -1,38 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership. The ASF licenses this file to you under the Apache License, Version
- * 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions
- * and limitations under the License.
- */
-
-package org.apache.storm.transactional.state;
-
-import java.util.List;
-import java.util.Map;
-import org.apache.storm.shade.org.apache.curator.framework.CuratorFramework;
-import org.apache.storm.shade.org.apache.zookeeper.CreateMode;
-import org.apache.storm.shade.org.apache.zookeeper.data.ACL;
-
-/**
- * Facilitates testing of non-public methods in the parent class.
- */
-public class TestTransactionalState extends TransactionalState {
-
- /**
- * Matching constructor in absence of a default constructor in the parent class.
- */
- protected TestTransactionalState(Map<String, Object> conf, String id, Map<String, Object> componentConf, String subroot) {
- super(conf, id, componentConf, subroot);
- }
-
- public static void createNode(CuratorFramework curator,
- String rootDir, byte[] data, List<ACL> acls, CreateMode mode)
- throws Exception {
- TransactionalState.createNode(curator, rootDir, data, acls, mode);
- }
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/da0bc6a7/storm-client/src/jvm/org/apache/storm/transactional/state/TransactionalState.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/transactional/state/TransactionalState.java b/storm-client/src/jvm/org/apache/storm/transactional/state/TransactionalState.java
deleted file mode 100644
index 4ebbc66..0000000
--- a/storm-client/src/jvm/org/apache/storm/transactional/state/TransactionalState.java
+++ /dev/null
@@ -1,185 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership. The ASF licenses this file to you under the Apache License, Version
- * 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions
- * and limitations under the License.
- */
-
-package org.apache.storm.transactional.state;
-
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import org.apache.storm.Config;
-import org.apache.storm.cluster.DaemonType;
-import org.apache.storm.serialization.KryoValuesDeserializer;
-import org.apache.storm.serialization.KryoValuesSerializer;
-import org.apache.storm.shade.org.apache.curator.framework.CuratorFramework;
-import org.apache.storm.shade.org.apache.curator.framework.api.PathAndBytesable;
-import org.apache.storm.shade.org.apache.curator.framework.api.ProtectACLCreateModePathAndBytesable;
-import org.apache.storm.shade.org.apache.zookeeper.CreateMode;
-import org.apache.storm.shade.org.apache.zookeeper.KeeperException;
-import org.apache.storm.shade.org.apache.zookeeper.data.ACL;
-import org.apache.storm.utils.CuratorUtils;
-import org.apache.storm.utils.Utils;
-import org.apache.storm.utils.ZookeeperAuthInfo;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class TransactionalState {
- public static final Logger LOG = LoggerFactory.getLogger(TransactionalState.class);
- CuratorFramework _curator;
- KryoValuesSerializer _ser;
- KryoValuesDeserializer _des;
- List<ACL> _zkAcls = null;
-
- protected TransactionalState(Map<String, Object> conf, String id, Map<String, Object> componentConf, String subroot) {
- try {
- conf = new HashMap<>(conf);
- // ensure that the serialization registrations are consistent with the declarations in this spout
- if (componentConf != null) {
- conf.put(Config.TOPOLOGY_KRYO_REGISTER,
- componentConf
- .get(Config.TOPOLOGY_KRYO_REGISTER));
- }
- String transactionalRoot = (String) conf.get(Config.TRANSACTIONAL_ZOOKEEPER_ROOT);
- String rootDir = transactionalRoot + "/" + id + "/" + subroot;
- List<String> servers =
- (List<String>) getWithBackup(conf, Config.TRANSACTIONAL_ZOOKEEPER_SERVERS, Config.STORM_ZOOKEEPER_SERVERS);
- Object port = getWithBackup(conf, Config.TRANSACTIONAL_ZOOKEEPER_PORT, Config.STORM_ZOOKEEPER_PORT);
- ZookeeperAuthInfo auth = new ZookeeperAuthInfo(conf);
- CuratorFramework initter = CuratorUtils.newCuratorStarted(conf, servers, port, auth, DaemonType.WORKER.getDefaultZkAcls(conf));
- _zkAcls = Utils.getWorkerACL(conf);
- try {
- TransactionalState.createNode(initter, transactionalRoot, null, null, null);
- } catch (KeeperException.NodeExistsException e) {
- }
- try {
- TransactionalState.createNode(initter, rootDir, null, _zkAcls, null);
- } catch (KeeperException.NodeExistsException e) {
- }
- initter.close();
-
- _curator = CuratorUtils.newCuratorStarted(conf, servers, port, rootDir, auth, DaemonType.WORKER.getDefaultZkAcls(conf));
- _ser = new KryoValuesSerializer(conf);
- _des = new KryoValuesDeserializer(conf);
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- }
-
- public static TransactionalState newUserState(Map<String, Object> conf, String id, Map<String, Object> componentConf) {
- return new TransactionalState(conf, id, componentConf, "user");
- }
-
- public static TransactionalState newCoordinatorState(Map<String, Object> conf, String id, Map<String, Object> componentConf) {
- return new TransactionalState(conf, id, componentConf, "coordinator");
- }
-
- protected static void forPath(PathAndBytesable<String> builder,
- String path, byte[] data) throws Exception {
- try {
- if (data == null) {
- builder.forPath(path);
- } else {
- builder.forPath(path, data);
- }
- } catch (KeeperException.NodeExistsException e) {
- LOG.info("Path {} already exists.", path);
- }
- }
-
- protected static void createNode(CuratorFramework curator, String path,
- byte[] data, List<ACL> acls, CreateMode mode) throws Exception {
- ProtectACLCreateModePathAndBytesable<String> builder =
- curator.create().creatingParentsIfNeeded();
-
- if (acls == null) {
- if (mode == null) {
- TransactionalState.forPath(builder, path, data);
- } else {
- TransactionalState.forPath(builder.withMode(mode), path, data);
- }
- return;
- }
-
- TransactionalState.forPath(builder.withACL(acls), path, data);
- }
-
- public void setData(String path, Object obj) {
- path = "/" + path;
- byte[] ser = _ser.serializeObject(obj);
- try {
- if (_curator.checkExists().forPath(path) != null) {
- _curator.setData().forPath(path, ser);
- } else {
- TransactionalState.createNode(_curator, path, ser, _zkAcls,
- CreateMode.PERSISTENT);
- }
- } catch (KeeperException.NodeExistsException nee) {
- LOG.warn("Path {} already exists.", path);
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- }
-
- public void delete(String path) {
- path = "/" + path;
- try {
- _curator.delete().forPath(path);
- } catch (KeeperException.NoNodeException nne) {
- // node was already deleted
- LOG.info("Path {} has already been deleted.", path);
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- }
-
- public List<String> list(String path) {
- path = "/" + path;
- try {
- if (_curator.checkExists().forPath(path) == null) {
- return new ArrayList<String>();
- } else {
- return _curator.getChildren().forPath(path);
- }
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- }
-
- public void mkdir(String path) {
- setData(path, 7);
- }
-
- public Object getData(String path) {
- path = "/" + path;
- try {
- if (_curator.checkExists().forPath(path) != null) {
- return _des.deserializeObject(_curator.getData().forPath(path));
- } else {
- return null;
- }
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- }
-
- public void close() {
- _curator.close();
- }
-
- private Object getWithBackup(Map amap, Object primary, Object backup) {
- Object ret = amap.get(primary);
- if (ret == null) {
- return amap.get(backup);
- }
- return ret;
- }
-}