| /* |
| * 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.parallel; |
| |
| 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.HashMap; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Set; |
| import java.util.concurrent.BlockingQueue; |
| |
| import org.apache.logging.log4j.Logger; |
| |
| import org.apache.geode.CancelException; |
| import org.apache.geode.DataSerializer; |
| import org.apache.geode.cache.CacheException; |
| import org.apache.geode.cache.EntryNotFoundException; |
| import org.apache.geode.cache.Region; |
| import org.apache.geode.cache.wan.GatewayEventFilter; |
| import org.apache.geode.cache.wan.GatewayQueueEvent; |
| import org.apache.geode.distributed.internal.ClusterDistributionManager; |
| import org.apache.geode.distributed.internal.PooledDistributionMessage; |
| import org.apache.geode.internal.cache.AbstractBucketRegionQueue; |
| import org.apache.geode.internal.cache.ForceReattemptException; |
| import org.apache.geode.internal.cache.InternalCache; |
| import org.apache.geode.internal.cache.LocalRegion; |
| import org.apache.geode.internal.cache.LocalRegion.InitializationLevel; |
| import org.apache.geode.internal.cache.PartitionedRegion; |
| import org.apache.geode.internal.cache.PartitionedRegionHelper; |
| import org.apache.geode.internal.cache.wan.AbstractGatewaySender; |
| import org.apache.geode.internal.cache.wan.GatewaySenderEventImpl; |
| import org.apache.geode.internal.logging.LogService; |
| |
| /** |
| * Removes a batch of events from the remote secondary queues |
| * |
| * @since GemFire 8.0 |
| */ |
| public class ParallelQueueRemovalMessage extends PooledDistributionMessage { |
| |
| private static final Logger logger = LogService.getLogger(); |
| |
| private HashMap regionToDispatchedKeysMap; |
| |
| public ParallelQueueRemovalMessage() {} |
| |
| public ParallelQueueRemovalMessage(HashMap rgnToDispatchedKeysMap) { |
| this.regionToDispatchedKeysMap = rgnToDispatchedKeysMap; |
| } |
| |
| @Override |
| public int getDSFID() { |
| return PARALLEL_QUEUE_REMOVAL_MESSAGE; |
| } |
| |
| @Override |
| public String toString() { |
| String cname = getShortClassName(); |
| final StringBuilder sb = new StringBuilder(cname); |
| sb.append("regionToDispatchedKeysMap=" + regionToDispatchedKeysMap); |
| sb.append(" sender=").append(getSender()); |
| return sb.toString(); |
| } |
| |
| @Override |
| protected void process(ClusterDistributionManager dm) { |
| final boolean isDebugEnabled = logger.isDebugEnabled(); |
| final InternalCache cache = dm.getCache(); |
| if (cache != null) { |
| final InitializationLevel oldLevel = |
| LocalRegion.setThreadInitLevelRequirement(BEFORE_INITIAL_IMAGE); |
| try { |
| for (Object name : regionToDispatchedKeysMap.keySet()) { |
| final String regionName = (String) name; |
| final PartitionedRegion region = (PartitionedRegion) cache.getRegion(regionName); |
| if (region == null) { |
| continue; |
| } else { |
| AbstractGatewaySender abstractSender = region.getParallelGatewaySender(); |
| // Find the map: bucketId to dispatchedKeys |
| // Find the bucket |
| // Destroy the keys |
| Map bucketIdToDispatchedKeys = (Map) this.regionToDispatchedKeysMap.get(regionName); |
| for (Object bId : bucketIdToDispatchedKeys.keySet()) { |
| final String bucketFullPath = |
| Region.SEPARATOR + PartitionedRegionHelper.PR_ROOT_REGION_NAME + Region.SEPARATOR |
| + region.getBucketName((Integer) bId); |
| AbstractBucketRegionQueue brq = |
| (AbstractBucketRegionQueue) cache.getRegionByPath(bucketFullPath); |
| if (isDebugEnabled) { |
| logger.debug( |
| "ParallelQueueRemovalMessage : The bucket in the cache is bucketRegionName : {} bucket: {}", |
| bucketFullPath, brq); |
| } |
| |
| List dispatchedKeys = (List) bucketIdToDispatchedKeys.get((Integer) bId); |
| if (dispatchedKeys != null) { |
| for (Object key : dispatchedKeys) { |
| // First, clear the Event from tempQueueEvents at AbstractGatewaySender level, if |
| // exists |
| // synchronize on AbstractGatewaySender.queuedEventsSync while doing so |
| abstractSender.removeFromTempQueueEvents(key); |
| |
| if (brq != null) { |
| if (brq.isInitialized()) { |
| if (isDebugEnabled) { |
| logger.debug( |
| "ParallelQueueRemovalMessage : The bucket {} is initialized. Destroying the key {} from BucketRegionQueue.", |
| bucketFullPath, key); |
| } |
| // fix for #48082 |
| afterAckForSecondary_EventInBucket(abstractSender, brq, key); |
| destroyKeyFromBucketQueue(brq, key, region); |
| } else { |
| // if bucket is not initialized, the event should either be in bucket or |
| // tempQueue |
| boolean isDestroyed = false; |
| if (isDebugEnabled) { |
| logger.debug( |
| "ParallelQueueRemovalMessage : The bucket {} is not yet initialized.", |
| bucketFullPath); |
| } |
| brq.getInitializationLock().readLock().lock(); |
| try { |
| if (brq.containsKey(key)) { |
| // fix for #48082 |
| afterAckForSecondary_EventInBucket(abstractSender, brq, key); |
| destroyKeyFromBucketQueue(brq, key, region); |
| isDestroyed = true; |
| } |
| |
| // Even if BucketRegionQueue does not have the key, it could be in the |
| // tempQueue |
| // remove it from there..defect #49196 |
| destroyFromTempQueue(brq.getPartitionedRegion(), (Integer) bId, key); |
| |
| // Finally, add the key to the failed batch removal keys so that it is |
| // definitely removed from the bucket region queue |
| brq.addToFailedBatchRemovalMessageKeys(key); |
| } finally { |
| brq.getInitializationLock().readLock().unlock(); |
| } |
| } |
| } else {// brq is null. Destroy the event from tempQueue. Defect #49196 |
| destroyFromTempQueue(region, (Integer) bId, key); |
| } |
| } |
| } |
| } |
| } |
| } // for loop regionToDispatchedKeysMap.keySet() |
| } finally { |
| LocalRegion.setThreadInitLevelRequirement(oldLevel); |
| } |
| } // cache != null |
| } |
| |
| // fix for #48082 |
| private void afterAckForSecondary_EventInBucket(AbstractGatewaySender abstractSender, |
| AbstractBucketRegionQueue brq, Object key) { |
| for (GatewayEventFilter filter : abstractSender.getGatewayEventFilters()) { |
| GatewayQueueEvent eventForFilter = (GatewayQueueEvent) brq.get(key); |
| try { |
| if (eventForFilter != null) { |
| filter.afterAcknowledgement(eventForFilter); |
| } |
| } catch (Exception e) { |
| logger.fatal(String.format( |
| "Exception occurred while handling call to %s.afterAcknowledgement for event %s:", |
| new Object[] {filter.toString(), eventForFilter}), |
| e); |
| } |
| } |
| } |
| |
| void destroyKeyFromBucketQueue(AbstractBucketRegionQueue brq, Object key, |
| PartitionedRegion prQ) { |
| final boolean isDebugEnabled = logger.isDebugEnabled(); |
| try { |
| brq.destroyKey(key); |
| if (!brq.getBucketAdvisor().isPrimary()) { |
| prQ.getParallelGatewaySender().getStatistics().decSecondaryQueueSize(); |
| prQ.getParallelGatewaySender().getStatistics().incEventsProcessedByPQRM(1); |
| } |
| if (isDebugEnabled) { |
| logger.debug("Destroyed the key {} for shadowPR {} for bucket {}", key, prQ.getName(), |
| brq.getId()); |
| } |
| } catch (EntryNotFoundException e) { |
| if (isDebugEnabled) { |
| logger.debug("Got EntryNotFoundException while destroying the key {} for bucket {}", key, |
| brq.getId()); |
| } |
| // add the key to failedBatchRemovalMessageQueue. |
| // This is to handle the last scenario in #49196 |
| // But if GII is already completed and FailedBatchRemovalMessageKeys |
| // are already cleared then no keys should be added to it as they will |
| // never be cleared and increase the memory footprint. |
| if (!brq.isFailedBatchRemovalMessageKeysClearedFlag()) { |
| brq.addToFailedBatchRemovalMessageKeys(key); |
| } |
| } catch (ForceReattemptException fe) { |
| if (isDebugEnabled) { |
| logger.debug( |
| "Got ForceReattemptException while getting bucket {} to destroyLocally the keys.", |
| brq.getId()); |
| } |
| } catch (CancelException e) { |
| return; // cache or DS is closing |
| } catch (CacheException e) { |
| logger.error(String.format( |
| "ParallelQueueRemovalMessage::process:Exception in processing the last disptached key for a ParallelGatewaySenderQueue's shadowPR. The problem is with key,%s for shadowPR with name=%s", |
| new Object[] {key, prQ.getName()}), |
| e); |
| } |
| } |
| |
| private boolean destroyFromTempQueue(PartitionedRegion qPR, int bId, Object key) { |
| boolean isDestroyed = false; |
| Set queues = qPR.getParallelGatewaySender().getQueues(); |
| if (queues != null) { |
| ConcurrentParallelGatewaySenderQueue prq = |
| (ConcurrentParallelGatewaySenderQueue) queues.toArray()[0]; |
| BlockingQueue<GatewaySenderEventImpl> tempQueue = prq.getBucketTmpQueue(bId); |
| if (tempQueue != null) { |
| Iterator<GatewaySenderEventImpl> itr = tempQueue.iterator(); |
| while (itr.hasNext()) { |
| GatewaySenderEventImpl eventForFilter = itr.next(); |
| // fix for #48082 |
| afterAckForSecondary_EventInTempQueue(qPR.getParallelGatewaySender(), eventForFilter); |
| if (eventForFilter.getShadowKey().equals(key)) { |
| itr.remove(); |
| eventForFilter.release(); // GEODE-1282 |
| isDestroyed = true; |
| } |
| } |
| } |
| } |
| return isDestroyed; |
| } |
| |
| // fix for #48082 |
| private void afterAckForSecondary_EventInTempQueue( |
| AbstractGatewaySender parallelGatewaySenderImpl, GatewaySenderEventImpl eventForFilter) { |
| for (GatewayEventFilter filter : parallelGatewaySenderImpl.getGatewayEventFilters()) { |
| try { |
| if (eventForFilter != null) { |
| filter.afterAcknowledgement(eventForFilter); |
| } |
| } catch (Exception e) { |
| logger.fatal(String.format( |
| "Exception occurred while handling call to %s.afterAcknowledgement for event %s:", |
| new Object[] {filter.toString(), eventForFilter}), |
| e); |
| } |
| } |
| } |
| |
| @Override |
| public void toData(DataOutput out) throws IOException { |
| super.toData(out); |
| DataSerializer.writeHashMap(this.regionToDispatchedKeysMap, out); |
| } |
| |
| @Override |
| public void fromData(DataInput in) throws IOException, ClassNotFoundException { |
| super.fromData(in); |
| this.regionToDispatchedKeysMap = DataSerializer.readHashMap(in); |
| } |
| } |
