geode-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From n...@apache.org
Subject [geode] branch develop updated: GEODE-4686: Remove ParallelQueueBatchRemovalMessage
Date Sat, 17 Feb 2018 00:56:59 GMT
This is an automated email from the ASF dual-hosted git repository.

nnag pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/geode.git


The following commit(s) were added to refs/heads/develop by this push:
     new ea79fcc  GEODE-4686: Remove ParallelQueueBatchRemovalMessage
ea79fcc is described below

commit ea79fcc5f1de3865853dc626409db4f6e3fc0bff
Author: nabarun <nnag@pivotal.io>
AuthorDate: Wed Feb 14 17:24:05 2018 -0800

    GEODE-4686: Remove ParallelQueueBatchRemovalMessage
    
    	* ParallelQueueBatchRemovalMessage is no more being used in the latest version of Apache
Geode.
    	* This commit cleans up the class from the code base.
---
 .../org/apache/geode/internal/DSFIDFactory.java    |   4 -
 .../parallel/ParallelQueueBatchRemovalMessage.java | 254 ---------------------
 2 files changed, 258 deletions(-)

diff --git a/geode-core/src/main/java/org/apache/geode/internal/DSFIDFactory.java b/geode-core/src/main/java/org/apache/geode/internal/DSFIDFactory.java
index e3379e5..204459f 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/DSFIDFactory.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/DSFIDFactory.java
@@ -401,8 +401,6 @@ import org.apache.geode.internal.cache.wan.GatewaySenderAdvisor;
 import org.apache.geode.internal.cache.wan.GatewaySenderEventCallbackArgument;
 import org.apache.geode.internal.cache.wan.GatewaySenderEventImpl;
 import org.apache.geode.internal.cache.wan.GatewaySenderQueueEntrySynchronizationOperation;
-import org.apache.geode.internal.cache.wan.parallel.ParallelQueueBatchRemovalMessage;
-import org.apache.geode.internal.cache.wan.parallel.ParallelQueueBatchRemovalMessage.BatchRemovalReplyMessage;
 import org.apache.geode.internal.cache.wan.parallel.ParallelQueueRemovalMessage;
 import org.apache.geode.internal.cache.wan.serial.BatchDestroyOperation;
 import org.apache.geode.management.internal.JmxManagerAdvisor.JmxManagerProfile;
@@ -887,8 +885,6 @@ public class DSFIDFactory implements DataSerializableFixedID {
     registerDSFID(FLOW_CONTROL_PERMIT_MESSAGE, FlowControlPermitMessage.class);
     registerDSFID(REQUEST_FILTERINFO_MESSAGE, InitialImageOperation.RequestFilterInfoMessage.class);
     registerDSFID(PARALLEL_QUEUE_REMOVAL_MESSAGE, ParallelQueueRemovalMessage.class);
-    registerDSFID(PARALLEL_QUEUE_BATCH_REMOVAL_MESSAGE, ParallelQueueBatchRemovalMessage.class);
-    registerDSFID(PARALLEL_QUEUE_BATCH_REMOVAL_REPLY, BatchRemovalReplyMessage.class);
     registerDSFID(BATCH_DESTROY_MESSAGE, BatchDestroyOperation.DestroyMessage.class);
     registerDSFID(FIND_REMOTE_TX_MESSAGE, FindRemoteTXMessage.class);
     registerDSFID(FIND_REMOTE_TX_REPLY, FindRemoteTXMessageReply.class);
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/ParallelQueueBatchRemovalMessage.java
b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/ParallelQueueBatchRemovalMessage.java
deleted file mode 100644
index b34fc3a..0000000
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/ParallelQueueBatchRemovalMessage.java
+++ /dev/null
@@ -1,254 +0,0 @@
-/*
- * 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 java.io.DataInput;
-import java.io.DataOutput;
-import java.io.IOException;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-
-import org.apache.logging.log4j.Logger;
-
-import org.apache.geode.DataSerializer;
-import org.apache.geode.cache.CacheException;
-import org.apache.geode.cache.EntryNotFoundException;
-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.DistributionManager;
-import org.apache.geode.distributed.internal.DistributionMessage;
-import org.apache.geode.distributed.internal.InternalDistributedSystem;
-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.NanoTimer;
-import org.apache.geode.internal.cache.BucketRegionQueue;
-import org.apache.geode.internal.cache.ForceReattemptException;
-import org.apache.geode.internal.cache.PartitionedRegion;
-import org.apache.geode.internal.cache.partitioned.PartitionMessage;
-import org.apache.geode.internal.i18n.LocalizedStrings;
-import org.apache.geode.internal.logging.LogService;
-import org.apache.geode.internal.logging.log4j.LocalizedMessage;
-import org.apache.geode.internal.logging.log4j.LogMarker;
-
-/**
- * Removes a batch of events from the remote secondary queues
- *
- *
- * @since GemFire 7.0
- *
- */
-public class ParallelQueueBatchRemovalMessage extends PartitionMessage {
-
-  private static final Logger logger = LogService.getLogger();
-
-  private Map<Integer, List> bucketToTailKey;
-
-  public ParallelQueueBatchRemovalMessage() {
-
-  }
-
-  public ParallelQueueBatchRemovalMessage(Set<InternalDistributedMember> recipient,
int regionId,
-      ReplyProcessor21 processor, Map bucketToTailKey) {
-    super(recipient, regionId, processor);
-    this.bucketToTailKey = bucketToTailKey;
-  }
-
-  @Override
-  public int getDSFID() {
-    return PARALLEL_QUEUE_BATCH_REMOVAL_MESSAGE;
-  }
-
-  @Override
-  protected boolean operateOnPartitionedRegion(ClusterDistributionManager dm, PartitionedRegion
pr,
-      long startTime) throws CacheException {
-    for (Integer bucketId : this.bucketToTailKey.keySet()) {
-      if (pr.getRegionAdvisor().getBucketAdvisor(bucketId).isHosting()) {
-        List dispatchedKeys = this.bucketToTailKey.get(bucketId);
-        try {
-          BucketRegionQueue bucketRegionQueue =
-              (BucketRegionQueue) pr.getDataStore().getInitializedBucketForId(null, bucketId);
-
-          for (Object key : dispatchedKeys) {
-            try {
-              for (GatewayEventFilter filter : pr.getParallelGatewaySender()
-                  .getGatewayEventFilters()) {
-                GatewayQueueEvent eventForFilter = (GatewayQueueEvent) bucketRegionQueue.get(key);
-                try {
-                  if (eventForFilter != null) {
-                    filter.afterAcknowledgement(eventForFilter);
-                  }
-                } catch (Exception e) {
-                  logger.fatal(LocalizedMessage.create(
-                      LocalizedStrings.GatewayEventFilter_EXCEPTION_OCCURRED_WHILE_HANDLING_CALL_TO_0_AFTER_ACKNOWLEDGEMENT_FOR_EVENT_1,
-                      new Object[] {filter.toString(), eventForFilter}), e);
-                }
-              }
-              bucketRegionQueue.destroyKey(key);
-            } catch (EntryNotFoundException e) {
-              if (logger.isDebugEnabled()) {
-                logger.debug(
-                    "WARNING! Got EntryNotFoundException while destroying the key {} for
bucket {}",
-                    key, bucketId);
-              }
-            }
-          }
-        } catch (ForceReattemptException fe) {
-          if (logger.isDebugEnabled()) {
-            logger.debug(
-                "Got ForceReattemptException while getting bucket {} to destroyLocally the
keys.",
-                bucketId);
-          }
-        }
-      }
-    }
-    BatchRemovalReplyMessage.sendWithException(getSender(), getProcessorId(), dm, null);
-    return false;
-  }
-
-  public static ParallelQueueBatchRemovalResponse send(Set<InternalDistributedMember>
recipients,
-      PartitionedRegion pr, Map<Integer, List> bucketToTailKey) {
-    Assert.assertTrue(recipients != null, "BatchRemovalResponse NULL recipient");
-
-    ParallelQueueBatchRemovalResponse response =
-        new ParallelQueueBatchRemovalResponse(pr.getSystem(), recipients, pr);
-    ParallelQueueBatchRemovalMessage msg =
-        new ParallelQueueBatchRemovalMessage(recipients, pr.getPRId(), response, bucketToTailKey);
-    msg.setTransactionDistributed(pr.getCache().getTxManager().isDistributed());
-
-    Set<InternalDistributedMember> failures = pr.getDistributionManager().putOutgoing(msg);
-
-    if (failures != null && failures.size() > 0) {
-      // throw new ForceReattemptException("Failed sending <" + msg + ">");
-      return null;
-    }
-    pr.getPrStats().incPartitionMessagesSent();
-    return response;
-  }
-
-  @Override
-  public void toData(DataOutput out) throws IOException {
-    super.toData(out);
-    DataSerializer.writeObject(bucketToTailKey, out);
-  }
-
-  @Override
-  public void fromData(DataInput in) throws IOException, ClassNotFoundException {
-    super.fromData(in);
-    this.bucketToTailKey = (Map) DataSerializer.readObject(in);
-  }
-
-  public static class BatchRemovalReplyMessage extends ReplyMessage {
-    /**
-     * Empty constructor to conform to DataSerializable interface
-     */
-    public BatchRemovalReplyMessage() {}
-
-    public BatchRemovalReplyMessage(DataInput in) throws IOException, ClassNotFoundException
{
-      fromData(in);
-    }
-
-    private BatchRemovalReplyMessage(int processorId, ReplyException re) {
-      setProcessorId(processorId);
-      setException(re);
-    }
-
-    @Override
-    public void process(DistributionManager dm, ReplyProcessor21 processor) {
-      final long startTime = getTimestamp();
-      if (logger.isTraceEnabled(LogMarker.DM)) {
-        logger.debug(
-            "BatchRemovalReplyMessage process invoking reply processor with processorId:
{}",
-            this.processorId);
-      }
-
-      if (processor == null) {
-        if (logger.isTraceEnabled(LogMarker.DM)) {
-          logger.debug("BatchRemovalReplyMessage processor not found");
-        }
-        return;
-      }
-      processor.process(this);
-
-      if (logger.isDebugEnabled()) {
-        logger.debug("{} processed {}", processor, this);
-      }
-      dm.getStats().incReplyMessageTime(NanoTimer.getTime() - startTime);
-    }
-
-    private static void sendWithException(InternalDistributedMember recipient, int processorId,
-        DistributionManager dm, ReplyException re) {
-      Assert.assertTrue(recipient != null, "BecomePrimaryBucketReplyMessage NULL recipient");
-      BatchRemovalReplyMessage m = new BatchRemovalReplyMessage(processorId, re);
-      m.setRecipient(recipient);
-      dm.putOutgoing(m);
-    }
-
-    @Override
-    public void toData(DataOutput out) throws IOException {
-      super.toData(out);
-    }
-
-    @Override
-    public int getDSFID() {
-      return PARALLEL_QUEUE_BATCH_REMOVAL_REPLY;
-    }
-
-    @Override
-    public void fromData(DataInput in) throws IOException, ClassNotFoundException {
-      super.fromData(in);
-    }
-
-    @Override
-    public String toString() {
-      StringBuffer sb = new StringBuffer();
-      sb.append("BatchRemovalReplyMessage ").append("processorid=").append(this.processorId)
-          .append(" reply to sender ").append(this.getSender());
-      return sb.toString();
-    }
-  }
-
-  public static class ParallelQueueBatchRemovalResponse extends PartitionResponse {
-
-    public ParallelQueueBatchRemovalResponse(InternalDistributedSystem dm,
-        Set<InternalDistributedMember> recipients) {
-      super(dm, recipients);
-    }
-
-    public ParallelQueueBatchRemovalResponse(InternalDistributedSystem ds,
-        Set<InternalDistributedMember> recipients, PartitionedRegion theRegion) {
-      super(ds, recipients);
-    }
-
-    @Override
-    public void process(DistributionMessage msg) {
-      super.process(msg);
-    }
-
-    public void waitForResponse() throws ForceReattemptException {
-      try {
-        waitForCacheException();
-      } catch (ForceReattemptException e) {
-        final String msg = "GetResponse got ForceReattemptException; rethrowing";
-        logger.debug(msg, e);
-        throw e;
-      }
-    }
-  }
-}

-- 
To stop receiving notification emails like this one, please contact
nnag@apache.org.

Mime
View raw message