jchen21 commented on a change in pull request #5124:
URL: https://github.com/apache/geode/pull/5124#discussion_r427468848
##########
File path:
geode-core/src/distributedTest/java/org/apache/geode/cache/PRCacheListenerDistributedTest.java
##########
@@ -38,28 +45,60 @@
@SuppressWarnings("serial")
public class PRCacheListenerDistributedTest extends
ReplicateCacheListenerDistributedTest {
- @Parameters(name = "{index}: redundancy={0}")
- public static Iterable<Integer> data() {
- return Arrays.asList(0, 3);
+ @Parameters
+ public static Collection<Object[]> data() {
+ return Arrays.asList(new Object[][] {
+ {1, Boolean.FALSE},
+ // {3, Boolean.TRUE},
Review comment:
Should this line be uncommented or removed?
##########
File path: geode-core/src/main/java/org/apache/geode/cache/Region.java
##########
@@ -1302,7 +1302,10 @@ Object selectValue(String queryPredicate) throws
FunctionDomainException, TypeMi
* @see java.util.Map#clear()
* @see CacheListener#afterRegionClear
* @see CacheWriter#beforeRegionClear
- * @throws UnsupportedOperationException If the region is a partitioned
region
+ * @throws PartitionedRegionPartialClearException when data is partially
cleared on partitioned
+ * region. Its caller responsibility to handle the partial data
clear either by retrying
Review comment:
I think you mean `It's` instead of `Its`.
##########
File path:
geode-core/src/distributedTest/java/org/apache/geode/internal/cache/PartitionedRegionClearWithExpirationDUnitTest.java
##########
@@ -408,22 +394,21 @@ public void
clearShouldFailWhenCoordinatorMemberIsBouncedAndExpirationTasksShoul
@TestCaseName("[{index}] {method}(Coordinator:{0}, RegionType:{1})")
public void
clearShouldSucceedAndRemoveRegisteredExpirationTasksWhenNonCoordinatorMemberIsBounced(
TestVM coordinatorVM, RegionShortcut regionShortcut) {
- final int entries = 1500;
+ final int entries = 500;
+
+ RegionShortcut rs = regionShortcut;
Review comment:
I am not sure why we need a new reference here.
##########
File path:
geode-core/src/main/java/org/apache/geode/internal/cache/BucketRegion.java
##########
@@ -575,16 +575,23 @@ public void cmnClearRegion(RegionEventImpl regionEvent,
boolean cacheWrite, bool
// get rvvLock
Set<InternalDistributedMember> participants =
getCacheDistributionAdvisor().adviseInvalidateRegion();
+ boolean isLockedAlready =
this.partitionedRegion.getPartitionedRegionClear()
+ .isLockedForListenerAndClientNotification();
+
try {
- obtainWriteLocksForClear(regionEvent, participants);
+ if (!isLockedAlready) {
+ obtainWriteLocksForClear(regionEvent, participants);
+ }
// no need to dominate my own rvv.
// Clear is on going here, there won't be GII for this member
clearRegionLocally(regionEvent, cacheWrite, null);
distributeClearOperation(regionEvent, rvv, participants);
// TODO: call reindexUserDataRegion if there're lucene indexes
} finally {
- releaseWriteLocksForClear(regionEvent, participants);
+ if (!isLockedAlready) {
Review comment:
My understanding is if it is locked already, release the write locks for
clear. I think the `!` should be removed.
##########
File path:
geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegionClearMessage.java
##########
@@ -0,0 +1,289 @@
+/*
+ * 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 java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.apache.geode.DataSerializer;
+import org.apache.geode.cache.CacheException;
+import org.apache.geode.cache.Operation;
+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.ReplySender;
+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.partitioned.PartitionMessage;
+import org.apache.geode.internal.logging.log4j.LogMarker;
+import org.apache.geode.internal.serialization.DeserializationContext;
+import org.apache.geode.internal.serialization.SerializationContext;
+import org.apache.geode.logging.internal.log4j.api.LogService;
+
+public class PartitionedRegionClearMessage extends PartitionMessage {
+
+ public enum OperationType {
+ OP_LOCK_FOR_PR_CLEAR, OP_UNLOCK_FOR_PR_CLEAR, OP_PR_CLEAR,
+ }
+
+ private Object cbArg;
+
+ private OperationType op;
+
+ private EventID eventID;
+
+ private PartitionedRegion partitionedRegion;
+
+ private Set<InternalDistributedMember> recipients;
+
+ private ArrayList bucketsCleared;
+
+ @Override
+ public EventID getEventID() {
+ return eventID;
+ }
+
+ public PartitionedRegionClearMessage() {}
+
+ PartitionedRegionClearMessage(Set recipients, PartitionedRegion region,
+ ReplyProcessor21 processor, PartitionedRegionClearMessage.OperationType
operationType,
+ final RegionEventImpl event) {
+ super(recipients, region.getPRId(), processor);
+ this.recipients = recipients;
+ partitionedRegion = region;
+ op = operationType;
+ cbArg = event.getRawCallbackArgument();
+ eventID = event.getEventId();
+ }
+
+ public OperationType getOp() {
+ return op;
+ }
+
+ public void send() {
+ Assert.assertTrue(recipients != null, "ClearMessage NULL recipients set");
+
setTransactionDistributed(partitionedRegion.getCache().getTxManager().isDistributed());
+ partitionedRegion.getDistributionManager().putOutgoing(this);
+ }
+
+ @Override
+ protected Throwable processCheckForPR(PartitionedRegion pr,
+ DistributionManager distributionManager) {
+ if (pr != null && !pr.getDistributionAdvisor().isInitialized()) {
+ Throwable thr = new ForceReattemptException(
+ String.format("%s : could not find partitioned region with Id %s",
+ distributionManager.getDistributionManagerId(),
+ pr.getRegionIdentifier()));
+ return thr;
+ }
+ return null;
+ }
+
+ @Override
+ protected boolean operateOnPartitionedRegion(ClusterDistributionManager dm,
+ PartitionedRegion partitionedRegion,
+ long startTime) throws CacheException {
+
+ if (partitionedRegion == null) {
+ return true;
+ }
+
+ if (partitionedRegion.isDestroyed()) {
Review comment:
This method never returns `false`. It always returns `true`. So there is
no need to do the `if` check.
##########
File path:
geode-core/src/main/java/org/apache/geode/internal/cache/BucketRegion.java
##########
@@ -2493,4 +2500,11 @@ void updateSenderIdMonitor() {
void checkSameSenderIdsAvailableOnAllNodes() {
// nothing needed on a bucket region
}
+
+ @Override
+ protected void basicClear(RegionEventImpl regionEvent) {
+ getDataView().checkSupportsRegionClear();
Review comment:
`LocalRegionDataView. checkSupportsRegionClear()` does nothing. The
other data views, including transactions related data views throw exception. Is
it expected?
##########
File path:
geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegionClearMessage.java
##########
@@ -0,0 +1,289 @@
+/*
+ * 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 java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.apache.geode.DataSerializer;
+import org.apache.geode.cache.CacheException;
+import org.apache.geode.cache.Operation;
+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.ReplySender;
+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.partitioned.PartitionMessage;
+import org.apache.geode.internal.logging.log4j.LogMarker;
+import org.apache.geode.internal.serialization.DeserializationContext;
+import org.apache.geode.internal.serialization.SerializationContext;
+import org.apache.geode.logging.internal.log4j.api.LogService;
+
+public class PartitionedRegionClearMessage extends PartitionMessage {
+
+ public enum OperationType {
+ OP_LOCK_FOR_PR_CLEAR, OP_UNLOCK_FOR_PR_CLEAR, OP_PR_CLEAR,
+ }
+
+ private Object cbArg;
+
+ private OperationType op;
+
+ private EventID eventID;
+
+ private PartitionedRegion partitionedRegion;
+
+ private Set<InternalDistributedMember> recipients;
Review comment:
This seems to be duplicate. The super class already has an `ArrayList`
of recipients.
##########
File path:
geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegion.java
##########
@@ -2148,69 +2150,71 @@ public void writeToDisk() {
throw new UnsupportedOperationException();
}
- @Override
- void basicClear(RegionEventImpl regionEvent, boolean cacheWrite) {
- final boolean isDebugEnabled = logger.isDebugEnabled();
- synchronized (clearLock) {
- final DistributedLockService lockService =
getPartitionedRegionLockService();
- try {
- lockService.lock("_clearOperation" + this.getFullPath().replace('/',
'_'), -1, -1);
- } catch (IllegalStateException e) {
- lockCheckReadiness();
- throw e;
- }
- try {
- if (cache.isCacheAtShutdownAll()) {
- throw cache.getCacheClosedException("Cache is shutting down");
- }
-
- // do cacheWrite
- cacheWriteBeforeRegionClear(regionEvent);
-
- // create ClearPRMessage per bucket
- List<ClearPRMessage> clearMsgList =
createClearPRMessages(regionEvent.getEventId());
- for (ClearPRMessage clearPRMessage : clearMsgList) {
- int bucketId = clearPRMessage.getBucketId();
- checkReadiness();
- long sendMessagesStartTime = 0;
- if (isDebugEnabled) {
- sendMessagesStartTime = System.currentTimeMillis();
- }
- try {
- sendClearMsgByBucket(bucketId, clearPRMessage);
- } catch (PartitionOfflineException poe) {
- // TODO add a PartialResultException
- logger.info("PR.sendClearMsgByBucket encountered
PartitionOfflineException at bucket "
- + bucketId, poe);
- } catch (Exception e) {
- logger.info("PR.sendClearMsgByBucket encountered exception at
bucket " + bucketId, e);
- }
-
- if (isDebugEnabled) {
- long now = System.currentTimeMillis();
- logger.debug("PR.sendClearMsgByBucket for bucket {} took {} ms",
bucketId,
- (now - sendMessagesStartTime));
- }
- // TODO add psStats
- }
- } finally {
- try {
- lockService.unlock("_clearOperation" +
this.getFullPath().replace('/', '_'));
- } catch (IllegalStateException e) {
- lockCheckReadiness();
- }
- }
-
- // notify bridge clients at PR level
- regionEvent.setEventType(EnumListenerEvent.AFTER_REGION_CLEAR);
- boolean hasListener = hasListener();
- if (hasListener) {
- dispatchListenerEvent(EnumListenerEvent.AFTER_REGION_CLEAR,
regionEvent);
- }
- notifyBridgeClients(regionEvent);
- logger.info("Partitioned region {} finsihed clear operation.",
this.getFullPath());
- }
- }
+ /* @Override */
Review comment:
If this section of code is not used, it's better to remove it.
##########
File path:
geode-core/src/test/java/org/apache/geode/internal/cache/PartitionedRegionTest.java
##########
@@ -221,22 +220,6 @@ public void
clearShouldNotThrowUnsupportedOperationException() {
spyPartitionedRegion.clear();
}
- @Test(expected = CacheClosedException.class)
- public void clearShouldThrowCacheClosedExceptionIfShutdownAll() {
Review comment:
Is the deleted test case covered else where? Or it no longer valid?
##########
File path:
geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegionClearMessage.java
##########
@@ -0,0 +1,289 @@
+/*
+ * 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 java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.apache.geode.DataSerializer;
+import org.apache.geode.cache.CacheException;
+import org.apache.geode.cache.Operation;
+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.ReplySender;
+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.partitioned.PartitionMessage;
+import org.apache.geode.internal.logging.log4j.LogMarker;
+import org.apache.geode.internal.serialization.DeserializationContext;
+import org.apache.geode.internal.serialization.SerializationContext;
+import org.apache.geode.logging.internal.log4j.api.LogService;
+
+public class PartitionedRegionClearMessage extends PartitionMessage {
+
+ public enum OperationType {
+ OP_LOCK_FOR_PR_CLEAR, OP_UNLOCK_FOR_PR_CLEAR, OP_PR_CLEAR,
+ }
+
+ private Object cbArg;
+
+ private OperationType op;
+
+ private EventID eventID;
+
+ private PartitionedRegion partitionedRegion;
+
+ private Set<InternalDistributedMember> recipients;
+
+ private ArrayList bucketsCleared;
+
+ @Override
+ public EventID getEventID() {
+ return eventID;
+ }
+
+ public PartitionedRegionClearMessage() {}
+
+ PartitionedRegionClearMessage(Set recipients, PartitionedRegion region,
+ ReplyProcessor21 processor, PartitionedRegionClearMessage.OperationType
operationType,
+ final RegionEventImpl event) {
+ super(recipients, region.getPRId(), processor);
+ this.recipients = recipients;
+ partitionedRegion = region;
+ op = operationType;
+ cbArg = event.getRawCallbackArgument();
+ eventID = event.getEventId();
+ }
+
+ public OperationType getOp() {
+ return op;
+ }
+
+ public void send() {
+ Assert.assertTrue(recipients != null, "ClearMessage NULL recipients set");
+
setTransactionDistributed(partitionedRegion.getCache().getTxManager().isDistributed());
+ partitionedRegion.getDistributionManager().putOutgoing(this);
+ }
+
+ @Override
+ protected Throwable processCheckForPR(PartitionedRegion pr,
+ DistributionManager distributionManager) {
+ if (pr != null && !pr.getDistributionAdvisor().isInitialized()) {
+ Throwable thr = new ForceReattemptException(
+ String.format("%s : could not find partitioned region with Id %s",
+ distributionManager.getDistributionManagerId(),
+ pr.getRegionIdentifier()));
+ return thr;
+ }
+ return null;
+ }
+
+ @Override
+ protected boolean operateOnPartitionedRegion(ClusterDistributionManager dm,
+ PartitionedRegion partitionedRegion,
+ long startTime) throws CacheException {
+
+ if (partitionedRegion == null) {
Review comment:
This method never returns `false`. It always returns `true`. So there is
no need to do the `if` check.
##########
File path:
geode-core/src/test/java/org/apache/geode/internal/cache/PartitionedRegionTest.java
##########
@@ -249,28 +232,6 @@ public void
createClearPRMessagesShouldCreateMessagePerBucket() {
assertThat(msgs.size()).isEqualTo(3);
}
- @Test
- public void sendEachMessagePerBucket() {
Review comment:
Is the deleted test case covered else where? Or it no longer valid?
##########
File path:
geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegionClearMessage.java
##########
@@ -0,0 +1,289 @@
+/*
+ * 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 java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.apache.geode.DataSerializer;
+import org.apache.geode.cache.CacheException;
+import org.apache.geode.cache.Operation;
+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.ReplySender;
+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.partitioned.PartitionMessage;
+import org.apache.geode.internal.logging.log4j.LogMarker;
+import org.apache.geode.internal.serialization.DeserializationContext;
+import org.apache.geode.internal.serialization.SerializationContext;
+import org.apache.geode.logging.internal.log4j.api.LogService;
+
+public class PartitionedRegionClearMessage extends PartitionMessage {
+
+ public enum OperationType {
+ OP_LOCK_FOR_PR_CLEAR, OP_UNLOCK_FOR_PR_CLEAR, OP_PR_CLEAR,
+ }
+
+ private Object cbArg;
+
+ private OperationType op;
+
+ private EventID eventID;
+
+ private PartitionedRegion partitionedRegion;
+
+ private Set<InternalDistributedMember> recipients;
+
+ private ArrayList bucketsCleared;
+
+ @Override
+ public EventID getEventID() {
+ return eventID;
+ }
+
+ public PartitionedRegionClearMessage() {}
+
+ PartitionedRegionClearMessage(Set recipients, PartitionedRegion region,
+ ReplyProcessor21 processor, PartitionedRegionClearMessage.OperationType
operationType,
+ final RegionEventImpl event) {
+ super(recipients, region.getPRId(), processor);
+ this.recipients = recipients;
+ partitionedRegion = region;
+ op = operationType;
+ cbArg = event.getRawCallbackArgument();
+ eventID = event.getEventId();
+ }
+
+ public OperationType getOp() {
+ return op;
+ }
+
+ public void send() {
+ Assert.assertTrue(recipients != null, "ClearMessage NULL recipients set");
+
setTransactionDistributed(partitionedRegion.getCache().getTxManager().isDistributed());
+ partitionedRegion.getDistributionManager().putOutgoing(this);
+ }
+
+ @Override
+ protected Throwable processCheckForPR(PartitionedRegion pr,
+ DistributionManager distributionManager) {
+ if (pr != null && !pr.getDistributionAdvisor().isInitialized()) {
+ Throwable thr = new ForceReattemptException(
+ String.format("%s : could not find partitioned region with Id %s",
+ distributionManager.getDistributionManagerId(),
+ pr.getRegionIdentifier()));
+ return thr;
+ }
+ return null;
+ }
+
+ @Override
+ protected boolean operateOnPartitionedRegion(ClusterDistributionManager dm,
+ PartitionedRegion partitionedRegion,
+ long startTime) throws CacheException {
+
+ if (partitionedRegion == null) {
+ return true;
+ }
+
+ if (partitionedRegion.isDestroyed()) {
+ return true;
+ }
+
+ if (op == OperationType.OP_LOCK_FOR_PR_CLEAR) {
+
partitionedRegion.getPartitionedRegionClear().obtainClearLockLocal(getSender());
+ } else if (op == OperationType.OP_UNLOCK_FOR_PR_CLEAR) {
+ partitionedRegion.getPartitionedRegionClear().releaseClearLockLocal();
+ } else {
+ RegionEventImpl event =
+ new RegionEventImpl(partitionedRegion, Operation.REGION_CLEAR,
this.cbArg, true,
+ partitionedRegion.getMyId(),
+ getEventID());
+ bucketsCleared =
partitionedRegion.getPartitionedRegionClear().clearRegionLocal(event);
+ }
+ return true;
+ }
+
+ @Override
+ protected void appendFields(StringBuilder buff) {
+ super.appendFields(buff);
+ buff.append(" cbArg=").append(this.cbArg).append(" op=").append(this.op);
+ }
+
+ @Override
+ public int getDSFID() {
+ return CLEAR_PARTITIONED_REGION_MESSAGE;
+ }
+
+ @Override
+ public void fromData(DataInput in,
+ DeserializationContext context) throws IOException,
ClassNotFoundException {
+ super.fromData(in, context);
+ this.cbArg = DataSerializer.readObject(in);
+ op = PartitionedRegionClearMessage.OperationType.values()[in.readByte()];
+ eventID = DataSerializer.readObject(in);
+ }
+
+ @Override
+ public void toData(DataOutput out,
+ SerializationContext context) throws IOException {
+ super.toData(out, context);
+ DataSerializer.writeObject(this.cbArg, out);
+ out.writeByte(op.ordinal());
+ DataSerializer.writeObject(eventID, out);
+ }
+
+ /**
+ * The response on which to wait for all the replies. This response ignores
any exceptions
+ * received from the "far side"
+ */
+ public static class PartitionedRegionClearResponse extends ReplyProcessor21 {
+ CopyOnWriteArrayList bucketsCleared = new CopyOnWriteArrayList();
+
+ public PartitionedRegionClearResponse(InternalDistributedSystem system,
Set initMembers) {
+ super(system, initMembers);
+ }
+
+ @Override
+ public void process(DistributionMessage msg) {
+ if (msg instanceof PartitionedRegionClearReplyMessage) {
+ List buckets = ((PartitionedRegionClearReplyMessage)
msg).bucketsCleared;
+ if (buckets != null) {
+ bucketsCleared.addAll(buckets);
+ }
+ }
+ super.process(msg, true);
+ }
+ }
+
+ @Override
+ protected void sendReply(InternalDistributedMember member, int processorId,
+ DistributionManager distributionManager, ReplyException ex,
+ PartitionedRegion partitionedRegion, long startTime) {
+ if (partitionedRegion != null) {
+ if (startTime > 0) {
+
partitionedRegion.getPrStats().endPartitionMessagesProcessing(startTime);
+ }
+ }
+ PartitionedRegionClearMessage.PartitionedRegionClearReplyMessage
+ .send(member, processorId, getReplySender(distributionManager), op,
bucketsCleared,
+ ex);
+ }
+
+ public static class PartitionedRegionClearReplyMessage extends ReplyMessage {
+
+ private ArrayList bucketsCleared;
+
+ private OperationType op;
+
+ @Override
+ public boolean getInlineProcess() {
+ return true;
+ }
+
+ /**
+ * Empty constructor to conform to DataSerializable interface
+ */
+ public PartitionedRegionClearReplyMessage() {}
+
+ private PartitionedRegionClearReplyMessage(int processorId, OperationType
op,
+ ArrayList bucketsCleared, ReplyException ex) {
+ super();
+ this.bucketsCleared = bucketsCleared;
+ this.op = op;
+ setProcessorId(processorId);
+ setException(ex);
+ }
+
+ /** Send an ack */
+ public static void send(InternalDistributedMember recipient, int
processorId, ReplySender dm,
+ OperationType op, ArrayList bucketsCleared, ReplyException ex) {
+
+ Assert.assertTrue(recipient != null, "partitionedRegionClearReplyMessage
NULL reply message");
+
+ PartitionedRegionClearMessage.PartitionedRegionClearReplyMessage m =
+ new
PartitionedRegionClearMessage.PartitionedRegionClearReplyMessage(processorId,
op,
+ bucketsCleared, ex);
+
+ m.setRecipient(recipient);
+ dm.putOutgoing(m);
+ }
+
+ /**
+ * Processes this message. This method is invoked by the receiver of the
message.
+ *
+ * @param dm the distribution manager that is processing the message.
+ */
+ @Override
+ public void process(final DistributionManager dm, final ReplyProcessor21
rp) {
+ final long startTime = getTimestamp();
+
+ if (rp == null) {
+ if (LogService.getLogger().isTraceEnabled(LogMarker.DM_VERBOSE)) {
+ LogService.getLogger().trace(LogMarker.DM_VERBOSE, "{}: processor
not found", this);
+ }
+ return;
+ }
+
+ rp.process(this);
+
+ dm.getStats().incReplyMessageTime(NanoTimer.getTime() - startTime);
+ }
+
+ @Override
+ public int getDSFID() {
+ return CLEAR_PARTITIONED_REGION_REPLY_MESSAGE;
+ }
+
+ @Override
+ public void fromData(DataInput in,
+ DeserializationContext context) throws IOException,
ClassNotFoundException {
+ super.fromData(in, context);
+ op = PartitionedRegionClearMessage.OperationType.values()[in.readByte()];
+ bucketsCleared = DataSerializer.readArrayList(in);
+ }
+
+ @Override
+ public void toData(DataOutput out,
+ SerializationContext context) throws IOException {
+ super.toData(out, context);
+ out.writeByte(op.ordinal());
+ DataSerializer.writeArrayList(bucketsCleared, out);
+ }
+
+ @Override
+ public String toString() {
Review comment:
The implementation can use `super.toString()` and append fields specific
to this class.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
