/*
 * 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.IOException;
import java.util.Collections;
import java.util.Random;
import java.util.Set;

import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.cache.partitioned.FetchKeysMessage;
import org.apache.geode.internal.cache.partitioned.FetchKeysMessage.FetchKeysResponse;
import org.apache.geode.internal.cache.partitioned.PRLocallyDestroyedException;
import org.apache.geode.internal.cache.tier.InterestType;

/**
 * Extracted from {@link PartitionedRegion}. This is a utility used by Hydra test code only.
 */
public class PartitionedRegionGetSomeKeys {

  /**
   * Test Method: Get a random set of keys from a randomly selected bucket using the provided
   * {@code Random} number generator.
   *
   * @return A set of keys from a randomly chosen bucket or {@link Collections#EMPTY_SET}
   */
  public static Set<?> getSomeKeys(PartitionedRegion partitionedRegion, Random random)
      throws IOException, ClassNotFoundException {
    Set<Integer> bucketIdSet = partitionedRegion.getRegionAdvisor().getBucketSet();

    if (bucketIdSet != null && !bucketIdSet.isEmpty()) {
      Integer[] bucketIds = bucketIdSet.toArray(new Integer[bucketIdSet.size()]);
      Set<?> someKeys;

      // Randomly pick a node to get some data from
      for (int i = 0; i < bucketIds.length; i++) {
        try {
          int whichBucket = random.nextInt(bucketIds.length);
          if (whichBucket >= bucketIds.length) {
            // The GSRandom.nextInt(int) may return a value that includes the maximum.
            whichBucket = bucketIds.length - 1;
          }
          int bucketId = bucketIds[whichBucket];

          InternalDistributedMember member = partitionedRegion.getNodeForBucketRead(bucketId);
          if (member != null) {
            if (member.equals(partitionedRegion.getMyId())) {
              someKeys = partitionedRegion.getDataStore().handleRemoteGetKeys(bucketId,
                  InterestType.REGULAR_EXPRESSION, ".*", false);
            } else {
              FetchKeysResponse fetchKeysResponse =
                  FetchKeysMessage.send(member, partitionedRegion, bucketId, false);
              someKeys = fetchKeysResponse.waitForKeys();
            }

            if (someKeys != null && !someKeys.isEmpty()) {
              return someKeys;
            }
          }
        } catch (ForceReattemptException | PRLocallyDestroyedException e) {
          throw new RuntimeException(e);
        }
      }
    }
    return Collections.emptySet();
  }
}

@@ -932,16 +932,57 @@
 l.Set;%0A%0A
+import org.apache.logging.log4j.Logger;%0A%0A
 import o
@@ -1056,16 +1056,16 @@
 Member;%0A
-
 import o
@@ -1349,16 +1349,69 @@
 estType;
+%0Aimport org.apache.geode.internal.logging.LogService;
 %0A%0A/**%0A *
@@ -1550,16 +1550,80 @@
 Keys %7B%0A%0A
+  private static final Logger logger = LogService.getLogger();%0A%0A
   /**%0A  
@@ -2139,23 +2139,22 @@
 %7B%0A      
-Integer
+Object
 %5B%5D bucke
@@ -2184,19 +2184,24 @@
 ray(
-new
+);%0A     
  Integer
 %5Bbuc
@@ -2200,17 +2200,17 @@
 eger
-%5B
+ 
 bucketId
 Set.
@@ -2209,20 +2209,15 @@
 etId
-Set.size()%5D)
+ = null
 ;%0A  
@@ -2623,20 +2623,16 @@
         
-int 
 bucketId
@@ -2633,16 +2633,26 @@
 cketId =
+ (Integer)
  bucketI
@@ -3403,10 +3403,363 @@
 ion 
-%7C 
+movinOn) %7B%0A          partitionedRegion.checkReadiness();%0A          logger.debug(%0A              %22Test hook getSomeKeys caught a ForceReattemptException for bucketId=%7B%7D%7B%7D%7B%7D. Moving on to another bucket%22,%0A              partitionedRegion.getPRId(), partitionedRegion.BUCKET_ID_SEPARATOR, bucketId,%0A              movinOn);%0A          continue;%0A        %7D catch (
 PRLo
@@ -3782,16 +3782,21 @@
 ception 
+ignor
 e) %7B%0A   
@@ -3806,37 +3806,142 @@
     
-throw new RuntimeException(e)
+logger.debug(%22getSomeKeys: Encountered PRLocallyDestroyedException%22);%0A          partitionedRegion.checkReadiness();%0A          continue
 ;%0A  

change back the behavior of test code of getSomeKeys to ignore exceptions. (#1564)