/*
 * 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 java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.stream.Collectors.toList;
import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS;
import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
import static org.apache.geode.distributed.ConfigurationProperties.NAME;
import static org.apache.geode.distributed.ConfigurationProperties.SECURITY_CLIENT_AUTHENTICATOR;
import static org.apache.geode.distributed.ConfigurationProperties.SECURITY_CLIENT_AUTH_INIT;
import static org.apache.geode.distributed.ConfigurationProperties.SERIALIZABLE_OBJECT_FILTER;
import static org.apache.geode.test.awaitility.GeodeAwaitility.getTimeout;
import static org.apache.geode.test.dunit.IgnoredException.addIgnoredException;
import static org.apache.geode.test.dunit.VM.getController;
import static org.apache.geode.test.dunit.VM.getVM;
import static org.apache.geode.test.dunit.VM.toArray;
import static org.apache.geode.util.internal.GeodeGlossary.GEMFIRE_PREFIX;
import static org.apache.geode.util.internal.UncheckedUtils.uncheckedCast;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.catchThrowable;

import java.io.IOException;
import java.io.Serializable;
import java.util.Comparator;
import java.util.HashSet;
import java.util.List;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.IntStream;

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

import org.apache.geode.cache.CacheClosedException;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.Scope;
import org.apache.geode.cache.client.ClientRegionShortcut;
import org.apache.geode.cache.client.Pool;
import org.apache.geode.cache.client.PoolFactory;
import org.apache.geode.cache.client.PoolManager;
import org.apache.geode.cache.client.internal.InternalClientCache;
import org.apache.geode.cache.execute.Function;
import org.apache.geode.cache.execute.FunctionContext;
import org.apache.geode.cache.execute.FunctionException;
import org.apache.geode.cache.execute.FunctionInvocationTargetException;
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.server.CacheServer;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.execute.util.TypedFunctionService;
import org.apache.geode.security.templates.DummyAuthenticator;
import org.apache.geode.security.templates.UserPasswordAuthInit;
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.rules.CacheRule;
import org.apache.geode.test.dunit.rules.ClientCacheRule;
import org.apache.geode.test.dunit.rules.DistributedRestoreSystemProperties;
import org.apache.geode.test.dunit.rules.DistributedRule;
import org.apache.geode.test.junit.categories.FunctionServiceTest;

@Category(FunctionServiceTest.class)
@SuppressWarnings("serial")
public class DistributedRegionFunctionExecutionDUnitTest implements Serializable {

  private static final AtomicReference<Region<?, ?>> REGION = new AtomicReference<>();
  private static final AtomicReference<CountDownLatch> LATCH = new AtomicReference<>();

  private final Set<String> filter = new HashSet<>();

  private String regionName;
  private String poolName;

  private VM empty;
  private VM normal;
  private VM replicate1;
  private VM replicate2;
  private VM replicate3;

  @Rule
  public CacheRule cacheRule = new CacheRule();

  @Rule
  public ClientCacheRule clientCacheRule = new ClientCacheRule();

  @Rule
  public DistributedRestoreSystemProperties restoreSystemProperties =
      new DistributedRestoreSystemProperties();

  @Before
  public void setUp() throws Exception {
    empty = getController();
    replicate1 = getVM(0);
    replicate2 = getVM(1);
    replicate3 = getVM(2);
    normal = getVM(3);

    for (VM vm : toArray(empty, normal, replicate1, replicate2, replicate3)) {
      // zero is no-op unless test replaces the reference
      vm.invoke(() -> LATCH.set(new CountDownLatch(0)));
    }

    regionName = getClass().getSimpleName() + "_region";
    poolName = getClass().getSimpleName() + "_pool";

    for (int i = 100; i < 120; i++) {
      filter.add("execKey-" + i);
    }
  }

  @After
  public void tearDown() {
    for (VM vm : toArray(empty, normal, replicate1, replicate2, replicate3)) {
      vm.invoke(() -> LATCH.get().countDown());
    }
  }

  @Test
  public void testDistributedRegionFunctionExecutionOnDataPolicyEmpty() {
    for (VM vm : toArray(empty, normal, replicate1, replicate2, replicate3)) {
      vm.invoke(() -> createCache(getDistributedSystemProperties()));
    }

    empty.invoke(() -> createRegion(DataPolicy.EMPTY));
    normal.invoke(() -> createRegion(DataPolicy.NORMAL));
    for (VM vm : toArray(replicate1, replicate2, replicate3)) {
      vm.invoke(() -> createRegion(DataPolicy.REPLICATE));
    }

    empty.invoke(() -> {
      populateRegion(200);

      executeDistributedRegionFunction();
    });
  }

  @Test
  public void testDistributedRegionFunctionExecutionOnDataPolicyEmpty_SendException() {
    for (VM vm : toArray(empty, normal, replicate1, replicate2, replicate3)) {
      vm.invoke(() -> createCache(getDistributedSystemProperties()));
    }

    empty.invoke(() -> createRegion(DataPolicy.EMPTY));
    normal.invoke(() -> createRegion(DataPolicy.NORMAL));
    for (VM vm : toArray(replicate1, replicate2, replicate3)) {
      vm.invoke(() -> createRegion(DataPolicy.REPLICATE));
    }

    empty.invoke(() -> {
      populateRegion(200);

      executeResultWithExceptionFunction();
    });
  }

  @Test
  public void testDistributedRegionFunctionExecutionOnDataPolicyEmpty_NoLastResult() {
    for (VM vm : toArray(empty, normal, replicate1, replicate2, replicate3)) {
      vm.invoke(() -> createCache(getDistributedSystemProperties()));
    }

    empty.invoke(() -> createRegion(DataPolicy.EMPTY));
    normal.invoke(() -> createRegion(DataPolicy.NORMAL));
    for (VM vm : toArray(replicate1, replicate2, replicate3)) {
      vm.invoke(() -> createRegion(DataPolicy.REPLICATE));
    }

    empty.invoke(() -> {
      populateRegion(200);

      executeNoLastResultFunction();
    });
  }

  @Test
  public void testDistributedRegionFunctionExecutionOnDataPolicyNormal() {
    for (VM vm : toArray(empty, normal, replicate1, replicate2, replicate3)) {
      vm.invoke(() -> createCache(getDistributedSystemProperties()));
    }

    empty.invoke(() -> createRegion(DataPolicy.EMPTY));
    normal.invoke(() -> createRegion(DataPolicy.NORMAL));
    for (VM vm : toArray(replicate1, replicate2, replicate3)) {
      vm.invoke(() -> createRegion(DataPolicy.REPLICATE));
    }

    empty.invoke(() -> populateRegion(200));

    normal.invoke(() -> {
      Throwable thrown = catchThrowable(this::executeDistributedRegionFunction);

      assertThat(thrown)
          .isInstanceOf(FunctionException.class)
          .hasMessage("Function execution on region with DataPolicy.NORMAL is not supported");
    });
  }

  @Test
  public void testDistributedRegionFunctionExecutionOnDataPolicyReplicate() {
    for (VM vm : toArray(empty, normal, replicate1, replicate2, replicate3)) {
      vm.invoke(() -> createCache(getDistributedSystemProperties()));
    }

    empty.invoke(() -> createRegion(DataPolicy.EMPTY));
    normal.invoke(() -> createRegion(DataPolicy.NORMAL));
    for (VM vm : toArray(replicate1, replicate2, replicate3)) {
      vm.invoke(() -> createRegion(DataPolicy.REPLICATE));
    }

    empty.invoke(() -> populateRegion(200));

    replicate1.invoke(this::executeDistributedRegionFunction);
  }

  @Test
  public void testDistributedRegionFunctionExecutionOnDataPolicyReplicateNotTimedOut()
      throws Exception {
    for (VM vm : toArray(empty, normal, replicate1, replicate2, replicate3)) {
      vm.invoke(() -> {
        createCache(getDistributedSystemProperties());

        LATCH.set(new CountDownLatch(1));

        FunctionService.registerFunction(new DistributedRegionFunction());
        FunctionService.registerFunction(new LongRunningFunction(LATCH.get()));
      });
    }

    empty.invoke(() -> createRegion(DataPolicy.EMPTY));
    normal.invoke(() -> createRegion(DataPolicy.NORMAL));
    for (VM vm : toArray(replicate1, replicate2, replicate3)) {
      vm.invoke(() -> createRegion(DataPolicy.REPLICATE));
    }

    empty.invoke(() -> populateRegion(200));

    AsyncInvocation<Void> executeFunctionInReplicate1 = replicate1.invokeAsync(() -> {
      ResultCollector<String, List<String>> resultCollector = TypedFunctionService
          .<Void, String, List<String>>onRegion(getRegion())
          .withFilter(filter)
          .execute(LongRunningFunction.class.getSimpleName(), getTimeout().toMillis(),
              MILLISECONDS);

      assertThat(resultCollector.getResult().get(0))
          .isEqualTo("LongRunningFunction completed");
    });

    // how long LongRunningFunction runs for is now controlled here:
    Thread.sleep(2000);

    for (VM vm : toArray(empty, normal, replicate1, replicate2, replicate3)) {
      vm.invoke(() -> {
        LATCH.get().countDown();
      });
    }

    executeFunctionInReplicate1.await();
  }

  @Test
  public void testDistributedRegionFunctionExecutionOnDataPolicyReplicateTimedOut() {
    for (VM vm : toArray(empty, normal, replicate1, replicate2, replicate3)) {
      vm.invoke(() -> {
        createCache(getDistributedSystemProperties());

        LATCH.set(new CountDownLatch(1));

        FunctionService.registerFunction(new DistributedRegionFunction());
        FunctionService.registerFunction(new LongRunningFunction(LATCH.get()));
      });
    }

    empty.invoke(() -> createRegion(DataPolicy.EMPTY));
    normal.invoke(() -> createRegion(DataPolicy.NORMAL));
    for (VM vm : toArray(replicate1, replicate2, replicate3)) {
      vm.invoke(() -> createRegion(DataPolicy.REPLICATE));
    }

    empty.invoke(() -> populateRegion(200));

    replicate1.invoke(() -> {
      Throwable thrown = catchThrowable(() -> {
        TypedFunctionService
            .<Void, String, List<String>>onRegion(getRegion())
            .withFilter(filter)
            .execute(LongRunningFunction.class.getSimpleName(), 1000, MILLISECONDS);
      });

      assertThat(thrown)
          .hasCauseInstanceOf(FunctionException.class)
          .hasMessageContaining("All results not received in time provided");
    });
  }

  @Test
  public void testDistributedRegionFunctionExecutionOnDataPolicyReplicate_SendException() {
    for (VM vm : toArray(empty, normal, replicate1, replicate2, replicate3)) {
      vm.invoke(() -> createCache(getDistributedSystemProperties()));
    }

    empty.invoke(() -> createRegion(DataPolicy.EMPTY));
    normal.invoke(() -> createRegion(DataPolicy.NORMAL));
    for (VM vm : toArray(replicate1, replicate2, replicate3)) {
      vm.invoke(() -> createRegion(DataPolicy.REPLICATE));
    }

    empty.invoke(() -> populateRegion(200));

    replicate1.invoke(this::executeResultWithExceptionFunction);
  }

  @Test
  public void testDistributedRegionFunctionExecutionOnDataPolicyReplicate_NoLastResult() {
    for (VM vm : toArray(empty, normal, replicate1, replicate2, replicate3)) {
      vm.invoke(() -> createCache(getDistributedSystemProperties()));
    }

    empty.invoke(() -> createRegion(DataPolicy.EMPTY));
    normal.invoke(() -> createRegion(DataPolicy.NORMAL));
    for (VM vm : toArray(replicate1, replicate2, replicate3)) {
      vm.invoke(() -> createRegion(DataPolicy.REPLICATE));
    }

    empty.invoke(() -> populateRegion(200));

    replicate1.invoke(this::executeNoLastResultFunction);
  }

  @Test
  public void testDistributedRegionFunctionExecutionWithFunctionInvocationTargetException() {
    for (VM vm : toArray(empty, normal, replicate1, replicate2, replicate3)) {
      vm.invoke(() -> createCache(getDistributedSystemProperties()));
    }

    empty.invoke(() -> createRegion(DataPolicy.EMPTY));
    normal.invoke(() -> createRegion(DataPolicy.NORMAL));
    for (VM vm : toArray(replicate1, replicate2, replicate3)) {
      vm.invoke(() -> createRegion(DataPolicy.REPLICATE));
    }

    for (VM vm : toArray(empty, normal, replicate1, replicate2, replicate3)) {
      vm.invoke(() -> registerThrowsFunctionInvocationTargetExceptionFunction(true, 5));
    }

    empty.invoke(() -> populateRegion(200));

    replicate1.invoke(() -> {
      try (IgnoredException ie = addIgnoredException("I have been thrown")) {
        executeFunctionFunctionInvocationTargetException();
      }
    });
  }

  @Test
  public void testDistributedRegionFunctionExecutionWithFunctionInvocationTargetException_WithoutHA() {
    for (VM vm : toArray(empty, normal, replicate1, replicate2, replicate3)) {
      vm.invoke(() -> createCache(getDistributedSystemProperties()));
    }

    empty.invoke(() -> createRegion(DataPolicy.EMPTY));
    normal.invoke(() -> createRegion(DataPolicy.NORMAL));
    for (VM vm : toArray(replicate1, replicate2, replicate3)) {
      vm.invoke(() -> createRegion(DataPolicy.REPLICATE));
    }

    for (VM vm : toArray(empty, normal, replicate1, replicate2, replicate3)) {
      vm.invoke(() -> registerThrowsFunctionInvocationTargetExceptionFunction(false, 0));
    }

    empty.invoke(() -> populateRegion(200));

    replicate1.invoke(() -> {
      try (IgnoredException ie = addIgnoredException("I have been thrown")) {
        executeFunctionFunctionInvocationTargetExceptionWithoutHA();
      }
    });
  }

  @Test
  public void testDistributedRegionFunctionExecutionWithFunctionInvocationTargetExceptionForEmptyDataPolicy() {
    for (VM vm : toArray(empty, normal, replicate1, replicate2, replicate3)) {
      vm.invoke(() -> createCache(getDistributedSystemProperties()));
    }

    empty.invoke(() -> createRegion(DataPolicy.EMPTY));
    normal.invoke(() -> createRegion(DataPolicy.NORMAL));
    for (VM vm : toArray(replicate1, replicate2, replicate3)) {
      vm.invoke(() -> createRegion(DataPolicy.REPLICATE));
    }

    for (VM vm : toArray(empty, normal, replicate1, replicate2, replicate3)) {
      vm.invoke(() -> registerThrowsFunctionInvocationTargetExceptionFunction(true, 5));
    }

    empty.invoke(() -> populateRegion(200));

    empty.invoke(() -> {
      try (IgnoredException ie = addIgnoredException("I have been thrown")) {
        executeFunctionFunctionInvocationTargetException();
      }
    });
  }

  @Test
  public void testDistributedRegionFunctionExecutionWithFunctionInvocationTargetExceptionForEmptyDataPolicy_WithoutHA() {
    for (VM vm : toArray(empty, normal, replicate1, replicate2, replicate3)) {
      vm.invoke(() -> createCache(getDistributedSystemProperties()));
    }

    empty.invoke(() -> createRegion(DataPolicy.EMPTY));
    normal.invoke(() -> createRegion(DataPolicy.NORMAL));
    for (VM vm : toArray(replicate1, replicate2, replicate3)) {
      vm.invoke(() -> createRegion(DataPolicy.REPLICATE));
    }

    for (VM vm : toArray(empty, normal, replicate1, replicate2, replicate3)) {
      vm.invoke(() -> registerThrowsFunctionInvocationTargetExceptionFunction(false, 0));
    }

    empty.invoke(() -> populateRegion(200));

    empty.invoke(() -> {
      try (IgnoredException ie = addIgnoredException("I have been thrown")) {
        executeFunctionFunctionInvocationTargetExceptionWithoutHA();
      }
    });
  }

  @Test
  public void testDistributedRegionFunctionExecutionHACacheClosedException() {
    for (VM vm : toArray(empty, replicate1)) {
      vm.invoke(() -> createCache(getDistributedSystemProperties()));
    }

    empty.invoke(() -> createRegion(DataPolicy.EMPTY));
    replicate1.invoke(() -> createRegion(DataPolicy.REPLICATE));

    empty.invoke(() -> populateRegion(200));

    replicate2.invoke(() -> {
      createCache(getDistributedSystemProperties());
      createRegion(DataPolicy.REPLICATE);
    });

    empty.invoke(() -> {
      List<Boolean> result = executeDistributedRegionFunction();

      assertThat(result)
          .hasSize(5001)
          .containsOnly(true);
    });
  }

  @Test
  public void testDistributedRegionFunctionExecutionHANodeFailure() throws Exception {
    for (VM vm : toArray(empty, replicate1)) {
      vm.invoke(() -> createCache(getDistributedSystemProperties()));
    }

    empty.invoke(() -> createRegion(DataPolicy.EMPTY));
    replicate1.invoke(() -> createRegion(DataPolicy.REPLICATE));

    empty.invoke(() -> populateRegion(200));

    AsyncInvocation<List<Boolean>> executeFunctionHaInEmptyVM =
        empty.invokeAsync(this::executeDistributedRegionFunction);

    replicate2.invoke(() -> {
      createCache(getDistributedSystemProperties());
      createRegion(DataPolicy.REPLICATE);
    });

    replicate1.invoke(() -> getCache().close());

    List<Boolean> result = executeFunctionHaInEmptyVM.get();

    assertThat(result)
        .hasSize(5001)
        .containsOnly(true);
  }

  @Test
  public void testDistributedRegionFunctionExecutionOnDataPolicyEmpty_ClientServer() {
    VM empty1 = replicate3;
    VM empty2 = normal;
    VM client = empty;

    for (VM vm : toArray(empty2, replicate1, replicate2, empty1)) {
      vm.invoke(() -> createCache(getDistributedSystemProperties()));
    }

    client.invoke(() -> createClientCache());

    int port1 = empty1.invoke(() -> {
      createRegion(DataPolicy.EMPTY);
      return createCacheServer();
    });
    int port2 = empty2.invoke(() -> {
      createRegion(DataPolicy.EMPTY);
      return createCacheServer();
    });

    for (VM vm : toArray(replicate1, replicate2)) {
      vm.invoke(() -> createRegion(DataPolicy.REPLICATE));
    }

    client.invoke(() -> {
      createClientRegion(port1, port2);
      populateClientRegion(200);

      executeDistributedRegionFunction();
    });
  }

  @Test
  public void testDistributedRegionFunctionExecutionOnDataPolicyEmpty_ClientServer_SendException() {
    VM empty1 = replicate3;
    VM empty2 = normal;
    VM client = empty;

    for (VM vm : toArray(empty2, replicate1, replicate2, empty1)) {
      vm.invoke(() -> createCache(getDistributedSystemProperties()));
    }

    client.invoke(() -> createClientCache());

    int port1 = empty1.invoke(() -> {
      createRegion(DataPolicy.EMPTY);
      return createCacheServer();
    });
    int port2 = empty2.invoke(() -> {
      createRegion(DataPolicy.EMPTY);
      return createCacheServer();
    });

    for (VM vm : toArray(replicate1, replicate2)) {
      vm.invoke(() -> createRegion(DataPolicy.REPLICATE));
    }

    client.invoke(() -> {
      createClientRegion(port1, port2);
      populateClientRegion(200);

      executeResultWithExceptionFunction();
    });
  }

  @Test
  public void testDistributedRegionFunctionExecutionOnDataPolicyEmpty_ClientServer_NoLastResult() {
    VM empty1 = replicate3;
    VM empty2 = normal;
    VM client = empty;

    for (VM vm : toArray(empty2, replicate1, replicate2, empty1)) {
      vm.invoke(() -> createCache(getDistributedSystemProperties()));
    }

    client.invoke(() -> createClientCache());

    int port1 = empty1.invoke(() -> {
      createRegion(DataPolicy.EMPTY);
      return createCacheServer();
    });
    int port2 = empty2.invoke(() -> {
      createRegion(DataPolicy.EMPTY);
      return createCacheServer();
    });

    for (VM vm : toArray(replicate1, replicate2)) {
      vm.invoke(() -> createRegion(DataPolicy.REPLICATE));
    }

    client.invoke(() -> {
      createClientRegion(port1, port2);
      populateClientRegion(200);

      try (IgnoredException ie1 = addIgnoredException("did not send last result", empty1);
          IgnoredException ie2 = addIgnoredException("did not send last result", empty2)) {
        executeNoLastResultFunction();
      }
    });
  }

  /**
   * If one server goes down while executing a function, that function should failover to other
   * available server.
   */
  @Test
  public void testServerFailoverWithTwoServerAliveHA() throws InterruptedException {
    VM emptyServer1 = replicate1;
    VM client = normal;

    for (VM vm : toArray(emptyServer1, replicate2, replicate3)) {
      vm.invoke(() -> createCache(getDistributedSystemProperties()));
    }

    int port1 = emptyServer1.invoke(() -> {
      createRegion(DataPolicy.EMPTY);
      return createCacheServer();
    });
    int port2 = replicate2.invoke(() -> {
      createRegion(DataPolicy.REPLICATE);
      return createCacheServer();
    });

    replicate3.invoke(() -> createRegion(DataPolicy.REPLICATE));

    client.invoke(() -> {
      LATCH.set(new CountDownLatch(1));
      createClientCache();
      createClientRegion(port1, port2);
      populateClientRegion(200);
    });

    replicate2.invoke(this::stopServerHA);

    AsyncInvocation<List<Boolean>> executeFunctionHaInClientVm =
        client.invokeAsync(this::executeDistributedRegionFunction);

    replicate2.invoke(() -> {
      startServerHA();
      client.invoke(() -> LATCH.get().countDown());
    });

    client.invoke(() -> {
      LATCH.get().await(getTimeout().toMillis(), MILLISECONDS);
    });

    emptyServer1.invoke(this::closeCacheHA);

    List<Boolean> result = executeFunctionHaInClientVm.get();

    assertThat(result)
        .hasSize(5001)
        .containsOnly(true);
  }

  @Test
  public void testDistributedRegionFunctionExecutionOnDataPolicyNormal_ClientServer() {
    VM client = empty;
    VM normal1 = normal;
    VM normal2 = replicate3;
    VM empty = replicate2;

    for (VM vm : toArray(normal1, replicate1, empty, normal2)) {
      vm.invoke(() -> createCache(getDistributedSystemProperties()));
    }

    client.invoke(() -> createClientCache());

    int port1 = normal1.invoke(() -> {
      createRegion(DataPolicy.NORMAL);
      return createCacheServer();
    });
    int port2 = normal2.invoke(() -> {
      createRegion(DataPolicy.NORMAL);
      return createCacheServer();
    });

    for (VM vm : toArray(empty, replicate1)) {
      vm.invoke(() -> createRegion(DataPolicy.EMPTY));
    }

    client.invoke(() -> {
      createClientRegion(port1, port2);
      populateClientRegion(200);

      // add expected exception
      try (IgnoredException ie = addIgnoredException(FunctionException.class)) {
        Throwable thrown = catchThrowable(this::executeDistributedRegionFunction);

        assertThat(thrown)
            .isInstanceOf(FunctionException.class)
            .hasMessageContaining(
                "Function execution on region with DataPolicy.NORMAL is not supported");
      }
    });
  }

  @Test
  public void testDistributedRegionFunctionExecutionOnDataPolicyReplicate_ClientServer() {
    VM client = empty;
    VM empty = replicate3;

    for (VM vm : toArray(normal, replicate1, replicate2, empty)) {
      vm.invoke(() -> createCache(getDistributedSystemProperties()));
    }

    client.invoke(() -> createClientCache());

    int port1 = replicate1.invoke(() -> {
      createRegion(DataPolicy.REPLICATE);
      return createCacheServer();
    });
    int port2 = replicate2.invoke(() -> {
      createRegion(DataPolicy.REPLICATE);
      return createCacheServer();
    });

    normal.invoke(() -> createRegion(DataPolicy.NORMAL));
    empty.invoke(() -> createRegion(DataPolicy.EMPTY));

    client.invoke(() -> {
      createClientRegion(port1, port2);
      populateClientRegion(200);

      executeUnregisteredFunction();
    });
  }

  @Test
  public void testDistributedRegionFunctionExecutionOnDataPolicyReplicate_ClientServer_WithoutRegister() {
    VM client = empty;
    VM empty = replicate3;

    for (VM vm : toArray(normal, replicate1, replicate2, empty)) {
      vm.invoke(() -> createCache(getDistributedSystemProperties()));
    }

    client.invoke(() -> createClientCache());

    int port1 = replicate1.invoke(() -> {
      createRegion(DataPolicy.REPLICATE);
      return createCacheServer();
    });
    int port2 = replicate2.invoke(() -> {
      createRegion(DataPolicy.REPLICATE);
      return createCacheServer();
    });

    normal.invoke(() -> createRegion(DataPolicy.NORMAL));
    empty.invoke(() -> createRegion(DataPolicy.EMPTY));

    client.invoke(() -> {
      createClientRegion(port1, port2);
      populateClientRegion(200);

      executeDistributedRegionFunction();
    });
  }

  @Test
  public void testDistributedRegionFunctionExecutionOnDataPolicyReplicate_ClientServer_FunctionInvocationTargetException() {
    VM client = empty;
    VM empty = replicate3;

    for (VM vm : toArray(normal, replicate1, replicate2, empty)) {
      vm.invoke(() -> createCache(getDistributedSystemProperties()));
    }

    int port1 = replicate1.invoke(() -> {
      createRegion(DataPolicy.REPLICATE);
      return createCacheServer();
    });
    int port2 = replicate2.invoke(() -> {
      createRegion(DataPolicy.REPLICATE);
      return createCacheServer();
    });

    client.invoke(() -> createClientCache());
    normal.invoke(() -> createRegion(DataPolicy.NORMAL));
    empty.invoke(() -> createRegion(DataPolicy.EMPTY));

    for (VM vm : toArray(client, replicate1, replicate2, normal, empty)) {
      vm.invoke(() -> registerThrowsFunctionInvocationTargetExceptionFunction(true, 5));
    }

    client.invoke(() -> {
      createClientRegion(port1, port2);
      populateClientRegion(200);

      // add expected exception to avoid suspect strings
      try (IgnoredException ie = addIgnoredException("I have been thrown")) {
        executeFunctionFunctionInvocationTargetException_ClientServer();
      }
    });
  }

  @Test
  public void testDistributedRegionFunctionExecutionOnDataPolicyReplicate_ClientServer_FunctionInvocationTargetException_WithoutHA() {
    VM client = empty;
    VM empty = replicate3;

    for (VM vm : toArray(normal, replicate1, replicate2, empty)) {
      vm.invoke(() -> createCache(getDistributedSystemProperties()));
    }

    int port1 = replicate1.invoke(() -> {
      createRegion(DataPolicy.REPLICATE);
      return createCacheServer();
    });
    int port2 = replicate2.invoke(() -> {
      createRegion(DataPolicy.REPLICATE);
      return createCacheServer();
    });

    client.invoke(() -> createClientCache());
    normal.invoke(() -> createRegion(DataPolicy.NORMAL));
    empty.invoke(() -> createRegion(DataPolicy.EMPTY));

    for (VM vm : toArray(client, replicate1, replicate2, normal, empty)) {
      vm.invoke(() -> registerThrowsFunctionInvocationTargetExceptionFunction(false, 0));
    }

    client.invoke(() -> {
      createClientRegion(port1, port2);
      populateClientRegion(200);

      // add expected exception to avoid suspect strings
      try (IgnoredException ie = addIgnoredException("I have been thrown")) {
        executeFunctionFunctionInvocationTargetException_ClientServer_WithoutHA();
      }
    });
  }

  @Test
  public void testDistributedRegionFunctionExecutionOnDataPolicyEmpty_ClientServer_FunctionInvocationTargetException() {
    VM client = empty;
    VM empty1 = replicate3;
    VM empty2 = normal;

    for (VM vm : toArray(empty2, replicate1, replicate2, empty1)) {
      vm.invoke(() -> createCache(getDistributedSystemProperties()));
    }

    int port1 = empty1.invoke(() -> {
      createRegion(DataPolicy.EMPTY);
      return createCacheServer();
    });
    int port2 = empty2.invoke(() -> {
      createRegion(DataPolicy.EMPTY);
      return createCacheServer();
    });

    client.invoke(() -> createClientCache());
    for (VM vm : toArray(replicate1, replicate2)) {
      vm.invoke(() -> createRegion(DataPolicy.REPLICATE));
    }

    client.invoke(() -> {
      createClientRegion(port1, port2);
      populateClientRegion(200);
    });

    for (VM vm : toArray(client, empty1, empty2, replicate1, replicate2)) {
      vm.invoke(() -> registerThrowsFunctionInvocationTargetExceptionFunction(true, 5));
    }

    client.invoke(() -> {
      // add expected exception to avoid suspect strings
      try (IgnoredException ie = addIgnoredException("I have been thrown")) {
        executeFunctionFunctionInvocationTargetException_ClientServer();
      }
    });
  }

  @Test
  public void testDistributedRegionFunctionExecutionOnDataPolicyEmpty_ClientServer_FunctionInvocationTargetException_WithoutHA() {
    VM client = empty;
    VM empty1 = replicate3;
    VM empty2 = normal;

    for (VM vm : toArray(empty2, replicate1, replicate2, empty1)) {
      vm.invoke(() -> createCache(getDistributedSystemProperties()));
    }

    client.invoke(() -> createClientCache());

    int port1 = empty1.invoke(() -> {
      createRegion(DataPolicy.EMPTY);
      return createCacheServer();
    });
    int port2 = empty2.invoke(() -> {
      createRegion(DataPolicy.EMPTY);
      return createCacheServer();
    });

    for (VM vm : toArray(replicate1, replicate2)) {
      vm.invoke(() -> createRegion(DataPolicy.REPLICATE));
    }

    client.invoke(() -> {
      createClientRegion(port1, port2);
      populateClientRegion(200);
    });

    for (VM vm : toArray(client, empty1, empty2, replicate1, replicate2)) {
      vm.invoke(() -> registerThrowsFunctionInvocationTargetExceptionFunction(false, 0));
    }

    client.invoke(() -> {
      // add expected exception to avoid suspect strings
      try (IgnoredException ie = addIgnoredException("I have been thrown")) {
        executeFunctionFunctionInvocationTargetException_ClientServer_WithoutHA();
      }
    });
  }

  @Test
  public void inlineFunctionIsUsedOnClientInsteadOfLookingUpFunctionById() {
    VM client = empty;
    VM empty = replicate3;

    for (VM vm : toArray(normal, empty, replicate1, replicate2)) {
      vm.invoke(() -> createCache(getDistributedSystemProperties()));
    }
    client.invoke(() -> createClientCache(getClientProperties()));

    int port1 = replicate1.invoke(() -> {
      createRegion(DataPolicy.REPLICATE);
      return createCacheServer();
    });
    int port2 = replicate2.invoke(() -> {
      createRegion(DataPolicy.REPLICATE);
      return createCacheServer();
    });

    for (VM vm : toArray(normal, empty)) {
      vm.invoke(() -> createRegion(DataPolicy.REPLICATE));
    }
    client.invoke(() -> {
      createClientRegion(port1, port2);
      populateClientRegion(200);
    });

    for (VM vm : toArray(normal, empty, replicate1, replicate2, client)) {
      vm.invoke(() -> FunctionService.registerFunction(inlineFunction("Failure", false)));
    }

    client.invoke(() -> {
      ResultCollector<Boolean, List<Boolean>> resultCollector = TypedFunctionService
          .<Boolean, Boolean, List<Boolean>>onRegion(getRegion())
          .setArguments(true)
          .execute(inlineFunction("Success", true));

      assertThat(resultCollector.getResult())
          .hasSize(1)
          .containsOnly(true);
    });
  }

  /**
   * Verify that AuthenticationRequiredException is not thrown when security-* properties are NOT
   * provided. We have to grep for this exception in logs for any occurrence.
   */
  @Test
  public void authenticationRequiredExceptionIsNotThrownWhenSecurityIsNotConfigured() {
    VM client = replicate1;
    VM server = replicate2;

    int port = server.invoke(() -> {
      createCacheWithSecurity();
      createRegion(DataPolicy.REPLICATE);
      return createCacheServer();
    });

    client.invoke(() -> {
      createClientCacheWithSecurity();
      createClientRegion(port);

      executeNoResultFunction();
      doPut();
    });
  }

  @Test
