You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by sr...@apache.org on 2017/03/15 22:04:24 UTC
[1/2] storm git commit: STORM-2407: KafkaTridentSpoutOpaque Doesn't
Poll Data From All Topic-Partitions When Parallelism Hint Not a Multiple
Total Topic-Partitions - Introduce logic to poll data from the topic
partitions assigned to each task
Repository: storm
Updated Branches:
refs/heads/1.x-branch 1deb9ab95 -> 262029b15
STORM-2407: KafkaTridentSpoutOpaque Doesn't Poll Data From All Topic-Partitions When Parallelism Hint Not a Multiple Total Topic-Partitions
- Introduce logic to poll data from the topic partitions assigned to each task
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/b8885411
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/b8885411
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/b8885411
Branch: refs/heads/1.x-branch
Commit: b88854110099edb778d5a906ff1f838737b673a3
Parents: 2a99f61
Author: Hugo Louro <hm...@gmail.com>
Authored: Fri Mar 10 15:13:31 2017 -0600
Committer: Hugo Louro <hm...@gmail.com>
Committed: Fri Mar 10 17:55:02 2017 -0600
----------------------------------------------------------------------
.../trident/OpaqueTridentEventHubEmitter.java | 20 ++++-
.../spout/trident/KafkaTridentSpoutEmitter.java | 94 +++++++-------------
.../spout/trident/KafkaTridentSpoutOpaque.java | 8 +-
.../kafka/trident/TridentKafkaEmitter.java | 19 ++--
.../spout/IOpaquePartitionedTridentSpout.java | 19 +++-
.../OpaquePartitionedTridentSpoutExecutor.java | 15 ++--
.../topology/state/TransactionalState.java | 4 +
7 files changed, 89 insertions(+), 90 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/b8885411/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/OpaqueTridentEventHubEmitter.java
----------------------------------------------------------------------
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/OpaqueTridentEventHubEmitter.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/OpaqueTridentEventHubEmitter.java
index ae21ab3..20375a2 100755
--- a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/OpaqueTridentEventHubEmitter.java
+++ b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/OpaqueTridentEventHubEmitter.java
@@ -17,16 +17,16 @@
*******************************************************************************/
package org.apache.storm.eventhubs.trident;
-import java.util.List;
-import java.util.Map;
-
import org.apache.storm.eventhubs.spout.EventHubSpoutConfig;
import org.apache.storm.eventhubs.spout.IEventHubReceiverFactory;
-
import org.apache.storm.trident.operation.TridentCollector;
import org.apache.storm.trident.spout.IOpaquePartitionedTridentSpout;
import org.apache.storm.trident.topology.TransactionAttempt;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
/**
* A thin wrapper of TransactionalTridentEventHubEmitter for OpaqueTridentEventHubSpout
*/
@@ -63,6 +63,18 @@ public class OpaqueTridentEventHubEmitter implements IOpaquePartitionedTridentSp
}
@Override
+ public List<Partition> getPartitionsForTask(int taskId, int numTasks, Partitions allPartitionInfo) {
+ final List<Partition> orderedPartitions = getOrderedPartitions(allPartitionInfo);
+ final List<Partition> taskPartitions = new ArrayList<>(orderedPartitions == null ? 0 : orderedPartitions.size());
+ if (orderedPartitions != null) {
+ for (int i = taskId; i < orderedPartitions.size(); i += numTasks) {
+ taskPartitions.add(orderedPartitions.get(i));
+ }
+ }
+ return taskPartitions;
+ }
+
+ @Override
public void refreshPartitions(List<Partition> partitionList) {
transactionalEmitter.refreshPartitions(partitionList);
}
http://git-wip-us.apache.org/repos/asf/storm/blob/b8885411/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutEmitter.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutEmitter.java b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutEmitter.java
index 79dfc60..8607853 100644
--- a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutEmitter.java
+++ b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutEmitter.java
@@ -35,7 +35,6 @@ import org.slf4j.LoggerFactory;
import java.io.Serializable;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
@@ -71,7 +70,7 @@ public class KafkaTridentSpoutEmitter<K, V> implements IOpaquePartitionedTrident
private TopologyContext topologyContext;
- public KafkaTridentSpoutEmitter(KafkaTridentSpoutManager<K,V> kafkaManager, TopologyContext topologyContext, Timer refreshSubscriptionTimer) {
+ public KafkaTridentSpoutEmitter(KafkaTridentSpoutManager<K, V> kafkaManager, TopologyContext topologyContext, Timer refreshSubscriptionTimer) {
this.kafkaConsumer = kafkaManager.createAndSubscribeKafkaConsumer(topologyContext);
this.kafkaManager = kafkaManager;
this.topologyContext = topologyContext;
@@ -87,14 +86,14 @@ public class KafkaTridentSpoutEmitter<K, V> implements IOpaquePartitionedTrident
/**
* Creates instance of this class with default 500 millisecond refresh subscription timer
*/
- public KafkaTridentSpoutEmitter(KafkaTridentSpoutManager<K,V> kafkaManager, TopologyContext topologyContext) {
+ public KafkaTridentSpoutEmitter(KafkaTridentSpoutManager<K, V> kafkaManager, TopologyContext topologyContext) {
this(kafkaManager, topologyContext, new Timer(500,
kafkaManager.getKafkaSpoutConfig().getPartitionRefreshPeriodMs(), TimeUnit.MILLISECONDS));
}
@Override
public KafkaTridentSpoutBatchMetadata<K, V> emitPartitionBatch(TransactionAttempt tx, TridentCollector collector,
- KafkaTridentSpoutTopicPartition currBatchPartition, KafkaTridentSpoutBatchMetadata<K, V> lastBatch) {
+ KafkaTridentSpoutTopicPartition currBatchPartition, KafkaTridentSpoutBatchMetadata<K, V> lastBatch) {
LOG.debug("Processing batch: [transaction = {}], [currBatchPartition = {}], [lastBatchMetadata = {}], [collector = {}]",
tx, currBatchPartition, lastBatch, collector);
@@ -105,10 +104,10 @@ public class KafkaTridentSpoutEmitter<K, V> implements IOpaquePartitionedTrident
Collection<TopicPartition> pausedTopicPartitions = Collections.emptySet();
if (assignments == null || !assignments.contains(currBatchPartition.getTopicPartition())) {
- LOG.warn("SKIPPING processing batch: [transaction = {}], [currBatchPartition = {}], [lastBatchMetadata = {}], " +
- "[collector = {}] because it is not assigned {} to consumer instance [{}] of consumer group [{}]",
- tx, currBatchPartition, lastBatch, collector, assignments, kafkaConsumer,
- kafkaManager.getKafkaSpoutConfig().getConsumerGroupId());
+ LOG.warn("SKIPPING processing batch [transaction = {}], [currBatchPartition = {}], [lastBatchMetadata = {}], " +
+ "[collector = {}] because it is not part of the assignments {} of consumer instance [{}] " +
+ "of consumer group [{}]", tx, currBatchPartition, lastBatch, collector, assignments,
+ kafkaConsumer, kafkaManager.getKafkaSpoutConfig().getConsumerGroupId());
} else {
try {
// pause other topic-partitions to only poll from current topic-partition
@@ -205,67 +204,40 @@ public class KafkaTridentSpoutEmitter<K, V> implements IOpaquePartitionedTrident
/**
* Computes ordered list of topic-partitions for this task taking into consideration that topic-partitions
* for this task must be assigned to the Kafka consumer running on this task.
+ *
* @param allPartitionInfo list of all partitions as returned by {@link KafkaTridentSpoutOpaqueCoordinator}
* @return ordered list of topic partitions for this task
*/
@Override
public List<KafkaTridentSpoutTopicPartition> getOrderedPartitions(final List<TopicPartition> allPartitionInfo) {
- final int numTopicPartitions = allPartitionInfo == null ? 0 : allPartitionInfo.size();
- final int taskIndex = topologyContext.getThisTaskIndex();
- final int numTasks = topologyContext.getComponentTasks(topologyContext.getThisComponentId()).size();
-
- LOG.debug("Computing task ordered list of topic-partitions from all partitions list {}, " +
- "for task with index [{}] of total tasks [{}] ", allPartitionInfo, taskIndex, numTasks);
-
- final Set<TopicPartition> assignment = kafkaConsumer.assignment();
- LOG.debug("Consumer [{}] has assigned topic-partitions {}", kafkaConsumer, assignment);
-
- List<KafkaTridentSpoutTopicPartition> taskOrderedTps = new ArrayList<>(numTopicPartitions);
-
- if (numTopicPartitions > 0) {
- final KafkaTridentSpoutTopicPartition[] tps = new KafkaTridentSpoutTopicPartition[numTopicPartitions];
- int tpTaskComputedIdx = taskIndex;
- /*
- * Put this task's Kafka consumer assigned topic-partitions in the right index locations such
- * that distribution by OpaquePartitionedTridentSpoutExecutor can be done correctly. This algorithm
- * does the distribution in exactly the same way as the one used in OpaquePartitionedTridentSpoutExecutor
- */
- for (TopicPartition assignedTp : assignment) {
- if (tpTaskComputedIdx >= numTopicPartitions) {
- LOG.warn("Ignoring attempt to add consumer [{}] assigned topic-partition [{}] to index [{}], " +
- "out of bounds [{}]. ", kafkaConsumer, assignedTp, tpTaskComputedIdx, numTopicPartitions);
- break;
- }
- tps[tpTaskComputedIdx] = new KafkaTridentSpoutTopicPartition(assignedTp);
- LOG.debug("Added consumer assigned topic-partition [{}] to position [{}] for task with index [{}]",
- assignedTp, tpTaskComputedIdx, taskIndex);
- tpTaskComputedIdx += numTasks;
- }
+ final List<KafkaTridentSpoutTopicPartition> allPartitions = newKafkaTridentSpoutTopicPartitions(allPartitionInfo);
+ LOG.debug("Returning all topic-partitions {} across all tasks. Current task index [{}]. Total tasks [{}] ",
+ allPartitions, topologyContext.getThisTaskIndex(), getNumTasks());
+ return allPartitions;
+ }
- // Put topic-partitions assigned to consumer instances running in different tasks in the empty slots
- int i = 0;
- for (TopicPartition tp : allPartitionInfo) {
- /*
- * Topic-partition not assigned to the Kafka consumer associated with this emitter task, hence not yet
- * added to the list of task ordered partitions. To be processed next.
- */
- if (!assignment.contains(tp)) {
- for (; i < numTopicPartitions; i++) {
- if (tps[i] == null) { // find empty slot to put the topic-partition
- tps[i] = new KafkaTridentSpoutTopicPartition(tp);
- LOG.debug("Added to position [{}] topic-partition [{}], which is assigned to a consumer " +
- "running on a task other than task with index [{}] ", i, tp, taskIndex);
- i++;
- break;
- }
- }
- }
+ @Override
+ public List<KafkaTridentSpoutTopicPartition> getPartitionsForTask(int taskId, int numTasks, List<TopicPartition> allPartitionInfo) {
+ final Set<TopicPartition> assignedTps = kafkaConsumer.assignment();
+ LOG.debug("Consumer [{}], running on task with index [{}], has assigned topic-partitions {}", kafkaConsumer, taskId, assignedTps);
+ final List<KafkaTridentSpoutTopicPartition> taskTps = newKafkaTridentSpoutTopicPartitions(assignedTps);
+ LOG.debug("Returning topic-partitions {} for task with index [{}]", taskTps, taskId);
+ return taskTps;
+ }
+
+ private List<KafkaTridentSpoutTopicPartition> newKafkaTridentSpoutTopicPartitions(Collection<TopicPartition> tps) {
+ final List<KafkaTridentSpoutTopicPartition> kttp = new ArrayList<>(tps == null ? 0 : tps.size());
+ if (tps != null) {
+ for (TopicPartition tp : tps) {
+ LOG.trace("Added topic-partition [{}]", tp);
+ kttp.add(new KafkaTridentSpoutTopicPartition(tp));
}
- taskOrderedTps = Arrays.asList(tps);
}
- LOG.debug("Returning ordered list of topic-partitions {} for task with index [{}], of total tasks [{}] ",
- taskOrderedTps, taskIndex, numTasks);
- return taskOrderedTps;
+ return kttp;
+ }
+
+ private int getNumTasks() {
+ return topologyContext.getComponentTasks(topologyContext.getThisComponentId()).size();
}
@Override
http://git-wip-us.apache.org/repos/asf/storm/blob/b8885411/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutOpaque.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutOpaque.java b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutOpaque.java
index 18d37d9..0f7f0af 100644
--- a/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutOpaque.java
+++ b/external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/trident/KafkaTridentSpoutOpaque.java
@@ -35,10 +35,7 @@ public class KafkaTridentSpoutOpaque<K,V> implements IOpaquePartitionedTridentSp
private static final Logger LOG = LoggerFactory.getLogger(KafkaTridentSpoutOpaque.class);
private final KafkaTridentSpoutManager<K, V> kafkaManager;
- private KafkaTridentSpoutEmitter<K, V> kafkaTridentSpoutEmitter;
- private KafkaTridentSpoutOpaqueCoordinator<K, V> coordinator;
-
public KafkaTridentSpoutOpaque(KafkaSpoutConfig<K, V> conf) {
this(new KafkaTridentSpoutManager<>(conf));
}
@@ -73,9 +70,6 @@ public class KafkaTridentSpoutOpaque<K,V> implements IOpaquePartitionedTridentSp
@Override
public String toString() {
return super.toString() +
- "{kafkaManager=" + kafkaManager +
- ", kafkaTridentSpoutEmitter=" + kafkaTridentSpoutEmitter +
- ", coordinator=" + coordinator +
- '}';
+ "{kafkaManager=" + kafkaManager + '}';
}
}
http://git-wip-us.apache.org/repos/asf/storm/blob/b8885411/external/storm-kafka/src/jvm/org/apache/storm/kafka/trident/TridentKafkaEmitter.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/org/apache/storm/kafka/trident/TridentKafkaEmitter.java b/external/storm-kafka/src/jvm/org/apache/storm/kafka/trident/TridentKafkaEmitter.java
index 136eb0b..2a407ca 100644
--- a/external/storm-kafka/src/jvm/org/apache/storm/kafka/trident/TridentKafkaEmitter.java
+++ b/external/storm-kafka/src/jvm/org/apache/storm/kafka/trident/TridentKafkaEmitter.java
@@ -161,11 +161,6 @@ public class TridentKafkaEmitter {
/**
* re-emit the batch described by the meta data provided
- *
- * @param attempt
- * @param collector
- * @param partition
- * @param meta
*/
private void reEmitPartitionBatch(TransactionAttempt attempt, TridentCollector collector, Partition partition, Map meta) {
LOG.info("re-emitting batch, attempt " + attempt);
@@ -177,7 +172,7 @@ public class TridentKafkaEmitter {
ByteBufferMessageSet msgs = null;
msgs = fetchMessages(consumer, partition, offset);
- if(msgs != null) {
+ if (msgs != null) {
for (MessageAndOffset msg : msgs) {
if (offset == nextOffset) {
break;
@@ -253,6 +248,18 @@ public class TridentKafkaEmitter {
}
@Override
+ public List<Partition> getPartitionsForTask(int taskId, int numTasks, List<GlobalPartitionInformation> allPartitionInfo) {
+ final List<Partition> orderedPartitions = getOrderedPartitions(allPartitionInfo);
+ final List<Partition> taskPartitions = new ArrayList<>(orderedPartitions == null ? 0 : orderedPartitions.size());
+ if (orderedPartitions != null) {
+ for (int i = taskId; i < orderedPartitions.size(); i += numTasks) {
+ taskPartitions.add(orderedPartitions.get(i));
+ }
+ }
+ return taskPartitions;
+ }
+
+ @Override
public void close() {
clear();
}
http://git-wip-us.apache.org/repos/asf/storm/blob/b8885411/storm-core/src/jvm/org/apache/storm/trident/spout/IOpaquePartitionedTridentSpout.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/trident/spout/IOpaquePartitionedTridentSpout.java b/storm-core/src/jvm/org/apache/storm/trident/spout/IOpaquePartitionedTridentSpout.java
index fc4d5ea..67cb361 100644
--- a/storm-core/src/jvm/org/apache/storm/trident/spout/IOpaquePartitionedTridentSpout.java
+++ b/storm-core/src/jvm/org/apache/storm/trident/spout/IOpaquePartitionedTridentSpout.java
@@ -52,13 +52,26 @@ public interface IOpaquePartitionedTridentSpout<Partitions, Partition extends IS
* 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);
+ void refreshPartitions(List<Partition> partitionResponsibilities);
+
+ /**
+ * @return The oredered list of partitions being processed by all the tasks
+ */
List<Partition> getOrderedPartitions(Partitions allPartitionInfo);
+
+ /**
+ * @return The list of partitions that are to be processed by the task with id {@code taskId}
+ */
+ List<Partition> getPartitionsForTask(int taskId, int numTasks, Partitions allPartitionInfo);
+
void close();
}
- Emitter<Partitions, Partition, M> getEmitter(Map conf, TopologyContext context);
- Coordinator getCoordinator(Map conf, TopologyContext context);
+ Emitter<Partitions, Partition, M> getEmitter(Map conf, TopologyContext context);
+
+ Coordinator getCoordinator(Map conf, TopologyContext context);
+
Map<String, Object> getComponentConfiguration();
+
Fields getOutputFields();
}
http://git-wip-us.apache.org/repos/asf/storm/blob/b8885411/storm-core/src/jvm/org/apache/storm/trident/spout/OpaquePartitionedTridentSpoutExecutor.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/trident/spout/OpaquePartitionedTridentSpoutExecutor.java b/storm-core/src/jvm/org/apache/storm/trident/spout/OpaquePartitionedTridentSpoutExecutor.java
index ea66acd..a6eeff0 100644
--- a/storm-core/src/jvm/org/apache/storm/trident/spout/OpaquePartitionedTridentSpoutExecutor.java
+++ b/storm-core/src/jvm/org/apache/storm/trident/spout/OpaquePartitionedTridentSpoutExecutor.java
@@ -27,7 +27,6 @@ import org.apache.storm.tuple.Fields;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -111,16 +110,14 @@ public class OpaquePartitionedTridentSpoutExecutor implements ICommitterTridentS
tx, coordinatorMeta, collector, this);
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));
+ final List<ISpoutPartition> taskPartitions = _emitter.getPartitionsForTask(_index, _numTasks, coordinatorMeta);
+ for (ISpoutPartition partition : taskPartitions) {
+ _partitionStates.put(partition.getId(), new EmitterPartitionState(new RotatingTransactionalState(_state, partition.getId()), partition));
}
- _emitter.refreshPartitions(myPartitions);
+
+ // refresh all partitions for backwards compatibility with old spout
+ _emitter.refreshPartitions(_emitter.getOrderedPartitions(coordinatorMeta));
_savedCoordinatorMeta = coordinatorMeta;
_changedMeta = true;
}
http://git-wip-us.apache.org/repos/asf/storm/blob/b8885411/storm-core/src/jvm/org/apache/storm/trident/topology/state/TransactionalState.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/trident/topology/state/TransactionalState.java b/storm-core/src/jvm/org/apache/storm/trident/topology/state/TransactionalState.java
index 8b63547..26ac404 100644
--- a/storm-core/src/jvm/org/apache/storm/trident/topology/state/TransactionalState.java
+++ b/storm-core/src/jvm/org/apache/storm/trident/topology/state/TransactionalState.java
@@ -37,6 +37,10 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
+/**
+ * Class that contains the logic to extract the transactional state info from zookeeper. All transactional state
+ * is kept in zookeeper. This class only contains references to Curator, which is used to get all info from zookeeper.
+ */
public class TransactionalState {
private static final Logger LOG = LoggerFactory.getLogger(TransactionalState.class);
[2/2] storm git commit: Merge branch
'1.x-branch_STORM-2407_KafkaTridentMissesData' of
https://github.com/hmcl/storm-apache into STORM-2407
Posted by sr...@apache.org.
Merge branch '1.x-branch_STORM-2407_KafkaTridentMissesData' of https://github.com/hmcl/storm-apache into STORM-2407
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/262029b1
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/262029b1
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/262029b1
Branch: refs/heads/1.x-branch
Commit: 262029b15f558969ffcc90658f43238b0d1dc31f
Parents: 1deb9ab b888541
Author: Sriharsha Chintalapani <ha...@hortonworks.com>
Authored: Wed Mar 15 12:45:28 2017 -0700
Committer: Sriharsha Chintalapani <ha...@hortonworks.com>
Committed: Wed Mar 15 12:45:28 2017 -0700
----------------------------------------------------------------------
.../trident/OpaqueTridentEventHubEmitter.java | 20 ++++-
.../spout/trident/KafkaTridentSpoutEmitter.java | 94 +++++++-------------
.../spout/trident/KafkaTridentSpoutOpaque.java | 8 +-
.../kafka/trident/TridentKafkaEmitter.java | 19 ++--
.../spout/IOpaquePartitionedTridentSpout.java | 19 +++-
.../OpaquePartitionedTridentSpoutExecutor.java | 15 ++--
.../topology/state/TransactionalState.java | 4 +
7 files changed, 89 insertions(+), 90 deletions(-)
----------------------------------------------------------------------