@SuppressWarnings("unchecked")
protected List<DiskStoreDetails> getDiskStoreListing() {
  final Execution membersFunctionExecutor = getMembersFunctionExecutor(getMembers(getCache()));

  if (membersFunctionExecutor instanceof AbstractExecution) {
    ((AbstractExecution) membersFunctionExecutor).setIgnoreDepartedMembers(true);
  }

  final ResultCollector<?, ?> resultCollector = membersFunctionExecutor.execute(new ListDiskStoresFunction());

  final List<?> results = (List<?>) resultCollector.getResult();
  final List<DiskStoreDetails> distributedSystemMemberDiskStores = new ArrayList<DiskStoreDetails>(results.size());

  for (final Object result : results) {
    if (result instanceof Set) { // ignore FunctionInvocationTargetExceptions and other Exceptions...
      distributedSystemMemberDiskStores.addAll((Set<DiskStoreDetails>) result);
    }
  }

  Collections.sort(distributedSystemMemberDiskStores);

  return distributedSystemMemberDiskStores;
}
 

public static Object executeFunctionHA() throws Exception {
  Region region = cache.getRegion(PartitionedRegionName);
  final HashSet testKeysSet = new HashSet();
  for (int i = (totalNumBuckets.intValue() * 10); i > 0; i--) {
    testKeysSet.add("execKey-" + i);
  }
  DistributedSystem.setThreadsSocketPolicy(false);
  Function function = new TestFunction(true, TestFunction.TEST_FUNCTION_HA);
  FunctionService.registerFunction(function);
  Execution dataSet = FunctionService.onRegion(region);
  ResultCollector rc1 = dataSet.withFilter(testKeysSet)
      .withArgs(Boolean.TRUE).execute(function.getId());
  List l = ((List)rc1.getResult());
  getLogWriter().info("Result size : " + l.size());
  return l;
}
 

public static Object executeFunctionHA() throws Exception {
  Region region = cache.getRegion(PartitionedRegionName);
  final HashSet testKeysSet = new HashSet();
  for (int i = (totalNumBuckets.intValue() * 10); i > 0; i--) {
    testKeysSet.add("execKey-" + i);
  }
  DistributedSystem.setThreadsSocketPolicy(false);
  Function function = new TestFunction(true, TestFunction.TEST_FUNCTION_HA);
  FunctionService.registerFunction(function);
  Execution dataSet = FunctionService.onRegion(region);
  ResultCollector rc1 = dataSet.withFilter(testKeysSet)
      .withArgs(Boolean.TRUE).execute(function.getId());
  List l = ((List)rc1.getResult());
  getLogWriter().info("Result size : " + l.size());
  return l;
}
 

protected void invalidateEntries() {
  hydra.Log.getLogWriter().info("Passing the region set " + regionSet);
  Execution dataSet = InternalFunctionService.onRegions(regionSet).withArgs(
      "invalidateEntries");

  try {
    dataSet.execute(new OnRegionsFunction()).getResult();
  }
  catch (Exception e) {
    throw new TestException("Caught Exception " , e);
  }

  for (Object aRegion : regionSet) {
    Set keySet = ((Region)aRegion).keySet();
    for (Object key : keySet) {
      Object value = ((Region)aRegion).get(key);
      if (value != null) {
        throw new TestException(
            "Values supposed to be invalidated after clear Region, but key "
                + key + " has value " + value);
      }
    }

  }
}
 

public static Object serverExecutionHAOneServerDown(Boolean isByName, Function function,
    Boolean toRegister) {

  DistributedSystem.setThreadsSocketPolicy(false);

  if (toRegister.booleanValue()) {
    FunctionService.registerFunction(function);
  }
  Execution member = FunctionService.onServer(pool);
  ResultCollector rs = null;
  try {
    ArrayList<String> args = new ArrayList<String>();
    args.add(retryRegionName);
    args.add("serverExecutionHAOneServerDown");
    rs = execute(member, args, function, isByName);
    assertEquals(retryRegionName, ((List)rs.getResult()).get(0));
  } catch (Exception ex) {
    ex.printStackTrace();
    getLogWriter().info("Exception : ", ex);
    fail("Test failed after the execute operation");
  }
  return rs.getResult();
}
 

@Override
public long export(Region<K, V> region, ExportSink sink, SnapshotOptions<K, V> options) throws IOException {
  try {
    ClientArgs<K, V> args = new ClientArgs<K, V>(region.getFullPath(), pool.getPRSingleHopEnabled(), options);
    ClientExportCollector results = new ClientExportCollector(sink);
    
    // For single hop we rely on tcp queuing to throttle the export; otherwise
    // we allow the WindowedExporter to provide back pressure. 
    Execution exec = pool.getPRSingleHopEnabled() 
        ? FunctionService.onRegion(region) 
        : FunctionService.onServer(pool);
        
    ResultCollector<?, ?> rc = exec
        .withArgs(args)
        .withCollector(results)
        .execute(new ProxyExportFunction<K, V>());
    
    // Our custom result collector is writing the data, but this will
    // check for errors.
    return (Long) rc.getResult();

  } catch (FunctionException e) {
    throw new IOException(e);
  }
}
 

@SuppressWarnings("unchecked")
protected List<IndexDetails> getIndexListing() {
  final Execution functionExecutor = getMembersFunctionExecutor(getMembers(getCache()));

  if (functionExecutor instanceof AbstractExecution) {
    ((AbstractExecution) functionExecutor).setIgnoreDepartedMembers(true);
  }

  final ResultCollector<?, ?> resultsCollector = functionExecutor.execute(new ListIndexFunction());

  final List<?> results = (List<?>) resultsCollector.getResult();
  final List<IndexDetails> indexDetailsList = new ArrayList<IndexDetails>(results.size());

  for (Object result : results) {
    if (result instanceof Set) { // ignore FunctionInvocationTargetExceptions and other Exceptions
      indexDetailsList.addAll((Set<IndexDetails>) result);
    }
  }

  Collections.sort(indexDetailsList);

  return indexDetailsList;
}
 

public void testNumOfRunningFunctions() throws Exception {
  initManagement(false);
  VM client = managedNodeList.get(2);    
  client.invokeAsync(new SerializableRunnable() {
    public void run() {
      Cache cache = getCache();        
      Function function = new TestFunction(true,
          TestFunction.TEST_FUNCTION_RUNNING_FOR_LONG_TIME);
      Execution execution = FunctionService.onMember(cache
          .getDistributedSystem().getDistributedMember());
      for (int i = 0; i < 100; i++) {
        execution.execute(function);
      }
    }
  });
  Integer numOfRunningFunctions = (Integer) managingNode.invoke(
      TestFunctionsDUnitTest.class, "getNumOfRunningFunction");
  getLogWriter().info(
      "TestNumOfFunctions numOfRunningFunctions= " + numOfRunningFunctions);
  assertTrue(numOfRunningFunctions > 0 ? true : false);
}
 

/**
 * Task to execute Random region Functions
 */
public static void doRandomFunctionExecutions(Region aRegion) {
  Set prKeys = getSomeKeys(aRegion);
  BitSet funcKey = new BitSet(funcKeys.length);
  funcKey.flip(KEYS_FIRST_OP, KEYS_LAST_OP + 1);
  HashSet keySet = ((FunctionServiceTest)testInstance).getKeySet(funcKey,
      prKeys);
  BitSet funcArg = new BitSet(funcArgs.length);
  funcArg.flip(ARGS_FIRST_OP, ARGS_LAST_OP + 1);
  String Args = ((FunctionServiceTest)testInstance).getArgs(funcArg);
  BitSet funcResultCollector = new BitSet(funcResultCollectors.length);
  funcResultCollector.flip(COLLECTOR_FIRST_OP, COLLECTOR_LAST_OP + 1);
  ResultCollector[] resultCollectors = ((FunctionServiceTest)testInstance)
      .getResultCollectors(funcResultCollector);
  BitSet funcDataSet = new BitSet(funcDataSets.length);
  funcDataSet.flip(DATASET_FIRST_OP, DATASET_LAST_OP + 1);
  Execution[] dataSets = ((FunctionServiceTest)testInstance).getDataSets(
      funcDataSet, aRegion);
  doRandomFunctions(keySet, Args, resultCollectors, dataSets);
}
 

public static void executeRegisteredFunction() {
  Region region = cache.getRegion(PartitionedRegionName);
  assertNotNull(region);
  final String testKey = "execKey";
  final Set testKeysSet = new HashSet();
  testKeysSet.add(testKey);
  Execution dataSet = FunctionService.onRegion(region);
  region.put(testKey, new Integer(1));
    ((AbstractExecution)dataSet).removeFunctionAttributes(TestFunction.TEST_FUNCTION2);
    ResultCollector rs = dataSet.withFilter(testKeysSet).withArgs(
        Boolean.TRUE).execute(TestFunction.TEST_FUNCTION2);
    assertEquals(Boolean.TRUE, ((List)rs.getResult()).get(0));
    byte[] functionAttributes = ((AbstractExecution)dataSet).getFunctionAttributes(TestFunction.TEST_FUNCTION2);
    assertNotNull(functionAttributes);
    
    rs = dataSet.withFilter(testKeysSet).withArgs(
        Boolean.TRUE).execute(TestFunction.TEST_FUNCTION2);
    assertEquals(Boolean.TRUE, ((List)rs.getResult()).get(0));
    assertNotNull(functionAttributes);
}
 

public static Object serverExecutionNonHA(Boolean isByName,
    Function function, Boolean toRegister) {

  DistributedSystem.setThreadsSocketPolicy(false);

  if (toRegister.booleanValue()) {
    FunctionService.registerFunction(function);
  }
  Execution member = FunctionService.onServer(pool);
  try {
    ArrayList<String> args = new ArrayList<String>();
    args.add(retryRegionName);
    args.add("serverExecutionNonHA");
    ResultCollector rs = execute(member, args, function, isByName);
    fail("Expected ServerConnectivityException not thrown!");
  } catch (Exception ex) {
    if (!(ex instanceof ServerConnectivityException)) {
      ex.printStackTrace();
      getLogWriter().info("Exception : ", ex);
      fail("Test failed after the execute operation");
    }
  }
  return null;
}
 

public static void serverExecutionHATwoServerDown(Boolean isByName, Function function,
    Boolean toRegister){
  DistributedSystem.setThreadsSocketPolicy(false);

  if (toRegister.booleanValue()) {
    FunctionService.registerFunction(function);
  }
  Execution member = FunctionService.onServer(pool);
  try {
    ArrayList<String> args = new ArrayList<String>();
    args.add(retryRegionName);
    args.add("serverExecutionHATwoServerDown");
    ResultCollector rs = execute(member, args, function, isByName);
    fail("Expected ServerConnectivityException not thrown!");
  } catch (Exception ex) {
    if (!(ex instanceof ServerConnectivityException)) {
      ex.printStackTrace();
      getLogWriter().info("Exception : ", ex);
      fail("Test failed after the execute operation");
    }
  }
}
 

public static Object executeFunctionHA() throws Exception {
  Region region = cache.getRegion(PartitionedRegionName);
  final HashSet testKeysSet = new HashSet();
  for (int i = (totalNumBuckets.intValue() * 10); i > 0; i--) {
    testKeysSet.add("execKey-" + i);
  }
  DistributedSystem.setThreadsSocketPolicy(false);
  Function function = new TestFunction(true, TestFunction.TEST_FUNCTION_HA);
  FunctionService.registerFunction(function);
  Execution dataSet = FunctionService.onRegion(region);
  ResultCollector rc1 = dataSet.withFilter(testKeysSet)
      .withArgs(Boolean.TRUE).execute(function.getId());
  List l = ((List)rc1.getResult());
  getLogWriter().info("Result size : " + l.size());
  return l;
}
 

/***
 * Executes a function with arguments on a set of members , ignores the departed members.
 * @param function Function to be executed.
 * @param args Arguments passed to the function, pass null if you wish to pass no arguments to the function.
 * @param targetMembers Set of members on which the function is to be executed.
 *
 * @return ResultCollector
 */
public static ResultCollector<?, ?> executeFunction(final Function function, Object args , final Set<DistributedMember> targetMembers) {
  Execution execution = null;

  if (args != null) {
    execution = FunctionService.onMembers(targetMembers).withArgs(args);
  } else {
    execution = FunctionService.onMembers(targetMembers);
  }

  ((AbstractExecution) execution).setIgnoreDepartedMembers(true);
  return execution.execute(function);
}
 

public Execution withCollector(ResultCollector rs) {
  if (rs == null) {
    throw new IllegalArgumentException(
        LocalizedStrings.ExecuteRegionFunction_THE_INPUT_0_FOR_THE_EXECUTE_FUNCTION_REQUEST_IS_NULL
            .toLocalizedString("Result Collector"));
  }
  return new MemberFunctionExecutor(this,rs);
}
 

public Execution withArgs(Object args) {
  if (args == null) {
    throw new FunctionException(
        LocalizedStrings.ExecuteRegionFunction_THE_INPUT_0_FOR_THE_EXECUTE_FUNCTION_REQUEST_IS_NULL
            .toLocalizedString("args"));
  }
  return new PartitionedRegionFunctionExecutor(this, args); 
}
 

public Execution withFilter(Set filter) {
  if (filter == null) {
    throw new FunctionException(
        LocalizedStrings.ExecuteRegionFunction_THE_INPUT_0_FOR_THE_EXECUTE_FUNCTION_REQUEST_IS_NULL
            .toLocalizedString("filter"));
  }
  
  return new DistributedRegionFunctionExecutor(this,filter);
}
 

public Execution withArgs(Object args) {
  if (args == null) {
    throw new IllegalArgumentException(
        LocalizedStrings.ExecuteRegionFunction_THE_INPUT_0_FOR_THE_EXECUTE_FUNCTION_REQUEST_IS_NULL
            .toLocalizedString("Args"));
  }
  return new DistributedRegionFunctionExecutor(this, args); 
}
 

public static void serverSingleKeyExecutionOnRegion_SingleConnection() {
  Region region = cache.getRegion(PartitionedRegionName);
  assertNotNull(region);
  for (int i = 0; i < 13; i++) {
    region.put(new Integer(i), "KB_"+i);
  }
  Function function = new TestFunction(false,TEST_FUNCTION2);
  Execution dataSet = FunctionService.onRegion(region);
  dataSet.withArgs(Boolean.TRUE).execute(function);
  region.put(new Integer(2), "KB_2");
  assertEquals("KB_2", region.get(new Integer(2)));    
}
 

public Execution withArgs(Object arguments) {
  if (arguments == null) {
    throw new IllegalArgumentException(
        LocalizedStrings.ExecuteRegionFunction_THE_INPUT_0_FOR_THE_EXECUTE_FUNCTION_REQUEST_IS_NULL
            .toLocalizedString("args"));
  }
  return new MemberFunctionExecutor(this,arguments);
}
 

public static void serverMultiKeyExecutionSocketTimeOut(Boolean isByName) {
  Region region = cache.getRegion(PartitionedRegionName);
  assertNotNull(region);
  final HashSet testKeysSet = new HashSet();
  for (int i = (totalNumBuckets.intValue() * 2); i > 0; i--) {
    testKeysSet.add("execKey-" + i);
  }
  DistributedSystem.setThreadsSocketPolicy(false);
  Function function = new TestFunction(true,TestFunction.TEST_FUNCTION_SOCKET_TIMEOUT);
  FunctionService.registerFunction(function);
  Execution dataSet = FunctionService.onRegion(region);
  try {
    int j = 0;
    HashSet origVals = new HashSet();
    for (Iterator i = testKeysSet.iterator(); i.hasNext();) {
      Integer val = new Integer(j++);
      origVals.add(val);
      region.put(i.next(), val);
    }
    List l = null;
    ResultCollector rc1 = execute(dataSet, testKeysSet, Boolean.TRUE,
        function, isByName);
    l = ((List)rc1.getResult());
    getLogWriter().info("Result size : " + l.size());
    assertEquals(3, l.size());
    for (Iterator i = l.iterator(); i.hasNext();) {
      assertEquals(Boolean.TRUE, i.next());
    }

  }catch(Exception e){
    fail("Test failed after the function execution", e);
    
  }
}
 

public void localDestroyFunction(Region aRegion, Object key) {
  Log.getLogWriter().info("Locally destroying " + key);
  checkContainsValueForKey(aRegion, key, true, "before local destroy");

  Function destroyFunction = new RegionOperationsFunction();
  // FunctionService.registerFunction(destroyFunction);
  Execution dataSet;

  if (aRegion instanceof PartitionedRegion) {
    Log.getLogWriter().info("Inside p2p function execution");
    dataSet = FunctionService.onRegion(aRegion);
  }
  else {
    Log.getLogWriter().info("Inside client server function execution");
    dataSet = FunctionService.onRegion(aRegion);
  }

  final Set keySet = new HashSet();
  keySet.add(key);
  ArrayList aList = new ArrayList();
  aList.add("localdestroy");
  aList.add(RemoteTestModule.getCurrentThread().getThreadId());

  Log.getLogWriter().info("Going to do local destroy execute");
  ResultCollector drc = dataSet.withFilter(keySet).withArgs(aList)
        .execute(destroyFunction.getId());
  drc.getResult();
}
 

public Execution withCollector(ResultCollector rs) {
  if (rs == null) {
    throw new FunctionException(
        LocalizedStrings.ExecuteRegionFunction_THE_INPUT_0_FOR_THE_EXECUTE_FUNCTION_REQUEST_IS_NULL
            .toLocalizedString("Result Collector"));
  }
  return new ServerRegionFunctionExecutor(this, rs);
}
 

public void localDestroyFunction(Region aRegion, Object key) {
  Log.getLogWriter().info("Locally destroying " + key);
  checkContainsValueForKey(aRegion, key, true, "before local destroy");

  Function destroyFunction = new RegionOperationsFunction();
  // FunctionService.registerFunction(destroyFunction);
  Execution dataSet;

  if (aRegion instanceof PartitionedRegion) {
    Log.getLogWriter().info("Inside p2p function execution");
    dataSet = FunctionService.onRegion(aRegion);
  }
  else {
    Log.getLogWriter().info("Inside client server function execution");
    dataSet = FunctionService.onRegion(aRegion);
  }

  final Set keySet = new HashSet();
  keySet.add(key);
  ArrayList aList = new ArrayList();
  aList.add("localdestroy");
  aList.add(RemoteTestModule.getCurrentThread().getThreadId());

  Log.getLogWriter().info("Going to do local destroy execute");
  ResultCollector drc = dataSet.withFilter(keySet).withArgs(aList)
        .execute(destroyFunction.getId());
  drc.getResult();
}
 

public Execution withCollector(ResultCollector rs) {
  if (rs == null) {
    throw new FunctionException(
        LocalizedStrings.ExecuteRegionFunction_THE_INPUT_0_FOR_THE_EXECUTE_FUNCTION_REQUEST_IS_NULL
            .toLocalizedString("Result Collector"));
  }
  return new ServerFunctionExecutor(this, rs);
}
 

public Execution withArgs(Object args) {
  if (args == null) {
    throw new IllegalArgumentException(
        LocalizedStrings.ExecuteRegionFunction_THE_INPUT_0_FOR_THE_EXECUTE_FUNCTION_REQUEST_IS_NULL
            .toLocalizedString("args"));
  }
  return new MultiRegionFunctionExecutor(this, args);
}
 

public Execution withFilter(Set filter) {
  if (filter == null) {
    throw new FunctionException(
        LocalizedStrings.ExecuteRegionFunction_THE_INPUT_0_FOR_THE_EXECUTE_FUNCTION_REQUEST_IS_NULL
            .toLocalizedString("filter"));
  }
  
  return new DistributedRegionFunctionExecutor(this,filter);
}
 

public static void validateRegNoFilterWithArgsWithRC(Execution dataSet,
    ResultCollector rc, String scenario) {

  ArrayList list = (ArrayList)(rc).getResult();
  int listSize = list.size();
  int numOfNodes = ((Integer)parReg.ParRegBB.getBB().getSharedMap().get(
      PR_TOTAL_DATASTORES)).intValue();

  int nodesExecuted = listSize / 10;

  if (!(nodesExecuted <= numOfNodes && nodesExecuted > 0)) {
    throw new TestException(
        "During the scenario "
            + scenario
            + " Function (SecondRandomFunction) should have executed atleast on one node and max on "
            + numOfNodes + " nodes (number of datastores)" + numOfNodes
            + " but executed on " + nodesExecuted + " and result size is " + listSize);

  }

  Iterator listItr = list.iterator();
  while (listItr.hasNext()) {
    if (!(listItr.next() instanceof Boolean)) {
      throw new TestException(
          "During the scenario "
              + scenario
              + " the function (SecondRandomFunction) returns boolean results, hence "
              + "expected the objects in the result list to be Boolean but it is not!");
    }
  }
}
 

private Execution getExecutionObject() {
  DistributedSystem ds = cache.getDistributedSystem();
  InternalDistributedMember localVM = ((InternalDistributedSystem) ds)
      .getDistributionManager().getDistributionManagerId();
  if (region == null)
    return FunctionService.onMembers(ds);
  else
    return FunctionService.onRegion(region);
  // TODO Add more types!!!
}
 

public static void serverMultiKeyExecutionSocketTimeOut(Boolean isByName) {
  Region region = cache.getRegion(PartitionedRegionName);
  assertNotNull(region);
  final HashSet testKeysSet = new HashSet();
  for (int i = (totalNumBuckets.intValue() * 2); i > 0; i--) {
    testKeysSet.add("execKey-" + i);
  }
  DistributedSystem.setThreadsSocketPolicy(false);
  Function function = new TestFunction(true,
      TestFunction.TEST_FUNCTION_SOCKET_TIMEOUT);
  FunctionService.registerFunction(function);
  Execution dataSet = FunctionService.onRegion(region);
  try {
    int j = 0;
    HashSet origVals = new HashSet();
    for (Iterator i = testKeysSet.iterator(); i.hasNext();) {
      Integer val = new Integer(j++);
      origVals.add(val);
      region.put(i.next(), val);
    }
    List l = null;
    ResultCollector rc1 = execute(dataSet, testKeysSet, Boolean.TRUE,
        function, isByName);
    l = ((List)rc1.getResult());
    getLogWriter().info("Result size : " + l.size());
    assertEquals(3, l.size());
    for (Iterator i = l.iterator(); i.hasNext();) {
      assertEquals(Boolean.TRUE, i.next());
    }

  }
  catch (Exception e) {
    fail("Test failed after the function execution", e);

  }
}