| /* |
| * 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.wan; |
| |
| import static org.apache.geode.internal.statistics.StatisticsClockFactory.disabledClock; |
| |
| import java.io.IOException; |
| import java.util.ArrayList; |
| import java.util.HashSet; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.Set; |
| import java.util.concurrent.ConcurrentLinkedQueue; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.locks.ReentrantReadWriteLock; |
| |
| import org.apache.logging.log4j.Logger; |
| |
| import org.apache.geode.CancelCriterion; |
| import org.apache.geode.CancelException; |
| import org.apache.geode.InternalGemFireError; |
| import org.apache.geode.annotations.Immutable; |
| import org.apache.geode.annotations.VisibleForTesting; |
| import org.apache.geode.annotations.internal.MutableForTesting; |
| import org.apache.geode.cache.CacheException; |
| import org.apache.geode.cache.DataPolicy; |
| import org.apache.geode.cache.Region; |
| import org.apache.geode.cache.RegionDestroyedException; |
| import org.apache.geode.cache.RegionExistsException; |
| import org.apache.geode.cache.RegionShortcut; |
| import org.apache.geode.cache.asyncqueue.AsyncEventListener; |
| import org.apache.geode.cache.client.internal.LocatorDiscoveryCallback; |
| import org.apache.geode.cache.client.internal.PoolImpl; |
| import org.apache.geode.cache.wan.GatewayEventFilter; |
| import org.apache.geode.cache.wan.GatewayEventSubstitutionFilter; |
| import org.apache.geode.cache.wan.GatewayQueueEvent; |
| import org.apache.geode.cache.wan.GatewaySender; |
| import org.apache.geode.cache.wan.GatewayTransportFilter; |
| import org.apache.geode.distributed.GatewayCancelledException; |
| import org.apache.geode.distributed.internal.DistributionAdvisee; |
| import org.apache.geode.distributed.internal.DistributionAdvisor; |
| import org.apache.geode.distributed.internal.DistributionAdvisor.Profile; |
| import org.apache.geode.distributed.internal.DistributionManager; |
| import org.apache.geode.distributed.internal.InternalDistributedSystem; |
| import org.apache.geode.distributed.internal.ResourceEvent; |
| import org.apache.geode.distributed.internal.ServerLocation; |
| import org.apache.geode.internal.cache.CachePerfStats; |
| import org.apache.geode.internal.cache.EntryEventImpl; |
| import org.apache.geode.internal.cache.EnumListenerEvent; |
| import org.apache.geode.internal.cache.HasCachePerfStats; |
| import org.apache.geode.internal.cache.InternalCache; |
| import org.apache.geode.internal.cache.InternalRegion; |
| import org.apache.geode.internal.cache.InternalRegionFactory; |
| import org.apache.geode.internal.cache.LocalRegion; |
| import org.apache.geode.internal.cache.PartitionedRegion; |
| import org.apache.geode.internal.cache.RegionQueue; |
| import org.apache.geode.internal.cache.execute.BucketMovedException; |
| import org.apache.geode.internal.cache.ha.ThreadIdentifier; |
| import org.apache.geode.internal.cache.wan.parallel.ConcurrentParallelGatewaySenderQueue; |
| import org.apache.geode.internal.cache.wan.parallel.WaitUntilParallelGatewaySenderFlushedCoordinator; |
| import org.apache.geode.internal.cache.wan.serial.ConcurrentSerialGatewaySenderEventProcessor; |
| import org.apache.geode.internal.cache.xmlcache.CacheCreation; |
| import org.apache.geode.internal.offheap.Releasable; |
| import org.apache.geode.internal.offheap.annotations.Released; |
| import org.apache.geode.internal.offheap.annotations.Retained; |
| import org.apache.geode.internal.offheap.annotations.Unretained; |
| import org.apache.geode.internal.statistics.StatisticsClock; |
| import org.apache.geode.logging.internal.executors.LoggingThread; |
| import org.apache.geode.logging.internal.log4j.api.LogService; |
| |
| /** |
| * Abstract implementation of both Serial and Parallel GatewaySender. It handles common |
| * functionality like initializing proxy. |
| * |
| * @since GemFire 7.0 |
| */ |
| public abstract class AbstractGatewaySender implements InternalGatewaySender, DistributionAdvisee { |
| |
| private static final Logger logger = LogService.getLogger(); |
| |
| protected InternalCache cache; |
| |
| protected String id; |
| |
| protected long startTime; |
| |
| protected PoolImpl proxy; |
| |
| protected int remoteDSId; |
| |
| protected String locName; |
| |
| protected int socketBufferSize; |
| |
| protected int socketReadTimeout; |
| |
| protected int queueMemory; |
| |
| protected int maxMemoryPerDispatcherQueue; |
| |
| protected int batchSize; |
| |
| protected int batchTimeInterval; |
| |
| protected boolean isConflation; |
| |
| protected boolean isPersistence; |
| |
| protected int alertThreshold; |
| |
| protected boolean manualStart; |
| |
| protected boolean isParallel; |
| |
| protected boolean groupTransactionEvents; |
| |
| protected boolean isForInternalUse; |
| |
| protected boolean isDiskSynchronous; |
| |
| protected String diskStoreName; |
| |
| protected List<GatewayEventFilter> eventFilters; |
| |
| protected List<GatewayTransportFilter> transFilters; |
| |
| protected List<AsyncEventListener> listeners; |
| |
| protected boolean forwardExpirationDestroy; |
| |
| protected GatewayEventSubstitutionFilter substitutionFilter; |
| |
| protected LocatorDiscoveryCallback locatorDiscoveryCallback; |
| |
| private final ReentrantReadWriteLock lifeCycleLock = new ReentrantReadWriteLock(); |
| |
| protected GatewaySenderAdvisor senderAdvisor; |
| |
| private int serialNumber; |
| |
| protected GatewaySenderStats statistics; |
| |
| private Stopper stopper; |
| |
| private OrderPolicy policy; |
| |
| private int dispatcherThreads; |
| |
| protected boolean isBucketSorted; |
| |
| protected boolean isMetaQueue; |
| |
| private int parallelismForReplicatedRegion; |
| |
| protected AbstractGatewaySenderEventProcessor eventProcessor; |
| |
| private org.apache.geode.internal.cache.GatewayEventFilter filter = |
| DefaultGatewayEventFilter.getInstance(); |
| |
| private ServerLocation serverLocation; |
| |
| private String expectedReceiverUniqueId = ""; |
| |
| protected Object queuedEventsSync = new Object(); |
| |
| protected volatile boolean enqueuedAllTempQueueEvents = false; |
| |
| protected volatile ConcurrentLinkedQueue<TmpQueueEvent> tmpQueuedEvents = |
| new ConcurrentLinkedQueue<>(); |
| |
| protected volatile ConcurrentLinkedQueue<EntryEventImpl> tmpDroppedEvents = |
| new ConcurrentLinkedQueue<>(); |
| /** |
| * The number of seconds to wait before stopping the GatewaySender. Default is 0 seconds. |
| */ |
| @MutableForTesting |
| public static int MAXIMUM_SHUTDOWN_WAIT_TIME = |
| Integer.getInteger("GatewaySender.MAXIMUM_SHUTDOWN_WAIT_TIME", 0).intValue(); |
| |
| /** |
| * The number of times to peek on shutdown before giving up and shutting down. |
| */ |
| protected static final int MAXIMUM_SHUTDOWN_PEEKS = |
| Integer.getInteger("GatewaySender.MAXIMUM_SHUTDOWN_PEEKS", 20).intValue(); |
| |
| public static final int QUEUE_SIZE_THRESHOLD = |
| Integer.getInteger("GatewaySender.QUEUE_SIZE_THRESHOLD", 5000).intValue(); |
| |
| @MutableForTesting |
| public static int TOKEN_TIMEOUT = |
| Integer.getInteger("GatewaySender.TOKEN_TIMEOUT", 120000).intValue(); |
| |
| /** |
| * The name of the DistributedLockService used when accessing the GatewaySender's meta data |
| * region. |
| */ |
| public static final String LOCK_SERVICE_NAME = "gatewayEventIdIndexMetaData_lockService"; |
| |
| /** |
| * The name of the GatewaySender's meta data region. |
| */ |
| protected static final String META_DATA_REGION_NAME = "gatewayEventIdIndexMetaData"; |
| |
| protected boolean startEventProcessorInPausedState = false; |
| |
| protected int myDSId = DEFAULT_DISTRIBUTED_SYSTEM_ID; |
| |
| protected int connectionIdleTimeOut = GATEWAY_CONNECTION_IDLE_TIMEOUT; |
| |
| private boolean removeFromQueueOnException = GatewaySender.REMOVE_FROM_QUEUE_ON_EXCEPTION; |
| |
| /** |
| * A unique (per <code>GatewaySender</code> id) index used when modifying <code>EventIDs</code>. |
| * Unlike the serialNumber, the eventIdIndex matches for the same <code>GatewaySender</code> |
| * across all members of the <code>DistributedSystem</code>. |
| */ |
| private int eventIdIndex; |
| |
| /** |
| * A <code>Region</code> used for storing <code>GatewaySender</code> event id indexes. This |
| * <code>Region</code> along with a <code>DistributedLock</code> facilitates creation of unique |
| * indexes across members. |
| */ |
| private Region<String, Integer> eventIdIndexMetaDataRegion; |
| |
| final Object lockForConcurrentDispatcher = new Object(); |
| |
| private final StatisticsClock statisticsClock; |
| |
| protected boolean enforceThreadsConnectSameReceiver; |
| |
| protected AbstractGatewaySender() { |
| statisticsClock = disabledClock(); |
| } |
| |
| public AbstractGatewaySender(InternalCache cache, StatisticsClock statisticsClock, |
| GatewaySenderAttributes attrs) { |
| this.cache = cache; |
| this.statisticsClock = statisticsClock; |
| this.id = attrs.getId(); |
| this.socketBufferSize = attrs.getSocketBufferSize(); |
| this.socketReadTimeout = attrs.getSocketReadTimeout(); |
| this.queueMemory = attrs.getMaximumQueueMemory(); |
| this.batchSize = attrs.getBatchSize(); |
| this.batchTimeInterval = attrs.getBatchTimeInterval(); |
| this.isConflation = attrs.isBatchConflationEnabled(); |
| this.isPersistence = attrs.isPersistenceEnabled(); |
| this.alertThreshold = attrs.getAlertThreshold(); |
| this.manualStart = attrs.isManualStart(); |
| this.isParallel = attrs.isParallel(); |
| this.groupTransactionEvents = attrs.mustGroupTransactionEvents(); |
| this.isForInternalUse = attrs.isForInternalUse(); |
| this.diskStoreName = attrs.getDiskStoreName(); |
| this.remoteDSId = attrs.getRemoteDSId(); |
| this.eventFilters = attrs.getGatewayEventFilters(); |
| this.transFilters = attrs.getGatewayTransportFilters(); |
| this.listeners = attrs.getAsyncEventListeners(); |
| this.substitutionFilter = attrs.getGatewayEventSubstitutionFilter(); |
| this.locatorDiscoveryCallback = attrs.getGatewayLocatoDiscoveryCallback(); |
| this.isDiskSynchronous = attrs.isDiskSynchronous(); |
| this.policy = attrs.getOrderPolicy(); |
| this.dispatcherThreads = attrs.getDispatcherThreads(); |
| this.parallelismForReplicatedRegion = attrs.getParallelismForReplicatedRegion(); |
| // divide the maximumQueueMemory of sender equally using number of dispatcher threads. |
| // if dispatcherThreads is 1 then maxMemoryPerDispatcherQueue will be same as maximumQueueMemory |
| // of sender |
| this.maxMemoryPerDispatcherQueue = this.queueMemory / this.dispatcherThreads; |
| this.serialNumber = DistributionAdvisor.createSerialNumber(); |
| this.isMetaQueue = attrs.isMetaQueue(); |
| this.enforceThreadsConnectSameReceiver = attrs.getEnforceThreadsConnectSameReceiver(); |
| if (!(this.cache instanceof CacheCreation)) { |
| this.myDSId = this.cache.getInternalDistributedSystem().getDistributionManager() |
| .getDistributedSystemId(); |
| this.stopper = new Stopper(cache.getCancelCriterion()); |
| this.senderAdvisor = GatewaySenderAdvisor.createGatewaySenderAdvisor(this); |
| if (!this.isForInternalUse()) { |
| this.statistics = new GatewaySenderStats(cache.getDistributedSystem(), |
| "gatewaySenderStats-", id, statisticsClock); |
| } |
| initializeEventIdIndex(); |
| } |
| this.isBucketSorted = attrs.isBucketSorted(); |
| this.forwardExpirationDestroy = attrs.isForwardExpirationDestroy(); |
| } |
| |
| public GatewaySenderAdvisor getSenderAdvisor() { |
| return senderAdvisor; |
| } |
| |
| @Override |
| public GatewaySenderStats getStatistics() { |
| return statistics; |
| } |
| |
| @Override |
| public StatisticsClock getStatisticsClock() { |
| return statisticsClock; |
| } |
| |
| public void initProxy() { |
| // no op |
| } |
| |
| @Override |
| public boolean isPrimary() { |
| return this.getSenderAdvisor().isPrimary(); |
| } |
| |
| public void setIsPrimary(boolean isPrimary) { |
| this.getSenderAdvisor().setIsPrimary(isPrimary); |
| } |
| |
| @Override |
| public InternalCache getCache() { |
| return this.cache; |
| } |
| |
| @Override |
| public int getAlertThreshold() { |
| return this.alertThreshold; |
| } |
| |
| @Override |
| public int getBatchSize() { |
| return this.batchSize; |
| } |
| |
| @Override |
| public int getBatchTimeInterval() { |
| return this.batchTimeInterval; |
| } |
| |
| @Override |
| public String getDiskStoreName() { |
| return this.diskStoreName; |
| } |
| |
| @Override |
| public List<GatewayEventFilter> getGatewayEventFilters() { |
| return this.eventFilters; |
| } |
| |
| @Override |
| public GatewayEventSubstitutionFilter getGatewayEventSubstitutionFilter() { |
| return this.substitutionFilter; |
| } |
| |
| @Override |
| public String getId() { |
| return this.id; |
| } |
| |
| public long getStartTime() { |
| return this.startTime; |
| } |
| |
| @Override |
| public int getRemoteDSId() { |
| return this.remoteDSId; |
| } |
| |
| @Override |
| public List<GatewayTransportFilter> getGatewayTransportFilters() { |
| return this.transFilters; |
| } |
| |
| public List<AsyncEventListener> getAsyncEventListeners() { |
| return this.listeners; |
| } |
| |
| public boolean hasListeners() { |
| return !this.listeners.isEmpty(); |
| } |
| |
| @Override |
| public boolean isForwardExpirationDestroy() { |
| return this.forwardExpirationDestroy; |
| } |
| |
| @Override |
| public boolean isManualStart() { |
| return this.manualStart; |
| } |
| |
| @Override |
| public int getMaximumQueueMemory() { |
| return this.queueMemory; |
| } |
| |
| public int getMaximumMemeoryPerDispatcherQueue() { |
| return this.maxMemoryPerDispatcherQueue; |
| } |
| |
| @Override |
| public int getSocketBufferSize() { |
| return this.socketBufferSize; |
| } |
| |
| @Override |
| public int getSocketReadTimeout() { |
| return this.socketReadTimeout; |
| } |
| |
| @Override |
| public boolean isBatchConflationEnabled() { |
| return this.isConflation; |
| } |
| |
| public void test_setBatchConflationEnabled(boolean enableConflation) { |
| this.isConflation = enableConflation; |
| } |
| |
| @Override |
| public boolean isPersistenceEnabled() { |
| return this.isPersistence; |
| } |
| |
| @Override |
| public boolean isDiskSynchronous() { |
| return this.isDiskSynchronous; |
| } |
| |
| @Override |
| public int getMaxParallelismForReplicatedRegion() { |
| return this.parallelismForReplicatedRegion; |
| } |
| |
| public LocatorDiscoveryCallback getLocatorDiscoveryCallback() { |
| return this.locatorDiscoveryCallback; |
| } |
| |
| @Override |
| public DistributionAdvisor getDistributionAdvisor() { |
| return this.senderAdvisor; |
| } |
| |
| @Override |
| public DistributionManager getDistributionManager() { |
| return getSystem().getDistributionManager(); |
| } |
| |
| @Override |
| public String getFullPath() { |
| return getId(); |
| } |
| |
| @Override |
| public String getName() { |
| return getId(); |
| } |
| |
| @Override |
| public DistributionAdvisee getParentAdvisee() { |
| return null; |
| } |
| |
| @Override |
| public int getDispatcherThreads() { |
| return this.dispatcherThreads; |
| } |
| |
| @Override |
| public OrderPolicy getOrderPolicy() { |
| return this.policy; |
| } |
| |
| @Override |
| public Profile getProfile() { |
| return this.senderAdvisor.createProfile(); |
| } |
| |
| @Override |
| public int getSerialNumber() { |
| return this.serialNumber; |
| } |
| |
| public boolean getBucketSorted() { |
| return this.isBucketSorted; |
| } |
| |
| @Override |
| public boolean getIsMetaQueue() { |
| return this.isMetaQueue; |
| } |
| |
| @Override |
| public InternalDistributedSystem getSystem() { |
| return this.cache.getInternalDistributedSystem(); |
| } |
| |
| public int getEventIdIndex() { |
| return this.eventIdIndex; |
| } |
| |
| @Override |
| public boolean getEnforceThreadsConnectSameReceiver() { |
| return this.enforceThreadsConnectSameReceiver; |
| } |
| |
| @Override |
| public boolean equals(Object obj) { |
| if (obj == null) { |
| return false; |
| } |
| if (this == obj) { |
| return true; |
| } |
| if (!(obj instanceof GatewaySender)) { |
| return false; |
| } |
| AbstractGatewaySender sender = (AbstractGatewaySender) obj; |
| if (sender.getId().equals(this.getId())) { |
| return true; |
| } |
| return false; |
| } |
| |
| @Override |
| public int hashCode() { |
| return this.getId().hashCode(); |
| } |
| |
| public PoolImpl getProxy() { |
| return proxy; |
| } |
| |
| @Override |
| public void removeGatewayEventFilter(GatewayEventFilter filter) { |
| this.eventFilters.remove(filter); |
| } |
| |
| @Override |
| public void addGatewayEventFilter(GatewayEventFilter filter) { |
| if (this.eventFilters.isEmpty()) { |
| this.eventFilters = new ArrayList<GatewayEventFilter>(); |
| } |
| if (filter == null) { |
| throw new IllegalStateException( |
| "null value can not be added to gateway-event-filters list"); |
| } |
| this.eventFilters.add(filter); |
| } |
| |
| @Override |
| public boolean isParallel() { |
| return this.isParallel; |
| } |
| |
| @Override |
| public boolean mustGroupTransactionEvents() { |
| return this.groupTransactionEvents; |
| } |
| |
| public boolean isForInternalUse() { |
| return this.isForInternalUse; |
| } |
| |
| @Override |
| public abstract void start(); |
| |
| @Override |
| public abstract void startWithCleanQueue(); |
| |
| @Override |
| public abstract void stop(); |
| |
| /** |
| * Destroys the GatewaySender. Before destroying the sender, caller needs to to ensure that the |
| * sender is stopped so that all the resources (threads, connection pool etc.) will be released |
| * properly. Stopping the sender is not handled in the destroy. Destroy is carried out in |
| * following steps: 1. Take the lifeCycleLock. 2. If the sender is attached to any application |
| * region, throw an exception. 3. Close the GatewaySenderAdvisor. 4. Remove the sender from the |
| * cache. 5. Destroy the region underlying the GatewaySender. |
| * |
| * In case of ParallelGatewaySender, the destroy operation does distributed destroy of the QPR. In |
| * case of SerialGatewaySender, the queue region is destroyed locally. |
| */ |
| @Override |
| public void destroy() { |
| destroy(true); |
| } |
| |
| @Override |
| public void destroy(boolean initiator) { |
| try { |
| this.getLifeCycleLock().writeLock().lock(); |
| // first, check if this sender is attached to any region. If so, throw |
| // GatewaySenderException |
| Set<InternalRegion> regions = this.cache.getApplicationRegions(); |
| Iterator regionItr = regions.iterator(); |
| while (regionItr.hasNext()) { |
| LocalRegion region = (LocalRegion) regionItr.next(); |
| |
| if (region.getAttributes().getGatewaySenderIds().contains(this.id)) { |
| throw new GatewaySenderException( |
| String.format( |
| "The GatewaySender %s could not be destroyed as it is still used by region(s).", |
| this)); |
| } |
| } |
| // close the GatewaySenderAdvisor |
| GatewaySenderAdvisor advisor = this.getSenderAdvisor(); |
| if (advisor != null) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("Stopping the GatewaySender advisor"); |
| } |
| advisor.close(); |
| } |
| |
| // remove the sender from the cache |
| this.cache.removeGatewaySender(this); |
| |
| // destroy the region underneath the sender's queue |
| if (initiator) { |
| Set<RegionQueue> regionQueues = getQueues(); |
| if (regionQueues != null) { |
| for (RegionQueue regionQueue : regionQueues) { |
| try { |
| if (regionQueue instanceof ConcurrentParallelGatewaySenderQueue) { |
| Set<PartitionedRegion> queueRegions = |
| ((ConcurrentParallelGatewaySenderQueue) regionQueue).getRegions(); |
| for (PartitionedRegion queueRegion : queueRegions) { |
| queueRegion.destroyRegion(); |
| } |
| } else {// For SerialGatewaySenderQueue, do local destroy |
| regionQueue.getRegion().localDestroyRegion(); |
| } |
| } |
| // Can occur in case of ParallelGatewaySenderQueue, when the region is |
| // being destroyed |
| // by several nodes simultaneously |
| catch (RegionDestroyedException e) { |
| // the region might have already been destroyed by other node. Just |
| // log |
| // the exception. |
| this.logger.info( |
| "Region {} that underlies the GatewaySender {} is already destroyed.", |
| e.getRegionFullPath(), this); |
| } |
| } |
| } // END if (regionQueues != null) |
| } |
| } finally { |
| this.getLifeCycleLock().writeLock().unlock(); |
| } |
| } |
| |
| @Override |
| public void rebalance() { |
| try { |
| // Pause the sender |
| pause(); |
| |
| // Rebalance the event processor if necessary |
| if (this.eventProcessor != null) { |
| this.eventProcessor.rebalance(); |
| } |
| } finally { |
| // Resume the sender |
| resume(); |
| } |
| logger.info( |
| "GatewaySender {} has been rebalanced", this); |
| } |
| |
| public boolean beforeEnqueue(GatewayQueueEvent gatewayEvent) { |
| boolean enqueue = true; |
| for (GatewayEventFilter filter : getGatewayEventFilters()) { |
| enqueue = filter.beforeEnqueue(gatewayEvent); |
| if (!enqueue) { |
| return enqueue; |
| } |
| } |
| return enqueue; |
| } |
| |
| protected void stopProcessing() { |
| // Stop the dispatcher |
| AbstractGatewaySenderEventProcessor ev = this.eventProcessor; |
| if (ev != null && !ev.isStopped()) { |
| ev.stopProcessing(); |
| } |
| |
| if (ev != null && ev.getDispatcher() != null) { |
| ev.getDispatcher().shutDownAckReaderConnection(); |
| } |
| } |
| |
| protected void stompProxyDead() { |
| Runnable stomper = new Runnable() { |
| @Override |
| public void run() { |
| PoolImpl bpi = proxy; |
| if (bpi != null) { |
| try { |
| bpi.destroy(); |
| } catch (Exception e) {/* ignore */ |
| } |
| } |
| } |
| }; |
| Thread t = new LoggingThread("GatewaySender Proxy Stomper", stomper); |
| t.start(); |
| try { |
| t.join(GATEWAY_SENDER_TIMEOUT * 1000); |
| return; |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| } |
| |
| logger.warn("Gateway <{}> is not closing cleanly; forcing cancellation.", this); |
| // OK, either we've timed out or been interrupted. Time for |
| // violence. |
| t.interrupt(); // give up |
| proxy.emergencyClose(); // VIOLENCE! |
| this.proxy = null; |
| } |
| |
| public int getMyDSId() { |
| return this.myDSId; |
| } |
| |
| /** |
| * @param removeFromQueueOnException the removeFromQueueOnException to set |
| */ |
| public void setRemoveFromQueueOnException(boolean removeFromQueueOnException) { |
| this.removeFromQueueOnException = removeFromQueueOnException; |
| } |
| |
| /** |
| * @return the removeFromQueueOnException |
| */ |
| public boolean isRemoveFromQueueOnException() { |
| return removeFromQueueOnException; |
| } |
| |
| public CancelCriterion getStopper() { |
| return this.stopper; |
| } |
| |
| @Override |
| public CancelCriterion getCancelCriterion() { |
| return stopper; |
| } |
| |
| public synchronized ServerLocation getServerLocation() { |
| return serverLocation; |
| } |
| |
| public synchronized boolean setServerLocation(ServerLocation location) { |
| this.serverLocation = location; |
| return true; |
| } |
| |
| private class Stopper extends CancelCriterion { |
| final CancelCriterion stper; |
| |
| Stopper(CancelCriterion stopper) { |
| this.stper = stopper; |
| } |
| |
| @Override |
| public String cancelInProgress() { |
| // checkFailure(); // done by stopper |
| return stper.cancelInProgress(); |
| } |
| |
| @Override |
| public RuntimeException generateCancelledException(Throwable e) { |
| RuntimeException result = stper.generateCancelledException(e); |
| return result; |
| } |
| } |
| |
| public RegionQueue getQueue() { |
| if (this.eventProcessor != null) { |
| if (!(this.eventProcessor instanceof ConcurrentSerialGatewaySenderEventProcessor)) { |
| return this.eventProcessor.getQueue(); |
| } else { |
| throw new IllegalArgumentException("getQueue() for concurrent serial gateway sender"); |
| } |
| } |
| return null; |
| } |
| |
| @Override |
| public Set<RegionQueue> getQueues() { |
| if (this.eventProcessor != null) { |
| if (!(this.eventProcessor instanceof ConcurrentSerialGatewaySenderEventProcessor)) { |
| Set<RegionQueue> queues = new HashSet<RegionQueue>(); |
| queues.add(this.eventProcessor.getQueue()); |
| return queues; |
| } |
| return ((ConcurrentSerialGatewaySenderEventProcessor) this.eventProcessor).getQueues(); |
| } |
| return null; |
| } |
| |
| protected void waitForRunningStatus() { |
| synchronized (this.eventProcessor.getRunningStateLock()) { |
| while (this.eventProcessor.getException() == null && this.eventProcessor.isStopped()) { |
| try { |
| this.eventProcessor.getRunningStateLock().wait(1000); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| } |
| } |
| Exception ex = this.eventProcessor.getException(); |
| if (ex != null) { |
| throw new GatewaySenderException( |
| String.format("Could not start a gateway sender %s because of exception %s", |
| new Object[] {this.getId(), ex.getMessage()}), |
| ex.getCause()); |
| } |
| } |
| } |
| |
| public boolean isStartEventProcessorInPausedState() { |
| return startEventProcessorInPausedState; |
| } |
| |
| @Override |
| public void setStartEventProcessorInPausedState() { |
| startEventProcessorInPausedState = true; |
| } |
| |
| /** |
| * This pause will set the pause flag even if the |
| * processor has not yet started. |
| */ |
| public void pauseEvenIfProcessorStopped() { |
| if (this.eventProcessor != null) { |
| this.getLifeCycleLock().writeLock().lock(); |
| try { |
| this.eventProcessor.pauseDispatching(); |
| InternalDistributedSystem system = |
| (InternalDistributedSystem) this.cache.getDistributedSystem(); |
| system.handleResourceEvent(ResourceEvent.GATEWAYSENDER_PAUSE, this); |
| logger.info("Paused {}", this); |
| |
| enqueueTempEvents(); |
| } finally { |
| this.getLifeCycleLock().writeLock().unlock(); |
| } |
| } |
| } |
| |
| @Override |
| public void pause() { |
| if (this.eventProcessor != null) { |
| this.getLifeCycleLock().writeLock().lock(); |
| try { |
| if (this.eventProcessor.isStopped()) { |
| return; |
| } |
| this.eventProcessor.pauseDispatching(); |
| |
| InternalDistributedSystem system = |
| (InternalDistributedSystem) this.cache.getDistributedSystem(); |
| system.handleResourceEvent(ResourceEvent.GATEWAYSENDER_PAUSE, this); |
| |
| logger.info("Paused {}", this); |
| |
| enqueueTempEvents(); |
| } finally { |
| this.getLifeCycleLock().writeLock().unlock(); |
| } |
| } |
| } |
| |
| @Override |
| public void resume() { |
| if (this.eventProcessor != null) { |
| this.getLifeCycleLock().writeLock().lock(); |
| try { |
| if (this.eventProcessor.isStopped()) { |
| return; |
| } |
| this.eventProcessor.resumeDispatching(); |
| |
| |
| InternalDistributedSystem system = |
| (InternalDistributedSystem) this.cache.getDistributedSystem(); |
| system.handleResourceEvent(ResourceEvent.GATEWAYSENDER_RESUME, this); |
| |
| logger.info("Resumed {}", this); |
| |
| enqueueTempEvents(); |
| } finally { |
| this.getLifeCycleLock().writeLock().unlock(); |
| } |
| } |
| } |
| |
| @Override |
| public boolean isPaused() { |
| if (this.eventProcessor != null) { |
| return this.eventProcessor.isPaused(); |
| } |
| return false; |
| } |
| |
| @Override |
| public boolean isRunning() { |
| if (this.eventProcessor != null) { |
| return !this.eventProcessor.isStopped(); |
| } |
| return false; |
| } |
| |
| @Override |
| public AbstractGatewaySenderEventProcessor getEventProcessor() { |
| return this.eventProcessor; |
| } |
| |
| /** |
| * Check if this event can be distributed by senders. |
| * |
| * @return boolean True if the event is allowed. |
| */ |
| private boolean checkForDistribution(EntryEventImpl event, GatewaySenderStats stats) { |
| if (event.getRegion().getDataPolicy().equals(DataPolicy.NORMAL)) { |
| return false; |
| } |
| // Check for eviction and expiration events. |
| if (event.getOperation().isLocal() || event.getOperation().isExpiration()) { |
| // Check if its AEQ and is configured to forward expiration destroy events. |
| if (event.getOperation().isExpiration() && this.isAsyncEventQueue() |
| && this.isForwardExpirationDestroy()) { |
| return true; |
| } |
| return false; |
| } |
| return true; |
| } |
| |
| public void distribute(EnumListenerEvent operation, EntryEventImpl event, |
| List<Integer> allRemoteDSIds) { |
| distribute(operation, event, allRemoteDSIds, false); |
| } |
| |
| public void distribute(EnumListenerEvent operation, EntryEventImpl event, |
| List<Integer> allRemoteDSIds, boolean isLastEventInTransaction) { |
| |
| final boolean isDebugEnabled = logger.isDebugEnabled(); |
| |
| // released by this method or transfers ownership to TmpQueueEvent |
| @Released |
| EntryEventImpl clonedEvent = new EntryEventImpl(event, false); |
| boolean freeClonedEvent = true; |
| try { |
| |
| final GatewaySenderStats stats = getStatistics(); |
| stats.incEventsReceived(); |
| |
| if (!checkForDistribution(event, stats)) { |
| stats.incEventsNotQueued(); |
| return; |
| } |
| |
| // this filter is defined by Asif which exist in old wan too. new wan has |
| // other GatewaEventFilter. Do we need to get rid of this filter. Cheetah is |
| // not considering this filter |
| if (!this.filter.enqueueEvent(event)) { |
| stats.incEventsFiltered(); |
| return; |
| } |
| |
| // start to distribute |
| setModifiedEventId(clonedEvent); |
| Object callbackArg = clonedEvent.getRawCallbackArgument(); |
| |
| if (isDebugEnabled) { |
| // We can't deserialize here for logging purposes so don't |
| // call getNewValue. |
| // event.getNewValue(); // to deserialize the value if necessary |
| logger.debug("{} : About to notify {} to perform operation {} for {} callback arg {}", |
| this.isPrimary(), getId(), operation, clonedEvent, callbackArg); |
| } |
| |
| if (callbackArg instanceof GatewaySenderEventCallbackArgument) { |
| GatewaySenderEventCallbackArgument seca = (GatewaySenderEventCallbackArgument) callbackArg; |
| if (isDebugEnabled) { |
| logger.debug( |
| "{}: Event originated in {}. My DS id is {}. The remote DS id is {}. The recipients are: {}", |
| this, seca.getOriginatingDSId(), this.getMyDSId(), this.getRemoteDSId(), |
| seca.getRecipientDSIds()); |
| } |
| if (seca.getOriginatingDSId() == DEFAULT_DISTRIBUTED_SYSTEM_ID) { |
| if (isDebugEnabled) { |
| logger.debug( |
| "{}: Event originated in {}. My DS id is {}. The remote DS id is {}. The recipients are: {}", |
| this, seca.getOriginatingDSId(), this.getMyDSId(), this.getRemoteDSId(), |
| seca.getRecipientDSIds()); |
| } |
| |
| seca.setOriginatingDSId(this.getMyDSId()); |
| seca.initializeReceipientDSIds(allRemoteDSIds); |
| |
| } else { |
| // if the dispatcher is GatewaySenderEventCallbackDispatcher (which is the case of WBCL), |
| // skip the below check of remoteDSId. |
| // Fix for #46517 |
| AbstractGatewaySenderEventProcessor ep = getEventProcessor(); |
| // if manual-start is true, ep is null |
| if (ep == null || !(ep.getDispatcher() instanceof GatewaySenderEventCallbackDispatcher)) { |
| if (seca.getOriginatingDSId() == this.getRemoteDSId()) { |
| if (isDebugEnabled) { |
| logger.debug( |
| "{}: Event originated in {}. My DS id is {}. It is being dropped as remote is originator.", |
| this, seca.getOriginatingDSId(), getMyDSId()); |
| } |
| return; |
| } else if (seca.getRecipientDSIds().contains(this.getRemoteDSId())) { |
| if (isDebugEnabled) { |
| logger.debug( |
| "{}: Event originated in {}. My DS id is {}. The remote DS id is {}.. It is being dropped as remote ds is already a recipient. Recipients are: {}", |
| this, seca.getOriginatingDSId(), getMyDSId(), this.getRemoteDSId(), |
| seca.getRecipientDSIds()); |
| } |
| return; |
| } |
| } |
| seca.getRecipientDSIds().addAll(allRemoteDSIds); |
| } |
| } else { |
| GatewaySenderEventCallbackArgument geCallbackArg = |
| new GatewaySenderEventCallbackArgument(callbackArg, this.getMyDSId(), allRemoteDSIds); |
| clonedEvent.setCallbackArgument(geCallbackArg); |
| } |
| |
| // If this gateway is not running, return |
| if (!isRunning()) { |
| if (this.isPrimary()) { |
| recordDroppedEvent(clonedEvent); |
| } |
| if (isDebugEnabled) { |
| logger.debug("Returning back without putting into the gateway sender queue:" + event); |
| } |
| return; |
| } |
| |
| if (!this.getLifeCycleLock().readLock().tryLock()) { |
| synchronized (this.queuedEventsSync) { |
| if (!this.enqueuedAllTempQueueEvents) { |
| if (!this.getLifeCycleLock().readLock().tryLock()) { |
| Object substituteValue = getSubstituteValue(clonedEvent, operation); |
| this.tmpQueuedEvents.add(new TmpQueueEvent(operation, clonedEvent, substituteValue)); |
| freeClonedEvent = false; |
| stats.incTempQueueSize(); |
| if (isDebugEnabled) { |
| logger.debug("Event : {} is added to TempQueue", clonedEvent); |
| } |
| return; |
| } |
| } |
| } |
| if (this.enqueuedAllTempQueueEvents) { |
| this.getLifeCycleLock().readLock().lock(); |
| } |
| } |
| try { |
| // If this gateway is not running, return |
| // The sender may have stopped, after we have checked the status in the beginning. |
| if (!isRunning()) { |
| if (isDebugEnabled) { |
| logger.debug("Returning back without putting into the gateway sender queue:" + event); |
| } |
| if (this.isPrimary()) { |
| recordDroppedEvent(clonedEvent); |
| } |
| return; |
| } |
| |
| try { |
| AbstractGatewaySenderEventProcessor ev = this.eventProcessor; |
| if (ev == null) { |
| getStopper().checkCancelInProgress(null); |
| this.getCache().getDistributedSystem().getCancelCriterion().checkCancelInProgress(null); |
| // event processor will be null if there was an authorization |
| // problem |
| // connecting to the other site (bug #40681) |
| if (ev == null) { |
| throw new GatewayCancelledException("Event processor thread is gone"); |
| } |
| } |
| |
| // Get substitution value to enqueue if necessary |
| Object substituteValue = getSubstituteValue(clonedEvent, operation); |
| |
| ev.enqueueEvent(operation, clonedEvent, substituteValue, isLastEventInTransaction); |
| } catch (CancelException e) { |
| logger.debug("caught cancel exception", e); |
| throw e; |
| } catch (RegionDestroyedException e) { |
| logger.warn(String.format( |
| "%s: An Exception occurred while queueing %s to perform operation %s for %s", |
| new Object[] {this, getId(), operation, clonedEvent}), |
| e); |
| } catch (Exception e) { |
| logger.fatal(String.format( |
| "%s: An Exception occurred while queueing %s to perform operation %s for %s", |
| new Object[] {this, getId(), operation, clonedEvent}), |
| e); |
| } |
| } finally { |
| this.getLifeCycleLock().readLock().unlock(); |
| } |
| } finally { |
| if (freeClonedEvent) { |
| clonedEvent.release(); // fix for bug 48035 |
| } |
| } |
| } |
| |
| private void recordDroppedEvent(EntryEventImpl event) { |
| if (this.eventProcessor != null) { |
| this.eventProcessor.registerEventDroppedInPrimaryQueue(event); |
| } else { |
| tmpDroppedEvents.add(event); |
| if (logger.isDebugEnabled()) { |
| logger.debug("added to tmpDroppedEvents event: {}", event); |
| } |
| } |
| } |
| |
| @VisibleForTesting |
| int getTmpDroppedEventSize() { |
| return tmpDroppedEvents.size(); |
| } |
| |
| /** |
| * During sender is getting started, if there are any cache operation on queue then that event |
| * will be stored in temp queue. Once sender is started, these event from tmp queue will be added |
| * to sender queue. |
| * |
| * Apart from sender's start() method, this method also gets called from |
| * ParallelGatewaySenderQueue.addPartitionedRegionForRegion(). This is done to support the |
| * postCreateRegion scenario i.e. the sender is already running and region is created later. The |
| * eventProcessor can be null when the method gets invoked through this flow: |
| * ParallelGatewaySenderImpl.start() -> ParallelGatewaySenderQueue.<init> -> |
| * ParallelGatewaySenderQueue.addPartitionedRegionForRegion |
| */ |
| public void enqueueTempEvents() { |
| if (this.eventProcessor != null) {// Fix for defect #47308 |
| // process tmpDroppedEvents |
| EntryEventImpl droppedEvent; |
| while ((droppedEvent = tmpDroppedEvents.poll()) != null) { |
| this.eventProcessor.registerEventDroppedInPrimaryQueue(droppedEvent); |
| } |
| |
| TmpQueueEvent nextEvent = null; |
| final GatewaySenderStats stats = getStatistics(); |
| try { |
| // Now finish emptying the queue with synchronization to make |
| // sure we don't miss any events. |
| synchronized (this.queuedEventsSync) { |
| while ((nextEvent = tmpQueuedEvents.poll()) != null) { |
| try { |
| if (logger.isDebugEnabled()) { |
| logger.debug("Event :{} is enqueued to GatewaySenderQueue from TempQueue", |
| nextEvent); |
| } |
| stats.decTempQueueSize(); |
| this.eventProcessor.enqueueEvent(nextEvent.getOperation(), nextEvent.getEvent(), |
| nextEvent.getSubstituteValue()); |
| } finally { |
| nextEvent.release(); |
| } |
| } |
| this.enqueuedAllTempQueueEvents = true; |
| } |
| } catch (CacheException e) { |
| logger.debug("caught cancel exception", e); |
| } catch (IOException e) { |
| logger.fatal(String.format( |
| "%s: An Exception occurred while queueing %s to perform operation %s for %s", |
| new Object[] {this, getId(), nextEvent.getOperation(), nextEvent}), |
| e); |
| } |
| } |
| } |
| |
| /** |
| * Removes the EntryEventImpl, whose tailKey matches with the provided tailKey, from |
| * tmpQueueEvents. |
| * |
| */ |
| public boolean removeFromTempQueueEvents(Object tailKey) { |
| synchronized (this.queuedEventsSync) { |
| Iterator<TmpQueueEvent> itr = this.tmpQueuedEvents.iterator(); |
| while (itr.hasNext()) { |
| TmpQueueEvent event = itr.next(); |
| if (tailKey.equals(event.getEvent().getTailKey())) { |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "shadowKey {} is found in tmpQueueEvents at AbstractGatewaySender level. Removing from there..", |
| tailKey); |
| } |
| event.release(); |
| itr.remove(); |
| return true; |
| } |
| } |
| return false; |
| } |
| } |
| |
| /** |
| * During sender is getting stopped, if there are any cache operation on queue then that event |
| * will be stored in temp queue. Once sender is started, these event from tmp queue will be |
| * cleared. |
| */ |
| public void clearTempEventsAfterSenderStopped() { |
| TmpQueueEvent nextEvent = null; |
| while ((nextEvent = tmpQueuedEvents.poll()) != null) { |
| nextEvent.release(); |
| } |
| synchronized (this.queuedEventsSync) { |
| while ((nextEvent = tmpQueuedEvents.poll()) != null) { |
| nextEvent.release(); |
| } |
| this.enqueuedAllTempQueueEvents = false; |
| } |
| |
| statistics.setQueueSize(0); |
| statistics.setSecondaryQueueSize(0); |
| statistics.setEventsProcessedByPQRM(0); |
| statistics.setTempQueueSize(0); |
| } |
| |
| public Object getSubstituteValue(EntryEventImpl clonedEvent, EnumListenerEvent operation) { |
| // Get substitution value to enqueue if necessary |
| Object substituteValue = null; |
| if (this.substitutionFilter != null) { |
| try { |
| substituteValue = this.substitutionFilter.getSubstituteValue(clonedEvent); |
| // If null is returned from the filter, null is set in the value |
| if (substituteValue == null) { |
| substituteValue = GatewaySenderEventImpl.TOKEN_NULL; |
| } |
| } catch (Exception e) { |
| // Log any exceptions that occur in the filter and use the original value. |
| logger.warn(String.format( |
| "%s: An Exception occurred while queueing %s to perform operation %s for %s", |
| new Object[] {this, getId(), operation, clonedEvent}), |
| e); |
| } |
| } |
| return substituteValue; |
| } |
| |
| protected void initializeEventIdIndex() { |
| final boolean isDebugEnabled = logger.isDebugEnabled(); |
| |
| boolean gotLock = false; |
| try { |
| // Obtain the distributed lock |
| gotLock = getCache().getGatewaySenderLockService().lock(META_DATA_REGION_NAME, -1, -1); |
| if (!gotLock) { |
| throw new IllegalStateException( |
| String.format("%s: Failed to lock gateway event id index metadata region", |
| this)); |
| } else { |
| if (isDebugEnabled) { |
| logger.debug("{}: Locked the metadata region", this); |
| } |
| // Get metadata region |
| Region<String, Integer> region = getEventIdIndexMetaDataRegion(); |
| |
| // Get or create the index |
| int index = 0; |
| String messagePrefix = null; |
| if (region.containsKey(getId())) { |
| index = region.get(getId()); |
| if (isDebugEnabled) { |
| messagePrefix = "Using existing"; |
| } |
| } else { |
| index = region.size(); |
| if (index > ThreadIdentifier.Bits.GATEWAY_ID.mask()) { |
| throw new IllegalStateException( |
| String.format( |
| "Cannot create GatewaySender %s because the maximum (%s) has been reached", |
| getId(), ThreadIdentifier.Bits.GATEWAY_ID.mask() + 1)); |
| } |
| region.put(getId(), index); |
| if (isDebugEnabled) { |
| messagePrefix = "Created new"; |
| } |
| } |
| |
| // Store the index locally |
| this.eventIdIndex = index; |
| if (logger.isDebugEnabled()) { |
| logger.debug("{}: {} event id index: {}", this, messagePrefix, this.eventIdIndex); |
| } |
| } |
| } finally { |
| // Unlock the lock if necessary |
| if (gotLock) { |
| getCache().getGatewaySenderLockService().unlock(META_DATA_REGION_NAME); |
| if (isDebugEnabled) { |
| logger.debug("{}: Unlocked the metadata region", this); |
| } |
| } |
| } |
| } |
| |
| private Region<String, Integer> getEventIdIndexMetaDataRegion() { |
| if (this.eventIdIndexMetaDataRegion == null) { |
| this.eventIdIndexMetaDataRegion = initializeEventIdIndexMetaDataRegion(this); |
| } |
| return this.eventIdIndexMetaDataRegion; |
| } |
| |
| private static synchronized Region<String, Integer> initializeEventIdIndexMetaDataRegion( |
| AbstractGatewaySender sender) { |
| final InternalCache cache = sender.getCache(); |
| Region<String, Integer> region = cache.getRegion(META_DATA_REGION_NAME); |
| if (region == null) { |
| InternalRegionFactory<String, Integer> factory = |
| cache.createInternalRegionFactory(RegionShortcut.REPLICATE); |
| |
| // Create a stats holder for the meta data stats |
| final HasCachePerfStats statsHolder = new HasCachePerfStats() { |
| @Override |
| public CachePerfStats getCachePerfStats() { |
| return new CachePerfStats(cache.getDistributedSystem(), |
| "RegionStats-" + META_DATA_REGION_NAME, sender.statisticsClock); |
| } |
| }; |
| factory.setIsUsedForMetaRegion(true); |
| factory.setCachePerfStatsHolder(statsHolder); |
| try { |
| region = factory.create(META_DATA_REGION_NAME); |
| } catch (RegionExistsException e) { |
| region = cache.getRegion(META_DATA_REGION_NAME); |
| } catch (Exception e) { |
| throw new IllegalStateException( |
| String.format( |
| "%s: Caught the following exception attempting to create gateway event id index metadata region:", |
| sender), |
| e); |
| } |
| } |
| return region; |
| } |
| |
| public abstract void setModifiedEventId(EntryEventImpl clonedEvent); |
| |
| public static class DefaultGatewayEventFilter |
| implements org.apache.geode.internal.cache.GatewayEventFilter { |
| @Immutable |
| private static final DefaultGatewayEventFilter singleton = new DefaultGatewayEventFilter(); |
| |
| private DefaultGatewayEventFilter() {} |
| |
| public static org.apache.geode.internal.cache.GatewayEventFilter getInstance() { |
| return singleton; |
| } |
| |
| @Override |
| public boolean enqueueEvent(EntryEventImpl event) { |
| return true; |
| } |
| } |
| |
| public int getTmpQueuedEventSize() { |
| if (tmpQueuedEvents != null) { |
| return tmpQueuedEvents.size(); |
| } |
| return 0; |
| } |
| |
| @Override |
| public int getEventQueueSize() { |
| AbstractGatewaySenderEventProcessor localProcessor = this.eventProcessor; |
| return localProcessor == null ? 0 : localProcessor.eventQueueSize(); |
| } |
| |
| public int getSecondaryEventQueueSize() { |
| AbstractGatewaySenderEventProcessor localProcessor = this.eventProcessor; |
| return localProcessor == null ? 0 : localProcessor.secondaryEventQueueSize(); |
| } |
| |
| public void setEnqueuedAllTempQueueEvents(boolean enqueuedAllTempQueueEvents) { |
| this.enqueuedAllTempQueueEvents = enqueuedAllTempQueueEvents; |
| } |
| |
| protected boolean isAsyncEventQueue() { |
| return this.getAsyncEventListeners() != null && !this.getAsyncEventListeners().isEmpty(); |
| } |
| |
| public Object getLockForConcurrentDispatcher() { |
| return this.lockForConcurrentDispatcher; |
| } |
| |
| public ReentrantReadWriteLock getLifeCycleLock() { |
| return lifeCycleLock; |
| } |
| |
| @Override |
| public boolean waitUntilFlushed(long timeout, TimeUnit unit) throws InterruptedException { |
| boolean result = false; |
| if (isParallel()) { |
| try { |
| WaitUntilParallelGatewaySenderFlushedCoordinator coordinator = |
| new WaitUntilParallelGatewaySenderFlushedCoordinator(this, timeout, unit, true); |
| result = coordinator.waitUntilFlushed(); |
| } catch (BucketMovedException | CancelException | RegionDestroyedException e) { |
| logger.warn( |
| "Caught the following exception attempting waitUntilFlushed and will retry:", |
| e); |
| throw e; |
| } catch (Throwable t) { |
| logger.warn( |
| "Caught the following exception attempting waitUntilFlushed and will return:", |
| t); |
| throw new InternalGemFireError(t); |
| } |
| return result; |
| } else { |
| // Serial senders are currently not supported |
| throw new UnsupportedOperationException( |
| "waitUntilFlushed is not currently supported for serial gateway senders"); |
| } |
| } |
| |
| public void setExpectedReceiverUniqueId(String expectedReceiverUniqueId) { |
| this.expectedReceiverUniqueId = expectedReceiverUniqueId; |
| } |
| |
| public String getExpectedReceiverUniqueId() { |
| return this.expectedReceiverUniqueId; |
| } |
| |
| /** |
| * Has a reference to a GatewayEventImpl and has a timeout value. |
| */ |
| public static class EventWrapper { |
| /** |
| * Timeout events received from secondary after 5 minutes |
| */ |
| private static final int EVENT_TIMEOUT = |
| Integer.getInteger("Gateway.EVENT_TIMEOUT", 5 * 60 * 1000).intValue(); |
| public final long timeout; |
| public final GatewaySenderEventImpl event; |
| |
| public EventWrapper(GatewaySenderEventImpl e) { |
| this.event = e; |
| this.timeout = System.currentTimeMillis() + EVENT_TIMEOUT; |
| } |
| } |
| |
| /** |
| * Instances of this class allow us to delay queuing an incoming event. What used to happen was |
| * that the tmpQ would have a GatewaySenderEventImpl added to it. But then when we took it out we |
| * had to ask it for its EntryEventImpl. Then we created another GatewaySenderEventImpl. As part |
| * of the off-heap work, the GatewaySenderEventImpl no longer has a EntryEventImpl. So this class |
| * allows us to defer creation of the GatewaySenderEventImpl until we are ready to actually |
| * enqueue it. The caller is responsible for giving us an EntryEventImpl that we own and that we |
| * will release. This is done by making a copy/clone of the original event. This fixes bug 52029. |
| */ |
| public static class TmpQueueEvent implements Releasable { |
| private final EnumListenerEvent operation; |
| private final @Retained EntryEventImpl event; |
| private final Object substituteValue; |
| |
| public TmpQueueEvent(EnumListenerEvent op, @Retained EntryEventImpl e, Object subValue) { |
| this.operation = op; |
| this.event = e; |
| this.substituteValue = subValue; |
| } |
| |
| public EnumListenerEvent getOperation() { |
| return this.operation; |
| } |
| |
| public @Unretained EntryEventImpl getEvent() { |
| return this.event; |
| } |
| |
| public Object getSubstituteValue() { |
| return this.substituteValue; |
| } |
| |
| @Override |
| public void release() { |
| this.event.release(); |
| } |
| } |
| |
| protected GatewayQueueEvent getSynchronizationEvent(Object key, long timestamp) { |
| GatewayQueueEvent event = null; |
| for (RegionQueue queue : getQueues()) { |
| Region region = queue.getRegion(); |
| if (region == null) { |
| continue; |
| } |
| for (Iterator i = region.values().iterator(); i.hasNext();) { |
| GatewaySenderEventImpl gsei = (GatewaySenderEventImpl) i.next(); |
| if (gsei.getKey().equals(key) && gsei.getVersionTimeStamp() == timestamp) { |
| event = gsei; |
| logger.info("{}: Providing synchronization event for key={}; timestamp={}: {}", |
| this, key, timestamp, event); |
| getStatistics().incSynchronizationEventsProvided(); |
| break; |
| } |
| } |
| } |
| return event; |
| } |
| |
| protected void putSynchronizationEvent(GatewayQueueEvent event) { |
| if (this.eventProcessor != null) { |
| this.lifeCycleLock.readLock().lock(); |
| try { |
| logger.info("{}: Enqueueing synchronization event: {}", |
| this, event); |
| this.eventProcessor.enqueueEvent(event); |
| this.statistics.incSynchronizationEventsEnqueued(); |
| } catch (Throwable t) { |
| logger.warn(String.format( |
| "%s: Caught the following exception attempting to enqueue synchronization event=%s:", |
| new Object[] {this, event}), |
| t); |
| } finally { |
| this.lifeCycleLock.readLock().unlock(); |
| } |
| } |
| } |
| } |
