| /* |
| * Licensed to the Apache Software Foundation (ASF) under one or more contributor license |
| * agreements. See the NOTICE file distributed with this work for additional information regarding |
| * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance with the License. You may obtain a |
| * copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software distributed under the License |
| * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express |
| * or implied. See the License for the specific language governing permissions and limitations under |
| * the License. |
| */ |
| package org.apache.geode.internal.cache; |
| |
| import static org.apache.geode.internal.cache.LocalRegion.InitializationLevel.BEFORE_INITIAL_IMAGE; |
| |
| import java.io.DataInput; |
| import java.io.DataOutput; |
| import java.io.IOException; |
| import java.util.ArrayList; |
| import java.util.Collections; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.IdentityHashMap; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.ServiceConfigurationError; |
| import java.util.Set; |
| |
| import org.apache.logging.log4j.Logger; |
| |
| import org.apache.geode.CancelException; |
| import org.apache.geode.DataSerializer; |
| import org.apache.geode.SystemFailure; |
| import org.apache.geode.annotations.Immutable; |
| import org.apache.geode.annotations.internal.MakeNotStatic; |
| import org.apache.geode.cache.Cache; |
| import org.apache.geode.cache.CacheClosedException; |
| import org.apache.geode.cache.CacheRuntimeException; |
| import org.apache.geode.cache.CommitDistributionException; |
| import org.apache.geode.cache.CommitIncompleteException; |
| import org.apache.geode.cache.Operation; |
| import org.apache.geode.cache.RegionDestroyedException; |
| import org.apache.geode.cache.RegionDistributionException; |
| import org.apache.geode.cache.TransactionId; |
| import org.apache.geode.cache.TransactionListener; |
| import org.apache.geode.distributed.internal.ClusterDistributionManager; |
| import org.apache.geode.distributed.internal.DistributionManager; |
| import org.apache.geode.distributed.internal.DistributionMessage; |
| import org.apache.geode.distributed.internal.MembershipListener; |
| import org.apache.geode.distributed.internal.MessageWithReply; |
| import org.apache.geode.distributed.internal.PooledDistributionMessage; |
| import org.apache.geode.distributed.internal.ReliableReplyProcessor21; |
| import org.apache.geode.distributed.internal.ReplyException; |
| import org.apache.geode.distributed.internal.ReplyMessage; |
| import org.apache.geode.distributed.internal.ReplyProcessor21; |
| import org.apache.geode.distributed.internal.membership.InternalDistributedMember; |
| import org.apache.geode.internal.Assert; |
| import org.apache.geode.internal.HeapDataOutputStream; |
| import org.apache.geode.internal.InternalDataSerializer; |
| import org.apache.geode.internal.cache.LocalRegion.InitializationLevel; |
| 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; |
| import org.apache.geode.internal.cache.versions.VersionTag; |
| import org.apache.geode.internal.offheap.annotations.Released; |
| import org.apache.geode.internal.serialization.DeserializationContext; |
| import org.apache.geode.internal.serialization.KnownVersion; |
| import org.apache.geode.internal.serialization.SerializationContext; |
| import org.apache.geode.internal.serialization.StaticSerialization; |
| import org.apache.geode.logging.internal.executors.LoggingThread; |
| import org.apache.geode.logging.internal.log4j.api.LogService; |
| |
| /** |
| * TXCommitMessage is the message that contains all the information that needs to be distributed, on |
| * commit, to other cache members. |
| * |
| * @since GemFire 4.0 |
| */ |
| public class TXCommitMessage extends PooledDistributionMessage |
| implements MembershipListener, MessageWithReply { |
| |
| private static final Logger logger = LogService.getLogger(); |
| |
| // Keep a 60 second history @ an estimated 1092 transactions/second ~= 16^4 |
| @MakeNotStatic |
| protected static final TXFarSideCMTracker txTracker = new TXFarSideCMTracker((60 * 1092)); |
| |
| private ArrayList<RegionCommit> regions; // list of RegionCommit instances |
| protected TXId txIdent; |
| protected int processorId; // 0 unless needsAck is true |
| protected TXLockIdImpl lockId; |
| protected HashSet<InternalDistributedMember> farSiders; |
| protected transient DistributionManager dm; // Used on the sending side of this message |
| private transient int sequenceNum = 0; |
| |
| // 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; |
| private transient boolean isProcessing; |
| private transient boolean dontProcess; |
| private transient boolean departureNoticed = false; |
| private transient boolean lockNeedsUpdate = false; |
| private transient boolean ackRequired = true; |
| /** |
| * List of operations to do when processing this tx. Valid on farside only. |
| */ |
| protected transient ArrayList<RegionCommit.FarSideEntryOp> farSideEntryOps; |
| private byte[] farsideBaseMembershipId; // only available on farside |
| private long farsideBaseThreadId; // only available on farside |
| private long farsideBaseSequenceId; // only available on farside |
| |
| /** |
| * (Nearside) true of any regions in this TX have required roles |
| */ |
| private transient boolean hasReliableRegions = false; |
| |
| /** |
| * Set of all caching exceptions produced while processing this tx |
| */ |
| private transient Set<Exception> processingExceptions = Collections.emptySet(); |
| |
| private ClientProxyMembershipID bridgeContext = null; |
| |
| /** |
| * Version of the client that this TXCommitMessage is being sent to. Used for backwards |
| * compatibility |
| */ |
| private transient KnownVersion clientVersion; |
| |
| /** |
| * A token to be put in TXManagerImpl#failoverMap to represent a CommitConflictException while |
| * committing a transaction |
| */ |
| @Immutable |
| public static final TXCommitMessage CMT_CONFLICT_MSG = new TXCommitMessage(); |
| /** |
| * A token to be put in TXManagerImpl#failoverMap to represent a |
| * TransactionDataNodeHasDepartedException |
| * while committing a transaction |
| */ |
| @Immutable |
| public static final TXCommitMessage REBALANCE_MSG = new TXCommitMessage(); |
| /** |
| * A token to be put in TXManagerImpl#failoverMap to represent an exception while committing a |
| * transaction |
| */ |
| @Immutable |
| public static final TXCommitMessage EXCEPTION_MSG = new TXCommitMessage(); |
| /** |
| * A token to be put in TXManagerImpl#failoverMap to represent a rolled back transaction |
| */ |
| @Immutable |
| public static final TXCommitMessage ROLLBACK_MSG = new TXCommitMessage(); |
| |
| public TXCommitMessage(TXId txIdent, DistributionManager dm, TXState txState) { |
| this.dm = dm; |
| this.txIdent = txIdent; |
| lockId = null; |
| regions = null; |
| this.txState = txState; |
| wasProcessed = false; |
| isProcessing = false; |
| dontProcess = false; |
| farSiders = null; |
| bridgeContext = txState.bridgeContext; |
| } |
| |
| public TXCommitMessage() { |
| // zero arg constructor for DataSerializer |
| } |
| |
| public static TXFarSideCMTracker getTracker() { |
| return TXCommitMessage.txTracker; |
| } |
| |
| /** |
| * Create and return an eventId given its offset. |
| * |
| * @since GemFire 5.7 |
| */ |
| protected EventID getEventId(int eventOffset) { |
| return new EventID(farsideBaseMembershipId, farsideBaseThreadId, |
| farsideBaseSequenceId + eventOffset); |
| } |
| |
| /** |
| * 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. |
| */ |
| public static TXCommitMessage waitForMessage(Object id, DistributionManager dm) { |
| TXFarSideCMTracker map = getTracker(); |
| return map.waitForMessage(id, dm); |
| } |
| |
| void startRegion(InternalRegion r, int maxSize) { |
| currentRegion = new RegionCommit(this, r, maxSize); |
| if (r.requiresReliabilityCheck()) { |
| hasReliableRegions = true; |
| } |
| } |
| |
| void finishRegion(Set<InternalDistributedMember> s) { |
| // make sure we have some changes and someone to send them to |
| if (!currentRegion.isEmpty() && s != null && !s.isEmpty()) { |
| // Get the persistent ids for the current region and save them |
| currentRegion.persistentIds = getPersistentIds(currentRegion.internalRegion); |
| |
| if (msgMap == null) { |
| msgMap = new HashMap<>(); |
| } |
| { |
| RegionCommitList newRCL = null; |
| for (final InternalDistributedMember recipient : s) { |
| if (!dm.getDistributionManagerIds().contains(recipient)) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("Skipping member {} due to dist list absence", recipient); |
| } |
| // skip this member since the dm no longer knows about it |
| continue; |
| } |
| RegionCommitList rcl = msgMap.get(recipient); |
| if (rcl == null) { |
| if (newRCL == null) { |
| rcl = new RegionCommitList(); |
| rcl.add(currentRegion); |
| newRCL = rcl; |
| } else { |
| rcl = newRCL; |
| } |
| msgMap.put(recipient, rcl); |
| } else if (rcl.get(rcl.size() - 1) != currentRegion) { |
| rcl.add(currentRegion); |
| } |
| } |
| } |
| |
| // Now deal with each existing recipient that does not care |
| // about this region |
| for (final Map.Entry<InternalDistributedMember, RegionCommitList> me : msgMap.entrySet()) { |
| if (!s.contains(me.getKey())) { |
| RegionCommitList rcl = me.getValue(); |
| RegionCommitList trimmedRcl = rcl.trim(currentRegion); |
| if (trimmedRcl != rcl) { |
| me.setValue(trimmedRcl); |
| } |
| } |
| } |
| } |
| currentRegion = null; |
| } |
| |
| private Map<InternalDistributedMember, PersistentMemberID> getPersistentIds(InternalRegion r) { |
| if (r instanceof DistributedRegion) { |
| return ((CacheDistributionAdvisee) r).getCacheDistributionAdvisor().advisePersistentMembers(); |
| } else { |
| return Collections.emptyMap(); |
| } |
| } |
| |
| void finishRegionComplete() { |
| // make sure we have some changes and someone to send them to |
| if (!currentRegion.isEmpty()) { |
| { |
| if (regions == null) { |
| regions = new RegionCommitList(); |
| } |
| regions.add(currentRegion); |
| } |
| } |
| currentRegion = null; |
| } |
| |
| Map<DistributedRegion, Long> 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, version); |
| } |
| |
| protected void releaseViewVersions() { |
| RuntimeException rte = null; |
| for (final Map.Entry<DistributedRegion, Long> e : viewVersions.entrySet()) { |
| final DistributedRegion dr = e.getKey(); |
| final Long viewVersion = e.getValue(); |
| // need to continue the iteration if one of the regions is destroyed |
| // since others may still be okay |
| try { |
| dr.getDistributionAdvisor().endOperation(viewVersion); |
| } catch (RuntimeException ex) { |
| rte = ex; |
| } |
| } |
| if (rte != null) { |
| throw rte; |
| } |
| } |
| |
| private boolean isEmpty() { |
| return msgMap == null || msgMap.isEmpty(); |
| } |
| |
| void addOp(InternalRegion r, Object key, TXEntryState entry) { |
| currentRegion.addOp(key, entry); |
| } |
| |
| void send(TXLockId lockId) { |
| if (isEmpty()) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("empty transaction - nothing to distribute"); |
| } |
| return; |
| } |
| Assert.assertTrue(txState != null, "Send must have transaction state."); |
| this.lockId = (TXLockIdImpl) lockId; |
| updateLockMembers(); |
| |
| // Map of RegionCommitList keys to Sets of receivers |
| IdentityHashMap<RegionCommitList, Set<InternalDistributedMember>> distMap = |
| new IdentityHashMap<>(); |
| HashSet<InternalDistributedMember> ackReceivers = null; |
| for (final Map.Entry<InternalDistributedMember, RegionCommitList> entry : msgMap.entrySet()) { |
| final RegionCommitList rcl = entry.getValue(); |
| if (rcl.getNeedsAck()) { |
| if (ackReceivers == null) { |
| ackReceivers = new HashSet<>(); |
| } |
| ackReceivers.add(entry.getKey()); |
| } |
| final Set<InternalDistributedMember> receivers = |
| distMap.computeIfAbsent(rcl, k -> new HashSet<>()); |
| receivers.add(entry.getKey()); |
| } |
| |
| CommitReplyProcessor processor = null; |
| { |
| if (ackReceivers != null) { |
| processor = new CommitReplyProcessor(dm, ackReceivers, msgMap); |
| if (ackReceivers.size() > 1) { |
| farSiders = ackReceivers; |
| } |
| processor.enableSevereAlertProcessing(); |
| } |
| { |
| for (final Map.Entry<RegionCommitList, Set<InternalDistributedMember>> me : distMap |
| .entrySet()) { |
| RegionCommitList rcl = me.getKey(); |
| Set<InternalDistributedMember> recipients = me.getValue(); |
| // now remove from the recipients any recipients that the dm no |
| // longer knows about |
| recipients.retainAll(dm.getDistributionManagerIds()); |
| if (!recipients.isEmpty()) { |
| if (txState.internalDuringIndividualSend != null) { |
| // Run in test mode, splitting out individual recipients, |
| // so we can control who gets what |
| for (final InternalDistributedMember recipient : recipients) { |
| txState.internalDuringIndividualSend.run(); |
| setRecipientsSendData(Collections.singleton(recipient), processor, rcl); |
| } |
| } else { |
| // Run in normal mode sending to multiple recipients in one shot |
| setRecipientsSendData(recipients, processor, rcl); |
| } |
| } |
| } |
| } |
| } |
| if (txState.internalAfterIndividualSend != null) { |
| txState.internalAfterIndividualSend.run(); |
| } |
| |
| if (processor != null) { |
| // Send the CommitProcessMessage |
| final CommitProcessMessage cpMsg; |
| if (this.lockId != null) { |
| cpMsg = new CommitProcessForLockIdMessage(this.lockId); |
| } else { |
| cpMsg = new CommitProcessForTXIdMessage(txIdent); |
| } |
| if (txState.internalDuringIndividualCommitProcess != null) { |
| // Run in test mode |
| for (final InternalDistributedMember ackReceiver : ackReceivers) { |
| txState.internalDuringIndividualCommitProcess.run(); |
| cpMsg.setRecipients(Collections.singleton(ackReceiver)); |
| dm.putOutgoing(cpMsg); |
| cpMsg.resetRecipients(); |
| } |
| } else { |
| // Run in normal mode |
| cpMsg.setRecipients(ackReceivers); |
| dm.putOutgoing(cpMsg); |
| } |
| |
| if (txState.internalAfterIndividualCommitProcess != null) { |
| // Testing callback |
| txState.internalAfterIndividualCommitProcess.run(); |
| } |
| |
| // for() loop removed for bug 36983 - you can't loop on waitForReplies() |
| dm.getCancelCriterion().checkCancelInProgress(null); |
| processor.waitForCommitCompletion(); |
| dm.getStats().incCommitWaits(); |
| } |
| if (hasReliableRegions) { |
| checkDistributionReliability(distMap, processor); |
| } |
| } |
| |
| @Override |
| public boolean containsRegionContentChange() { |
| return true; |
| } |
| |
| /** |
| * Checks reliable regions and throws CommitDistributionException if any required roles may not |
| * have received the commit message. |
| * |
| * @param distMap map of RegionCommitList keys to Sets of receivers |
| * @param processor the reply processor |
| * @throws CommitDistributionException if any required roles may not have received the commit |
| * message |
| */ |
| private void checkDistributionReliability( |
| IdentityHashMap<RegionCommitList, Set<InternalDistributedMember>> distMap, |
| CommitReplyProcessor processor) { |
| // key=RegionCommit, value=Set of recipients |
| Map<RegionCommit, Set<InternalDistributedMember>> regionToRecipients = new IdentityHashMap<>(); |
| |
| // build up the keys in regionToRecipients and add all receivers |
| for (final Map.Entry<RegionCommitList, Set<InternalDistributedMember>> me : distMap |
| .entrySet()) { |
| RegionCommitList rcl = me.getKey(); |
| Set<InternalDistributedMember> recipients = me.getValue(); |
| |
| for (RegionCommit rc : rcl) { |
| // skip region if no required roles |
| if (!rc.internalRegion.requiresReliabilityCheck()) { |
| continue; |
| } |
| |
| final Set<InternalDistributedMember> recipientsForRegion = |
| regionToRecipients.computeIfAbsent(rc, k -> new HashSet<>()); |
| // get the receiver Set for rcl and perform addAll |
| if (recipients != null) { |
| recipientsForRegion.addAll(recipients); |
| } |
| } |
| } |
| |
| Set<InternalDistributedMember> cacheClosedMembers = |
| (processor == null) ? Collections.emptySet() : processor.getCacheClosedMembers(); |
| Set<InternalDistributedMember> departedMembers = |
| (processor == null) ? Collections.emptySet() : processor.getDepartedMembers(); |
| |
| // check reliability on each region |
| @SuppressWarnings("deprecation") |
| Set<RegionDistributionException> regionDistributionExceptions = Collections.emptySet(); |
| Set<String> failedRegionNames = Collections.emptySet(); |
| for (final Map.Entry<RegionCommit, Set<InternalDistributedMember>> me : regionToRecipients |
| .entrySet()) { |
| final RegionCommit rc = me.getKey(); |
| final Set<InternalDistributedMember> successfulRecipients = new HashSet<>(msgMap.keySet()); |
| successfulRecipients.removeAll(departedMembers); |
| |
| // remove members who destroyed that region or closed their cache |
| Set<InternalDistributedMember> regionDestroyedMembers = |
| (processor == null) ? Collections.emptySet() |
| : processor.getRegionDestroyedMembers(rc.internalRegion.getFullPath()); |
| |
| successfulRecipients.removeAll(cacheClosedMembers); |
| successfulRecipients.removeAll(regionDestroyedMembers); |
| |
| try { |
| rc.internalRegion.handleReliableDistribution(successfulRecipients); |
| } catch (@SuppressWarnings("deprecation") RegionDistributionException e) { |
| if (regionDistributionExceptions.isEmpty()) { |
| regionDistributionExceptions = new HashSet<>(); |
| failedRegionNames = new HashSet<>(); |
| } |
| regionDistributionExceptions.add(e); |
| failedRegionNames.add(rc.internalRegion.getFullPath()); |
| } |
| } |
| |
| if (!regionDistributionExceptions.isEmpty()) { |
| throw new CommitDistributionException( |
| String.format( |
| "These regions experienced reliability failure during distribution of the operation: %s", |
| failedRegionNames), |
| regionDistributionExceptions); |
| } |
| } |
| |
| /** |
| * Helper method for send |
| */ |
| private void setRecipientsSendData(Set<InternalDistributedMember> recipients, |
| ReplyProcessor21 processor, |
| RegionCommitList rcl) { |
| setRecipients(recipients); |
| regions = rcl; |
| if (rcl.getNeedsAck()) { |
| processorId = processor.getProcessorId(); |
| } else { |
| processorId = 0; |
| } |
| dm.getStats().incSentCommitMessages(1L); |
| sequenceNum++; |
| dm.putOutgoing(this); |
| resetRecipients(); |
| } |
| |
| @Override |
| protected void process(ClusterDistributionManager dm) { |
| this.dm = dm; |
| // Remove this node from the set of recipients |
| if (farSiders != null) { |
| farSiders.remove(dm.getId()); |
| } |
| |
| if (processorId != 0) { |
| TXLockService.createDTLS(this.dm.getSystem()); // fix bug 38843; no-op if already created |
| synchronized (this) { |
| // Handle potential origin departure |
| this.dm.addMembershipListener(this); |
| // Assume ACK mode, defer processing until we receive a |
| // CommitProcess message |
| if (logger.isDebugEnabled()) { |
| final Object key = getTrackerKey(); |
| logger.debug("Adding key:{} class{} to tracker list", key, key.getClass().getName()); |
| } |
| txTracker.add(this); |
| } |
| if (!this.dm.getDistributionManagerIds().contains(getSender())) { |
| memberDeparted(this.dm, getSender(), false /* don't care */); |
| } |
| |
| } else { |
| basicProcess(); |
| } |
| } |
| |
| /** |
| * Adds an entry op for this tx to do on the far side |
| */ |
| void addFarSideEntryOp(RegionCommit.FarSideEntryOp entryOp) { |
| farSideEntryOps.add(entryOp); |
| } |
| |
| protected void addProcessingException(Exception e) { |
| // clear all previous exceptions if e is a CacheClosedException |
| if (processingExceptions.isEmpty() || e instanceof CancelException) { |
| processingExceptions = new HashSet<>(); |
| } |
| processingExceptions.add(e); |
| } |
| |
| public void setDM(DistributionManager dm) { |
| this.dm = dm; |
| } |
| |
| public void basicProcess() { |
| final DistributionManager dm = this.dm; |
| |
| synchronized (this) { |
| if (isProcessing()) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("TXCommitMessage {} is already in process, returning", this); |
| } |
| return; |
| } else { |
| setIsProcessing(true); |
| } |
| } |
| |
| if (logger.isDebugEnabled()) { |
| logger.debug("begin processing TXCommitMessage for {}", txIdent); |
| } |
| final InitializationLevel oldLevel = |
| LocalRegion.setThreadInitLevelRequirement(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 { |
| TXRmtEvent txEvent = null; |
| final Cache cache = dm.getExistingCache(); |
| if (cache == null) { |
| addProcessingException(new CacheClosedException()); |
| // return ... this cache is closed so we can't do anything. |
| return; |
| } |
| final TransactionListener[] tls = cache.getCacheTransactionManager().getListeners(); |
| if (tls.length > 0) { |
| txEvent = new TXRmtEvent(txIdent, cache); |
| } |
| try { |
| // Pre-process each Region in the tx |
| try { |
| Iterator<RegionCommit> it = regions.iterator(); |
| while (it.hasNext()) { |
| boolean failedBeginProcess = true; |
| RegionCommit rc = it.next(); |
| try { |
| failedBeginProcess = !rc.beginProcess(dm, txIdent, txEvent); |
| } catch (CacheRuntimeException problem) { |
| processCacheRuntimeException(problem); |
| } finally { |
| if (failedBeginProcess) { |
| rc.internalRegion = null; // Cause related FarSideEntryOps to skip processing |
| it.remove(); // Skip endProcessing as well |
| } |
| } |
| } |
| basicProcessOps(); |
| } finally { |
| // post-process each Region in the tx |
| Iterator<RegionCommit> it = regions.iterator(); |
| while (it.hasNext()) { |
| try { |
| RegionCommit rc = it.next(); |
| rc.endProcess(); |
| if (rc.isForceFireEvent(dm)) { |
| forceListener = true; |
| } |
| } catch (CacheRuntimeException problem) { |
| processCacheRuntimeException(problem); |
| } |
| } |
| } |
| |
| /* |
| * We need to make sure that we should fire a TX afterCommit event. |
| */ |
| boolean internalEvent = (txEvent != null && txEvent.hasOnlyInternalEvents()); |
| if (!disableListeners && !internalEvent |
| && (forceListener || (txEvent != null && !txEvent.isEmpty()))) { |
| for (final TransactionListener tl : tls) { |
| try { |
| tl.afterCommit(txEvent); |
| } catch (VirtualMachineError err) { |
| SystemFailure.initiateFailure(err); |
| // If this ever returns, rethrow the error. We're poisoned |
| // now, so don't let this thread continue. |
| throw err; |
| } catch (Throwable t) { |
| // Whenever you catch Error or Throwable, you must also |
| // catch VirtualMachineError (see above). However, there is |
| // _still_ a possibility that you are dealing with a cascading |
| // error condition, so you also need to check to see if the JVM |
| // is still usable: |
| SystemFailure.checkFailure(); |
| logger.error("Exception occurred in TransactionListener", |
| t); |
| } |
| } |
| } |
| } catch (CancelException e) { |
| processCacheRuntimeException(e); |
| } finally { |
| if (txEvent != null) { |
| txEvent.freeOffHeapResources(); |
| } |
| } |
| } finally { |
| LocalRegion.setThreadInitLevelRequirement(oldLevel); |
| if (isAckRequired()) { |
| ack(); |
| } |
| if (!dm.getExistingCache().isClient() && bridgeContext != null) { |
| getTracker().saveTXForClientFailover(txIdent, this); |
| } |
| if (logger.isDebugEnabled()) { |
| logger.debug("completed processing TXCommitMessage for {}", txIdent); |
| } |
| } |
| } |
| |
| public void basicProcessOps() { |
| List<EntryEventImpl> pendingCallbacks = new ArrayList<>(farSideEntryOps.size()); |
| farSideEntryOps.sort(null); |
| Iterator<RegionCommit.FarSideEntryOp> it = farSideEntryOps.iterator(); |
| while (it.hasNext()) { |
| try { |
| RegionCommit.FarSideEntryOp entryOp = it.next(); |
| entryOp.process(pendingCallbacks); |
| } catch (CacheRuntimeException problem) { |
| processCacheRuntimeException(problem); |
| } catch (Exception e) { |
| addProcessingException(e); |
| } |
| } |
| firePendingCallbacks(pendingCallbacks); |
| } |
| |
| void firePendingCallbacks(List<EntryEventImpl> callbacks) { |
| boolean isConfigError = false; |
| EntryEventImpl lastTransactionEvent = null; |
| try { |
| lastTransactionEvent = getLastTransactionEvent(callbacks); |
| } catch (ServiceConfigurationError ex) { |
| logger.error(ex.getMessage()); |
| isConfigError = true; |
| } |
| |
| for (EntryEventImpl ee : callbacks) { |
| boolean isLastTransactionEvent = TXLastEventInTransactionUtils |
| .isLastTransactionEvent(isConfigError, lastTransactionEvent, ee); |
| try { |
| if (ee.getOperation().isDestroy()) { |
| ee.getRegion().invokeTXCallbacks(EnumListenerEvent.AFTER_DESTROY, ee, true, |
| isLastTransactionEvent); |
| } else if (ee.getOperation().isInvalidate()) { |
| ee.getRegion().invokeTXCallbacks(EnumListenerEvent.AFTER_INVALIDATE, ee, true, |
| isLastTransactionEvent); |
| } else if (ee.getOperation().isCreate()) { |
| ee.getRegion().invokeTXCallbacks(EnumListenerEvent.AFTER_CREATE, ee, true, |
| isLastTransactionEvent); |
| } else { |
| if (!ee.hasNewValue()) { // GEODE-8964, fixes GII and TX create conflict that |
| ee.getRegion(). // produces an Update with null value |
| invokeTXCallbacks(EnumListenerEvent.AFTER_CREATE, ee, true, isLastTransactionEvent); |
| } else { |
| ee.getRegion().invokeTXCallbacks(EnumListenerEvent.AFTER_UPDATE, ee, true, |
| isLastTransactionEvent); |
| } |
| } |
| } finally { |
| ee.release(); |
| } |
| } |
| } |
| |
| EntryEventImpl getLastTransactionEvent(List<EntryEventImpl> callbacks) { |
| return TXLastEventInTransactionUtils.getLastTransactionEventInGroupedTxForWANSender(callbacks, |
| dm.getCache()); |
| } |
| |
| protected void processCacheRuntimeException(CacheRuntimeException problem) { |
| if (problem instanceof RegionDestroyedException) { // catch RegionDestroyedException |
| addProcessingException(problem); |
| } else if (problem instanceof CancelException) { // catch CacheClosedException |
| addProcessingException(problem); |
| throw problem; |
| } else { // catch CacheRuntimeException |
| addProcessingException(problem); |
| logger.error( |
| "Transaction message {} from sender {} failed processing, unknown transaction state: {}", |
| new Object[] {this, getSender(), problem}); |
| } |
| } |
| |
| private void ack() { |
| if (processorId != 0) { |
| CommitReplyException replyEx = null; |
| if (!processingExceptions.isEmpty()) { |
| replyEx = new CommitReplyException( |
| String.format("Commit operation generated one or more exceptions from %s", |
| getSender()), |
| processingExceptions); |
| } |
| ReplyMessage.send(getSender(), processorId, replyEx, dm); |
| } |
| } |
| |
| @Override |
| public int getDSFID() { |
| return TX_COMMIT_MESSAGE; |
| } |
| |
| @Override |
| public void fromData(DataInput in, DeserializationContext context) |
| throws IOException, ClassNotFoundException { |
| int pId = in.readInt(); |
| |
| if (isAckRequired()) { |
| processorId = pId; |
| ReplyProcessor21.setMessageRPId(processorId); |
| } else { |
| processorId = -1; |
| } |
| |
| txIdent = TXId.createFromData(in); |
| if (in.readBoolean()) { |
| lockId = TXLockIdImpl.createFromData(in); |
| } |
| int totalMaxSize = in.readInt(); |
| |
| farsideBaseMembershipId = DataSerializer.readByteArray(in); |
| farsideBaseThreadId = in.readLong(); |
| farsideBaseSequenceId = in.readLong(); |
| |
| needsLargeModCount = in.readBoolean(); |
| |
| final boolean hasShadowKeys = hasFlagsField(in) ? in.readBoolean() : useShadowKey(); |
| |
| int regionsSize = in.readInt(); |
| regions = new ArrayList<>(regionsSize); |
| farSideEntryOps = new ArrayList<>(totalMaxSize); |
| for (int i = 0; i < regionsSize; i++) { |
| RegionCommit rc = new RegionCommit(this); |
| try { |
| rc.fromData(in, hasShadowKeys); |
| } catch (CacheClosedException cce) { |
| addProcessingException(cce); |
| // return to avoid serialization error being sent in reply |
| return; |
| } |
| regions.add(rc); |
| } |
| |
| bridgeContext = ClientProxyMembershipID.readCanonicalized(in); |
| farSiders = DataSerializer.readHashSet(in); |
| } |
| |
| /** |
| * Return true if a distributed ack message is required. On the client side of a transaction, this |
| * returns false, while returning true elsewhere. |
| * |
| * @return requires ack message or not |
| */ |
| private boolean isAckRequired() { |
| return ackRequired; |
| } |
| |
| |
| /** |
| * Indicate whether an ack is required. Defaults to true. |
| * |
| * @param a true if we require an ack. false if not. false on clients. |
| */ |
| public void setAckRequired(boolean a) { |
| ackRequired = a; |
| if (!a) { |
| processorId = -1; |
| } |
| } |
| |
| @Override |
| public void toData(DataOutput out, SerializationContext context) throws IOException { |
| out.writeInt(processorId); |
| InternalDataSerializer.invokeToData(txIdent, out); |
| { |
| boolean hasLockId = lockId != null; |
| out.writeBoolean(hasLockId); |
| if (hasLockId) { |
| InternalDataSerializer.invokeToData(lockId, out); |
| } |
| } |
| int regionsSize = 0; |
| { |
| int totalMaxSize = 0; |
| if (regions != null) { |
| regionsSize = regions.size(); |
| for (RegionCommit region : regions) { |
| totalMaxSize += region.maxSize; |
| } |
| } |
| out.writeInt(totalMaxSize); |
| } |
| |
| if (txState != null) { |
| DataSerializer.writeByteArray(txState.getBaseMembershipId(), out); |
| out.writeLong(txState.getBaseThreadId()); |
| out.writeLong(txState.getBaseSequenceId()); |
| } else { |
| DataSerializer.writeByteArray(farsideBaseMembershipId, out); |
| out.writeLong(farsideBaseThreadId); |
| out.writeLong(farsideBaseSequenceId); |
| } |
| |
| if (txState != null) { |
| DataSerializer.writeBoolean(txState.needsLargeModCount(), out); |
| } else { |
| DataSerializer.writeBoolean(needsLargeModCount, out); |
| } |
| |
| final boolean useShadowKey = useShadowKey(); |
| if (hasFlagsField(out)) { |
| out.writeBoolean(useShadowKey); |
| } |
| |
| out.writeInt(regionsSize); |
| { |
| if (regionsSize > 0) { |
| for (RegionCommit region : regions) { |
| region.toData(out, context, useShadowKey); |
| } |
| } |
| } |
| |
| DataSerializer.writeObject(bridgeContext, out); |
| |
| DataSerializer.writeHashSet(farSiders, out); |
| } |
| |
| private boolean hasFlagsField(final DataOutput out) { |
| return hasFlagsField(StaticSerialization.getVersionForDataStream(out)); |
| } |
| |
| private boolean hasFlagsField(final DataInput in) { |
| return hasFlagsField(StaticSerialization.getVersionForDataStream(in)); |
| } |
| |
| private boolean hasFlagsField(final KnownVersion version) { |
| return version.isNotOlderThan(KnownVersion.GEODE_1_7_0); |
| } |
| |
| private boolean useShadowKey() { |
| return null == clientVersion; |
| } |
| |
| @Override |
| public String toString() { |
| StringBuilder result = new StringBuilder(256); |
| result.append("TXCommitMessage@").append(System.identityHashCode(this)).append("#") |
| .append(sequenceNum).append(" processorId=").append(processorId).append(" txId=") |
| .append(txIdent); |
| |
| if (farSiders != null) { |
| Iterator<InternalDistributedMember> fs = farSiders.iterator(); |
| result.append(" farSiders="); |
| while (fs.hasNext()) { |
| result.append(fs.next()); |
| if (fs.hasNext()) { |
| result.append(' '); |
| } |
| } |
| } else { |
| result.append(" farSiders=<null>"); |
| } |
| if (regions != null) { |
| for (final RegionCommit region : regions) { |
| result.append(' ').append(region); |
| } |
| } |
| return result.toString(); |
| } |
| |
| /** |
| * Combines a set of small TXCommitMessages that belong to one transaction into a txCommitMessage |
| * 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. |
| * |
| * @return the complete txCommitMessage |
| */ |
| public static TXCommitMessage combine(Set<TXCommitMessage> msgSet) { |
| assert msgSet != null; |
| TXCommitMessage firstPart = null; |
| Iterator<TXCommitMessage> it = msgSet.iterator(); |
| while (it.hasNext()) { |
| if (firstPart == null) { |
| firstPart = it.next(); |
| continue; |
| } |
| firstPart.combine(it.next()); |
| } |
| return firstPart; |
| } |
| |
| /** |
| * Combines the other TXCommitMessage into this message. Used to compute complete TXCommitMessage |
| * from parts. |
| */ |
| public void combine(TXCommitMessage other) { |
| assert other != null; |
| final Map<String, RegionCommit> regionCommits = new HashMap<>(); |
| for (RegionCommit commit : regions) { |
| regionCommits.put(commit.getRegionPath(), commit); |
| } |
| for (RegionCommit commit : other.regions) { |
| if (!regionCommits.containsKey(commit.getRegionPath())) { |
| commit.msg = this; |
| regions.add(commit); |
| regionCommits.put(commit.getRegionPath(), commit); |
| } |
| } |
| } |
| |
| public static class RegionCommitList extends ArrayList<RegionCommit> { |
| private static final long serialVersionUID = -8910813949027683641L; |
| private transient boolean needsAck = false; |
| private transient RegionCommit trimRC = null; |
| private transient RegionCommitList trimChild = null; |
| |
| public RegionCommitList() { |
| super(); |
| } |
| |
| public RegionCommitList(RegionCommitList c) { |
| super(c); |
| } |
| |
| public boolean getNeedsAck() { |
| return needsAck; |
| } |
| |
| @Override |
| public boolean add(RegionCommit regionCommit) { |
| regionCommit.incRefCount(); |
| if (!needsAck && regionCommit.needsAck()) { |
| needsAck = true; |
| } |
| return super.add(regionCommit); |
| } |
| |
| /** |
| * Creates a new list, if needed, that contains all the elements of the specified old list |
| * except the last one if it is 'rc'. Also recomputes needsAck field. |
| */ |
| public RegionCommitList trim(RegionCommit rc) { |
| if (get(size() - 1) != rc) { |
| // no need to trim because it does not contain rc |
| return this; |
| } |
| if (trimRC == rc) { |
| return trimChild; |
| } |
| RegionCommitList result = new RegionCommitList(this); |
| trimRC = rc; |
| trimChild = result; |
| result.remove(result.size() - 1); |
| { |
| for (final RegionCommit itrc : result) { |
| itrc.incRefCount(); |
| if (itrc.needsAck()) { |
| result.needsAck = true; |
| } |
| } |
| } |
| return result; |
| } |
| |
| @Override |
| public String toString() { |
| return "@" + System.identityHashCode(this) + ' ' + super.toString(); |
| } |
| } |
| |
| public static class RegionCommit { |
| private final TxCallbackEventFactory txCallbackEventFactory = new TxCallbackEventFactoryImpl(); |
| /** |
| * The region that this commit represents. Valid on both nearside and farside. |
| */ |
| protected transient InternalRegion internalRegion; |
| /** |
| * Valid only on farside. |
| */ |
| private String regionPath; |
| private String parentRegionPath; |
| /** |
| * The message this region commit is a part of. Valid on both farside and nearside. |
| */ |
| private transient TXCommitMessage msg; |
| /** |
| * Number of RegionCommitList instances that have this RegionCommit in them Valid only on |
| * nearside. |
| */ |
| private transient int refCount = 0; |
| /** |
| * Valid only on nearside. |
| */ |
| private transient HeapDataOutputStream preserializedBuffer = null; |
| /** |
| * Upperbound on the number of operations this region could possibly have Valid only on |
| * nearside. |
| */ |
| transient int maxSize; |
| /** |
| * A list of Object; each one is the entry key for a distributed operation done by this |
| * transaction. The list must be kept in sync with opKeys. Valid only on nearside. |
| */ |
| private transient ArrayList<Object> opKeys; |
| /** |
| * A list of TXEntryState; each one is the entry info for a distributed operation done by this |
| * transaction. The list must be kept in sync with opKeys. Valid only on nearside. |
| */ |
| private transient ArrayList<Object> opEntries; |
| |
| private transient VersionSource memberId; |
| |
| /** |
| * The persistent ids of the peers for this region. Used to mark peers as offline if they do not |
| * apply the commit due to a cache close. |
| */ |
| public Map<InternalDistributedMember, PersistentMemberID> persistentIds; |
| |
| /** |
| * Used on nearside |
| */ |
| RegionCommit(TXCommitMessage msg, InternalRegion r, int maxSize) { |
| this.msg = msg; |
| internalRegion = r; |
| this.maxSize = maxSize; |
| } |
| |
| /** |
| * Used on farside who inits r later and never sets maxSize |
| */ |
| RegionCommit(TXCommitMessage msg) { |
| this.msg = msg; |
| } |
| |
| public String getRegionPath() { |
| return regionPath; |
| } |
| |
| public void incRefCount() { |
| refCount++; |
| } |
| |
| /** |
| * Valid on farside after beginProcess. Used to remember what to do at region cleanup time |
| */ |
| private boolean needsUnlock; |
| /** |
| * Valid on farside after beginProcess. Used to remember what to do at region cleanup time |
| */ |
| private boolean needsLRUEnd; |
| /** |
| * Valid on farside after beginProcess This is the txEvent that should be used by this |
| * RegionCommit |
| */ |
| private TXRmtEvent txEvent; |
| |
| /** |
| * Called to setup a region commit so its entryOps can be processed |
| * |
| * @return true if region should be processed; false if it can be ignored |
| * @throws CacheClosedException if the cache has been closed |
| */ |
| boolean beginProcess(DistributionManager dm, TransactionId txIdent, TXRmtEvent txEvent) |
| throws CacheClosedException { |
| if (logger.isDebugEnabled()) { |
| logger.debug("begin processing TXCommitMessage {} for region {}", txIdent, regionPath); |
| } |
| try { |
| if (!hookupRegion(dm)) { |
| return false; |
| } |
| if (msg.isAckRequired() |
| && (internalRegion == null || !internalRegion.getScope().isDistributed())) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("Received unneeded commit data for region {}", regionPath); |
| } |
| msg.addProcessingException(new RegionDestroyedException( |
| "Region not found", |
| regionPath)); |
| internalRegion = null; |
| return false; |
| } |
| needsUnlock = internalRegion.lockGII(); |
| internalRegion.txLRUStart(); |
| needsLRUEnd = true; |
| if (internalRegion.isInitialized()) { |
| // We don't want the txEvent to know anything about our regions |
| // that are still doing gii. |
| this.txEvent = txEvent; |
| } |
| } catch (RegionDestroyedException e) { |
| msg.addProcessingException(e); |
| // Region destroyed: Update cancelled |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "Received unneeded commit data for region {} because the region was destroyed.", |
| regionPath, e); |
| } |
| internalRegion = null; |
| } |
| return internalRegion != null; |
| } |
| |
| private boolean hookupRegion(DistributionManager dm) { |
| internalRegion = getRegionByPath(dm, regionPath); |
| if (internalRegion == null && parentRegionPath != null) { |
| internalRegion = getRegionByPath(dm, parentRegionPath); |
| regionPath = parentRegionPath; |
| } |
| // If there are additional regions that the server enlisted in the tx, |
| // which the client does not have, the client can just ignore the region |
| return internalRegion != null || !dm.getSystem().isLoner(); |
| } |
| |
| LocalRegion getRegionByPath(DistributionManager dm, String regionPath) { |
| InternalCache cache = dm.getCache(); |
| return cache == null ? null : (LocalRegion) cache.getInternalRegionByPath(regionPath); |
| } |
| |
| /** |
| * Called when processing is complete; only needs to be called if beginProcess returned true. |
| */ |
| void endProcess() { |
| if (internalRegion != null) { |
| try { |
| if (needsLRUEnd) { |
| needsLRUEnd = false; |
| internalRegion.txLRUEnd(); |
| } |
| } finally { |
| if (needsUnlock) { |
| needsUnlock = false; |
| internalRegion.unlockGII(); |
| } |
| } |
| } |
| } |
| |
| /** |
| * Returns the eventId to use for the give farside entry op. |
| * |
| * @since GemFire 5.7 |
| */ |
| private EventID getEventId(FarSideEntryOp entryOp) { |
| return msg.getEventId(entryOp.eventOffset); |
| } |
| |
| |
| /** |
| * Apply a single tx entry op on the far side |
| */ |
| @SuppressWarnings("synthetic-access") |
| protected void txApplyEntryOp(FarSideEntryOp entryOp, List<EntryEventImpl> pendingCallbacks) { |
| if (internalRegion == null) { |
| return; |
| } |
| EventID eventID = getEventId(entryOp); |
| boolean isDuplicate = internalRegion.hasSeenEvent(eventID); |
| if (internalRegion instanceof PartitionedRegion) { |
| /* |
| * This happens when we don't have the bucket and are getting adjunct notification |
| */ |
| long tailKey = entryOp.tailKey; |
| if (entryOp.filterRoutingInfo != null) { |
| if (entryOp.filterRoutingInfo.getMembers().contains(internalRegion.getMyId())) { |
| tailKey = -1; |
| } |
| } |
| |
| // No need to release because it is added to pendingCallbacks and they will be released |
| // later |
| EntryEventImpl eei = |
| txCallbackEventFactory.createCallbackEvent(internalRegion, entryOp.op, |
| entryOp.key, |
| entryOp.value, msg.txIdent, txEvent, getEventId(entryOp), entryOp.callbackArg, |
| entryOp.filterRoutingInfo, msg.bridgeContext, null, entryOp.versionTag, tailKey); |
| |
| if (entryOp.filterRoutingInfo != null) { |
| eei.setLocalFilterInfo( |
| entryOp.filterRoutingInfo.getFilterInfo(internalRegion.getCache().getMyId())); |
| } |
| if (isDuplicate) { |
| eei.setPossibleDuplicate(true); |
| } |
| if (logger.isDebugEnabled()) { |
| logger.debug("invoking transactional callbacks for {} key={} needsUnlock={} event={}", |
| entryOp.op, entryOp.key, needsUnlock, eei); |
| } |
| // we reach this spot because the event is either delivered to this member |
| // as an "adjunct" message or because the bucket was being created when |
| // the message was sent and already reflects the change caused by this event. |
| // In the latter case we need to invoke listeners |
| final boolean skipListeners = !isDuplicate; |
| eei.setInvokePRCallbacks(!skipListeners); |
| pendingCallbacks.add(eei); |
| return; |
| } |
| if (logger.isDebugEnabled()) { |
| logger.debug("applying transactional {} key={} needsUnlock={} eventId {} with routing {}", |
| entryOp.op, entryOp.key, needsUnlock, getEventId(entryOp), |
| entryOp.filterRoutingInfo); |
| } |
| if (entryOp.versionTag != null) { |
| entryOp.versionTag.replaceNullIDs(msg.getSender()); |
| } |
| if (entryOp.op.isDestroy()) { |
| internalRegion.txApplyDestroy(entryOp.key, msg.txIdent, txEvent, |
| needsUnlock, |
| entryOp.op, getEventId(entryOp), entryOp.callbackArg, pendingCallbacks, |
| entryOp.filterRoutingInfo, msg.bridgeContext, false /* origin remote */, |
| null/* txEntryState */, entryOp.versionTag, entryOp.tailKey); |
| } else if (entryOp.op.isInvalidate()) { |
| internalRegion.txApplyInvalidate(entryOp.key, Token.INVALID, entryOp.didDestroy, |
| msg.txIdent, |
| txEvent, false /* localOp */, getEventId(entryOp), entryOp.callbackArg, |
| pendingCallbacks, entryOp.filterRoutingInfo, msg.bridgeContext, |
| null/* txEntryState */, entryOp.versionTag, entryOp.tailKey); |
| } else { |
| internalRegion.txApplyPut(entryOp.op, entryOp.key, entryOp.value, entryOp.didDestroy, |
| msg.txIdent, txEvent, getEventId(entryOp), entryOp.callbackArg, |
| pendingCallbacks, entryOp.filterRoutingInfo, msg.bridgeContext, |
| null/* txEntryState */, entryOp.versionTag, entryOp.tailKey); |
| } |
| } |
| |
| /** |
| * Apply a single tx entry op on the far side |
| */ |
| @SuppressWarnings("synthetic-access") |
| protected void txApplyEntryOpAdjunctOnly(FarSideEntryOp entryOp) { |
| if (internalRegion == null) { |
| return; |
| } |
| EventID eventID = getEventId(entryOp); |
| boolean isDuplicate = internalRegion.hasSeenEvent(eventID); |
| if (internalRegion instanceof PartitionedRegion) { |
| |
| PartitionedRegion pr = (PartitionedRegion) internalRegion; |
| BucketRegion br = pr.getBucketRegion(entryOp.key); |
| Set<InternalDistributedMember> bucketOwners = br.getBucketOwners(); |
| InternalDistributedMember thisMember = internalRegion.getDistributionManager().getId(); |
| if (bucketOwners.contains(thisMember)) { |
| return; |
| } |
| |
| /* |
| * This happens when we don't have the bucket and are getting adjunct notification |
| */ |
| @Released |
| EntryEventImpl eei = |
| txCallbackEventFactory.createCallbackEvent(internalRegion, entryOp.op, |
| entryOp.key, |
| entryOp.value, msg.txIdent, txEvent, getEventId(entryOp), entryOp.callbackArg, |
| entryOp.filterRoutingInfo, msg.bridgeContext, null, entryOp.versionTag, |
| entryOp.tailKey); |
| try { |
| if (entryOp.filterRoutingInfo != null) { |
| eei.setLocalFilterInfo( |
| entryOp.filterRoutingInfo.getFilterInfo(internalRegion.getCache().getMyId())); |
| } |
| if (isDuplicate) { |
| eei.setPossibleDuplicate(true); |
| } |
| if (logger.isDebugEnabled()) { |
| logger.debug("invoking transactional callbacks for {} key={} needsUnlock={} event={}", |
| entryOp.op, entryOp.key, needsUnlock, eei); |
| } |
| // we reach this spot because the event is either delivered to this member |
| // as an "adjunct" message or because the bucket was being created when |
| // the message was sent and already reflects the change caused by this event. |
| // In the latter case we need to invoke listeners |
| final boolean skipListeners = !isDuplicate; |
| eei.invokeCallbacks(internalRegion, skipListeners, true); |
| } finally { |
| eei.release(); |
| } |
| } |
| } |
| |
| boolean isEmpty() { |
| return opKeys == null; |
| } |
| |
| boolean needsAck() { |
| return internalRegion.getScope().isDistributedAck(); |
| } |
| |
| void addOp(Object key, TXEntryState entry) { |
| if (opKeys == null) { |
| opKeys = new ArrayList<>(maxSize); |
| opEntries = new ArrayList<>(maxSize); |
| } |
| opKeys.add(key); |
| opEntries.add(entry); |
| } |
| |
| |
| public boolean isForceFireEvent(DistributionManager dm) { |
| LocalRegion r = getRegionByPath(dm, regionPath); |
| return !(r instanceof PartitionedRegion) && (r == null || !r |
| .isUsedForPartitionedRegionBucket()); |
| } |
| |
| public void fromData(DataInput in, boolean hasShadowKey) |
| throws IOException, ClassNotFoundException { |
| regionPath = DataSerializer.readString(in); |
| parentRegionPath = DataSerializer.readString(in); |
| |
| int size = in.readInt(); |
| if (size > 0) { |
| opKeys = new ArrayList<>(size); |
| opEntries = new ArrayList<>(size); |
| final boolean largeModCount = in.readBoolean(); |
| memberId = DataSerializer.readObject(in); |
| for (int i = 0; i < size; i++) { |
| FarSideEntryOp entryOp = new FarSideEntryOp(); |
| // shadowkey is not being sent to clients |
| entryOp.fromData(in, largeModCount, hasShadowKey); |
| if (entryOp.versionTag != null && memberId != null) { |
| entryOp.versionTag.setMemberID(memberId); |
| } |
| msg.addFarSideEntryOp(entryOp); |
| opKeys.add(entryOp.key); |
| opEntries.add(entryOp); |
| } |
| } |
| } |
| |
| @Override |
| public String toString() { |
| StringBuilder result = new StringBuilder(64); |
| if (regionPath != null) { |
| result.append(regionPath); |
| } else { |
| result.append(internalRegion.getFullPath()); |
| } |
| if (refCount > 0) { |
| result.append(" refCount=").append(refCount); |
| } |
| return result.toString(); |
| } |
| |
| private void basicToData(DataOutput out, |
| SerializationContext context, |
| boolean useShadowKey) throws IOException { |
| if (internalRegion != null) { |
| DataSerializer.writeString(internalRegion.getFullPath(), out); |
| if (internalRegion instanceof BucketRegion) { |
| DataSerializer.writeString( |
| ((Bucket) internalRegion).getPartitionedRegion().getFullPath(), out); |
| } else { |
| DataSerializer.writeString(null, out); |
| } |
| } else { |
| DataSerializer.writeString(regionPath, out); |
| DataSerializer.writeString(parentRegionPath, out); |
| } |
| |
| if (isEmpty() || opKeys.size() == 0) { |
| out.writeInt(0); |
| } else { |
| int size = opKeys.size(); |
| out.writeInt(size); |
| |
| final boolean largeModCount; |
| if (msg.txState != null) { |
| largeModCount = msg.txState.needsLargeModCount(); |
| } else { |
| largeModCount = msg.needsLargeModCount; |
| } |
| out.writeBoolean(largeModCount); |
| |
| VersionSource member = memberId; |
| if (member == null) { |
| if (internalRegion == null) { |
| Assert.assertTrue(msg.txState == null); |
| } else { |
| member = internalRegion.getVersionMember(); |
| } |
| } |
| DataSerializer.writeObject(member, out); |
| for (int i = 0; i < size; i++) { |
| DataSerializer.writeObject(opKeys.get(i), out); |
| if (msg.txState != null) { |
| /* we are still on tx node and have the entry state */ |
| ((TXEntryState) opEntries.get(i)).toFarSideData(out, context, largeModCount, |
| true, useShadowKey); |
| } else { |
| ((FarSideEntryOp) opEntries.get(i)).toData(out, largeModCount, true, |
| useShadowKey); |
| } |
| } |
| } |
| } |
| |
| |
| public void toData(DataOutput out, SerializationContext context, boolean useShadowKey) |
| throws IOException { |
| if (preserializedBuffer != null) { |
| preserializedBuffer.rewind(); |
| preserializedBuffer.sendTo(out); |
| } else if (refCount > 1) { |
| KnownVersion v = StaticSerialization.getVersionForDataStream(out); |
| HeapDataOutputStream hdos = new HeapDataOutputStream(1024, v); |
| basicToData(hdos, context, useShadowKey); |
| preserializedBuffer = hdos; |
| preserializedBuffer.sendTo(out); |
| } else { |
| basicToData(out, context, useShadowKey); |
| } |
| } |
| |
| /** |
| * Holds data that describes a tx entry op on the far side. |
| * |
| * @since GemFire 5.0 |
| */ |
| public class FarSideEntryOp implements Comparable<FarSideEntryOp> { |
| public Operation op; |
| public int modSerialNum; |
| public int eventOffset; |
| public Object key; |
| public Object value; |
| public boolean didDestroy; |
| public Object callbackArg; |
| private FilterRoutingInfo filterRoutingInfo; |
| private VersionTag versionTag; |
| private long tailKey; |
| |
| /** |
| * Create a new representation of a tx entry op on the far side. All init will be done by a |
| * call to fromData |
| */ |
| public FarSideEntryOp() {} |
| |
| /** |
| * Creates and returns a new instance of a tx entry op on the far side. The "toData" that this |
| * should match is {@link TXEntryState#toFarSideData}. |
| * |
| * @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 |
| */ |
| public void fromData(DataInput in, boolean largeModCount, boolean readShadowKey) |
| throws IOException, ClassNotFoundException { |
| key = DataSerializer.readObject(in); |
| op = Operation.fromOrdinal(in.readByte()); |
| if (largeModCount) { |
| modSerialNum = in.readInt(); |
| } else { |
| modSerialNum = in.readByte(); |
| } |
| callbackArg = DataSerializer.readObject(in); |
| filterRoutingInfo = DataSerializer.readObject(in); |
| versionTag = DataSerializer.readObject(in); |
| if (readShadowKey) { |
| tailKey = in.readLong(); |
| } |
| eventOffset = in.readInt(); |
| if (!op.isDestroy()) { |
| didDestroy = in.readBoolean(); |
| if (!op.isInvalidate()) { |
| boolean isTokenOrByteArray = in.readBoolean(); |
| if (isTokenOrByteArray) { |
| // token or byte[] |
| value = DataSerializer.readObject(in); |
| } else { |
| // CachedDeserializable, Object, or PDX |
| value = CachedDeserializableFactory.create(DataSerializer.readByteArray(in), |
| GemFireCacheImpl.getInstance()); |
| } |
| } |
| } |
| } |
| |
| public void toData(DataOutput out, boolean largeModCount, boolean sendVersionTag, |
| boolean sendShadowKey) throws IOException { |
| // DataSerializer.writeObject(this.key,out); |
| /* Don't serialize key because caller did that already */ |
| |
| out.writeByte(op.ordinal); |
| if (largeModCount) { |
| out.writeInt(modSerialNum); |
| } else { |
| out.writeByte(modSerialNum); |
| } |
| DataSerializer.writeObject(callbackArg, out); |
| DataSerializer.writeObject(filterRoutingInfo, out); |
| if (sendVersionTag) { |
| DataSerializer.writeObject(versionTag, out); |
| } |
| if (sendShadowKey) { |
| out.writeLong(tailKey); |
| } |
| out.writeInt(eventOffset); |
| if (!op.isDestroy()) { |
| out.writeBoolean(didDestroy); |
| if (!op.isInvalidate()) { |
| boolean sendObject = Token.isInvalidOrRemoved(value); |
| sendObject = sendObject || value instanceof byte[]; |
| out.writeBoolean(sendObject); |
| if (sendObject) { |
| DataSerializer.writeObject(value, out); |
| } else { |
| DataSerializer.writeObjectAsByteArray(value, out); |
| } |
| } |
| } |
| } |
| |
| |
| /** |
| * Performs this entryOp on the farside of a tx commit. |
| */ |
| public void process(List<EntryEventImpl> pendingCallbacks) { |
| txApplyEntryOp(this, pendingCallbacks); |
| } |
| |
| public void processAdjunctOnly() { |
| txApplyEntryOpAdjunctOnly(this); |
| } |
| |
| public RegionCommit getRegionCommit() { |
| return RegionCommit.this; |
| } |
| |
| /** |
| * Returns the value to use to sort us |
| */ |
| private int getSortValue() { |
| return modSerialNum; |
| } |
| |
| @Override |
| public int compareTo(FarSideEntryOp o) { |
| return getSortValue() - o.getSortValue(); |
| } |
| |
| @Override |
| public boolean equals(Object o) { |
| if (!(o instanceof FarSideEntryOp)) { |
| return false; |
| } |
| return compareTo((FarSideEntryOp) o) == 0; |
| } |
| |
| @Override |
| public int hashCode() { |
| return getSortValue(); |
| } |
| } |
| } |
| |
| Object getTrackerKey() { |
| if (lockId != null) { |
| return lockId; |
| } else { |
| return txIdent; |
| } |
| } |
| |
| /** |
| * Used to prevent processing of the message if we have reported to other FarSiders that we did |
| * not received the CommitProcessMessage |
| */ |
| boolean dontProcess() { |
| return dontProcess; |
| } |
| |
| /** |
| * Indicate that this message should not be processed if we receive CommitProcessMessage (late) |
| */ |
| void setDontProcess() { |
| dontProcess = true; |
| } |
| |
| boolean isProcessing() { |
| return isProcessing; |
| } |
| |
| private void setIsProcessing(boolean isProcessing) { |
| this.isProcessing = isProcessing; |
| } |
| |
| boolean wasProcessed() { |
| return wasProcessed; |
| } |
| |
| void setProcessed(boolean wasProcessed) { |
| this.wasProcessed = wasProcessed; |
| } |
| |
| /** |
| * 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. |
| */ |
| public static class CommitProcessForLockIdMessage extends CommitProcessMessage { |
| private TXLockId lockId; |
| |
| public CommitProcessForLockIdMessage() { |
| // Zero arg constructor for DataSerializer |
| } |
| |
| public CommitProcessForLockIdMessage(TXLockId lockId) { |
| this.lockId = lockId; |
| Assert.assertTrue(this.lockId != null, |
| "CommitProcessForLockIdMessage must have a non-null lockid!"); |
| } |
| |
| @Override |
| protected void process(ClusterDistributionManager dm) { |
| final TXCommitMessage mess = waitForMessage(lockId, dm); |
| Assert.assertTrue(mess != null, "Commit data for TXLockId: " + lockId + " not found"); |
| basicProcess(mess, dm); |
| } |
| |
| @Override |
| public int getDSFID() { |
| return COMMIT_PROCESS_FOR_LOCKID_MESSAGE; |
| } |
| |
| @Override |
| public void toData(DataOutput out, |
| SerializationContext context) throws IOException { |
| InternalDataSerializer.invokeToData(lockId, out); |
| } |
| |
| @Override |
| public void fromData(DataInput in, |
| DeserializationContext context) throws IOException, ClassNotFoundException { |
| lockId = TXLockIdImpl.createFromData(in); |
| Assert.assertTrue(lockId != null, |
| "CommitProcessForLockIdMessage must have a non-null lockid!"); |
| } |
| |
| @Override |
| public String toString() { |
| return "CommitProcessForLockIdMessage@" + System.identityHashCode(this) |
| + " lockId=" + lockId; |
| } |
| } |
| |
| /** |
| * The CommitProcessForTXIdMessaage is sent by the Distributed ACK TX origin to the recipients |
| * (aka FarSiders) to indicate that a previously received RegionCommit that contained a TXId |
| * should commence processing. RegionCommit messages that contain a TXId (and no TXLockId) are |
| * typically sent if all the TX changes are a result of load/netsearch/netload values (thus no |
| * lockid) |
| */ |
| public static class CommitProcessForTXIdMessage extends CommitProcessMessage { |
| private TXId txId; |
| |
| public CommitProcessForTXIdMessage() { |
| // Zero arg constructor for DataSerializer |
| } |
| |
| public CommitProcessForTXIdMessage(TXId txId) { |
| this.txId = txId; |
| Assert.assertTrue(this.txId != null, |
| "CommitProcessMessageForTXId must have a non-null txid!"); |
| } |
| |
| @Override |
| protected void process(ClusterDistributionManager dm) { |
| final TXCommitMessage mess = waitForMessage(txId, dm); |
| Assert.assertTrue(mess != null, "Commit data for TXId: " + txId + " not found"); |
| basicProcess(mess, dm); |
| } |
| |
| @Override |
| public int getDSFID() { |
| return COMMIT_PROCESS_FOR_TXID_MESSAGE; |
| } |
| |
| @Override |
| public void toData(DataOutput out, |
| SerializationContext context) throws IOException { |
| InternalDataSerializer.invokeToData(txId, out); |
| } |
| |
| @Override |
| public void fromData(DataInput in, |
| DeserializationContext context) throws IOException, ClassNotFoundException { |
| txId = TXId.createFromData(in); |
| Assert.assertTrue(txId != null, |
| "CommitProcessMessageForTXId must have a non-null txid!"); |
| } |
| |
| @Override |
| public String toString() { |
| return "CommitProcessForTXIdMessage@" + System.identityHashCode(this) |
| + " txId=" + txId; |
| } |
| } |
| |
| public abstract static class CommitProcessMessage extends PooledDistributionMessage { |
| protected void basicProcess(final TXCommitMessage mess, final ClusterDistributionManager dm) { |
| dm.removeMembershipListener(mess); |
| synchronized (mess) { |
| if (mess.dontProcess()) { |
| return; |
| } |
| } |
| try { |
| mess.basicProcess(); |
| } finally { |
| txTracker.processed(mess); |
| } |
| } |
| } |
| |
| /** |
| * 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 |
| * this message is attempting to query other potential fellow FarSiders (aka recipients) who may |
| * have received the CommitProcess message. |
| * |
| * Since the occurance of this message will be rare (hopefully), it was decided to be general |
| * 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 |
| */ |
| public static class CommitProcessQueryMessage extends PooledDistributionMessage { |
| private Object trackerKey; // Either a TXLockId or a TXId |
| private int processorId; |
| |
| public CommitProcessQueryMessage() { |
| // Zero arg constructor for DataSerializer |
| } |
| |
| public CommitProcessQueryMessage(Object trackerKey, int processorId) { |
| this.trackerKey = trackerKey; |
| this.processorId = processorId; |
| } |
| |
| @Override |
| protected void process(ClusterDistributionManager dm) { |
| final boolean processMsgReceived = txTracker.commitProcessReceived(trackerKey); |
| if (!processMsgReceived) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("CommitProcessQuery did not find {} in the history", trackerKey); |
| } |
| } |
| |
| // Reply to the fellow FarSider as to whether the |
| // CommitProcess message was received |
| CommitProcessQueryReplyMessage resp = new CommitProcessQueryReplyMessage(processMsgReceived); |
| resp.setProcessorId(processorId); |
| resp.setRecipient(getSender()); |
| dm.putOutgoing(resp); |
| } |
| |
| @Override |
| public void toData(DataOutput out, |
| SerializationContext context) throws IOException { |
| DataSerializer.writeObject(trackerKey, out); |
| out.writeInt(processorId); |
| } |
| |
| @Override |
| public int getDSFID() { |
| return COMMIT_PROCESS_QUERY_MESSAGE; |
| } |
| |
| @Override |
| public void fromData(DataInput in, |
| DeserializationContext context) throws IOException, ClassNotFoundException { |
| trackerKey = DataSerializer.readObject(in); |
| processorId = in.readInt(); |
| } |
| |
| @Override |
| public String toString() { |
| return "CommitProcessQueryMessage@" + System.identityHashCode(this) |
| + " trackerKeyClass=" + trackerKey.getClass().getName() |
| + " trackerKey=" + trackerKey + " processorId=" |
| + processorId; |
| } |
| } |
| |
| /********************* Commit Process Query Response Message **********************************/ |
| public static class CommitProcessQueryReplyMessage extends ReplyMessage { |
| private boolean wasReceived; |
| |
| public CommitProcessQueryReplyMessage(boolean wasReceived) { |
| this.wasReceived = wasReceived; |
| } |
| |
| public CommitProcessQueryReplyMessage() { |
| // zero arg constructor for DataSerializer |
| } |
| |
| public boolean wasReceived() { |
| return wasReceived; |
| } |
| |
| @Override |
| public int getDSFID() { |
| return COMMIT_PROCESS_QUERY_REPLY_MESSAGE; |
| } |
| |
| @Override |
| public void fromData(DataInput in, |
| DeserializationContext context) throws IOException, ClassNotFoundException { |
| super.fromData(in, context); |
| wasReceived = in.readBoolean(); |
| } |
| |
| @Override |
| public void toData(DataOutput out, |
| SerializationContext context) throws IOException { |
| super.toData(out, context); |
| out.writeBoolean(wasReceived); |
| } |
| |
| @Override |
| public String toString() { |
| return "CommitProcessQueryReplyMessage@" + System.identityHashCode(this) |
| + " wasReceived=" + wasReceived + " processorId=" |
| + processorId + " from " + getSender(); |
| } |
| } |
| |
| /********************* Commit Process Query Response Processor *********************************/ |
| public static class CommitProcessQueryReplyProcessor extends ReplyProcessor21 { |
| public boolean receivedOnePositive; |
| |
| CommitProcessQueryReplyProcessor(DistributionManager dm, |
| Set<InternalDistributedMember> members) { |
| super(dm, members); |
| receivedOnePositive = false; |
| } |
| |
| @Override |
| public void process(DistributionMessage msg) { |
| CommitProcessQueryReplyMessage ccMess = (CommitProcessQueryReplyMessage) msg; |
| if (ccMess.wasReceived()) { |
| receivedOnePositive = true; |
| } |
| super.process(msg); |
| } |
| |
| @Override |
| protected boolean canStopWaiting() { |
| return receivedOnePositive; |
| } |
| |
| public boolean receivedACommitProcessMessage() { |
| return receivedOnePositive; |
| } |
| } |
| |
| /********************* MembershipListener Implementation ***************************************/ |
| @Override |
| public void memberJoined(DistributionManager distributionManager, InternalDistributedMember id) { |
| // do nothing |
| } |
| |
| @Override |
| public void memberSuspect(DistributionManager distributionManager, InternalDistributedMember id, |
| InternalDistributedMember whoSuspected, String reason) {} |
| |
| @Override |
| public void quorumLost(DistributionManager distributionManager, |
| Set<InternalDistributedMember> failures, |
| List<InternalDistributedMember> remaining) {} |
| |
| /** |
| * return true if the member initiating this transaction has left the cluster |
| */ |
| public boolean isDepartureNoticed() { |
| return departureNoticed; |
| } |
| |
| private void doOriginDepartedCommit() { |
| try { |
| // Set processor to zero to avoid the ack to the now departed origin |
| processorId = 0; |
| basicProcess(); |
| } finally { |
| txTracker.processed(this); |
| } |
| } |
| |
| @Override |
| public void memberDeparted(DistributionManager distributionManager, |
| final InternalDistributedMember id, boolean crashed) { |
| |
| if (!getSender().equals(id)) { |
| return; |
| } |
| distributionManager.removeMembershipListener(this); |
| |
| synchronized (this) { |
| if (isProcessing() || departureNoticed) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("Member departed: Commit data is already being processed for lockid: {}", |
| lockId); |
| } |
| return; |
| } |
| departureNoticed = true; |
| } |
| |
| // Send message to fellow FarSiders (aka recipients), if any, to |
| // determine if any one of them have received a CommitProcessMessage |
| if (getFarSiders() != null && !getFarSiders().isEmpty()) { |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "Member departed: {} sending query for CommitProcess message to other recipients.", id); |
| } |
| |
| // Create a new thread, send the CommitProcessQuery, wait for a response and potentially |
| // process |
| // Should I use a thread pool?, Darrel suggests look in DM somewhere or introduce a zero |
| // sized thread pool |
| Thread fellowFarSidersQuery = new LoggingThread("CommitProcessQuery Thread", |
| () -> doCommitProcessQuery(id)); |
| fellowFarSidersQuery.start(); |
| } else { |
| if (logger.isDebugEnabled()) { |
| logger.debug("Member departed: {}. Processing commit data.", getSender()); |
| } |
| |
| // Optimimal case where we are the only FarSider, assume we |
| // will never get the CommitProcess message, but it |
| // doesn't matter since we can commit anyway. |
| // Start a new thread to process the commit |
| Thread originDepartedCommit = new LoggingThread("Origin Departed Commit", |
| this::doOriginDepartedCommit); |
| originDepartedCommit.start(); |
| } |
| } |
| |
| HashSet<InternalDistributedMember> getFarSiders() { |
| return farSiders; |
| } |
| |
| DistributionManager getDistributionManager() { |
| return dm; |
| } |
| |
| void doCommitProcessQuery(final InternalDistributedMember id) { |
| CommitProcessQueryReplyProcessor replyProcessor = createReplyProcessor(); |
| CommitProcessQueryMessage queryMessage = createQueryMessage(replyProcessor); |
| queryMessage.setRecipients(farSiders); |
| getDistributionManager().putOutgoing(queryMessage); |
| // Wait for any one positive response or all negative responses. |
| // (while() loop removed for bug 36983 - you can't loop on waitForReplies() |
| getDistributionManager().getCancelCriterion().checkCancelInProgress(null); |
| try { |
| replyProcessor.waitForRepliesUninterruptibly(); |
| } catch (ReplyException e) { |
| e.handleCause(); |
| } |
| if (replyProcessor.receivedACommitProcessMessage()) { |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "Transaction associated with lockID: {} from orign {} is processing due to a received \"commit process\" message", |
| lockId, id); |
| } |
| |
| try { |
| // Set processor to zero to avoid the ack to the now departed origin |
| processorId = 0; |
| basicProcess(); |
| } finally { |
| txTracker.processed(this); |
| } |
| } else { |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "Transaction associated with lockID: {} from origin {} ignored. No other recipients received \"commit process\" message", |
| lockId, id); |
| } |
| txTracker.removeMessage(this); |
| } |
| } |
| |
| CommitProcessQueryReplyProcessor createReplyProcessor() { |
| return new CommitProcessQueryReplyProcessor(dm, farSiders); |
| } |
| |
| CommitProcessQueryMessage createQueryMessage(CommitProcessQueryReplyProcessor replyProcessor) { |
| return new CommitProcessQueryMessage(getTrackerKey(), replyProcessor.getProcessorId()); |
| } |
| |
| void setUpdateLockMembers() { |
| lockNeedsUpdate = true; |
| } |
| |
| /** |
| * Intended to be called after TXState.applyChanges when the potential for a different set of TX |
| * members has been determined and it is safe to ignore any new members because the changes have |
| * been applied to committed state. This was added as the solution to bug 32999 and the recovery |
| * when the TXLock Lessor (the sending VM) crashes/departs before or while sending the |
| * TXCommitMessage. |
| * |
| * @see TXState#commit() |
| * @see org.apache.geode.internal.cache.locks.TXLockBatch#getBatchId() |
| */ |
| private void updateLockMembers() { |
| if (lockNeedsUpdate && lockId != null) { |
| TXLockService.createDTLS(dm.getSystem()).updateParticipants(lockId, |
| msgMap.keySet()); |
| } |
| } |
| |
| /** |
| * Reply processor which collects all CommitReplyExceptions and emits a detailed failure exception |
| * if problems occur |
| * |
| * @since GemFire 5.7 |
| */ |
| private class CommitReplyProcessor extends ReliableReplyProcessor21 { |
| private final Map<InternalDistributedMember, RegionCommitList> msgMap; |
| |
| public CommitReplyProcessor(DistributionManager dm, Set<InternalDistributedMember> initMembers, |
| Map<InternalDistributedMember, RegionCommitList> msgMap) { |
| super(dm, initMembers); |
| this.msgMap = msgMap; |
| } |
| |
| public void waitForCommitCompletion() { |
| try { |
| waitForRepliesUninterruptibly(); |
| } catch (CommitExceptionCollectingException e) { |
| e.handlePotentialCommitFailure(msgMap); |
| } |
| } |
| |
| @Override |
| protected synchronized void processException(DistributionMessage msg, ReplyException ex) { |
| if (msg instanceof ReplyMessage) { |
| synchronized (this) { |
| if (exception == null) { |
| // Exception Container |
| exception = new CommitExceptionCollectingException(txIdent); |
| } |
| CommitExceptionCollectingException cce = |
| (CommitExceptionCollectingException) exception; |
| if (ex instanceof CommitReplyException) { |
| CommitReplyException cre = (CommitReplyException) ex; |
| cce.addExceptionsFromMember(msg.getSender(), cre.getExceptions()); |
| } else { |
| cce.addExceptionsFromMember(msg.getSender(), Collections.singleton(ex)); |
| } |
| } |
| } |
| } |
| |
| @Override |
| protected boolean stopBecauseOfExceptions() { |
| return false; |
| } |
| |
| public Set<InternalDistributedMember> getCacheClosedMembers() { |
| if (exception != null) { |
| CommitExceptionCollectingException cce = |
| (CommitExceptionCollectingException) exception; |
| return cce.getCacheClosedMembers(); |
| } else { |
| return Collections.emptySet(); |
| } |
| } |
| |
| public Set<InternalDistributedMember> getRegionDestroyedMembers(String regionFullPath) { |
| if (exception != null) { |
| CommitExceptionCollectingException cce = |
| (CommitExceptionCollectingException) exception; |
| return cce.getRegionDestroyedMembers(regionFullPath); |
| } else { |
| return Collections.emptySet(); |
| } |
| } |
| } |
| |
| /** |
| * An Exception that collects many remote CommitExceptions |
| * |
| * @since GemFire 5.7 |
| */ |
| public static class CommitExceptionCollectingException extends ReplyException { |
| private static final long serialVersionUID = 589384721273797822L; |
| /** |
| * Set of members that threw CacheClosedExceptions |
| */ |
| private final Set<InternalDistributedMember> cacheExceptions = new HashSet<>(); |
| /** |
| * key=region path, value=Set of members |
| */ |
| private final Map<String, Set<InternalDistributedMember>> regionExceptions = new HashMap<>(); |
| /** |
| * List of exceptions that were unexpected and caused the tx to fail |
| */ |
| private final Map<InternalDistributedMember, List<Exception>> fatalExceptions = new HashMap<>(); |
| |
| private final TXId id; |
| |
| public CommitExceptionCollectingException(TXId txIdent) { |
| id = txIdent; |
| } |
| |
| /** |
| * Determine if the commit processing was incomplete, if so throw a detailed exception |
| * indicating the source of the problem |
| */ |
| public void handlePotentialCommitFailure( |
| Map<InternalDistributedMember, RegionCommitList> msgMap) { |
| if (fatalExceptions.size() > 0) { |
| StringBuilder errorMessage = new StringBuilder("Incomplete commit of transaction ") |
| .append(id).append(". Caused by the following exceptions: "); |
| for (final Map.Entry<InternalDistributedMember, List<Exception>> me : fatalExceptions |
| .entrySet()) { |
| InternalDistributedMember mem = me.getKey(); |
| errorMessage.append(" From member: ").append(mem).append(" "); |
| List<Exception> exceptions = me.getValue(); |
| for (Iterator<Exception> ei = exceptions.iterator(); ei.hasNext();) { |
| Exception e = ei.next(); |
| errorMessage.append(e); |
| for (StackTraceElement ste : e.getStackTrace()) { |
| errorMessage.append("\n\tat ").append(ste); |
| } |
| if (ei.hasNext()) { |
| errorMessage.append("\nAND\n"); |
| } |
| } |
| errorMessage.append("."); |
| } |
| throw new CommitIncompleteException(errorMessage.toString()); |
| } |
| |
| // Mark any persistent members as offline |
| handleClosedMembers(msgMap); |
| handleRegionDestroyed(msgMap); |
| } |
| |
| /** |
| * Mark peers as offline for regions that the peer returned a RegionDestroyedException |
| */ |
| private void handleRegionDestroyed(Map<InternalDistributedMember, RegionCommitList> msgMap) { |
| if (regionExceptions.isEmpty()) { |
| return; |
| } |
| |
| for (Map.Entry<InternalDistributedMember, RegionCommitList> memberMap : msgMap.entrySet()) { |
| InternalDistributedMember member = memberMap.getKey(); |
| RegionCommitList rcl = memberMap.getValue(); |
| for (RegionCommit region : rcl) { |
| Set<InternalDistributedMember> failedMembers = |
| regionExceptions.get(region.internalRegion.getFullPath()); |
| if (failedMembers != null && failedMembers.contains(member)) { |
| markMemberOffline(member, region); |
| } |
| } |
| } |
| |
| } |
| |
| /** |
| * Mark peers as offline that returned a cache closed exception |
| */ |
| private void handleClosedMembers(Map<InternalDistributedMember, RegionCommitList> msgMap) { |
| for (InternalDistributedMember member : getCacheClosedMembers()) { |
| RegionCommitList rcl = msgMap.get(member); |
| |
| for (RegionCommit region : rcl) { |
| markMemberOffline(member, region); |
| } |
| } |
| } |
| |
| private void markMemberOffline(InternalDistributedMember member, RegionCommit region) { |
| if (region.persistentIds == null) { |
| return; |
| } |
| |
| PersistentMemberID persistentId = region.persistentIds.get(member); |
| /// iterate over the list and mark the members offline |
| if (persistentId != null) { |
| // Fix for bug 42142 - In order for recovery to work, |
| // we must either |
| // 1) persistent the region operation successfully on the peer |
| // 2) record that the peer is offline |
| // or |
| // 3) fail the operation |
| |
| // if we have started to shutdown, we don't want to mark the peer |
| // as offline, or we will think we have newer data when in fact we don't |
| region.internalRegion.getCancelCriterion().checkCancelInProgress(null); |
| |
| // Otherwise, mark the peer as offline, because it didn't complete |
| // the operation. |
| ((DistributedRegion) region.internalRegion).getPersistenceAdvisor().markMemberOffline( |
| member, |
| persistentId); |
| } |
| } |
| |
| public Set<InternalDistributedMember> getCacheClosedMembers() { |
| return cacheExceptions; |
| } |
| |
| public Set<InternalDistributedMember> getRegionDestroyedMembers(String regionFullPath) { |
| Set<InternalDistributedMember> members = regionExceptions.get(regionFullPath); |
| if (members == null) { |
| members = Collections.emptySet(); |
| } |
| return members; |
| } |
| |
| /** |
| * Protected by (this) |
| */ |
| public void addExceptionsFromMember(InternalDistributedMember member, |
| Set<Exception> exceptions) { |
| for (final Exception ex : exceptions) { |
| if (ex instanceof CancelException) { |
| cacheExceptions.add(member); |
| } else if (ex instanceof RegionDestroyedException) { |
| final String r = ((RegionDestroyedException) ex).getRegionFullPath(); |
| regionExceptions.computeIfAbsent(r, k -> new HashSet<>()).add(member); |
| } else { |
| fatalExceptions.computeIfAbsent(member, k -> new ArrayList<>(2)).add(ex); |
| } |
| } |
| } |
| } |
| |
| public void hookupRegions(DistributionManager dm) { |
| if (regions != null) { |
| for (final RegionCommit rc : regions) { |
| rc.hookupRegion(dm); |
| } |
| } |
| |
| } |
| |
| |
| /** |
| * Disable firing of TX Listeners. Currently on used on clients. |
| * |
| * @param disableListeners disable the listeners |
| */ |
| public void setDisableListeners(boolean disableListeners) { |
| this.disableListeners = disableListeners; |
| } |
| |
| public KnownVersion getClientVersion() { |
| return clientVersion; |
| } |
| |
| public void setClientVersion(KnownVersion clientVersion) { |
| this.clientVersion = clientVersion; |
| } |
| |
| } |
