/*
 * 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.execute;

import static org.apache.geode.cache.Region.SEPARATOR;
import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS;
import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
import static org.apache.geode.distributed.ConfigurationProperties.SERIALIZABLE_OBJECT_FILTER;
import static org.apache.geode.internal.cache.functions.TestFunction.TEST_FUNCTION1;
import static org.apache.geode.internal.cache.functions.TestFunction.TEST_FUNCTION2;
import static org.apache.geode.internal.cache.functions.TestFunction.TEST_FUNCTION3;
import static org.apache.geode.internal.cache.functions.TestFunction.TEST_FUNCTION7;
import static org.apache.geode.internal.cache.functions.TestFunction.TEST_FUNCTION_BUCKET_FILTER;
import static org.apache.geode.internal.cache.functions.TestFunction.TEST_FUNCTION_HA;
import static org.apache.geode.internal.cache.functions.TestFunction.TEST_FUNCTION_LASTRESULT;
import static org.apache.geode.internal.cache.functions.TestFunction.TEST_FUNCTION_REEXECUTE_EXCEPTION;
import static org.apache.geode.test.dunit.Host.getHost;
import static org.apache.geode.test.dunit.Invoke.invokeInEveryVM;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.junit.Assert.assertEquals;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.TimeUnit;

import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.junit.experimental.categories.Category;

import org.apache.geode.cache.AttributesFactory;
import org.apache.geode.cache.EntryOperation;
import org.apache.geode.cache.PartitionAttributes;
import org.apache.geode.cache.PartitionAttributesFactory;
import org.apache.geode.cache.PartitionResolver;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.Region.Entry;
import org.apache.geode.cache.RegionAttributes;
import org.apache.geode.cache.RegionFactory;
import org.apache.geode.cache.RegionShortcut;
import org.apache.geode.cache.execute.Execution;
import org.apache.geode.cache.execute.Function;
import org.apache.geode.cache.execute.FunctionAdapter;
import org.apache.geode.cache.execute.FunctionContext;
import org.apache.geode.cache.execute.FunctionException;
import org.apache.geode.cache.execute.FunctionService;
import org.apache.geode.cache.execute.RegionFunctionContext;
import org.apache.geode.cache.execute.ResultCollector;
import org.apache.geode.cache.partition.PartitionRegionHelper;
import org.apache.geode.distributed.DistributedSystem;
import org.apache.geode.internal.cache.EntrySnapshot;
import org.apache.geode.internal.cache.LocalDataSet;
import org.apache.geode.internal.cache.PartitionedRegion;
import org.apache.geode.internal.cache.execute.data.CustId;
import org.apache.geode.internal.cache.execute.data.Customer;
import org.apache.geode.internal.cache.execute.data.Order;
import org.apache.geode.internal.cache.execute.data.OrderId;
import org.apache.geode.internal.cache.functions.TestFunction;
import org.apache.geode.test.dunit.AsyncInvocation;
import org.apache.geode.test.dunit.IgnoredException;
import org.apache.geode.test.dunit.VM;
import org.apache.geode.test.dunit.cache.CacheTestCase;
import org.apache.geode.test.junit.categories.FunctionServiceTest;

@Category({FunctionServiceTest.class})
public class PRFunctionExecutionDUnitTest extends CacheTestCase {

  private static final String STRING_KEY = "execKey";

  private String regionName;

  private String regionNameTop;
  private String regionNameColocated1;
  private String regionNameColocated2;

  @Before
  public void setUp() {
    regionName = getUniqueName();

    regionNameTop = regionName + "_top";
    regionNameColocated1 = regionName + "_colo1";
    regionNameColocated2 = regionName + "_colo2";
  }

  @After
  public void tearDown() {
    invokeInEveryVM(() -> DistributedSystem.setThreadsSocketPolicy(false));
    DistributedSystem.setThreadsSocketPolicy(false);
  }

  @Override
  public Properties getDistributedSystemProperties() {
    Properties config = new Properties();
    config.put(SERIALIZABLE_OBJECT_FILTER,
        "org.apache.geode.test.junit.rules.**;org.apache.geode.internal.cache.execute.**;org.apache.geode.internal.cache.functions.**;org.apache.geode.test.dunit.**");
    return config;
  }

  /**
   * Test to validate that the function execution is successful on PR with Loner Distributed System
   */
  @Test
  public void testFunctionExecution() throws Exception {
    Properties config = getDistributedSystemProperties();
    config.setProperty(MCAST_PORT, "0");
    config.setProperty(LOCATORS, "");
    getCache(config);

    Region<String, Integer> region = createPartitionedRegion(regionName, 10, 0);

    Function<Boolean> function = new TestFunction<>(true, TEST_FUNCTION1);
    FunctionService.registerFunction(function);

    Execution<Boolean, Boolean, Collection<Boolean>> execution = FunctionService.onRegion(region);

    ResultCollector<Boolean, Collection<Boolean>> resultCollector =
        execution.setArguments(true).withFilter(createKeySet(STRING_KEY)).execute(function);

    assertThat(resultCollector.getResult()).containsExactly(true);
  }

  @Test
  public void testHAFunctionExecution() throws Exception {
    Region<String, Integer> region = createPartitionedRegion(regionName, 10, 0);

    Function<Void> function = new TestFunction<>(false, TestFunction.TEST_FUNCTION10);

    assertThatThrownBy(() -> FunctionService.registerFunction(function))
        .isInstanceOf(FunctionException.class)
        .hasMessageContaining("For Functions with isHA true, hasResult must also be true.");

    Execution<String, Void, Void> dataSet = FunctionService.onRegion(region);
    assertThatThrownBy(() -> dataSet.withFilter(createKeySet(STRING_KEY)).setArguments(STRING_KEY)
        .execute(function)).isInstanceOf(FunctionException.class)
            .hasMessageContaining("For Functions with isHA true, hasResult must also be true.");
  }

  /**
   * Test remote execution by a pure accessor which doesn't have the function factory present.
   */
  @Test
  public void testRemoteSingleKeyExecution_byName() throws Exception {
    VM accessor = getHost(0).getVM(2);
    VM datastore = getHost(0).getVM(3);

    accessor.invoke(() -> {
      createPartitionedRegion(regionName, 0, 0);
    });

    datastore.invoke(() -> {
      createPartitionedRegion(regionName, 10, 0);
      FunctionService.registerFunction(new TestFunction(true, TEST_FUNCTION2));
    });

    accessor.invoke(() -> {
      Region<String, Integer> region = getPartitionedRegion(regionName);
      region.put(STRING_KEY, 1);

      Function<Boolean> function = new TestFunction<>(true, TEST_FUNCTION2);
      FunctionService.registerFunction(function);
      Execution<Object, Object, List<Object>> dataSet = FunctionService.onRegion(region);
      dataSet.withFilter(createKeySet(STRING_KEY)).setArguments(true).execute(function.getId());

      ResultCollector<Object, List<Object>> resultCollector =
          dataSet.withFilter(createKeySet(STRING_KEY)).setArguments(true).execute(function.getId());

      assertThat(resultCollector.getResult()).hasSize(1).containsExactly(true);

      resultCollector = dataSet.withFilter(createKeySet(STRING_KEY)).setArguments(STRING_KEY)
          .execute(function.getId());

      assertThat(resultCollector.getResult()).hasSize(1).containsExactly(1);

      Map<String, Integer> putData = createEntryMap(2, STRING_KEY, 1, 2);
      resultCollector = dataSet.withFilter(createKeySet(STRING_KEY)).setArguments(putData)
          .execute(function.getId());

      assertThat(resultCollector.getResult()).hasSize(1).containsExactly(true);
      assertThat(region.get(STRING_KEY + "1")).isEqualTo(2);
      assertThat(region.get(STRING_KEY + "2")).isEqualTo(3);
    });
  }

  /**
   * Test remote execution with timeout by sending function id where timeout does not expire.
   */
  @Test
  public void testRemoteSingleKeyExecution_byName_NotTimedOut() throws Exception {
    VM accessor = getHost(0).getVM(2);
    VM datastore = getHost(0).getVM(3);

    accessor.invoke(() -> {
      createPartitionedRegion(regionName, 0, 0);
    });

    datastore.invoke(() -> {
      createPartitionedRegion(regionName, 10, 0);
      FunctionService.registerFunction(
          new TestFunction(true, TestFunction.TEST_FUNCTION_RUNNING_FOR_LONG_TIME));
    });

    accessor.invoke(() -> {
      Region<String, Integer> region = getPartitionedRegion(regionName);
      region.put(STRING_KEY, 1);

      Function<Boolean> function =
          new TestFunction<>(true, TestFunction.TEST_FUNCTION_RUNNING_FOR_LONG_TIME);
      FunctionService.registerFunction(function);
      Execution<Object, Object, List<Object>> dataSet = FunctionService.onRegion(region);
      dataSet.withFilter(createKeySet(STRING_KEY)).setArguments(true).execute(function.getId());

      long timeout = 8000;
      TimeUnit unit = TimeUnit.MILLISECONDS;
      ResultCollector<Object, List<Object>> resultCollector =
          dataSet.withFilter(createKeySet(STRING_KEY)).setArguments(true).execute(function.getId(),
              timeout, unit);

      List li = resultCollector.getResult();
      assertEquals(li.get(0), "Ran executeFunctionRunningForLongTime for 2000");

    });
  }

  /**
   * Test remote execution with timeout by sending function id where timeout expires.
   */
  @Test
  public void testRemoteSingleKeyExecution_byName_TimedOut() throws Exception {
    VM accessor = getHost(0).getVM(2);
    VM datastore = getHost(0).getVM(3);

    accessor.invoke(() -> {
      createPartitionedRegion(regionName, 0, 0);
    });

    datastore.invoke(() -> {
      createPartitionedRegion(regionName, 10, 0);
      FunctionService.registerFunction(
          new TestFunction(true, TestFunction.TEST_FUNCTION_RUNNING_FOR_LONG_TIME));
    });

    accessor.invoke(() -> {
      Region<String, Integer> region = getPartitionedRegion(regionName);
      region.put(STRING_KEY, 1);

      Function<Boolean> function =
          new TestFunction<>(true, TestFunction.TEST_FUNCTION_RUNNING_FOR_LONG_TIME);
      FunctionService.registerFunction(function);
      Execution<Object, Object, List<Object>> dataSet = FunctionService.onRegion(region)
          .withFilter(createKeySet(STRING_KEY)).setArguments(STRING_KEY);

      long timeout = 1000;
      TimeUnit unit = TimeUnit.MILLISECONDS;

      assertThatThrownBy(() -> {
        dataSet.execute(function.getId(), timeout, unit);
      }).hasCauseInstanceOf(FunctionException.class)
          .hasMessageContaining("All results not received in time provided");
    });
  }

  /**
   * Test local execution by a datastore Function throws the FunctionInvocationTargetException. As
   * this is the case of HA then system should retry the function execution. After 5th attempt
   * function will send Boolean as last result. factory present.
   */
  @Test
  public void testLocalSingleKeyExecution_byName_FunctionInvocationTargetException()
      throws Exception {
    Region<String, Integer> region = createPartitionedRegion(regionName, 10, 0);
    region.put(STRING_KEY, 1);

    Function<Integer> function = new TestFunction<>(true, TEST_FUNCTION_REEXECUTE_EXCEPTION);
    FunctionService.registerFunction(function);
    Execution<Boolean, Integer, List<Integer>> dataSet = FunctionService.onRegion(region);

    ResultCollector<Integer, List<Integer>> resultCollector =
        dataSet.withFilter(createKeySet(STRING_KEY)).setArguments(true).execute(function.getId());

    List<Integer> list = resultCollector.getResult();
    assertThat(list.get(0)).isEqualTo(5);
  }

  /**
   * Test remote execution by a pure accessor which doesn't have the function factory
   * present.Function throws the FunctionInvocationTargetException. As this is the case of HA then
   * system should retry the function execution. After 5th attempt function will send Boolean as
   * last result.
   */
  @Test
  public void testRemoteSingleKeyExecution_byName_FunctionInvocationTargetException()
      throws Exception {
    VM accessor = getHost(0).getVM(2);
    VM datastore = getHost(0).getVM(3);

    accessor.invoke(() -> {
      createPartitionedRegion(regionName, 0, 0);
    });

    datastore.invoke(() -> {
      createPartitionedRegion(regionName, 10, 0);
      Function function = new TestFunction(true, TEST_FUNCTION_REEXECUTE_EXCEPTION);
      FunctionService.registerFunction(function);
    });

    accessor.invoke(() -> {
      Region<String, Integer> region = getPartitionedRegion(regionName);
      region.put(STRING_KEY, 1);

      Function<Integer> function = new TestFunction<>(true, TEST_FUNCTION_REEXECUTE_EXCEPTION);
      FunctionService.registerFunction(function);
      Execution<Boolean, Integer, List<Integer>> dataSet = FunctionService.onRegion(region);

      ResultCollector<Integer, List<Integer>> resultCollector =
          dataSet.withFilter(createKeySet(STRING_KEY)).setArguments(true).execute(function.getId());

      List<Integer> list = resultCollector.getResult();
      assertThat(list.get(0)).isEqualTo(5);
    });
  }

  /**
   * Test remote execution by a pure accessor which doesn't have the function factory present.
   */
  @Test
  public void testRemoteSingleKeyExecution_byInstance() throws Exception {
    VM accessor = getHost(0).getVM(2);
    VM datastore = getHost(0).getVM(3);

    accessor.invoke(() -> {
      createPartitionedRegion(regionName, 0, 0);
    });

    datastore.invoke(() -> {
      createPartitionedRegion(regionName, 10, 0);

      Function function = new TestFunction(true, TEST_FUNCTION2);
      FunctionService.registerFunction(function);
    });

    accessor.invoke(() -> {
      Region<String, Integer> region = getPartitionedRegion(regionName);
      region.put(STRING_KEY, 1);

      Function function = new TestFunction(true, TEST_FUNCTION2);
      FunctionService.registerFunction(function);

      Execution<Boolean, Boolean, List<Boolean>> booleanDataSet = FunctionService.onRegion(region);
      ResultCollector<Boolean, List<Boolean>> booleanResultCollector =
          booleanDataSet.withFilter(createKeySet(STRING_KEY)).setArguments(true).execute(function);
      assertThat(booleanResultCollector.getResult().get(0)).isTrue();

      Execution<String, Integer, List<Integer>> integerDataSet = FunctionService.onRegion(region);
      ResultCollector<Integer, List<Integer>> integerResultCollector = integerDataSet
          .withFilter(createKeySet(STRING_KEY)).setArguments(STRING_KEY).execute(function);
      assertThat(integerResultCollector.getResult().get(0)).isEqualTo(1);

      Map<String, Integer> putData = createEntryMap(2, STRING_KEY, 1, 2);
      Execution<Map<String, Integer>, Boolean, List<Boolean>> mapDataSet =
          FunctionService.onRegion(region);
      ResultCollector<Boolean, List<Boolean>> mapResultCollector =
          mapDataSet.withFilter(createKeySet(STRING_KEY)).setArguments(putData).execute(function);
      assertThat(mapResultCollector.getResult().get(0)).isTrue();

      assertThat(region.get(STRING_KEY + "1")).isEqualTo(2);
      assertThat(region.get(STRING_KEY + "2")).isEqualTo(3);
    });
  }

  /**
   * Test remote execution with timeout by sending function instance where timeout does not expire.
   */
  @Test
  public void testRemoteSingleKeyExecution_byInstance_NotTimedOut() throws Exception {
    VM accessor = getHost(0).getVM(2);
    VM datastore = getHost(0).getVM(3);

    accessor.invoke(() -> {
      createPartitionedRegion(regionName, 0, 0);
    });

    datastore.invoke(() -> {
      createPartitionedRegion(regionName, 10, 0);
      FunctionService.registerFunction(
          new TestFunction(true, TestFunction.TEST_FUNCTION_RUNNING_FOR_LONG_TIME));
    });

    accessor.invoke(() -> {
      Region<String, Integer> region = getPartitionedRegion(regionName);
      region.put(STRING_KEY, 1);

      Function<Boolean> function =
          new TestFunction<>(true, TestFunction.TEST_FUNCTION_RUNNING_FOR_LONG_TIME);
      FunctionService.registerFunction(function);
      Execution<Object, Object, List<Object>> dataSet = FunctionService.onRegion(region);
      dataSet.withFilter(createKeySet(STRING_KEY)).setArguments(true).execute(function);

      long timeout = 8000;
      TimeUnit unit = TimeUnit.MILLISECONDS;
      ResultCollector<Object, List<Object>> resultCollector =
          dataSet.withFilter(createKeySet(STRING_KEY)).setArguments(true).execute(function.getId(),
              timeout, unit);

      List li = resultCollector.getResult();
      assertEquals(li.get(0), "Ran executeFunctionRunningForLongTime for 2000");

    });
  }

  /**
   * Test remote execution with timeout by sending function instance where timeout expires.
   */
  @Test
  public void testRemoteSingleKeyExecution_byInstance_TimedOut() throws Exception {
    VM accessor = getHost(0).getVM(2);
    VM datastore = getHost(0).getVM(3);

    accessor.invoke(() -> {
      createPartitionedRegion(regionName, 0, 0);
    });

    datastore.invoke(() -> {
      createPartitionedRegion(regionName, 10, 0);
      FunctionService.registerFunction(
          new TestFunction(true, TestFunction.TEST_FUNCTION_RUNNING_FOR_LONG_TIME));
    });

    accessor.invoke(() -> {
      Region<String, Integer> region = getPartitionedRegion(regionName);
      region.put(STRING_KEY, 1);

      Function<Boolean> function =
          new TestFunction<>(true, TestFunction.TEST_FUNCTION_RUNNING_FOR_LONG_TIME);
      FunctionService.registerFunction(function);
      Execution<Object, Object, List<Object>> dataSet = FunctionService.onRegion(region)
          .withFilter(createKeySet(STRING_KEY)).setArguments(STRING_KEY);

      long timeout = 1000;
      TimeUnit unit = TimeUnit.MILLISECONDS;

      assertThatThrownBy(() -> {
        dataSet.execute(function, timeout, unit);
      }).hasCauseInstanceOf(FunctionException.class)
          .hasMessageContaining("All results not received in time provided");
    });
  }

  /**
   * Test remote execution of inline function by a pure accessor
   */
  @Test
  public void testRemoteSingleKeyExecution_byInlineFunction() throws Exception {
    VM accessor = getHost(0).getVM(2);
    VM datastore = getHost(0).getVM(3);

    accessor.invoke(() -> {
      createPartitionedRegion(regionName, 0, 0);
    });

    datastore.invoke(() -> {
      createPartitionedRegion(regionName, 10, 0);
    });

    accessor.invoke(() -> {
      Region<String, Integer> region = getPartitionedRegion(regionName);
      region.put(STRING_KEY, 1);

      Execution<Boolean, Boolean, List<Boolean>> dataSet = FunctionService.onRegion(region);
      ResultCollector<Boolean, List<Boolean>> rs1 = dataSet.withFilter(createKeySet(STRING_KEY))
          .setArguments(true).execute(new BooleanFunction(true));

      assertThat(rs1.getResult().get(0)).isTrue();
    });
  }

  /**
   * Test multi-key remote execution by a pure accessor which doesn't have the function factory
   * present. ResultCollector = DefaultResultCollector haveResults = true;
   */
  @Test
  public void testRemoteMultiKeyExecution_byName() throws Exception {
    VM accessor = getHost(0).getVM(3);
    VM datastore0 = getHost(0).getVM(0);
    VM datastore1 = getHost(0).getVM(1);
    VM datastore2 = getHost(0).getVM(2);

    accessor.invoke(() -> {
      createPartitionedRegion(regionName, 0, 0);
    });

    datastore0.invoke(() -> {
      createPartitionedRegion(regionName, 10, 0);
      FunctionService.registerFunction(new TestFunction(true, TEST_FUNCTION2));
    });

    datastore1.invoke(() -> {
      createPartitionedRegion(regionName, 10, 0);
      FunctionService.registerFunction(new TestFunction(true, TEST_FUNCTION2));
    });

    datastore2.invoke(() -> {
      createPartitionedRegion(regionName, 10, 0);
      FunctionService.registerFunction(new TestFunction(true, TEST_FUNCTION2));
    });

    accessor.invoke(() -> {
      PartitionedRegion pr = getPartitionedRegion(regionName);

      Set<String> keySet = new HashSet<>();
      for (int i = pr.getTotalNumberOfBuckets() * 2; i > 0; i--) {
        keySet.add(STRING_KEY + i);
      }

      Collection<Integer> expectedValues = new HashSet<>();
      int valueIndex = 0;
      for (String key : keySet) {
        int value = valueIndex++;
        expectedValues.add(value);
        pr.put(key, value);
      }

      Function function = new TestFunction(true, TEST_FUNCTION2);
      FunctionService.registerFunction(function);

      Execution<Boolean, Boolean, List<Boolean>> booleanExecution = FunctionService.onRegion(pr);
      ResultCollector<Boolean, List<Boolean>> booleanResultCollector =
          booleanExecution.withFilter(keySet).setArguments(true).execute(function.getId());

      List<Boolean> booleanResultList = booleanResultCollector.getResult();
      assertThat(booleanResultList).hasSize(3).containsExactly(true, true, true);

      Execution<Set<String>, List<Integer>, List<List<Integer>>> integerExecution =
          FunctionService.onRegion(pr);
      ResultCollector<List<Integer>, List<List<Integer>>> valuesResultCollector =
          integerExecution.withFilter(keySet).setArguments(keySet).execute(function.getId());

      List<List<Integer>> valuesResults = valuesResultCollector.getResult();
      assertThat(valuesResults).hasSize(3).isInstanceOf(List.class);
      assertThat(valuesResults.get(0)).isInstanceOf(List.class);

      Collection<Integer> actualValues = new HashSet<>();
      for (List<Integer> values : valuesResults) {
        assertThat(values.size()).isGreaterThan(0);
        for (int value : values) {
          assertThat(actualValues.add(value)).isTrue();
        }
      }
      assertThat(actualValues).isEqualTo(expectedValues);
    });
  }

  @Test
  public void testRemoteMultiKeyExecution_BucketMoved() throws Exception {
    VM accessor = getHost(0).getVM(3);
    VM datastore0 = getHost(0).getVM(0);
    VM datastore1 = getHost(0).getVM(1);
    VM datastore2 = getHost(0).getVM(2);

    accessor.invoke(() -> {
      createPartitionedRegion(regionName, createPartitionAttributes(0, 1, 0, 113));
    });

    datastore0.invoke(() -> {
      createPartitionedRegion(regionName, createPartitionAttributes(40, 1, 0, 113));
      FunctionService.registerFunction(new TestFunction(true, TEST_FUNCTION_LASTRESULT));
    });

    datastore1.invoke(() -> {
      createPartitionedRegion(regionName, createPartitionAttributes(40, 1, 0, 113));
      FunctionService.registerFunction(new TestFunction(true, TEST_FUNCTION_LASTRESULT));
    });

    accessor.invoke(() -> {
      Region<Integer, String> region = getPartitionedRegion(regionName);
      for (int i = 0; i < 113; i++) {
        region.put(i, STRING_KEY + i);
      }
    });

    datastore2.invoke(() -> {
      createPartitionedRegion(regionName, createPartitionAttributes(40, 1, 0, 113));
      FunctionService.registerFunction(new TestFunction(true, TEST_FUNCTION_LASTRESULT));
    });

    accessor.invoke(() -> {
      PartitionedRegion pr = getPartitionedRegion(regionName);

      Function<Boolean> function = new TestFunction<>(true, TEST_FUNCTION_LASTRESULT);
      FunctionService.registerFunction(function);
      Execution<Boolean, Boolean, List<Boolean>> dataSet = FunctionService.onRegion(pr);

      ResultCollector<Boolean, List<Boolean>> resultCollector =
          dataSet.setArguments(true).execute(function.getId());

      List<Boolean> resultList = resultCollector.getResult();
      assertThat(resultList).hasSize(2).containsExactly(true, true);
    });
  }

  @Test
  public void testLocalMultiKeyExecution_BucketMoved() throws Exception {
    IgnoredException.addIgnoredException("BucketMovedException");

    VM datastore0 = getHost(0).getVM(0);
    VM datastore1 = getHost(0).getVM(1);

    datastore0.invoke(() -> {
      createPartitionedRegion(regionName, createPartitionAttributes(40, 0, 0, 113));
      FunctionService.registerFunction(new TestFunction(true, TEST_FUNCTION_LASTRESULT));
    });

    datastore0.invoke(() -> {
      Region<Integer, String> region = getPartitionedRegion(regionName);
      for (int i = 0; i < 113; i++) {
        region.put(i, STRING_KEY + i);
      }
    });

    datastore1.invoke(() -> {
      createPartitionedRegion(regionName, createPartitionAttributes(40, 0, 0, 113));
      FunctionService.registerFunction(new TestFunction(true, TEST_FUNCTION_LASTRESULT));
    });

    datastore0.invoke(() -> {
      Region<Integer, String> region = getPartitionedRegion(regionName);

      Function<Boolean> function = new TestFunction<>(true, TEST_FUNCTION_LASTRESULT);
      FunctionService.registerFunction(function);
      Execution<Boolean, Boolean, List<Boolean>> dataSet = FunctionService.onRegion(region);

      ResultCollector<Boolean, List<Boolean>> resultCollector =
          dataSet.setArguments(true).execute(function.getId());

      List<Boolean> resultList = resultCollector.getResult();
      assertThat(resultList).hasSize(2).containsExactly(true, true);
    });
  }

  /**
   * Test remote execution by a pure accessor which doesn't have the function factory
   * present.Function throws the FunctionInvocationTargetException. As this is the case of HA then
   * system should retry the function execution. After 5th attempt function will send Boolean as
   * last result.
   */
  @Test
  public void testRemoteMultipleKeyExecution_byName_FunctionInvocationTargetException()
      throws Exception {
    VM accessor = getHost(0).getVM(3);
    VM datastore0 = getHost(0).getVM(0);
    VM datastore1 = getHost(0).getVM(1);
    VM datastore2 = getHost(0).getVM(2);

    accessor.invoke(() -> {
      createPartitionedRegion(regionName, 0, 0);
    });

    datastore0.invoke(() -> {
      createPartitionedRegion(regionName, createPartitionAttributes(10, 0));
      FunctionService.registerFunction(new TestFunction(true, TEST_FUNCTION_REEXECUTE_EXCEPTION));
    });

    datastore1.invoke(() -> {
      createPartitionedRegion(regionName, createPartitionAttributes(10, 0));
      FunctionService.registerFunction(new TestFunction(true, TEST_FUNCTION_REEXECUTE_EXCEPTION));
    });

    datastore2.invoke(() -> {
      createPartitionedRegion(regionName, createPartitionAttributes(10, 0));
      FunctionService.registerFunction(new TestFunction(true, TEST_FUNCTION_REEXECUTE_EXCEPTION));
    });

    accessor.invoke(() -> {
      PartitionedRegion pr = getPartitionedRegion(regionName);

      Set<String> keySet = new HashSet<>();
      for (int i = pr.getTotalNumberOfBuckets() * 2; i > 0; i--) {
        keySet.add(STRING_KEY + i);
      }

      Function<Integer> function = new TestFunction<>(true, TEST_FUNCTION_REEXECUTE_EXCEPTION);
      FunctionService.registerFunction(function);
      Execution<Boolean, Integer, List<Integer>> dataSet = FunctionService.onRegion(pr);

      int valueIndex = 0;
      for (String key : keySet) {
        int value = valueIndex++;
        pr.put(key, value);
      }

      ResultCollector<Integer, List<Integer>> resultCollector =
          dataSet.withFilter(keySet).setArguments(true).execute(function.getId());

      List<Integer> resultList = resultCollector.getResult();
      assertThat(resultList).hasSize(3).containsExactly(5, 5, 5);
    });
  }

  @Test
  public void testRemoteMultiKeyExecutionHA_CacheClose() throws Exception {
    VM accessor = getHost(0).getVM(3);
    VM datastore0 = getHost(0).getVM(0);
    VM datastore1 = getHost(0).getVM(1);
    VM datastore2 = getHost(0).getVM(2);

    accessor.invoke(() -> {
      createPartitionedRegion(regionName, 0, 1);
    });

    datastore0.invoke(() -> {
      createPartitionedRegion(regionName, 10, 1);
      FunctionService.registerFunction(new TestFunction(true, TEST_FUNCTION_HA));
    });

    datastore1.invoke(() -> {
      createPartitionedRegion(regionName, 10, 1);
      FunctionService.registerFunction(new TestFunction(true, TEST_FUNCTION_HA));
    });

    datastore2.invoke(() -> {
      createPartitionedRegion(regionName, 10, 1);
      FunctionService.registerFunction(new TestFunction(true, TEST_FUNCTION_HA));
    });

    accessor.invoke(() -> {
      PartitionedRegion pr = getPartitionedRegion(regionName);

      Collection<String> keySet = new HashSet<>();
      for (int i = pr.getTotalNumberOfBuckets() * 2; i > 0; i--) {
        keySet.add(STRING_KEY + i);
      }

      int valueIndex = 0;
      for (String key : keySet) {
        int value = valueIndex++;
        pr.put(key, value);
      }
    });

    AsyncInvocation<List<Boolean>> async = accessor.invokeAsync(this::executeFunction);

    datastore0.invoke(() -> {
      Thread.sleep(3_000);
      getCache().close();
    });

    List<Boolean> resultList = async.get();
    assertThat(resultList).hasSize(2).containsExactly(true, true);
  }

  @Test
  public void testRemoteMultiKeyExecutionHA_Disconnect() throws Exception {
    VM accessor = getHost(0).getVM(3);
    VM datastore0 = getHost(0).getVM(0);
    VM datastore1 = getHost(0).getVM(1);
    VM datastore2 = getHost(0).getVM(2);

    accessor.invoke(() -> {
      createPartitionedRegion(regionName, 0, 1);
    });

    datastore0.invoke(() -> {
      createPartitionedRegion(regionName, 10, 1);
      FunctionService.registerFunction(new TestFunction(true, TEST_FUNCTION_HA));
    });

    datastore1.invoke(() -> {
      createPartitionedRegion(regionName, 10, 1);
      FunctionService.registerFunction(new TestFunction(true, TEST_FUNCTION_HA));
    });

    datastore2.invoke(() -> {
      createPartitionedRegion(regionName, 10, 1);
      FunctionService.registerFunction(new TestFunction(true, TEST_FUNCTION_HA));
    });

    accessor.invoke(() -> {
      PartitionedRegion pr = getPartitionedRegion(regionName);

      Collection<String> keySet = new HashSet<>();
      for (int i = pr.getTotalNumberOfBuckets() * 2; i > 0; i--) {
        keySet.add(STRING_KEY + i);
      }

      int valueIndex = 0;
      for (String key : keySet) {
        int value = valueIndex++;
        pr.put(key, value);
      }
    });

    AsyncInvocation<List<Boolean>> async = accessor.invokeAsync(this::executeFunction);

    datastore0.invoke(() -> {
      Thread.sleep(3000);
      getCache().getDistributedSystem().disconnect();
    });

    List<Boolean> resultList = async.get();
    assertThat(resultList).hasSize(2).containsExactly(true, true);
  }

  /**
   * Test multi-key remote execution of inline function by a pure accessor ResultCollector =
   * DefaultResultCollector haveResults = true;
   */
  @Test
  public void testRemoteMultiKeyExecution_byInlineFunction() throws Exception {
    VM accessor = getHost(0).getVM(3);
    VM datastore0 = getHost(0).getVM(0);
    VM datastore1 = getHost(0).getVM(1);
    VM datastore2 = getHost(0).getVM(2);

    accessor.invoke(() -> {
      createPartitionedRegion(regionName, 0, 0);
    });

    datastore0.invoke(() -> {
      createPartitionedRegion(regionName, 10, 0);
    });

    datastore1.invoke(() -> {
      createPartitionedRegion(regionName, 10, 0);
    });

    datastore2.invoke(() -> {
      createPartitionedRegion(regionName, 10, 0);
    });

    accessor.invoke(() -> {
      PartitionedRegion pr = getPartitionedRegion(regionName);

      Set<String> keySet = new HashSet<>();
      for (int i = pr.getTotalNumberOfBuckets() * 2; i > 0; i--) {
        keySet.add(STRING_KEY + i);
      }

      int valueIndex = 0;
      for (String key : keySet) {
        int value = valueIndex++;
        pr.put(key, value);
      }

      Execution<Boolean, Boolean, List<Boolean>> execution = FunctionService.onRegion(pr);

      ResultCollector<Boolean, List<Boolean>> resultCollector =
