You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by kl...@apache.org on 2017/05/01 23:13:18 UTC
[10/37] geode git commit: GEODE-2632: change dependencies on
GemFireCacheImpl to InternalCache
http://git-wip-us.apache.org/repos/asf/geode/blob/d319d129/geode-core/src/main/java/org/apache/geode/internal/cache/TXCommitMessage.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/TXCommitMessage.java b/geode-core/src/main/java/org/apache/geode/internal/cache/TXCommitMessage.java
index 7c2a3e3..1113c82 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/TXCommitMessage.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/TXCommitMessage.java
@@ -12,14 +12,12 @@
* or implied. See the License for the specific language governing permissions and limitations under
* the License.
*/
-
package org.apache.geode.internal.cache;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
@@ -65,6 +63,7 @@ import org.apache.geode.internal.Version;
import org.apache.geode.internal.cache.locks.TXLockId;
import org.apache.geode.internal.cache.locks.TXLockIdImpl;
import org.apache.geode.internal.cache.locks.TXLockService;
+import org.apache.geode.internal.cache.partitioned.Bucket;
import org.apache.geode.internal.cache.persistence.PersistentMemberID;
import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID;
import org.apache.geode.internal.cache.versions.VersionSource;
@@ -80,7 +79,6 @@ import org.apache.geode.internal.offheap.annotations.Released;
* commit, to other cache members.
*
* @since GemFire 4.0
- *
*/
public class TXCommitMessage extends PooledDistributionMessage
implements MembershipListener, MessageWithReply {
@@ -98,12 +96,10 @@ public class TXCommitMessage extends PooledDistributionMessage
protected transient DM dm; // Used on the sending side of this message
private transient int sequenceNum = 0;
- private transient HashMap<InternalDistributedMember, RegionCommitList> msgMap = null; // Maps
- // receiver
- // Serializables
- // to
- // RegionCommitList
- // instances
+
+ // Maps receiver Serializables to RegionCommitList instances
+ private transient HashMap<InternalDistributedMember, RegionCommitList> msgMap = null;
+
private transient RegionCommit currentRegion;
protected transient TXState txState = null;
private transient boolean wasProcessed;
@@ -124,7 +120,7 @@ public class TXCommitMessage extends PooledDistributionMessage
private transient boolean hasReliableRegions = false;
/** Set of all caching exceptions produced hile processing this tx */
- private transient Set processingExceptions = Collections.EMPTY_SET;
+ private transient Set processingExceptions = Collections.emptySet();
private transient ClientProxyMembershipID bridgeContext = null;
@@ -149,15 +145,6 @@ public class TXCommitMessage extends PooledDistributionMessage
* transaction
*/
public static final TXCommitMessage EXCEPTION_MSG = new TXCommitMessage();
- /*
- * /** this is set if this message should deserialize the WAN shadowKey sent by the sender. Sender
- * will not send shadowKeys when there is a mix of 7.0 and 7.0.1 members
- *
- * private transient boolean shouldReadShadowKey; /** this is set if the sender has decided to
- * send WAN shadowKey for 7.0.1 members
- *
- * private transient boolean shouldWriteShadowKey;
- */
public TXCommitMessage(TXId txIdent, DM dm, TXState txState) {
this.dm = dm;
@@ -176,7 +163,7 @@ public class TXCommitMessage extends PooledDistributionMessage
// zero arg constructor for DataSerializer
}
- static public TXFarSideCMTracker getTracker() {
+ public static TXFarSideCMTracker getTracker() {
return TXCommitMessage.txTracker;
}
@@ -194,7 +181,7 @@ public class TXCommitMessage extends PooledDistributionMessage
* Return the TXCommitMessage we have already received that is associated with id. Note because of
* bug 37657 we may need to wait for it to show up.
*/
- static public TXCommitMessage waitForMessage(Object id, DM dm) {
+ public static TXCommitMessage waitForMessage(Object id, DM dm) {
TXFarSideCMTracker map = getTracker();
return map.waitForMessage(id, dm);
}
@@ -210,12 +197,10 @@ public class TXCommitMessage extends PooledDistributionMessage
// make sure we have some changes and someone to send them to
if (!this.currentRegion.isEmpty() && s != null && !s.isEmpty()) {
// Get the persistent ids for the current region and save them
- Map<InternalDistributedMember, PersistentMemberID> persistentIds =
- getPersistentIds(this.currentRegion.r);
- this.currentRegion.persistentIds = persistentIds;
+ this.currentRegion.persistentIds = getPersistentIds(this.currentRegion.r);
if (this.msgMap == null) {
- this.msgMap = new HashMap<InternalDistributedMember, RegionCommitList>();
+ this.msgMap = new HashMap<>();
}
{
RegionCommitList newRCL = null;
@@ -245,18 +230,18 @@ public class TXCommitMessage extends PooledDistributionMessage
}
}
}
- { // Now deal with each existing recipient that does not care
- // about this region
- Iterator<Map.Entry<InternalDistributedMember, RegionCommitList>> it =
- this.msgMap.entrySet().iterator();
- while (it.hasNext()) {
- Map.Entry<InternalDistributedMember, RegionCommitList> me = it.next();
- if (!s.contains(me.getKey())) {
- RegionCommitList rcl = me.getValue();
- RegionCommitList trimmedRcl = rcl.trim(this.currentRegion);
- if (trimmedRcl != rcl) {
- me.setValue(trimmedRcl);
- }
+
+ // Now deal with each existing recipient that does not care
+ // about this region
+ Iterator<Map.Entry<InternalDistributedMember, RegionCommitList>> it =
+ this.msgMap.entrySet().iterator();
+ while (it.hasNext()) {
+ Map.Entry<InternalDistributedMember, RegionCommitList> me = it.next();
+ if (!s.contains(me.getKey())) {
+ RegionCommitList rcl = me.getValue();
+ RegionCommitList trimmedRcl = rcl.trim(this.currentRegion);
+ if (trimmedRcl != rcl) {
+ me.setValue(trimmedRcl);
}
}
}
@@ -264,13 +249,11 @@ public class TXCommitMessage extends PooledDistributionMessage
this.currentRegion = null;
}
-
-
private Map<InternalDistributedMember, PersistentMemberID> getPersistentIds(LocalRegion r) {
if (r instanceof DistributedRegion) {
- return ((DistributedRegion) r).getCacheDistributionAdvisor().advisePersistentMembers();
+ return ((CacheDistributionAdvisee) r).getCacheDistributionAdvisor().advisePersistentMembers();
} else {
- return Collections.EMPTY_MAP;
+ return Collections.emptyMap();
}
}
@@ -287,17 +270,15 @@ public class TXCommitMessage extends PooledDistributionMessage
this.currentRegion = null;
}
-
Map viewVersions = new HashMap();
private Boolean needsLargeModCount;
private transient boolean disableListeners = false;
-
/** record CacheDistributionAdvisor.startOperation versions for later cleanup */
protected void addViewVersion(DistributedRegion dr, long version) {
- viewVersions.put(dr, Long.valueOf(version));
+ viewVersions.put(dr, version);
}
protected void releaseViewVersions() {
@@ -309,7 +290,7 @@ public class TXCommitMessage extends PooledDistributionMessage
// need to continue the iteration if one of the regions is destroyed
// since others may still be okay
try {
- long newv = dr.getDistributionAdvisor().endOperation(viewVersion.longValue());
+ long newv = dr.getDistributionAdvisor().endOperation(viewVersion);
} catch (RuntimeException ex) {
rte = ex;
}
@@ -504,13 +485,13 @@ public class TXCommitMessage extends PooledDistributionMessage
}
Set cacheClosedMembers =
- (processor == null) ? Collections.EMPTY_SET : processor.getCacheClosedMembers();
+ (processor == null) ? Collections.emptySet() : processor.getCacheClosedMembers();
Set departedMembers =
- (processor == null) ? Collections.EMPTY_SET : processor.getDepartedMembers();
+ (processor == null) ? Collections.emptySet() : processor.getDepartedMembers();
// check reliability on each region
- Set regionDistributionExceptions = Collections.EMPTY_SET;
- Set failedRegionNames = Collections.EMPTY_SET;
+ Set regionDistributionExceptions = Collections.emptySet();
+ Set failedRegionNames = Collections.emptySet();
for (Iterator iter = regionToRecipients.entrySet().iterator(); iter.hasNext();) {
Map.Entry me = (Map.Entry) iter.next();
final RegionCommit rc = (RegionCommit) me.getKey();
@@ -519,7 +500,7 @@ public class TXCommitMessage extends PooledDistributionMessage
successfulRecipients.removeAll(departedMembers);
// remove members who destroyed that region or closed their cache
- Set regionDestroyedMembers = (processor == null) ? Collections.EMPTY_SET
+ Set regionDestroyedMembers = (processor == null) ? Collections.emptySet()
: processor.getRegionDestroyedMembers(rc.r.getFullPath());
successfulRecipients.removeAll(cacheClosedMembers);
@@ -528,7 +509,7 @@ public class TXCommitMessage extends PooledDistributionMessage
try {
rc.r.handleReliableDistribution(successfulRecipients);
} catch (RegionDistributionException e) {
- if (regionDistributionExceptions == Collections.EMPTY_SET) {
+ if (regionDistributionExceptions == Collections.emptySet()) {
regionDistributionExceptions = new HashSet();
failedRegionNames = new HashSet();
}
@@ -545,11 +526,10 @@ public class TXCommitMessage extends PooledDistributionMessage
}
}
-
/**
* Helper method for send
*/
- private final void setRecipientsSendData(Set recipients, ReplyProcessor21 processor,
+ private void setRecipientsSendData(Set recipients, ReplyProcessor21 processor,
RegionCommitList rcl) {
setRecipients(recipients);
this.regions = rcl;
@@ -601,22 +581,21 @@ public class TXCommitMessage extends PooledDistributionMessage
this.farSideEntryOps.add(entryOp);
}
- protected final void addProcessingException(Exception e) {
+ protected void addProcessingException(Exception e) {
// clear all previous exceptions if e is a CacheClosedException
- if (this.processingExceptions == Collections.EMPTY_SET || e instanceof CancelException) {
+ if (this.processingExceptions == Collections.emptySet() || e instanceof CancelException) {
this.processingExceptions = new HashSet();
}
this.processingExceptions.add(e);
}
-
public void setDM(DM dm) {
this.dm = dm;
}
public DM getDM() {
if (this.dm == null) {
- GemFireCacheImpl cache = GemFireCacheImpl.getExisting("Applying TXCommit");
+ InternalCache cache = GemFireCacheImpl.getExisting("Applying TXCommit");
this.dm = cache.getDistributionManager();
}
return this.dm;
@@ -639,12 +618,9 @@ public class TXCommitMessage extends PooledDistributionMessage
if (logger.isDebugEnabled()) {
logger.debug("begin processing TXCommitMessage for {}", this.txIdent);
}
+ // do this before CacheFactory.getInstance for bug 33471
final int oldLevel =
- LocalRegion.setThreadInitLevelRequirement(LocalRegion.BEFORE_INITIAL_IMAGE); // do this
- // before
- // CacheFactory.getInstance
- // for bug
- // 33471
+ LocalRegion.setThreadInitLevelRequirement(LocalRegion.BEFORE_INITIAL_IMAGE);
boolean forceListener = false; // this gets flipped if we need to fire tx listener
// it needs to default to false because we don't want to fire listeners on pr replicates
try {
@@ -662,20 +638,18 @@ public class TXCommitMessage extends PooledDistributionMessage
try {
// Pre-process each Region in the tx
try {
- {
- Iterator it = this.regions.iterator();
- while (it.hasNext()) {
- boolean failedBeginProcess = true;
- RegionCommit rc = (RegionCommit) it.next();
- try {
- failedBeginProcess = !rc.beginProcess(dm, this.txIdent, txEvent);
- } catch (CacheRuntimeException problem) {
- processCacheRuntimeException(problem);
- } finally {
- if (failedBeginProcess) {
- rc.r = null; // Cause related FarSideEntryOps to skip processing
- it.remove(); // Skip endProcessing as well
- }
+ Iterator it = this.regions.iterator();
+ while (it.hasNext()) {
+ boolean failedBeginProcess = true;
+ RegionCommit rc = (RegionCommit) it.next();
+ try {
+ failedBeginProcess = !rc.beginProcess(dm, this.txIdent, txEvent);
+ } catch (CacheRuntimeException problem) {
+ processCacheRuntimeException(problem);
+ } finally {
+ if (failedBeginProcess) {
+ rc.r = null; // Cause related FarSideEntryOps to skip processing
+ it.remove(); // Skip endProcessing as well
}
}
}
@@ -746,22 +720,20 @@ public class TXCommitMessage extends PooledDistributionMessage
}
public void basicProcessOps() {
- {
- List<EntryEventImpl> pendingCallbacks = new ArrayList<>(this.farSideEntryOps.size());
- Collections.sort(this.farSideEntryOps);
- Iterator it = this.farSideEntryOps.iterator();
- while (it.hasNext()) {
- try {
- RegionCommit.FarSideEntryOp entryOp = (RegionCommit.FarSideEntryOp) it.next();
- entryOp.process(pendingCallbacks);
- } catch (CacheRuntimeException problem) {
- processCacheRuntimeException(problem);
- } catch (Exception e) {
- addProcessingException(e);
- }
+ List<EntryEventImpl> pendingCallbacks = new ArrayList<>(this.farSideEntryOps.size());
+ Collections.sort(this.farSideEntryOps);
+ Iterator it = this.farSideEntryOps.iterator();
+ while (it.hasNext()) {
+ try {
+ RegionCommit.FarSideEntryOp entryOp = (RegionCommit.FarSideEntryOp) it.next();
+ entryOp.process(pendingCallbacks);
+ } catch (CacheRuntimeException problem) {
+ processCacheRuntimeException(problem);
+ } catch (Exception e) {
+ addProcessingException(e);
}
- firePendingCallbacks(pendingCallbacks);
}
+ firePendingCallbacks(pendingCallbacks);
}
private void firePendingCallbacks(List<EntryEventImpl> callbacks) {
@@ -963,7 +935,7 @@ public class TXCommitMessage extends PooledDistributionMessage
@Override
public String toString() {
- StringBuffer result = new StringBuffer(256);
+ StringBuilder result = new StringBuilder(256);
result.append("TXCommitMessage@").append(System.identityHashCode(this)).append("#")
.append(this.sequenceNum).append(" processorId=").append(this.processorId).append(" txId=")
.append(this.txIdent);
@@ -994,7 +966,6 @@ public class TXCommitMessage extends PooledDistributionMessage
* that represents an entire transaction. At commit time the txCommitMessage sent to each node can
* be a subset of the transaction, this method will combine those subsets into a complete message.
*
- * @param msgSet
* @return the complete txCommitMessage
*/
public static TXCommitMessage combine(Set<TXCommitMessage> msgSet) {
@@ -1014,8 +985,6 @@ public class TXCommitMessage extends PooledDistributionMessage
/**
* Combines the other TXCommitMessage into this message. Used to compute complete TXCommitMessage
* from parts.
- *
- * @param other
*/
public void combine(TXCommitMessage other) {
assert other != null;
@@ -1032,7 +1001,7 @@ public class TXCommitMessage extends PooledDistributionMessage
}
}
- public final static class RegionCommitList extends ArrayList<RegionCommit> {
+ public static class RegionCommitList extends ArrayList<RegionCommit> {
private static final long serialVersionUID = -8910813949027683641L;
private transient boolean needsAck = false;
private transient RegionCommit trimRC = null;
@@ -1091,7 +1060,7 @@ public class TXCommitMessage extends PooledDistributionMessage
@Override
public String toString() {
- StringBuffer result = new StringBuffer(256);
+ StringBuilder result = new StringBuilder(256);
result.append('@').append(System.identityHashCode(this)).append(' ').append(super.toString());
return result.toString();
}
@@ -1390,8 +1359,6 @@ public class TXCommitMessage extends PooledDistributionMessage
}
}
-
-
boolean isEmpty() {
return this.opKeys == null;
}
@@ -1444,7 +1411,7 @@ public class TXCommitMessage extends PooledDistributionMessage
@Override
public String toString() {
- StringBuffer result = new StringBuffer(64);
+ StringBuilder result = new StringBuilder(64);
if (this.regionPath != null) {
result.append(this.regionPath);
} else {
@@ -1460,8 +1427,7 @@ public class TXCommitMessage extends PooledDistributionMessage
if (this.r != null) {
DataSerializer.writeString(this.r.getFullPath(), out);
if (this.r instanceof BucketRegion) {
- DataSerializer.writeString(((BucketRegion) this.r).getPartitionedRegion().getFullPath(),
- out);
+ DataSerializer.writeString(((Bucket) this.r).getPartitionedRegion().getFullPath(), out);
} else {
DataSerializer.writeString(null, out);
}
@@ -1557,7 +1523,6 @@ public class TXCommitMessage extends PooledDistributionMessage
* @param in the data input that is used to read the data for this entry op
* @param largeModCount true if the mod count is a int instead of a byte.
* @param readShadowKey true if a long shadowKey should be read
- * @throws ClassNotFoundException
*/
public void fromData(DataInput in, boolean largeModCount, boolean readShadowKey)
throws IOException, ClassNotFoundException {
@@ -1665,7 +1630,7 @@ public class TXCommitMessage extends PooledDistributionMessage
}
}
- final Object getTrackerKey() {
+ Object getTrackerKey() {
if (this.lockId != null) {
return this.lockId;
} else {
@@ -1677,41 +1642,39 @@ public class TXCommitMessage extends PooledDistributionMessage
* Used to prevent processing of the message if we have reported to other FarSiders that we did
* not received the CommitProcessMessage
*/
- final boolean dontProcess() {
+ boolean dontProcess() {
return this.dontProcess;
}
/**
* Indicate that this message should not be processed if we receive CommitProcessMessage (late)
*/
- final void setDontProcess() {
+ void setDontProcess() {
this.dontProcess = true;
}
- final boolean isProcessing() {
+ boolean isProcessing() {
return this.isProcessing;
}
- private final void setIsProcessing(boolean isProcessing) {
+ private void setIsProcessing(boolean isProcessing) {
this.isProcessing = isProcessing;
}
- final boolean wasProcessed() {
+ boolean wasProcessed() {
return this.wasProcessed;
}
- final void setProcessed(boolean wasProcessed) {
+ void setProcessed(boolean wasProcessed) {
this.wasProcessed = wasProcessed;
}
- /********************* Region Commit Process Messages ***************************************/
-
/**
* The CommitProcessForLockIDMessaage is sent by the Distributed ACK TX origin to the recipients
* (aka FarSiders) to indicate that a previously received RegionCommit that contained a lockId
* should commence processing.
*/
- static final public class CommitProcessForLockIdMessage extends CommitProcessMessage {
+ public static class CommitProcessForLockIdMessage extends CommitProcessMessage {
private TXLockId lockId;
public CommitProcessForLockIdMessage() {
@@ -1749,7 +1712,7 @@ public class TXCommitMessage extends PooledDistributionMessage
@Override
public String toString() {
- StringBuffer result = new StringBuffer(128);
+ StringBuilder result = new StringBuilder(128);
result.append("CommitProcessForLockIdMessage@").append(System.identityHashCode(this))
.append(" lockId=").append(this.lockId);
return result.toString();
@@ -1763,7 +1726,7 @@ public class TXCommitMessage extends PooledDistributionMessage
* typically sent if all the TX changes are a result of load/netsearch/netload values (thus no
* lockid)
*/
- static final public class CommitProcessForTXIdMessage extends CommitProcessMessage {
+ public static class CommitProcessForTXIdMessage extends CommitProcessMessage {
private TXId txId;
public CommitProcessForTXIdMessage() {
@@ -1801,14 +1764,15 @@ public class TXCommitMessage extends PooledDistributionMessage
@Override
public String toString() {
- StringBuffer result = new StringBuffer(128);
+ StringBuilder result = new StringBuilder(128);
result.append("CommitProcessForTXIdMessage@").append(System.identityHashCode(this))
.append(" txId=").append(this.txId);
return result.toString();
}
}
- static abstract public class CommitProcessMessage extends PooledDistributionMessage {
- protected final void basicProcess(final TXCommitMessage mess, final DistributionManager dm) {
+
+ public abstract static class CommitProcessMessage extends PooledDistributionMessage {
+ protected void basicProcess(final TXCommitMessage mess, final DistributionManager dm) {
dm.removeMembershipListener(mess);
synchronized (mess) {
if (mess.dontProcess()) {
@@ -1823,8 +1787,6 @@ public class TXCommitMessage extends PooledDistributionMessage
}
}
- /********************* Commit Process Query Message ***************************************/
-
/**
* The CommitProcessQueryMessage is used to attempt to recover - in the Distributed ACK TXs - when
* the origin of the CommitProcess messages departed from the distributed system. The sender of
@@ -1835,9 +1797,8 @@ public class TXCommitMessage extends PooledDistributionMessage
* about the the tracker key - opting not to have specific messages for each type like
* CommitProcessFor<Lock/TX>Id - and take the performance penalty of an extra call to
* DataSerializer
- *
*/
- static final public class CommitProcessQueryMessage extends PooledDistributionMessage {
+ public static class CommitProcessQueryMessage extends PooledDistributionMessage {
private Object trackerKey; // Either a TXLockId or a TXId
private int processorId;
@@ -1885,7 +1846,7 @@ public class TXCommitMessage extends PooledDistributionMessage
@Override
public String toString() {
- StringBuffer result = new StringBuffer(128);
+ StringBuilder result = new StringBuilder(128);
result.append("CommitProcessQueryMessage@").append(System.identityHashCode(this))
.append(" trackerKeyClass=").append(this.trackerKey.getClass().getName())
.append(" trackerKey=").append(this.trackerKey).append(" processorId=")
@@ -1895,7 +1856,7 @@ public class TXCommitMessage extends PooledDistributionMessage
}
/********************* Commit Process Query Response Message **********************************/
- static final public class CommitProcessQueryReplyMessage extends ReplyMessage {
+ public static class CommitProcessQueryReplyMessage extends ReplyMessage {
private boolean wasReceived;
public CommitProcessQueryReplyMessage(boolean wasReceived) {
@@ -1929,7 +1890,7 @@ public class TXCommitMessage extends PooledDistributionMessage
@Override
public String toString() {
- StringBuffer result = new StringBuffer(128);
+ StringBuilder result = new StringBuilder(128);
result.append("CommitProcessQueryReplyMessage@").append(System.identityHashCode(this))
.append(" wasReceived=").append(this.wasReceived).append(" processorId=")
.append(this.processorId).append(" from ").append(this.getSender());
@@ -1938,7 +1899,7 @@ public class TXCommitMessage extends PooledDistributionMessage
}
/********************* Commit Process Query Response Processor *********************************/
- static final public class CommitProcessQueryReplyProcessor extends ReplyProcessor21 {
+ public static class CommitProcessQueryReplyProcessor extends ReplyProcessor21 {
public boolean receivedOnePositive;
CommitProcessQueryReplyProcessor(DM dm, Set members) {
@@ -1956,17 +1917,19 @@ public class TXCommitMessage extends PooledDistributionMessage
}
@Override
- final protected boolean canStopWaiting() {
+ protected boolean canStopWaiting() {
return this.receivedOnePositive;
}
- final public boolean receivedACommitProcessMessage() {
+ public boolean receivedACommitProcessMessage() {
return this.receivedOnePositive;
}
}
/********************* MembershipListener Implementation ***************************************/
- public void memberJoined(InternalDistributedMember id) {}
+ public void memberJoined(InternalDistributedMember id) {
+ // do nothing
+ }
public void memberSuspect(InternalDistributedMember id, InternalDistributedMember whoSuspected,
String reason) {}
@@ -2118,58 +2081,6 @@ public class TXCommitMessage extends PooledDistributionMessage
}
}
- // /** Custom subclass that keeps all ReplyExceptions */
- // private class ReliableCommitReplyProcessor extends ReliableReplyProcessor21 {
- //
- // /** Set of members that threw CacheClosedExceptions */
- // private Set cacheExceptions = new HashSet();
- // /** key=region path, value=Set of members */
- // private Map regionExceptions = new HashMap();
- //
- // public ReliableCommitReplyProcessor(DM dm,
- // Set initMembers) {
- // super(dm, initMembers);
- // }
- // protected synchronized void processException(DistributionMessage msg,
- // ReplyException re) {
- // // only interested in CommitReplyException
- // if (re instanceof CommitReplyException) {
- // CommitReplyException cre = (CommitReplyException) re;
- // Set exceptions = cre.getExceptions();
- // for (Iterator iter = exceptions.iterator(); iter.hasNext();) {
- // Exception ex = (Exception) iter.next();
- // if (ex instanceof CacheClosedException) {
- // cacheExceptions.add(msg.getSender());
- // }
- // else if (ex instanceof RegionDestroyedException) {
- // String r = ((RegionDestroyedException)ex).getRegionFullPath();
- // Set members = (Set) regionExceptions.get(r);
- // if (members == null) {
- // members = new HashSet();
- // regionExceptions.put(r, members);
- // }
- // members.add(msg.getSender());
- // }
- // }
- // }
- // else {
- // // allow superclass to handle all other exceptions
- // super.processException(msg, re);
- // }
- // }
- // // these two accessors should be called after wait for replies completes
- // protected Set getCacheClosedMembers() {
- // return this.cacheExceptions;
- // }
- // protected Set getRegionDestroyedMembers(String regionFullPath) {
- // Set members = (Set) this.regionExceptions.get(regionFullPath);
- // if (members == null) {
- // members = Collections.EMPTY_SET;
- // }
- // return members;
- // }
- // }
-
/**
* Reply processor which collects all CommitReplyExceptions and emits a detailed failure exception
* if problems occur
@@ -2223,7 +2134,7 @@ public class TXCommitMessage extends PooledDistributionMessage
(CommitExceptionCollectingException) this.exception;
return cce.getCacheClosedMembers();
} else {
- return Collections.EMPTY_SET;
+ return Collections.emptySet();
}
}
@@ -2233,7 +2144,7 @@ public class TXCommitMessage extends PooledDistributionMessage
(CommitExceptionCollectingException) this.exception;
return cce.getRegionDestroyedMembers(regionFullPath);
} else {
- return Collections.EMPTY_SET;
+ return Collections.emptySet();
}
}
}
@@ -2264,14 +2175,12 @@ public class TXCommitMessage extends PooledDistributionMessage
/**
* Determine if the commit processing was incomplete, if so throw a detailed exception
* indicating the source of the problem
- *
- * @param msgMap
*/
public void handlePotentialCommitFailure(
HashMap<InternalDistributedMember, RegionCommitList> msgMap) {
if (fatalExceptions.size() > 0) {
- StringBuffer errorMessage = new StringBuffer("Incomplete commit of transaction ").append(id)
- .append(". Caused by the following exceptions: ");
+ StringBuilder errorMessage = new StringBuilder("Incomplete commit of transaction ")
+ .append(id).append(". Caused by the following exceptions: ");
for (Iterator i = fatalExceptions.entrySet().iterator(); i.hasNext();) {
Map.Entry me = (Map.Entry) i.next();
DistributedMember mem = (DistributedMember) me.getKey();
@@ -2366,16 +2275,13 @@ public class TXCommitMessage extends PooledDistributionMessage
public Set getRegionDestroyedMembers(String regionFullPath) {
Set members = (Set) this.regionExceptions.get(regionFullPath);
if (members == null) {
- members = Collections.EMPTY_SET;
+ members = Collections.emptySet();
}
return members;
}
/**
* Protected by (this)
- *
- * @param member
- * @param exceptions
*/
public void addExceptionsFromMember(InternalDistributedMember member, Set exceptions) {
for (Iterator iter = exceptions.iterator(); iter.hasNext();) {
http://git-wip-us.apache.org/repos/asf/geode/blob/d319d129/geode-core/src/main/java/org/apache/geode/internal/cache/TXManagerImpl.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/TXManagerImpl.java b/geode-core/src/main/java/org/apache/geode/internal/cache/TXManagerImpl.java
index 2948a48..a0a4d7c 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/TXManagerImpl.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/TXManagerImpl.java
@@ -12,16 +12,52 @@
* or implied. See the License for the specific language governing permissions and limitations under
* the License.
*/
-
package org.apache.geode.internal.cache;
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import java.util.concurrent.locks.LockSupport;
+
+import org.apache.logging.log4j.Logger;
+
import org.apache.geode.DataSerializer;
import org.apache.geode.GemFireException;
import org.apache.geode.InternalGemFireError;
import org.apache.geode.SystemFailure;
-import org.apache.geode.cache.*;
+import org.apache.geode.cache.CacheTransactionManager;
+import org.apache.geode.cache.CommitConflictException;
+import org.apache.geode.cache.TransactionDataRebalancedException;
+import org.apache.geode.cache.TransactionId;
+import org.apache.geode.cache.TransactionInDoubtException;
+import org.apache.geode.cache.TransactionListener;
+import org.apache.geode.cache.TransactionWriter;
+import org.apache.geode.cache.UnsupportedOperationInTransactionException;
import org.apache.geode.distributed.TXManagerCancelledException;
-import org.apache.geode.distributed.internal.*;
+import org.apache.geode.distributed.internal.DM;
+import org.apache.geode.distributed.internal.DistributionConfig;
+import org.apache.geode.distributed.internal.DistributionManager;
+import org.apache.geode.distributed.internal.HighPriorityDistributionMessage;
+import org.apache.geode.distributed.internal.InternalDistributedSystem;
+import org.apache.geode.distributed.internal.MembershipListener;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.SystemTimer.SystemTimerTask;
import org.apache.geode.internal.cache.tier.sockets.Message;
@@ -32,30 +68,14 @@ import org.apache.geode.internal.logging.log4j.LocalizedMessage;
import org.apache.geode.internal.util.concurrent.CustomEntryConcurrentHashMap;
import org.apache.geode.internal.util.concurrent.CustomEntryConcurrentHashMap.HashEntry;
import org.apache.geode.internal.util.concurrent.CustomEntryConcurrentHashMap.MapCallback;
-import org.apache.logging.log4j.Logger;
-
-import java.io.DataInput;
-import java.io.DataOutput;
-import java.io.IOException;
-import java.util.*;
-import java.util.Map.Entry;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentLinkedQueue;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
-import java.util.concurrent.locks.LockSupport;
/**
- * <p>
* The internal implementation of the {@link CacheTransactionManager} interface returned by
- * {@link GemFireCacheImpl#getCacheTransactionManager}. Internal operations
+ * {@link InternalCache#getCacheTransactionManager}. Internal operations
*
- * </code>TransactionListener</code> invocation, Region synchronization, transaction statistics and
+ * {@code TransactionListener} invocation, Region synchronization, transaction statistics and
*
* transaction logging are handled here
- *
*
* @since GemFire 4.0
*
@@ -67,12 +87,14 @@ public class TXManagerImpl implements CacheTransactionManager, MembershipListene
// Thread specific context container
private final ThreadLocal<TXStateProxy> txContext;
+
private static TXManagerImpl currentInstance = null;
+
// The unique transaction ID for this Manager
private final AtomicInteger uniqId;
private final DM dm;
- private final Cache cache;
+ private final InternalCache cache;
// The DistributionMemberID used to construct TXId's
private final InternalDistributedMember distributionMgrId;
@@ -86,8 +108,10 @@ public class TXManagerImpl implements CacheTransactionManager, MembershipListene
*/
public static final int NOTX = -1;
- private final ArrayList<TransactionListener> txListeners = new ArrayList<TransactionListener>(8);
+ private final List<TransactionListener> txListeners = new ArrayList<>(8);
+
public TransactionWriter writer = null;
+
private volatile boolean closed = false;
private final Map<TXId, TXStateProxy> hostedTXStates;
@@ -95,7 +119,7 @@ public class TXManagerImpl implements CacheTransactionManager, MembershipListene
/**
* the number of client initiated transactions to store for client failover
*/
- public final static int FAILOVER_TX_MAP_SIZE =
+ public static final int FAILOVER_TX_MAP_SIZE =
Integer.getInteger(DistributionConfig.GEMFIRE_PREFIX + "transactionFailoverMapSize", 1000);
/**
@@ -106,6 +130,7 @@ public class TXManagerImpl implements CacheTransactionManager, MembershipListene
@SuppressWarnings("unchecked")
private Map<TXId, TXCommitMessage> failoverMap =
Collections.synchronizedMap(new LinkedHashMap<TXId, TXCommitMessage>() {
+ // TODO: inner class is serializable but outer class is not
private static final long serialVersionUID = -4156018226167594134L;
protected boolean removeEldestEntry(Entry eldest) {
@@ -114,7 +139,7 @@ public class TXManagerImpl implements CacheTransactionManager, MembershipListene
eldest.getKey(), (size() > FAILOVER_TX_MAP_SIZE));
}
return size() > FAILOVER_TX_MAP_SIZE;
- };
+ }
});
/**
@@ -126,8 +151,7 @@ public class TXManagerImpl implements CacheTransactionManager, MembershipListene
/**
* this keeps track of all the transactions that were initiated locally.
*/
- private ConcurrentMap<TXId, TXStateProxy> localTxMap =
- new ConcurrentHashMap<TXId, TXStateProxy>();
+ private ConcurrentMap<TXId, TXStateProxy> localTxMap = new ConcurrentHashMap<>();
/**
* the time in minutes after which any suspended transaction are rolled back. default is 30
@@ -152,49 +176,44 @@ public class TXManagerImpl implements CacheTransactionManager, MembershipListene
/**
* Constructor that implements the {@link CacheTransactionManager} interface. Only only one
* instance per {@link org.apache.geode.cache.Cache}
- *
- * @param cachePerfStats
*/
- public TXManagerImpl(CachePerfStats cachePerfStats, Cache cache) {
+ public TXManagerImpl(CachePerfStats cachePerfStats, InternalCache cache) {
this.cache = cache;
this.dm = ((InternalDistributedSystem) cache.getDistributedSystem()).getDistributionManager();
this.distributionMgrId = this.dm.getDistributionManagerId();
this.uniqId = new AtomicInteger(0);
this.cachePerfStats = cachePerfStats;
- this.hostedTXStates = new HashMap<TXId, TXStateProxy>();
- this.txContext = new ThreadLocal<TXStateProxy>();
- this.isTXDistributed = new ThreadLocal<Boolean>();
+ this.hostedTXStates = new HashMap<>();
+ this.txContext = new ThreadLocal<>();
+ this.isTXDistributed = new ThreadLocal<>();
this.transactionTimeToLive = Integer
.getInteger(DistributionConfig.GEMFIRE_PREFIX + "cacheServer.transactionTimeToLive", 180);
currentInstance = this;
}
- final Cache getCache() {
+ InternalCache getCache() {
return this.cache;
}
-
/**
* Get the TransactionWriter for the cache
*
* @return the current TransactionWriter
* @see TransactionWriter
*/
- public final TransactionWriter getWriter() {
+ public TransactionWriter getWriter() {
return writer;
}
-
- public final void setWriter(TransactionWriter writer) {
- if (((GemFireCacheImpl) this.cache).isClient()) {
+ public void setWriter(TransactionWriter writer) {
+ if (this.cache.isClient()) {
throw new IllegalStateException(
LocalizedStrings.TXManager_NO_WRITER_ON_CLIENT.toLocalizedString());
}
this.writer = writer;
}
-
- public final TransactionListener getListener() {
+ public TransactionListener getListener() {
synchronized (this.txListeners) {
if (this.txListeners.isEmpty()) {
return null;
@@ -280,7 +299,7 @@ public class TXManagerImpl implements CacheTransactionManager, MembershipListene
}
}
- final CachePerfStats getCachePerfStats() {
+ CachePerfStats getCachePerfStats() {
return this.cachePerfStats;
}
@@ -396,7 +415,7 @@ public class TXManagerImpl implements CacheTransactionManager, MembershipListene
noteCommitSuccess(opStart, lifeTime, tx);
}
- final void noteCommitFailure(long opStart, long lifeTime, TXStateInterface tx) {
+ void noteCommitFailure(long opStart, long lifeTime, TXStateInterface tx) {
long opEnd = CachePerfStats.getStatTime();
this.cachePerfStats.txFailure(opEnd - opStart, lifeTime, tx.getChanges());
TransactionListener[] listeners = getListeners();
@@ -428,7 +447,7 @@ public class TXManagerImpl implements CacheTransactionManager, MembershipListene
}
}
- final void noteCommitSuccess(long opStart, long lifeTime, TXStateInterface tx) {
+ void noteCommitSuccess(long opStart, long lifeTime, TXStateInterface tx) {
long opEnd = CachePerfStats.getStatTime();
this.cachePerfStats.txSuccess(opEnd - opStart, lifeTime, tx.getChanges());
TransactionListener[] listeners = getListeners();
@@ -497,7 +516,7 @@ public class TXManagerImpl implements CacheTransactionManager, MembershipListene
noteRollbackSuccess(opStart, lifeTime, tx);
}
- final void noteRollbackSuccess(long opStart, long lifeTime, TXStateInterface tx) {
+ void noteRollbackSuccess(long opStart, long lifeTime, TXStateInterface tx) {
long opEnd = CachePerfStats.getStatTime();
this.cachePerfStats.txRollback(opEnd - opStart, lifeTime, tx.getChanges());
TransactionListener[] listeners = getListeners();
@@ -597,7 +616,7 @@ public class TXManagerImpl implements CacheTransactionManager, MembershipListene
return retVal;
}
- public final void setTXState(TXStateProxy val) {
+ public void setTXState(TXStateProxy val) {
txContext.set(val);
}
@@ -625,11 +644,9 @@ public class TXManagerImpl implements CacheTransactionManager, MembershipListene
for (TXStateProxy proxy : this.localTxMap.values()) {
proxy.close();
}
- {
- TransactionListener[] listeners = getListeners();
- for (int i = 0; i < listeners.length; i++) {
- closeListener(listeners[i]);
- }
+ TransactionListener[] listeners = getListeners();
+ for (int i = 0; i < listeners.length; i++) {
+ closeListener(listeners[i]);
}
}
@@ -660,7 +677,7 @@ public class TXManagerImpl implements CacheTransactionManager, MembershipListene
* @return the state of the transaction or null. Pass this value to {@link TXManagerImpl#resume}
* to reactivate the suspended transaction.
*/
- public final TXStateProxy internalSuspend() {
+ public TXStateProxy internalSuspend() {
TXStateProxy result = getTXState();
if (result != null) {
result.suspend();
@@ -691,26 +708,26 @@ public class TXManagerImpl implements CacheTransactionManager, MembershipListene
/**
* @deprecated use internalResume instead
*/
- public final void resume(TXStateProxy tx) {
+ @Deprecated
+ public void resume(TXStateProxy tx) {
internalResume(tx);
}
- public final boolean isClosed() {
+ public boolean isClosed() {
return this.closed;
}
- private final void checkClosed() {
+ private void checkClosed() {
cache.getCancelCriterion().checkCancelInProgress(null);
if (this.closed) {
throw new TXManagerCancelledException("This transaction manager is closed.");
}
}
- final DM getDM() {
+ DM getDM() {
return this.dm;
}
-
public static int getCurrentTXUniqueId() {
if (currentInstance == null) {
return NOTX;
@@ -718,9 +735,7 @@ public class TXManagerImpl implements CacheTransactionManager, MembershipListene
return currentInstance.getMyTXUniqueId();
}
-
-
- public final static TXStateProxy getCurrentTXState() {
+ public static TXStateProxy getCurrentTXState() {
if (currentInstance == null) {
return null;
}
@@ -747,9 +762,7 @@ public class TXManagerImpl implements CacheTransactionManager, MembershipListene
* on the txState, on which this thread operates. Some messages like SizeMessage should not create
* a new txState.
*
- * @param msg
* @return {@link TXStateProxy} the txProxy for the transactional message
- * @throws InterruptedException
*/
public TXStateProxy masqueradeAs(TransactionMessage msg) throws InterruptedException {
if (msg.getTXUniqId() == NOTX || !msg.canParticipateInTransaction()) {
@@ -828,11 +841,8 @@ public class TXManagerImpl implements CacheTransactionManager, MembershipListene
* on the txState, on which this thread operates. Some messages like SizeMessage should not create
* a new txState.
*
- * @param msg
- * @param memberId
* @param probeOnly - do not masquerade; just look up the TX state
* @return {@link TXStateProxy} the txProxy for the transactional message
- * @throws InterruptedException
*/
public TXStateProxy masqueradeAs(Message msg, InternalDistributedMember memberId,
boolean probeOnly) throws InterruptedException {
@@ -846,8 +856,7 @@ public class TXManagerImpl implements CacheTransactionManager, MembershipListene
synchronized (this.hostedTXStates) {
val = this.hostedTXStates.get(key);
if (val == null) {
- // [sjigyasu] TODO: Conditionally create object based on distributed or non-distributed tx
- // mode
+ // TODO: Conditionally create object based on distributed or non-distributed tx mode
if (msg instanceof TransactionMessage
&& ((TransactionMessage) msg).isTransactionDistributed()) {
val = new DistTXStateProxyImplOnDatanode(this, key, memberId);
@@ -894,8 +903,6 @@ public class TXManagerImpl implements CacheTransactionManager, MembershipListene
/**
* Remove the association created by {@link #masqueradeAs(TransactionMessage)}
- *
- * @param tx
*/
public void unmasquerade(TXStateProxy tx) {
if (tx != null) {
@@ -907,7 +914,6 @@ public class TXManagerImpl implements CacheTransactionManager, MembershipListene
/**
* Cleanup the remote txState after commit and rollback
*
- * @param txId
* @return the TXStateProxy
*/
public TXStateProxy removeHostedTXState(TXId txId) {
@@ -942,7 +948,6 @@ public class TXManagerImpl implements CacheTransactionManager, MembershipListene
/**
* Used to verify if a transaction with a given id is hosted by this txManager.
*
- * @param txId
* @return true if the transaction is in progress, false otherwise
*/
public boolean isHostedTxInProgress(TXId txId) {
@@ -1104,7 +1109,6 @@ public class TXManagerImpl implements CacheTransactionManager, MembershipListene
* If the given transaction is already being completed by another thread this will wait for that
* completion to finish and will ensure that the result is saved in the client failover map.
*
- * @param txId
* @return true if a wait was performed
*/
public boolean waitForCompletingTransaction(TXId txId) {
@@ -1132,7 +1136,6 @@ public class TXManagerImpl implements CacheTransactionManager, MembershipListene
/**
* Returns the TXCommitMessage for a transaction that has been successfully completed.
*
- * @param txId
* @return the commit message or an exception token e.g {@link TXCommitMessage#CMT_CONFLICT_MSG}
* if the transaction threw an exception
* @see #isExceptionToken(TXCommitMessage)
@@ -1142,7 +1145,6 @@ public class TXManagerImpl implements CacheTransactionManager, MembershipListene
}
/**
- * @param msg
* @return true if msg is an exception token, false otherwise
*/
public boolean isExceptionToken(TXCommitMessage msg) {
@@ -1158,7 +1160,6 @@ public class TXManagerImpl implements CacheTransactionManager, MembershipListene
* during transaction execution.
*
* @param msg the token that represents the exception
- * @param txId
* @return the exception
*/
public RuntimeException getExceptionForToken(TXCommitMessage msg, TXId txId) {
@@ -1209,13 +1210,12 @@ public class TXManagerImpl implements CacheTransactionManager, MembershipListene
@Override
protected void process(DistributionManager dm) {
- GemFireCacheImpl cache = GemFireCacheImpl.getInstance();
+ InternalCache cache = GemFireCacheImpl.getInstance();
if (cache != null) {
TXManagerImpl mgr = cache.getTXMgr();
mgr.removeTransactions(this.txIds, false);
}
}
-
}
private ConcurrentMap<TransactionId, TXStateProxy> suspendedTXs =
@@ -1290,8 +1290,7 @@ public class TXManagerImpl implements CacheTransactionManager, MembershipListene
SystemTimerTask task = this.expiryTasks.remove(txProxy.getTransactionId());
if (task != null) {
if (task.cancel()) {
- GemFireCacheImpl cache = (GemFireCacheImpl) this.cache;
- cache.purgeCCPTimer();
+ this.cache.purgeCCPTimer();
}
}
}
@@ -1300,8 +1299,7 @@ public class TXManagerImpl implements CacheTransactionManager, MembershipListene
* this map keeps track of all the threads that are waiting in
* {@link #tryResume(TransactionId, long, TimeUnit)} for a particular transactionId
*/
- private ConcurrentMap<TransactionId, Queue<Thread>> waitMap =
- new ConcurrentHashMap<TransactionId, Queue<Thread>>();
+ private ConcurrentMap<TransactionId, Queue<Thread>> waitMap = new ConcurrentHashMap<>();
public boolean tryResume(TransactionId transactionId, long time, TimeUnit unit) {
if (transactionId == null || getTXState() != null || !exists(transactionId)) {
@@ -1383,11 +1381,9 @@ public class TXManagerImpl implements CacheTransactionManager, MembershipListene
/**
* schedules the transaction to expire after {@link #suspendedTXTimeout}
*
- * @param txId
* @param expiryTimeUnit the time unit to use when scheduling the expiration
*/
private void scheduleExpiry(TransactionId txId, TimeUnit expiryTimeUnit) {
- final GemFireCacheImpl cache = (GemFireCacheImpl) this.cache;
if (suspendedTXTimeout < 0) {
if (logger.isDebugEnabled()) {
logger.debug("TX: transaction: {} not scheduled to expire", txId);
@@ -1452,9 +1448,13 @@ public class TXManagerImpl implements CacheTransactionManager, MembershipListene
}
private static class RefCountMapEntry
implements HashEntry<AbstractRegionEntry, RefCountMapEntry> {
+
private final AbstractRegionEntry key;
+
private HashEntry<AbstractRegionEntry, RefCountMapEntry> next;
+
private volatile int refCount;
+
private static final AtomicIntegerFieldUpdater<RefCountMapEntry> refCountUpdater =
AtomicIntegerFieldUpdater.newUpdater(RefCountMapEntry.class, "refCount");
@@ -1561,7 +1561,7 @@ public class TXManagerImpl implements CacheTransactionManager, MembershipListene
}
};
- public static final void incRefCount(AbstractRegionEntry re) {
+ public static void incRefCount(AbstractRegionEntry re) {
TXManagerImpl mgr = currentInstance;
if (mgr != null) {
mgr.refCountMap.create(re, incCallback, null, null, true);
@@ -1571,7 +1571,7 @@ public class TXManagerImpl implements CacheTransactionManager, MembershipListene
/**
* Return true if refCount went to zero.
*/
- public static final boolean decRefCount(AbstractRegionEntry re) {
+ public static boolean decRefCount(AbstractRegionEntry re) {
TXManagerImpl mgr = currentInstance;
if (mgr != null) {
return mgr.refCountMap.removeConditionally(re, decCallback, null, null) != null;
@@ -1628,9 +1628,9 @@ public class TXManagerImpl implements CacheTransactionManager, MembershipListene
}
};
try {
- ((GemFireCacheImpl) this.cache).getCCPTimer().schedule(task, timeout);
+ this.cache.getCCPTimer().schedule(task, timeout);
} catch (IllegalStateException ise) {
- if (!((GemFireCacheImpl) this.cache).isClosed()) {
+ if (!this.cache.isClosed()) {
throw ise;
}
// task not able to be scheduled due to cache is closing,
@@ -1716,7 +1716,7 @@ public class TXManagerImpl implements CacheTransactionManager, MembershipListene
LocalizedStrings.TXManagerImpl_CANNOT_CHANGE_TRANSACTION_MODE_WHILE_TRANSACTIONS_ARE_IN_PROGRESS
.toLocalizedString());
} else {
- isTXDistributed.set(new Boolean(flag));
+ isTXDistributed.set(flag);
}
}
@@ -1726,14 +1726,13 @@ public class TXManagerImpl implements CacheTransactionManager, MembershipListene
* default value of this property.
*/
public boolean isDistributed() {
-
Boolean value = isTXDistributed.get();
// This can be null if not set in setDistributed().
if (value == null) {
InternalDistributedSystem ids = (InternalDistributedSystem) cache.getDistributedSystem();
return ids.getOriginalConfig().getDistributedTransactions();
} else {
- return value.booleanValue();
+ return value;
}
}
http://git-wip-us.apache.org/repos/asf/geode/blob/d319d129/geode-core/src/main/java/org/apache/geode/internal/cache/TXMessage.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/TXMessage.java b/geode-core/src/main/java/org/apache/geode/internal/cache/TXMessage.java
index fd53fb1..24cbaa2 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/TXMessage.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/TXMessage.java
@@ -39,20 +39,20 @@ import org.apache.geode.internal.cache.partitioned.PartitionMessage;
import org.apache.geode.internal.i18n.LocalizedStrings;
import org.apache.geode.internal.logging.LogService;
-/**
- *
- *
- */
public abstract class TXMessage extends SerialDistributionMessage
implements MessageWithReply, TransactionMessage {
private static final Logger logger = LogService.getLogger();
private int processorId;
+
private int txUniqId;
+
private InternalDistributedMember txMemberId = null;
- public TXMessage() {}
+ public TXMessage() {
+ // nothing
+ }
public TXMessage(int txUniqueId, InternalDistributedMember onBehalfOfMember,
ReplyProcessor21 processor) {
@@ -73,7 +73,7 @@ public abstract class TXMessage extends SerialDistributionMessage
if (logger.isDebugEnabled()) {
logger.debug("processing {}", this);
}
- GemFireCacheImpl cache = GemFireCacheImpl.getInstance();
+ InternalCache cache = GemFireCacheImpl.getInstance();
if (checkCacheClosing(cache) || checkDSClosing(cache.getInternalDistributedSystem())) {
thr = new CacheClosedException(LocalizedStrings.PartitionMessage_REMOTE_CACHE_IS_CLOSED_0
.toLocalizedString(dm.getId()));
@@ -130,7 +130,7 @@ public abstract class TXMessage extends SerialDistributionMessage
return distributedSystem == null || distributedSystem.isDisconnecting();
}
- private boolean checkCacheClosing(GemFireCacheImpl cache) {
+ private boolean checkCacheClosing(InternalCache cache) {
return cache == null || cache.isClosed();
}
@@ -160,7 +160,7 @@ public abstract class TXMessage extends SerialDistributionMessage
* Transaction operations override this method to do actual work
*
* @param txId The transaction Id to operate on
- * @return true if {@link TXMessage} should send a reply false otherwise
+ * @return true if TXMessage should send a reply false otherwise
*/
protected abstract boolean operateOnTx(TXId txId, DistributionManager dm)
throws RemoteOperationException;
@@ -192,7 +192,6 @@ public abstract class TXMessage extends SerialDistributionMessage
return txMemberId;
}
-
@Override
public int getProcessorId() {
return this.processorId;
http://git-wip-us.apache.org/repos/asf/geode/blob/d319d129/geode-core/src/main/java/org/apache/geode/internal/cache/TXRegionLockRequestImpl.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/TXRegionLockRequestImpl.java b/geode-core/src/main/java/org/apache/geode/internal/cache/TXRegionLockRequestImpl.java
index eefa27c..1e586aa 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/TXRegionLockRequestImpl.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/TXRegionLockRequestImpl.java
@@ -12,7 +12,6 @@
* or implied. See the License for the specific language governing permissions and limitations under
* the License.
*/
-
package org.apache.geode.internal.cache;
import java.io.DataInput;
@@ -26,23 +25,23 @@ import org.apache.logging.log4j.Logger;
import org.apache.geode.DataSerializer;
import org.apache.geode.cache.CacheClosedException;
import org.apache.geode.internal.InternalDataSerializer;
+import org.apache.geode.internal.cache.locks.TXRegionLockRequest;
import org.apache.geode.internal.logging.LogService;
import org.apache.geode.internal.logging.log4j.LogMarker;
/**
* TXRegionLockRequest represents all the locks that need to be made for a single region.
*
- *
* @since GemFire 4.0
- *
*/
-public class TXRegionLockRequestImpl
- implements org.apache.geode.internal.cache.locks.TXRegionLockRequest {
+public class TXRegionLockRequestImpl implements TXRegionLockRequest {
private static final long serialVersionUID = 5840033961584078082L;
private static final Logger logger = LogService.getLogger();
private transient LocalRegion r;
+
private String regionPath;
+
private Set<Object> entryKeys;
public TXRegionLockRequestImpl() {
@@ -93,26 +92,26 @@ public class TXRegionLockRequestImpl
this.entryKeys.add(key);
}
- public final void fromData(DataInput in) throws IOException, ClassNotFoundException {
+ public void fromData(DataInput in) throws IOException, ClassNotFoundException {
this.regionPath = DataSerializer.readString(in);
- final GemFireCacheImpl cache = getCache(false);
+ final InternalCache cache = getCache(false);
try {
final int size = InternalDataSerializer.readArrayLength(in);
if (cache != null && size > 0) {
this.r = (LocalRegion) cache.getRegion(this.regionPath);
}
this.entryKeys = readEntryKeySet(size, in);
- } catch (CacheClosedException cce) {
+ } catch (CacheClosedException ignore) {
// don't throw in deserialization
this.entryKeys = null;
}
}
- private final Set<Object> readEntryKeySet(final int size, final DataInput in)
+ private Set<Object> readEntryKeySet(final int size, final DataInput in)
throws IOException, ClassNotFoundException {
- if (logger.isDebugEnabled()) {
+ if (logger.isTraceEnabled()) {
logger.trace(LogMarker.SERIALIZER, "Reading HashSet with size {}", size);
}
@@ -135,21 +134,21 @@ public class TXRegionLockRequestImpl
InternalDataSerializer.writeSet(this.entryKeys, out);
}
- public static final TXRegionLockRequestImpl createFromData(DataInput in)
+ public static TXRegionLockRequestImpl createFromData(DataInput in)
throws IOException, ClassNotFoundException {
TXRegionLockRequestImpl result = new TXRegionLockRequestImpl();
InternalDataSerializer.invokeFromData(result, in);
return result;
}
- public final String getRegionFullPath() {
+ public String getRegionFullPath() {
if (this.regionPath == null) {
this.regionPath = this.r.getFullPath();
}
return this.regionPath;
}
- public final Set<Object> getKeys() {
+ public Set<Object> getKeys() {
if (this.entryKeys == null) {
// check for cache closed/closing
getCache(true);
@@ -157,8 +156,8 @@ public class TXRegionLockRequestImpl
return this.entryKeys;
}
- private final GemFireCacheImpl getCache(boolean throwIfClosing) {
- final GemFireCacheImpl cache = GemFireCacheImpl.getInstance();
+ private InternalCache getCache(boolean throwIfClosing) {
+ final InternalCache cache = GemFireCacheImpl.getInstance();
if (cache != null && !cache.isClosed()) {
if (throwIfClosing) {
cache.getCancelCriterion().checkCancelInProgress(null);
@@ -175,7 +174,7 @@ public class TXRegionLockRequestImpl
* Only safe to call in the vm that creates this request. Once it is serialized this method will
* return null.
*/
- public final LocalRegion getLocalRegion() {
+ public LocalRegion getLocalRegion() {
return this.r;
}
http://git-wip-us.apache.org/repos/asf/geode/blob/d319d129/geode-core/src/main/java/org/apache/geode/internal/cache/TXRegionState.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/TXRegionState.java b/geode-core/src/main/java/org/apache/geode/internal/cache/TXRegionState.java
index 167f1c1..496a812 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/TXRegionState.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/TXRegionState.java
@@ -97,7 +97,7 @@ public class TXRegionState {
public TXEntryState createReadEntry(LocalRegion r, Object entryKey, RegionEntry re, Object vId,
Object pendingValue) {
- GemFireCacheImpl cache = r.getCache();
+ InternalCache cache = r.getCache();
boolean isDistributed = false;
if (cache.getTxManager().getTXState() != null) {
isDistributed = cache.getTxManager().getTXState().isDistTx();
http://git-wip-us.apache.org/repos/asf/geode/blob/d319d129/geode-core/src/main/java/org/apache/geode/internal/cache/TXRemoteCommitMessage.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/TXRemoteCommitMessage.java b/geode-core/src/main/java/org/apache/geode/internal/cache/TXRemoteCommitMessage.java
index 725ad64..6a1eeed 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/TXRemoteCommitMessage.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/TXRemoteCommitMessage.java
@@ -40,16 +40,13 @@ import org.apache.geode.internal.cache.RemoteOperationMessage.RemoteOperationRes
import org.apache.geode.internal.logging.LogService;
import org.apache.geode.internal.logging.log4j.LogMarker;
-/**
- *
- *
- */
public class TXRemoteCommitMessage extends TXMessage {
-
private static final Logger logger = LogService.getLogger();
/** for deserialization */
- public TXRemoteCommitMessage() {}
+ public TXRemoteCommitMessage() {
+ // nothing
+ }
public TXRemoteCommitMessage(int txUniqId, InternalDistributedMember onBehalfOfClientMember,
ReplyProcessor21 processor) {
@@ -76,7 +73,7 @@ public class TXRemoteCommitMessage extends TXMessage {
@Override
protected boolean operateOnTx(TXId txId, DistributionManager dm) throws RemoteOperationException {
- GemFireCacheImpl cache = GemFireCacheImpl.getInstance();
+ InternalCache cache = GemFireCacheImpl.getInstance();
TXManagerImpl txMgr = cache.getTXMgr();
if (logger.isDebugEnabled()) {
@@ -124,8 +121,6 @@ public class TXRemoteCommitMessage extends TXMessage {
return true;
}
-
-
/**
* This message is used for the reply to a remote commit operation: a commit from a stub to the tx
* host. This is the reply to a {@link TXRemoteCommitMessage}.
@@ -133,7 +128,9 @@ public class TXRemoteCommitMessage extends TXMessage {
* @since GemFire 6.5
*/
public static final class TXRemoteCommitReplyMessage extends ReplyMessage {
+
private transient TXCommitMessage commitMessage;
+
/*
* Used on the fromData side to transfer the value bytes to the requesting thread
*/
@@ -142,7 +139,9 @@ public class TXRemoteCommitMessage extends TXMessage {
/**
* Empty constructor to conform to DataSerializable interface
*/
- public TXRemoteCommitReplyMessage() {}
+ public TXRemoteCommitReplyMessage() {
+ // nothing
+ }
public TXRemoteCommitReplyMessage(DataInput in) throws IOException, ClassNotFoundException {
fromData(in);
@@ -219,7 +218,7 @@ public class TXRemoteCommitMessage extends TXMessage {
@Override
public String toString() {
- StringBuffer sb = new StringBuffer();
+ StringBuilder sb = new StringBuilder();
sb.append("TXRemoteCommitReplyMessage ").append("processorid=").append(this.processorId)
.append(" reply to sender ").append(this.getSender());
return sb.toString();
http://git-wip-us.apache.org/repos/asf/geode/blob/d319d129/geode-core/src/main/java/org/apache/geode/internal/cache/TXRemoteRollbackMessage.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/TXRemoteRollbackMessage.java b/geode-core/src/main/java/org/apache/geode/internal/cache/TXRemoteRollbackMessage.java
index e416e11..13b783f 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/TXRemoteRollbackMessage.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/TXRemoteRollbackMessage.java
@@ -30,15 +30,12 @@ import org.apache.geode.distributed.internal.membership.InternalDistributedMembe
import org.apache.geode.internal.i18n.LocalizedStrings;
import org.apache.geode.internal.logging.LogService;
-/**
- *
- *
- */
public class TXRemoteRollbackMessage extends TXMessage {
-
private static final Logger logger = LogService.getLogger();
- public TXRemoteRollbackMessage() {}
+ public TXRemoteRollbackMessage() {
+ // nothing
+ }
public TXRemoteRollbackMessage(int txUniqId, InternalDistributedMember onBehalfOfClientMember,
ReplyProcessor21 processor) {
@@ -60,7 +57,7 @@ public class TXRemoteRollbackMessage extends TXMessage {
@Override
protected boolean operateOnTx(TXId txId, DistributionManager dm) {
- GemFireCacheImpl cache = GemFireCacheImpl.getInstance();
+ InternalCache cache = GemFireCacheImpl.getInstance();
if (cache == null) {
throw new CacheClosedException(
LocalizedStrings.CacheFactory_A_CACHE_HAS_NOT_YET_BEEN_CREATED.toLocalizedString());
http://git-wip-us.apache.org/repos/asf/geode/blob/d319d129/geode-core/src/main/java/org/apache/geode/internal/cache/TXStateProxyImpl.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/TXStateProxyImpl.java b/geode-core/src/main/java/org/apache/geode/internal/cache/TXStateProxyImpl.java
index 49922a0..22b95f3 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/TXStateProxyImpl.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/TXStateProxyImpl.java
@@ -12,9 +12,6 @@
* or implied. See the License for the specific language governing permissions and limitations under
* the License.
*/
-/**
- * File comment
- */
package org.apache.geode.internal.cache;
import java.util.Collection;
@@ -48,11 +45,7 @@ import org.apache.geode.internal.i18n.LocalizedStrings;
import org.apache.geode.internal.logging.LogService;
import org.apache.geode.internal.logging.log4j.LocalizedMessage;
-/**
- *
- */
public class TXStateProxyImpl implements TXStateProxy {
-
private static final Logger logger = LogService.getLogger();
protected static final AtomicBoolean txDistributedClientWarningIssued = new AtomicBoolean();
@@ -63,6 +56,7 @@ public class TXStateProxyImpl implements TXStateProxy {
protected DistributedMember target;
private boolean commitRequestedByOwner;
private boolean isJCATransaction;
+
/**
* for client/server JTA transactions we need to have a single thread handle both beforeCompletion
* and afterCompletion so that beforeC can obtain locks for the afterC step. This is that thread
@@ -88,39 +82,26 @@ public class TXStateProxyImpl implements TXStateProxy {
return this.synchRunnable;
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.internal.cache.TXStateInterface#getSemaphore()
- */
public ReentrantLock getLock() {
return this.lock;
}
-
- /**
- * @return the isJTA
- */
final boolean isJTA() {
return isJTA;
}
- /**
- * @return the txId
- */
final public TXId getTxId() {
return txId;
}
- /**
- * @return the txMgr
- */
public final TXManagerImpl getTxMgr() {
return txMgr;
}
protected volatile TXStateInterface realDeal;
+
protected boolean inProgress = true;
+
protected InternalDistributedMember onBehalfOfClientMember = null;
/**
@@ -184,10 +165,6 @@ public class TXStateProxyImpl implements TXStateProxy {
return this.realDeal;
}
- /**
- * @param managerImpl
- * @param id
- */
public TXStateProxyImpl(TXManagerImpl managerImpl, TXId id,
InternalDistributedMember clientMember) {
this.txMgr = managerImpl;
@@ -196,11 +173,6 @@ public class TXStateProxyImpl implements TXStateProxy {
this.onBehalfOfClientMember = clientMember;
}
- /**
- * @param managerImpl
- * @param id
- * @param isjta
- */
public TXStateProxyImpl(TXManagerImpl managerImpl, TXId id, boolean isjta) {
this.txMgr = managerImpl;
this.txId = id;
@@ -219,16 +191,6 @@ public class TXStateProxyImpl implements TXStateProxy {
this.isJTA = isJTA;
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.internal.cache.TXStateInterface#checkJTA(java.lang.String)
- */
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.internal.cache.TXStateProxyInterface#checkJTA(java.lang.String)
- */
public void checkJTA(String errmsg) throws IllegalStateException {
if (isJTA()) {
throw new IllegalStateException(errmsg);
@@ -243,11 +205,6 @@ public class TXStateProxyImpl implements TXStateProxy {
.toLocalizedString("precommit"));
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.internal.cache.TXStateInterface#commit()
- */
public void commit() throws CommitConflictException {
boolean preserveTx = false;
try {
@@ -284,12 +241,6 @@ public class TXStateProxyImpl implements TXStateProxy {
return (TransactionException) e;
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.internal.cache.TXStateInterface#containsValueForKey(java.lang.Object,
- * org.apache.geode.internal.cache.LocalRegion)
- */
public boolean containsValueForKey(KeyInfo keyInfo, LocalRegion region) {
try {
this.operationCount++;
@@ -302,7 +253,6 @@ public class TXStateProxyImpl implements TXStateProxy {
}
private void trackBucketForTx(KeyInfo keyInfo) {
- GemFireCacheImpl cache = (GemFireCacheImpl) txMgr.getCache();
if (keyInfo.getBucketId() >= 0) {
if (logger.isDebugEnabled()) {
logger.debug("adding bucket:{} for tx:{}", keyInfo.getBucketId(), getTransactionId());
@@ -313,13 +263,6 @@ public class TXStateProxyImpl implements TXStateProxy {
}
}
- /*
- * (non-Javadoc)
- *
- * @see
- * org.apache.geode.internal.cache.TXStateInterface#destroyExistingEntry(org.apache.geode.internal
- * .cache.EntryEventImpl, boolean, java.lang.Object)
- */
public void destroyExistingEntry(EntryEventImpl event, boolean cacheWrite,
Object expectedOldValue) throws EntryNotFoundException {
try {
@@ -332,40 +275,19 @@ public class TXStateProxyImpl implements TXStateProxy {
}
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.internal.cache.TXStateInterface#getBeginTime()
- */
public long getBeginTime() {
return getRealDeal(null, null).getBeginTime();
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.internal.cache.TXStateInterface#getCache()
- */
public Cache getCache() {
return txMgr.getCache();
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.internal.cache.TXStateInterface#getChanges()
- */
public int getChanges() {
assertBootstrapped();
return getRealDeal(null, null).getChanges();
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.internal.cache.TXStateInterface#getDeserializedValue(java.lang.Object,
- * org.apache.geode.internal.cache.LocalRegion, boolean)
- */
public Object getDeserializedValue(KeyInfo keyInfo, LocalRegion localRegion, boolean updateStats,
boolean disableCopyOnRead, boolean preferCD, EntryEventImpl clientEvent,
boolean returnTombstones, boolean retainResult) {
@@ -379,12 +301,6 @@ public class TXStateProxyImpl implements TXStateProxy {
return val;
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.internal.cache.TXStateInterface#getEntry(java.lang.Object,
- * org.apache.geode.internal.cache.LocalRegion)
- */
public Entry getEntry(KeyInfo keyInfo, LocalRegion region, boolean allowTombstones) {
try {
this.operationCount++;
@@ -396,51 +312,25 @@ public class TXStateProxyImpl implements TXStateProxy {
}
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.internal.cache.TXStateInterface#getEvent()
- */
public TXEvent getEvent() {
assertBootstrapped();
return getRealDeal(null, null).getEvent();
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.internal.cache.TXStateInterface#getEvents()
- */
public List getEvents() {
assertBootstrapped();
return getRealDeal(null, null).getEvents();
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.internal.cache.TXStateInterface#getRegions()
- */
public Collection<LocalRegion> getRegions() {
assertBootstrapped();
return getRealDeal(null, null).getRegions();
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.internal.cache.TXStateInterface#getTransactionId()
- */
public TransactionId getTransactionId() {
return txId;
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.internal.cache.TXStateInterface#invalidateExistingEntry(org.apache.geode.
- * internal.cache.EntryEventImpl, boolean, boolean)
- */
public void invalidateExistingEntry(EntryEventImpl event, boolean invokeCallbacks,
boolean forceNewEntry) {
try {
@@ -453,11 +343,6 @@ public class TXStateProxyImpl implements TXStateProxy {
}
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.internal.cache.TXStateInterface#isInProgress()
- */
public boolean isInProgress() {
return inProgress;
}
@@ -467,54 +352,26 @@ public class TXStateProxyImpl implements TXStateProxy {
this.inProgress = progress;
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.internal.cache.TXStateInterface#needsLargeModCount()
- */
public boolean needsLargeModCount() {
assertBootstrapped();
return getRealDeal(null, null).needsLargeModCount();
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.internal.cache.TXStateInterface#nextModSerialNum()
- */
public int nextModSerialNum() {
assertBootstrapped();
return getRealDeal(null, null).nextModSerialNum();
}
- /*
- * (non-Javadoc)
- *
- * @see
- * org.apache.geode.internal.cache.TXStateInterface#readRegion(org.apache.geode.internal.cache.
- * LocalRegion)
- */
public TXRegionState readRegion(LocalRegion r) {
assertBootstrapped();
return getRealDeal(null, r).readRegion(r);
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.internal.cache.TXStateInterface#rmRegion(org.apache.geode.internal.cache.
- * LocalRegion)
- */
public void rmRegion(LocalRegion r) {
assertBootstrapped();
getRealDeal(null, r).rmRegion(r);
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.internal.cache.TXStateInterface#rollback()
- */
public void rollback() {
try {
getRealDeal(null, null).rollback();
@@ -526,13 +383,6 @@ public class TXStateProxyImpl implements TXStateProxy {
}
}
- /*
- * (non-Javadoc)
- *
- * @see
- * org.apache.geode.internal.cache.TXStateInterface#txPutEntry(org.apache.geode.internal.cache.
- * EntryEventImpl, boolean, boolean, boolean)
- */
public boolean txPutEntry(EntryEventImpl event, boolean ifNew, boolean requireOldValue,
boolean checkResources, Object expectedOldValue) {
try {
@@ -546,12 +396,6 @@ public class TXStateProxyImpl implements TXStateProxy {
}
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.internal.cache.TXStateInterface#txReadEntry(java.lang.Object,
- * org.apache.geode.internal.cache.LocalRegion, boolean)
- */
public TXEntryState txReadEntry(KeyInfo keyInfo, LocalRegion localRegion, boolean rememberRead,
boolean createTxEntryIfAbsent) {
try {
@@ -565,36 +409,15 @@ public class TXStateProxyImpl implements TXStateProxy {
}
}
- /*
- * (non-Javadoc)
- *
- * @see
- * org.apache.geode.internal.cache.TXStateInterface#txReadRegion(org.apache.geode.internal.cache.
- * LocalRegion)
- */
public TXRegionState txReadRegion(LocalRegion localRegion) {
assertBootstrapped();
return getRealDeal(null, localRegion).txReadRegion(localRegion);
}
- /*
- * (non-Javadoc)
- *
- * @see
- * org.apache.geode.internal.cache.TXStateInterface#txWriteRegion(org.apache.geode.internal.cache.
- * LocalRegion, java.lang.Object)
- */
public TXRegionState txWriteRegion(LocalRegion localRegion, KeyInfo entryKey) {
return getRealDeal(entryKey, localRegion).txWriteRegion(localRegion, entryKey);
}
- /*
- * (non-Javadoc)
- *
- * @see
- * org.apache.geode.internal.cache.TXStateInterface#writeRegion(org.apache.geode.internal.cache.
- * LocalRegion)
- */
public TXRegionState writeRegion(LocalRegion r) {
assertBootstrapped();
return getRealDeal(null, r).writeRegion(r);
@@ -604,11 +427,6 @@ public class TXStateProxyImpl implements TXStateProxy {
assert realDeal != null;
}
- /*
- * (non-Javadoc)
- *
- * @see javax.transaction.Synchronization#afterCompletion(int)
- */
public void afterCompletion(int status) {
assertBootstrapped();
try {
@@ -621,22 +439,11 @@ public class TXStateProxyImpl implements TXStateProxy {
}
}
- /*
- * (non-Javadoc)
- *
- * @see javax.transaction.Synchronization#beforeCompletion()
- */
public void beforeCompletion() {
assertBootstrapped();
getRealDeal(null, null).beforeCompletion();
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.internal.cache.InternalDataView#containsKey(java.lang.Object,
- * org.apache.geode.internal.cache.LocalRegion)
- */
public boolean containsKey(KeyInfo keyInfo, LocalRegion localRegion) {
try {
this.operationCount++;
@@ -648,13 +455,6 @@ public class TXStateProxyImpl implements TXStateProxy {
}
}
- /*
- * (non-Javadoc)
- *
- * @see
- * org.apache.geode.internal.cache.InternalDataView#entryCount(org.apache.geode.internal.cache.
- * LocalRegion)
- */
@edu.umd.cs.findbugs.annotations.SuppressWarnings(value = "UL_UNRELEASED_LOCK",
justification = "This method unlocks and then conditionally undoes the unlock in the finally-block. Review again at later time.")
public int entryCount(LocalRegion localRegion) {
@@ -684,13 +484,6 @@ public class TXStateProxyImpl implements TXStateProxy {
}
}
- /*
- * (non-Javadoc)
- *
- * @see
- * org.apache.geode.internal.cache.InternalDataView#findObject(org.apache.geode.internal.cache.
- * LocalRegion, java.lang.Object, java.lang.Object, boolean, boolean, java.lang.Object)
- */
public Object findObject(KeyInfo key, LocalRegion r, boolean isCreate, boolean generateCallbacks,
Object value, boolean disableCopyOnRead, boolean preferCD,
ClientProxyMembershipID requestingClient, EntryEventImpl clientEvent,
@@ -706,13 +499,6 @@ public class TXStateProxyImpl implements TXStateProxy {
}
}
- /*
- * (non-Javadoc)
- *
- * @see
- * org.apache.geode.internal.cache.InternalDataView#getAdditionalKeysForIterator(org.apache.geode.
- * internal.cache.LocalRegion)
- */
public Set getAdditionalKeysForIterator(LocalRegion currRgn) {
if (this.realDeal == null) {
return null;
@@ -720,13 +506,6 @@ public class TXStateProxyImpl implements TXStateProxy {
return getRealDeal(null, currRgn).getAdditionalKeysForIterator(currRgn);
}
- /*
- * (non-Javadoc)
- *
- * @see
- * org.apache.geode.internal.cache.InternalDataView#getEntryForIterator(org.apache.geode.internal.
- * cache.LocalRegion, java.lang.Object, boolean)
- */
public Object getEntryForIterator(KeyInfo key, LocalRegion currRgn, boolean rememberReads,
boolean allowTombstones) {
boolean resetTxState = this.realDeal == null;
@@ -745,15 +524,8 @@ public class TXStateProxyImpl implements TXStateProxy {
getTxMgr().internalResume(txp);
}
}
-
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.internal.cache.InternalDataView#getKeyForIterator(java.lang.Object,
- * org.apache.geode.internal.cache.LocalRegion, boolean)
- */
public Object getKeyForIterator(KeyInfo keyInfo, LocalRegion currRgn, boolean rememberReads,
boolean allowTombstones) {
boolean resetTxState = this.realDeal == null;
@@ -775,33 +547,16 @@ public class TXStateProxyImpl implements TXStateProxy {
}
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.internal.cache.InternalDataView#getValueInVM(java.lang.Object,
- * org.apache.geode.internal.cache.LocalRegion, boolean)
- */
public Object getValueInVM(KeyInfo keyInfo, LocalRegion localRegion, boolean rememberRead) {
this.operationCount++;
return getRealDeal(keyInfo, localRegion).getValueInVM(keyInfo, localRegion, rememberRead);
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.internal.cache.InternalDataView#isDeferredStats()
- */
public boolean isDeferredStats() {
assertBootstrapped();
return getRealDeal(null, null).isDeferredStats();
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.internal.cache.InternalDataView#putEntry(org.apache.geode.internal.cache.
- * EntryEventImpl, boolean, boolean, java.lang.Object, boolean, long, boolean)
- */
public boolean putEntry(EntryEventImpl event, boolean ifNew, boolean ifOld,
Object expectedOldValue, boolean requireOldValue, long lastModified,
boolean overwriteDestroyed) {
@@ -816,34 +571,14 @@ public class TXStateProxyImpl implements TXStateProxy {
}
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.internal.cache.TXStateInterface#isInProgressAndSameAs(org.apache.geode.
- * internal.cache.TXStateInterface)
- */
public boolean isInProgressAndSameAs(TXStateInterface otherState) {
return isInProgress() && otherState == this;
}
- /*
- * (non-Javadoc)
- *
- * @see
- * org.apache.geode.internal.cache.TXStateProxy#setLocalTXState(org.apache.geode.internal.cache.
- * TXState)
- */
public void setLocalTXState(TXStateInterface state) {
this.realDeal = state;
}
- /*
- * (non-Javadoc)
- *
- * @see
- * org.apache.geode.internal.cache.InternalDataView#getSerializedValue(org.apache.geode.internal.
- * cache.LocalRegion, java.lang.Object, java.lang.Object)
- */
public Object getSerializedValue(LocalRegion localRegion, KeyInfo key, boolean doNotLockEntry,
ClientProxyMembershipID requestingClient, EntryEventImpl clientEvent,
boolean returnTombstones) throws DataLocationException {
@@ -852,13 +587,6 @@ public class TXStateProxyImpl implements TXStateProxy {
requestingClient, clientEvent, returnTombstones);
}
- /*
- * (non-Javadoc)
- *
- * @see
- * org.apache.geode.internal.cache.InternalDataView#putEntryOnRemote(org.apache.geode.internal.
- * cache.EntryEventImpl, boolean, boolean, java.lang.Object, boolean, long, boolean)
- */
public boolean putEntryOnRemote(EntryEventImpl event, boolean ifNew, boolean ifOld,
Object expectedOldValue, boolean requireOldValue, long lastModified,
boolean overwriteDestroyed) throws DataLocationException {
@@ -873,12 +601,6 @@ public class TXStateProxyImpl implements TXStateProxy {
return getRealDeal(null, null).isFireCallbacks();
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.internal.cache.InternalDataView#destroyOnRemote(java.lang.Integer,
- * org.apache.geode.internal.cache.EntryEventImpl, java.lang.Object)
- */
public void destroyOnRemote(EntryEventImpl event, boolean cacheWrite, Object expectedOldValue)
throws DataLocationException {
this.operationCount++;
@@ -887,13 +609,6 @@ public class TXStateProxyImpl implements TXStateProxy {
tx.destroyOnRemote(event, cacheWrite, expectedOldValue);
}
- /*
- * (non-Javadoc)
- *
- * @see
- * org.apache.geode.internal.cache.InternalDataView#invalidateOnRemote(org.apache.geode.internal.
- * cache.EntryEventImpl, boolean, boolean)
- */
public void invalidateOnRemote(EntryEventImpl event, boolean invokeCallbacks,
boolean forceNewEntry) throws DataLocationException {
this.operationCount++;
@@ -919,13 +634,6 @@ public class TXStateProxyImpl implements TXStateProxy {
LocalizedStrings.TXState_REGION_CLEAR_NOT_SUPPORTED_IN_A_TRANSACTION.toLocalizedString());
}
- /*
- * (non-Javadoc)
- *
- * @see
- * org.apache.geode.internal.cache.InternalDataView#getBucketKeys(org.apache.geode.internal.cache.
- * LocalRegion, int)
- */
public Set getBucketKeys(LocalRegion localRegion, int bucketId, boolean allowTombstones) {
// if this the first operation in a transaction, reset txState
boolean resetTxState = this.realDeal == null;
@@ -945,12 +653,6 @@ public class TXStateProxyImpl implements TXStateProxy {
}
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.internal.cache.InternalDataView#getEntryOnRemote(java.lang.Object,
- * org.apache.geode.internal.cache.LocalRegion)
- */
public Entry getEntryOnRemote(KeyInfo keyInfo, LocalRegion localRegion, boolean allowTombstones)
throws DataLocationException {
this.operationCount++;
@@ -963,33 +665,15 @@ public class TXStateProxyImpl implements TXStateProxy {
getRealDeal(null, null);
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.internal.cache.TXStateProxy#getTarget()
- */
public DistributedMember getTarget() {
return this.target;
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.internal.cache.TXStateProxy#setTarget(org.apache.geode.distributed.
- * DistributedMember)
- */
public void setTarget(DistributedMember target) {
assert this.target == null;
getRealDeal(target);
}
- /*
- * (non-Javadoc)
- *
- * @see
- * org.apache.geode.internal.cache.InternalDataView#getRegionKeysForIteration(org.apache.geode.
- * internal.cache.LocalRegion)
- */
public Collection<?> getRegionKeysForIteration(LocalRegion currRegion) {
if (currRegion.isUsedForPartitionedRegionBucket()) {
return currRegion.getRegionKeysForIteration();
@@ -998,20 +682,10 @@ public class TXStateProxyImpl implements TXStateProxy {
}
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.internal.cache.TXStateProxy#isCommitRequestedByOwner()
- */
public boolean isCommitOnBehalfOfRemoteStub() {
return this.commitRequestedByOwner;
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.internal.cache.TXStateProxy#setCommitRequestedByOwner()
- */
public boolean setCommitOnBehalfOfRemoteStub(boolean requestedByOwner) {
return this.commitRequestedByOwner = requestedByOwner;
}
@@ -1039,7 +713,8 @@ public class TXStateProxyImpl implements TXStateProxy {
public String toString() {
StringBuilder builder = new StringBuilder();
builder.append("TXStateProxyImpl@").append(System.identityHashCode(this)).append(" txId:")
- .append(this.txId).append(" realDeal:" + this.realDeal).append(" isJTA:").append(isJTA);
+ .append(this.txId).append(" realDeal:").append(this.realDeal).append(" isJTA:")
+ .append(isJTA);
return builder.toString();
}
@@ -1051,7 +726,6 @@ public class TXStateProxyImpl implements TXStateProxy {
}
}
-
public boolean isMemberIdForwardingRequired() {
if (this.realDeal == null) {
return false;
@@ -1060,7 +734,6 @@ public class TXStateProxyImpl implements TXStateProxy {
}
}
-
public TXCommitMessage getCommitMessage() {
if (this.realDeal == null) {
return null;
@@ -1069,7 +742,6 @@ public class TXStateProxyImpl implements TXStateProxy {
}
}
-
public void postPutAll(DistributedPutAllOperation putallOp, VersionedObjectList successfulPuts,
LocalRegion region) {
if (putallOp.putAllData.length == 0) {
@@ -1159,7 +831,6 @@ public class TXStateProxyImpl implements TXStateProxy {
// Do nothing. Not applicable for transactions.
}
-
public void close() {
if (this.realDeal != null) {
this.realDeal.close();