You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by es...@apache.org on 2019/07/16 23:42:18 UTC
[geode] 01/01: GEODE-6975: Reply with exception if fail to persit
sender's new id.
This is an automated email from the ASF dual-hosted git repository.
eshu11 pushed a commit to branch feature/GEODE-6975
in repository https://gitbox.apache.org/repos/asf/geode.git
commit a4f98c78dcd1f624afa5745ee65299d40e080df5
Author: eshu <es...@pivotal.io>
AuthorDate: Tue Jul 16 16:32:59 2019 -0700
GEODE-6975: Reply with exception if fail to persit sender's new id.
* Reply with ReplyException when processing PrepareNewPersistentMemberMessage
failed.
* Retry again if remote member failed to persit the new persistent member id.
* Make sure atomicCreation flag is reset, so that redundancy could be satisfied
during bucket creation retry if previous attempt failed.
---
.../internal/cache/BucketPersistenceAdvisor.java | 13 ++-
.../internal/cache/PRHARedundancyProvider.java | 22 ++++-
.../PrepareNewPersistentMemberMessage.java | 23 ++++-
.../cache/BucketPersistenceAdvisorTest.java | 49 ++++++++--
.../PrepareNewPersistentMemberMessageTest.java | 106 +++++++++++++++++++++
5 files changed, 196 insertions(+), 17 deletions(-)
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/BucketPersistenceAdvisor.java b/geode-core/src/main/java/org/apache/geode/internal/cache/BucketPersistenceAdvisor.java
index 0b0a889..a794340 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/BucketPersistenceAdvisor.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/BucketPersistenceAdvisor.java
@@ -461,7 +461,18 @@ public class BucketPersistenceAdvisor extends PersistenceAdvisorImpl {
}
atomicCreation = false;
}
- super.setOnline(false, true, newId);
+ try {
+ superSetOnline(false, true, newId);
+ } catch (ReplyException replyException) {
+ synchronized (lock) {
+ atomicCreation = true;
+ }
+ throw replyException;
+ }
+ }
+
+ void superSetOnline(boolean didGII, boolean atomicCreation, PersistentMemberID newId) {
+ super.setOnline(didGII, atomicCreation, newId);
}
public boolean isAtomicCreation() {
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/PRHARedundancyProvider.java b/geode-core/src/main/java/org/apache/geode/internal/cache/PRHARedundancyProvider.java
index cf70f2f..fc63118 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/PRHARedundancyProvider.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/PRHARedundancyProvider.java
@@ -55,6 +55,7 @@ import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.internal.DistributionManager;
import org.apache.geode.distributed.internal.LonerDistributionManager;
import org.apache.geode.distributed.internal.MembershipListener;
+import org.apache.geode.distributed.internal.ReplyException;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.NanoTimer;
@@ -460,7 +461,7 @@ public class PRHARedundancyProvider {
partitionedRegion.getRegionAdvisor().adviseFixedPrimaryPartitionDataStore(bucketId);
}
- InternalDistributedMember memberHostingBucket;
+ InternalDistributedMember memberHostingBucket = null;
Collection<InternalDistributedMember> attempted = new HashSet<>();
do {
partitionedRegion.checkReadiness();
@@ -491,7 +492,14 @@ public class PRHARedundancyProvider {
}
CreateBucketMessage.NodeResponse response =
CreateBucketMessage.send(targetMember, partitionedRegion, bucketId, size);
- memberHostingBucket = response.waitForResponse();
+ try {
+ memberHostingBucket = response.waitForResponse();
+ } catch (ReplyException replyException) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("create bucket failed on member {}: ", targetMember, replyException);
+ }
+ // Will start over and retry again.
+ }
if (memberHostingBucket != null) {
return memberHostingBucket;
}
@@ -679,7 +687,15 @@ public class PRHARedundancyProvider {
if (redundancySatisfied || exhaustedPotentialCandidates) {
// Tell one of the members to become primary.
// The rest of the members will be allowed to volunteer for primary.
- endBucketCreation(bucketId, acceptedMembers, bucketPrimary, partitionName);
+ try {
+ endBucketCreation(bucketId, acceptedMembers, bucketPrimary, partitionName);
+ } catch (ReplyException replyException) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Will retry as endBucketCreation failed with: ", replyException);
+ }
+ // One of the accepted members failed. Retry again.
+ continue;
+ }
int expectedRemoteHosts = acceptedMembers.size()
- (acceptedMembers.contains(partitionedRegion.getMyId()) ? 1 : 0);
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/persistence/PrepareNewPersistentMemberMessage.java b/geode-core/src/main/java/org/apache/geode/internal/cache/persistence/PrepareNewPersistentMemberMessage.java
index 8eba7eb..15d31aa 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/persistence/PrepareNewPersistentMemberMessage.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/persistence/PrepareNewPersistentMemberMessage.java
@@ -38,6 +38,7 @@ 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.InternalDataSerializer;
+import org.apache.geode.internal.Version;
import org.apache.geode.internal.cache.DistributedRegion;
import org.apache.geode.internal.cache.LocalRegion;
import org.apache.geode.internal.cache.LocalRegion.InitializationLevel;
@@ -106,9 +107,21 @@ public class PrepareNewPersistentMemberMessage extends HighPriorityDistributionM
}
} catch (RegionDestroyedException e) {
- logger.debug("<RegionDestroyed> {}", this);
+ if (logger.isDebugEnabled()) {
+ logger.debug("<RegionDestroyed> {}", this);
+ }
+ if (getSender().getVersionObject().compareTo(Version.GEODE_1_10_0) >= 0) {
+ // Reply with exception to sender to indicate we have not persisted its id.
+ exception = new ReplyException(e);
+ }
} catch (CancelException e) {
- logger.debug("<CancelException> {}", this);
+ if (logger.isDebugEnabled()) {
+ logger.debug("<CancelException> {}", this);
+ }
+ if (getSender().getVersionObject().compareTo(Version.GEODE_1_10_0) >= 0) {
+ // Reply with exception to sender to indicate we have not persisted its id.
+ exception = new ReplyException(e);
+ }
} catch (VirtualMachineError e) {
SystemFailure.initiateFailure(e);
throw e;
@@ -117,7 +130,7 @@ public class PrepareNewPersistentMemberMessage extends HighPriorityDistributionM
exception = new ReplyException(t);
} finally {
LocalRegion.setThreadInitLevelRequirement(oldLevel);
- ReplyMessage replyMsg = new ReplyMessage();
+ ReplyMessage replyMsg = createReplyMessage();
replyMsg.setRecipient(getSender());
replyMsg.setProcessorId(processorId);
if (exception != null) {
@@ -127,6 +140,10 @@ public class PrepareNewPersistentMemberMessage extends HighPriorityDistributionM
}
}
+ ReplyMessage createReplyMessage() {
+ return new ReplyMessage();
+ }
+
@Override
public int getDSFID() {
return PREPARE_NEW_PERSISTENT_MEMBER_REQUEST;
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/BucketPersistenceAdvisorTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/BucketPersistenceAdvisorTest.java
index cb4232e..ccd630d 100644
--- a/geode-core/src/test/java/org/apache/geode/internal/cache/BucketPersistenceAdvisorTest.java
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/BucketPersistenceAdvisorTest.java
@@ -14,14 +14,20 @@
*/
package org.apache.geode.internal.cache;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.catchThrowable;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.when;
import org.junit.Test;
import org.apache.geode.distributed.DistributedLockService;
+import org.apache.geode.distributed.internal.ReplyException;
import org.apache.geode.internal.cache.PartitionedRegion.BucketLock;
import org.apache.geode.internal.cache.persistence.PersistentMemberID;
import org.apache.geode.internal.cache.persistence.PersistentMemberManager;
@@ -29,6 +35,13 @@ import org.apache.geode.internal.cache.persistence.PersistentMemberView;
public class BucketPersistenceAdvisorTest {
+ private final PersistentMemberID persistentID = mock(PersistentMemberID.class);
+ private final PersistentMemberView mockStorage = mock(PersistentMemberView.class);
+ private final BucketPersistenceAdvisor bucketPersistenceAdvisor = new BucketPersistenceAdvisor(
+ mock(CacheDistributionAdvisor.class), mock(DistributedLockService.class), mockStorage,
+ "/region", mock(DiskRegionStats.class), mock(PersistentMemberManager.class),
+ mock(BucketLock.class), mock(ProxyBucketRegion.class));
+
@Test
public void shouldBeMockable() throws Exception {
BucketPersistenceAdvisor mockBucketAdvisor = mock(BucketPersistenceAdvisor.class);
@@ -38,16 +51,32 @@ public class BucketPersistenceAdvisorTest {
@Test
public void atomicCreationShouldNotBeSetForPersistentRegion() throws Exception {
- PersistentMemberID mockPersistentID = mock(PersistentMemberID.class);
- PersistentMemberView mockStorage = mock(PersistentMemberView.class);
- when(mockStorage.getMyPersistentID()).thenReturn(mockPersistentID);
-
- BucketPersistenceAdvisor bpa = new BucketPersistenceAdvisor(
- mock(CacheDistributionAdvisor.class), mock(DistributedLockService.class), mockStorage,
- "/region", mock(DiskRegionStats.class), mock(PersistentMemberManager.class),
- mock(BucketLock.class), mock(ProxyBucketRegion.class));
- bpa.setAtomicCreation(true);
- assertFalse(bpa.isAtomicCreation());
+ when(mockStorage.getMyPersistentID()).thenReturn(persistentID);
+
+ bucketPersistenceAdvisor.setAtomicCreation(true);
+ assertFalse(bucketPersistenceAdvisor.isAtomicCreation());
+ }
+
+ @Test
+ public void atomicCreationShouldNotBeSetAfterEndBucketCreation() {
+ BucketPersistenceAdvisor spy = spy(bucketPersistenceAdvisor);
+ spy.setAtomicCreation(true);
+ doNothing().when(spy).superSetOnline(false, true, persistentID);
+
+ spy.endBucketCreation(persistentID);
+
+ assertFalse(spy.isAtomicCreation());
+ }
+
+ @Test
+ public void atomicCreationShouldBeResetAfterEndBucketCreation() {
+ BucketPersistenceAdvisor spy = spy(bucketPersistenceAdvisor);
+ spy.setAtomicCreation(true);
+ doThrow(new ReplyException()).when(spy).superSetOnline(false, true, persistentID);
+
+ Throwable thrown = catchThrowable(() -> spy.endBucketCreation(persistentID));
+ assertThat(thrown).isInstanceOf(ReplyException.class);
+ assertTrue(spy.isAtomicCreation());
}
}
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/persistence/PrepareNewPersistentMemberMessageTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/persistence/PrepareNewPersistentMemberMessageTest.java
new file mode 100644
index 0000000..156891c
--- /dev/null
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/persistence/PrepareNewPersistentMemberMessageTest.java
@@ -0,0 +1,106 @@
+/*
+ * 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.persistence;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import org.junit.Test;
+
+import org.apache.geode.cache.CacheClosedException;
+import org.apache.geode.cache.RegionDestroyedException;
+import org.apache.geode.distributed.internal.ClusterDistributionManager;
+import org.apache.geode.distributed.internal.ReplyMessage;
+import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
+import org.apache.geode.internal.Version;
+import org.apache.geode.internal.cache.InternalCache;
+
+public class PrepareNewPersistentMemberMessageTest {
+ private final ClusterDistributionManager manager = mock(ClusterDistributionManager.class);
+ private final ReplyMessage replyMessage = mock(ReplyMessage.class);
+ private final InternalCache cache = mock(InternalCache.class);
+ private final String regionPath = "regionPath";
+ private final InternalDistributedMember sender = mock(InternalDistributedMember.class);
+ private final Version version = Version.CURRENT;
+ private final Version oldVersion = mock(Version.class);
+
+ private static final Version VERSION = Version.GEODE_1_10_0;
+
+ @Test
+ public void replyWithExceptionIfFailedWithCancelledException() {
+ PrepareNewPersistentMemberMessage message = spy(new PrepareNewPersistentMemberMessage());
+ when(manager.getExistingCache()).thenThrow(new CacheClosedException());
+ doReturn(sender).when(message).getSender();
+ doReturn(replyMessage).when(message).createReplyMessage();
+ when(sender.getVersionObject()).thenReturn(version);
+
+ message.process(manager);
+
+ verify(replyMessage).setException(any());
+ }
+
+ @Test
+ public void doNotReplyWithExceptionToOldVersionMemberIfFailedWithCancelledException() {
+ PrepareNewPersistentMemberMessage message = spy(new PrepareNewPersistentMemberMessage());
+ when(manager.getExistingCache()).thenThrow(new CacheClosedException());
+ doReturn(sender).when(message).getSender();
+ doReturn(replyMessage).when(message).createReplyMessage();
+ when(sender.getVersionObject()).thenReturn(oldVersion);
+ when(oldVersion.compareTo(VERSION)).thenReturn(-1);
+
+ message.process(manager);
+
+ verify(replyMessage, never()).setException(any());
+ }
+
+ @Test
+ public void replyWithExceptionIfFailedWithRegionDestroyedException() {
+ PrepareNewPersistentMemberMessage message =
+ spy(new PrepareNewPersistentMemberMessage(regionPath, null, null, 1));
+ when(manager.getExistingCache()).thenReturn(cache);
+ when(cache.getRegion(regionPath)).thenThrow(new RegionDestroyedException("", ""));
+ doReturn(sender).when(message).getSender();
+ doReturn(replyMessage).when(message).createReplyMessage();
+ when(sender.getVersionObject()).thenReturn(version);
+
+ message.process(manager);
+
+ verify(replyMessage).setException(any());
+ }
+
+ @Test
+ public void doNotReplyWithExceptionToOldVersionMemberIfFailedWithRegionDestroyedException() {
+ PrepareNewPersistentMemberMessage message =
+ spy(new PrepareNewPersistentMemberMessage(regionPath, null, null, 1));
+ when(manager.getExistingCache()).thenReturn(cache);
+ when(cache.getRegion(regionPath)).thenThrow(new RegionDestroyedException("", ""));
+ doReturn(sender).when(message).getSender();
+ doReturn(replyMessage).when(message).createReplyMessage();
+ when(sender.getVersionObject()).thenReturn(oldVersion);
+ when(oldVersion.compareTo(VERSION)).thenReturn(-1);
+
+ message.process(manager);
+
+ verify(replyMessage, never()).setException(any());
+ }
+
+
+
+}