GEODE-9806: added skip recovery of recovered bucket (apache#7107)

* GEODE-9806: added skip recovery of recovered bucket

* GEODE-9806: added test

* GEODE-9806: Update UT after comments

geode-core/src/main/java/org/apache/geode/internal/cache/ProxyBucketRegion.java

@@ -29,6 +29,7 @@
 
 import org.apache.geode.CancelCriterion;
 import org.apache.geode.InternalGemFireError;
+import org.apache.geode.annotations.VisibleForTesting;
 import org.apache.geode.cache.DiskAccessException;
 import org.apache.geode.cache.EvictionAttributes;
 import org.apache.geode.cache.RegionAttributes;
@@ -71,6 +72,7 @@ public class ProxyBucketRegion implements Bucket {
   private final Set<DistributedMember> sickHosts = new HashSet<>();
   private final DiskRegion diskRegion;
   private final BucketLock bucketLock;
+  private int recoverFromDiskCnt;
 
   /**
    * Note that LocalRegion has a version of this name spelled "NO_PARTITITON". So if code is written
@@ -95,6 +97,7 @@ public ProxyBucketRegion(int bid, PartitionedRegion partitionedRegion,
         BucketAdvisor.createBucketAdvisor(this, internalRegionArgs.getPartitionedRegionAdvisor());
 
     this.bucketLock = this.partitionedRegion.getBucketLock(this.bid);
+    this.recoverFromDiskCnt = 0;
 
     if (this.partitionedRegion.getDataPolicy().withPersistence()) {
 
@@ -427,6 +430,11 @@ public void recoverFromDisk() {
       logger.debug("{} coming to recover from disk. wasHosting {}", getFullPath(),
           persistenceAdvisor.wasHosting());
     }
+    if (!persistenceAdvisor.isRecovering()) {
+      return;
+    }
+    recoverFromDiskCnt++;
+
     try {
       if (persistenceAdvisor.wasHosting()) {
         if (isDebugEnabled) {
@@ -501,6 +509,11 @@ public void recoverFromDisk() {
     }
   }
 
+  @VisibleForTesting
+  public int getRecoverFromDiskCnt() {
+    return recoverFromDiskCnt;
+  }
+
   boolean hasPersistentChildRegion() {
     return ColocationHelper.hasPersistentChildRegion(partitionedRegion);
   }

geode-core/src/test/java/org/apache/geode/internal/cache/ProxyBucketRegionTest.java

@@ -15,11 +15,23 @@
 package org.apache.geode.internal.cache;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
 import org.junit.Test;
 
+import org.apache.geode.CancelCriterion;
+import org.apache.geode.cache.DataPolicy;
+import org.apache.geode.cache.PartitionAttributes;
+import org.apache.geode.cache.RegionAttributes;
+import org.apache.geode.distributed.internal.DistributionConfig;
+import org.apache.geode.distributed.internal.DistributionManager;
+import org.apache.geode.distributed.internal.InternalDistributedSystem;
+import org.apache.geode.internal.cache.partitioned.RegionAdvisor;
 
 public class ProxyBucketRegionTest {
 
@@ -32,4 +44,54 @@ public void shouldBeMockable() throws Exception {
 
     assertThat(mockProxyBucketRegion.getBucketAdvisor()).isSameAs(mockBucketAdvisor);
   }
+
+  @Test
+  public void testTwoTimesCallToRecoverFromDiskWillExecuteOneRecovery() {
+    PartitionedRegion partitionedRegion = mock(PartitionedRegion.class);
+    InternalRegionArguments internalRegionArguments = mock(InternalRegionArguments.class);
+    RegionAdvisor regionAdvisor = mock(RegionAdvisor.class);
+    PartitionAttributes partitionAttributes = mock(PartitionAttributes.class);
+    InternalCache cache = mock(InternalCache.class);
+    InternalDistributedSystem ids = mock(InternalDistributedSystem.class);
+    DataPolicy dp = mock(DataPolicy.class);
+    RegionAttributes ra = mock(RegionAttributes.class);
+    DiskStoreImpl ds = mock(DiskStoreImpl.class);
+    DiskInitFile dif = mock(DiskInitFile.class);
+    DiskRegion dr = mock(DiskRegion.class);
+    DistributionManager dm = mock(DistributionManager.class);
+    DistributionConfig config = mock(DistributionConfig.class);
+    CancelCriterion cancel = mock(CancelCriterion.class);
+
+    when(internalRegionArguments.getPartitionedRegionAdvisor()).thenReturn(regionAdvisor);
+    when(regionAdvisor.getPartitionedRegion()).thenReturn(partitionedRegion);
+    when(partitionedRegion.getPartitionAttributes()).thenReturn(partitionAttributes);
+    when(partitionAttributes.getColocatedWith()).thenReturn(null);
+    when(partitionedRegion.getCache()).thenReturn(cache);
+    when(partitionedRegion.getGemFireCache()).thenReturn(cache);
+    when(cache.getInternalDistributedSystem()).thenReturn(ids);
+
+    when(ids.getDistributionManager()).thenReturn(dm);
+    when(partitionedRegion.getDataPolicy()).thenReturn(dp);
+    when(dp.withPersistence()).thenReturn(true);
+    when(cache.getInternalDistributedSystem()).thenReturn(ids);
+    when(partitionedRegion.getAttributes()).thenReturn(ra);
+    when(ra.getEvictionAttributes()).thenReturn(null);
+    when(partitionedRegion.getDiskStore()).thenReturn(ds);
+    when(ds.getDiskInitFile()).thenReturn(dif);
+    when(dif.createDiskRegion(any(), anyString(), anyBoolean(), anyBoolean(), anyBoolean(),
+        anyBoolean(), any(), any(), any(), any(), any(), anyString(), anyInt(), any(),
+        anyBoolean())).thenReturn(dr);
+    when(dm.getConfig()).thenReturn(config);
+    when(config.getAckWaitThreshold()).thenReturn(10);
+    when(cache.getCancelCriterion()).thenReturn(cancel);
+    when(regionAdvisor.isInitialized()).thenReturn(true);
+
+    ProxyBucketRegion proxyBucketRegion =
+        new ProxyBucketRegion(0, partitionedRegion, internalRegionArguments);
+
+    proxyBucketRegion.recoverFromDisk();
+    proxyBucketRegion.recoverFromDisk();
+    assertThat(proxyBucketRegion.getRecoverFromDiskCnt()).isEqualTo(1);
+  }
+
 }