/*
 * 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 static org.apache.geode.distributed.ConfigurationProperties.ENABLE_NETWORK_PARTITION_DETECTION;
import static org.apache.geode.distributed.ConfigurationProperties.LOG_LEVEL;
import static org.apache.geode.test.dunit.Assert.assertEquals;
import static org.apache.geode.test.dunit.Assert.assertFalse;
import static org.apache.geode.test.dunit.Assert.assertNotNull;
import static org.apache.geode.test.dunit.Assert.assertNull;
import static org.apache.geode.test.dunit.Assert.assertTrue;
import static org.apache.geode.test.dunit.Assert.fail;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.HashSet;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;

import org.junit.Ignore;
import org.junit.Test;

import org.apache.geode.DataSerializable;
import org.apache.geode.Delta;
import org.apache.geode.InvalidDeltaException;
import org.apache.geode.cache.CacheListener;
import org.apache.geode.cache.EntryEvent;
import org.apache.geode.cache.Operation;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionEvent;
import org.apache.geode.cache.RegionShortcut;
import org.apache.geode.cache.client.ClientCache;
import org.apache.geode.cache.client.ClientCacheFactory;
import org.apache.geode.cache.client.ClientRegionFactory;
import org.apache.geode.cache.client.ClientRegionShortcut;
import org.apache.geode.cache.client.internal.DestroyOp;
import org.apache.geode.cache.server.CacheServer;
import org.apache.geode.cache.util.CacheListenerAdapter;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.internal.membership.MembershipManager;
import org.apache.geode.distributed.internal.membership.gms.MembershipManagerHelper;
import org.apache.geode.internal.AvailablePort;
import org.apache.geode.test.awaitility.GeodeAwaitility;
import org.apache.geode.test.dunit.Host;
import org.apache.geode.test.dunit.IgnoredException;
import org.apache.geode.test.dunit.LogWriterUtils;
import org.apache.geode.test.dunit.SerializableCallable;
import org.apache.geode.test.dunit.SerializableRunnable;
import org.apache.geode.test.dunit.VM;
import org.apache.geode.test.dunit.WaitCriterion;
import org.apache.geode.test.dunit.cache.internal.JUnit4CacheTestCase;

/**
 * tests for the concurrentMapOperations. there are more tests in ClientServerMiscDUnitTest
 */

public class ConcurrentMapOpsDUnitTest extends JUnit4CacheTestCase {

  private static final String REP_REG_NAME = "repRegion";
  private static final String PR_REG_NAME = "prRegion";
  private static final int MAX_ENTRIES = 113;

  enum OP {
    PUTIFABSENT, REPLACE, REMOVE
  }

  @Override
  public Properties getDistributedSystemProperties() {
    Properties result = super.getDistributedSystemProperties();
    result.put(ENABLE_NETWORK_PARTITION_DETECTION, "false");
    return result;
  }

  private void createRegions(VM vm) {
    vm.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        createReplicateRegion();
        createPartitionedRegion();
        return null;
      }
    });
  }

  private void createRedundantRegions(VM vm) {
    vm.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        getCache().createRegionFactory(RegionShortcut.REPLICATE).setConcurrencyChecksEnabled(true)
            .create(REP_REG_NAME);
        getCache().createRegionFactory(RegionShortcut.PARTITION_REDUNDANT)
            .setConcurrencyChecksEnabled(true).create(PR_REG_NAME);
        return null;
      }
    });
  }

  private Region createReplicateRegion() {
    return getCache().createRegionFactory(RegionShortcut.REPLICATE)
        .setConcurrencyChecksEnabled(true).create(REP_REG_NAME);
  }

  private Region createPartitionedRegion() {
    return getCache().createRegionFactory(RegionShortcut.PARTITION)
        .setConcurrencyChecksEnabled(true).create(PR_REG_NAME);
  }

  private Integer createRegionsAndStartServer(VM vm) {
    return createRegionsAndStartServer(vm, false);
  }

  private Integer createRegionsAndStartServer(VM vm, final boolean withRedundancy) {
    return (Integer) vm.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        getCache().createRegionFactory(RegionShortcut.REPLICATE).create(REP_REG_NAME);
        if (withRedundancy) {
          getCache().createRegionFactory(RegionShortcut.PARTITION_REDUNDANT).create(PR_REG_NAME);
        } else {
          getCache().createRegionFactory(RegionShortcut.PARTITION).create(PR_REG_NAME);
        }
        int port = AvailablePort.getRandomAvailablePort(AvailablePort.SOCKET);
        CacheServer s = getCache().addCacheServer();
        s.setPort(port);
        s.start();
        return port;
      }
    });
  }

  private void createEmptyRegion(VM vm) {
    vm.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        getCache().createRegionFactory(RegionShortcut.REPLICATE_PROXY)
            .setConcurrencyChecksEnabled(true).create(REP_REG_NAME);
        getCache().createRegionFactory(RegionShortcut.PARTITION_PROXY)
            .setConcurrencyChecksEnabled(true).create(PR_REG_NAME);
        return null;
      }
    });
  }

  private void createClientRegionWithRI(VM vm, final int port, final boolean isEmpty) {
    createClientRegion(vm, port, isEmpty, true, -1);
  }

  private void createClientRegion(VM vm, final int port1, final boolean isEmpty, int port2) {
    createClientRegion(vm, port1, isEmpty, false, port2);
  }

  private void createClientRegion(VM vm, final int port1, final boolean isEmpty, final boolean ri,
      final int port2) {
    vm.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        ClientCacheFactory ccf = new ClientCacheFactory();
        ccf.addPoolServer("localhost"/* getServerHostName(Host.getHost(0)) */, port1);
        if (port2 > 0) {
          ccf.addPoolServer("localhost", port2);
        }
        ccf.setPoolSubscriptionEnabled(true);
        ccf.set(LOG_LEVEL, LogWriterUtils.getDUnitLogLevel());
        ClientCache cCache = getClientCache(ccf);
        ClientRegionFactory<Integer, String> crf = cCache.createClientRegionFactory(
            isEmpty ? ClientRegionShortcut.PROXY : ClientRegionShortcut.CACHING_PROXY);
        Region<Integer, String> r = crf.create(REP_REG_NAME);
        Region<Integer, String> pr = crf.create(PR_REG_NAME);
        if (ri) {
          r.registerInterestRegex(".*");
          pr.registerInterestRegex(".*");
        }
        return null;
      }
    });
  }

  private abstract static class AbstractConcMapOpsListener
      implements CacheListener<Integer, String> {
    @Override
    public void afterCreate(EntryEvent<Integer, String> event) {
      validate(event);
    }

    @Override
    public void afterDestroy(EntryEvent<Integer, String> event) {
      validate(event);
    }

    @Override
    public void afterInvalidate(EntryEvent<Integer, String> event) {
      validate(event);
    }

    @Override
    public void afterRegionClear(RegionEvent<Integer, String> event) {}

    @Override
    public void afterRegionCreate(RegionEvent<Integer, String> event) {}

    @Override
    public void afterRegionDestroy(RegionEvent<Integer, String> event) {}

    @Override
    public void afterRegionInvalidate(RegionEvent<Integer, String> event) {}

    @Override
    public void afterRegionLive(RegionEvent<Integer, String> event) {}

    @Override
    public void afterUpdate(EntryEvent<Integer, String> event) {
      validate(event);
    }

    @Override
    public void close() {}

    abstract void validate(EntryEvent event);
  }

  private static class NotInvokedListener extends AbstractConcMapOpsListener {
    @Override
    void validate(EntryEvent event) {
      fail("should not be called.  Event=" + event);
    }
  }

  private static class InitialCreatesListener extends AbstractConcMapOpsListener {
    AtomicInteger numCreates = new AtomicInteger();

    @Override
    void validate(EntryEvent event) {
      if (!event.getOperation().isCreate()) {
        fail("expected only create events");
      }
      numCreates.incrementAndGet();
    }
  }

  // test for bug #42164
  @Test
  public void testListenerNotInvokedForRejectedOperation() {
    Host host = Host.getHost(0);
    VM vm1 = host.getVM(0);
    VM vm2 = host.getVM(1);
    VM client1 = host.getVM(2);
    VM client2 = host.getVM(3);
    int port1 = createRegionsAndStartServer(vm1);
    int port2 = createRegionsAndStartServer(vm2);
    createClientRegionWithRI(client1, port1, true);
    createClientRegionWithRI(client2, port2, true);

    SerializableCallable addListenerToClientForInitialCreates = new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        Region r = getCache().getRegion(REP_REG_NAME);
        r.getAttributesMutator().addCacheListener(new InitialCreatesListener());
        Region pr = getCache().getRegion(PR_REG_NAME);
        pr.getAttributesMutator().addCacheListener(new InitialCreatesListener());
        return null;
      }
    };
    client1.invoke(addListenerToClientForInitialCreates);
    client2.invoke(addListenerToClientForInitialCreates);

    vm1.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        Region<Integer, String> r = getGemfireCache().getRegion(REP_REG_NAME);
        Region<Integer, String> pr = getGemfireCache().getRegion(PR_REG_NAME);
        for (int i = 0; i < MAX_ENTRIES; i++) {
          r.put(i, "value" + i);
          pr.put(i, "value" + i);
        }
        return null;
      }
    });

    SerializableCallable waitForInitialCreates = new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        Region<Integer, String> r = getGemfireCache().getRegion(REP_REG_NAME);
        Region<Integer, String> pr = getGemfireCache().getRegion(PR_REG_NAME);
        waitForCreates(r);
        waitForCreates(pr);
        return null;
      }

      private void waitForCreates(Region region) {
        CacheListener[] listeners = region.getAttributes().getCacheListeners();
        boolean listenerFound = false;
        for (CacheListener listener : listeners) {
          if (listener instanceof InitialCreatesListener) {
            listenerFound = true;
            final InitialCreatesListener initialCreatesListener = (InitialCreatesListener) listener;
            WaitCriterion wc = new WaitCriterion() {
              @Override
              public boolean done() {
                return initialCreatesListener.numCreates.get() == MAX_ENTRIES;
              }

              @Override
              public String description() {
                return "Client expected to get " + MAX_ENTRIES + " creates, but got "
                    + initialCreatesListener.numCreates.get();
              }
            };
            GeodeAwaitility.await().untilAsserted(wc);
          }
        }
        if (!listenerFound) {
          fail("Client listener should have been found");
        }
      }
    };
    client1.invoke(waitForInitialCreates);
    client2.invoke(waitForInitialCreates);

    SerializableCallable addListener = new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        Region r = getCache().getRegion(REP_REG_NAME);
        r.getAttributesMutator().addCacheListener(new NotInvokedListener());
        Region pr = getCache().getRegion(PR_REG_NAME);
        pr.getAttributesMutator().addCacheListener(new NotInvokedListener());
        return null;
      }
    };
    vm1.invoke(addListener);
    vm2.invoke(addListener);
    SerializableCallable addListenerToClient = new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        Region r = getCache().getRegion(REP_REG_NAME);
        r.getAttributesMutator().addCacheListener(new NotInvokedListener());
        Region pr = getCache().getRegion(PR_REG_NAME);
        pr.getAttributesMutator().addCacheListener(new NotInvokedListener());
        return null;
      }
    };
    client1.invoke(addListenerToClient);
    client2.invoke(addListenerToClient);
    vm1.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        Region r = getCache().getRegion(REP_REG_NAME);
        Region pr = getCache().getRegion(PR_REG_NAME);
        for (int i = 0; i < MAX_ENTRIES; i++) {
          assertEquals("value" + i, r.putIfAbsent(i, "piavalue"));
          assertEquals("value" + i, pr.putIfAbsent(i, "piavalue"));
        }
        for (int i = 0; i < MAX_ENTRIES; i++) {
          assertFalse(r.replace(i, "value", "replace1Value"));
          assertFalse(pr.replace(i, "value", "replace1Value"));
        }
        for (int i = MAX_ENTRIES + 1; i < MAX_ENTRIES * 2; i++) {
          assertNull(r.replace(i, "replace2value" + i));
          assertNull(pr.replace(i, "replace2value" + i));
        }
        for (int i = MAX_ENTRIES + 1; i < MAX_ENTRIES * 2; i++) {
          assertFalse(r.remove(i, "removeValue" + i));
          assertFalse(pr.remove(i, "removeValue" + i));
        }
        return null;
      }
    });
  }

  @Test
  public void testBug42162() {
    dotestConcOps(false);
  }

  @Test
  public void testBug42162EmptyClient() {
    dotestConcOps(true);
  }

  private void dotestConcOps(final boolean emptyClient) {
    Host host = Host.getHost(0);
    VM server = host.getVM(0);
    VM client = host.getVM(2);
    int port1 = createRegionsAndStartServer(server);

    createClientRegion(client, port1, emptyClient, -1);
    client.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        final Region r = getCache().getRegion(REP_REG_NAME);
        final Region pr = getCache().getRegion(PR_REG_NAME);
        r.registerInterestRegex(".*");
        pr.registerInterestRegex(".*");
        return null;
      }
    });
    server.invoke(new SerializableCallable() {
      @Override
      public Object call() {
        final Region r = getCache().getRegion(REP_REG_NAME);
        final Region pr = getCache().getRegion(PR_REG_NAME);
        r.put("key0", "value");
        pr.put("key0", "value");
        assertNull(r.putIfAbsent("keyForNull", null));
        assertNull(pr.putIfAbsent("keyForNull", null));
        assertEquals("value", r.putIfAbsent("key0", null));
        assertEquals("value", pr.putIfAbsent("key0", null));
        assertTrue(r.containsKey("keyForNull"));
        assertTrue(pr.containsKey("keyForNull"));
        assertFalse(r.containsValueForKey("keyForNull"));
        assertFalse(pr.containsValueForKey("keyForNull"));
        r.put("key0", "value");
        pr.put("key0", "value");
        return null;
      }
    });
    client.invoke(new SerializableCallable() {
      @Override
      public Object call() {
        final Region r = getCache().getRegion(REP_REG_NAME);
        final Region pr = getCache().getRegion(PR_REG_NAME);
        WaitCriterion wc = new WaitCriterion() {
          AssertionError e = null;

          @Override
          public boolean done() {
            try {
              if (!emptyClient) {
                assertTrue(r.containsKey("key0"));
                assertTrue(pr.containsKey("key0"));
                assertTrue(r.containsKey("keyForNull"));
                assertTrue(pr.containsKey("keyForNull"));
                assertFalse(r.containsValueForKey("keyForNull"));
                assertFalse(pr.containsValueForKey("keyForNull"));
              }
              assertEquals("value", r.putIfAbsent("key0", null));
              assertEquals("value", pr.putIfAbsent("key0", null));
              assertNull(r.putIfAbsent("keyForNull", null));
              assertNull(pr.putIfAbsent("keyForNull", null));
              assertNull(r.putIfAbsent("clientNullKey", null));
              assertNull(pr.putIfAbsent("clientNullKey", null));
            } catch (AssertionError ex) {
              r.getCache().getLogger().fine("SWAP:caught ", ex);
              e = ex;
              return false;
            }
            return true;
          }

          @Override
          public String description() {
            return "timeout " + e;
          }
        };
        GeodeAwaitility.await().untilAsserted(wc);
        return null;
      }
    });
    server.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        final Region r = getCache().getRegion(REP_REG_NAME);
        final Region pr = getCache().getRegion(PR_REG_NAME);
        assertTrue(r.containsKey("clientNullKey"));
        assertTrue(pr.containsKey("clientNullKey"));
        assertFalse(r.containsValueForKey("clientNullKey"));
        assertFalse(pr.containsValueForKey("clientNullKey"));
        assertNotNull(r.replace("key0", "value2"));
        assertNotNull(pr.replace("key0", "value2"));
        assertTrue(r.replace("keyForNull", null, "newValue"));
        assertTrue(pr.replace("keyForNull", null, "newValue"));
        return null;
      }
    });
    client.invoke(new SerializableCallable() {
      @Override
      public Object call() {
        final Region r = getCache().getRegion(REP_REG_NAME);
        final Region pr = getCache().getRegion(PR_REG_NAME);
        WaitCriterion wc = new WaitCriterion() {
          AssertionError e = null;

          @Override
          public boolean done() {
            try {
              assertEquals("value2", r.putIfAbsent("key0", null));
              assertEquals("value2", pr.putIfAbsent("key0", null));
              assertEquals("newValue", r.putIfAbsent("keyForNull", null));
              assertEquals("newValue", pr.putIfAbsent("keyForNull", null));
              // replace from client
              assertEquals("value2", r.replace("key0", "value"));
              assertEquals("value2", pr.replace("key0", "value"));
              assertNull(r.replace("NoKeyOnServer", "value"));
              assertNull(r.replace("NoKeyOnServer", "value"));
              assertTrue(r.replace("clientNullKey", null, "newValue"));
              assertTrue(pr.replace("clientNullKey", null, "newValue"));
            } catch (AssertionError ex) {
              e = ex;
              return false;
            }
            return true;
          }

          @Override
          public String description() {
            return "timeout " + e.getMessage();
          }
        };
        GeodeAwaitility.await().untilAsserted(wc);
        return null;
      }
    });
    server.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        final Region r = getCache().getRegion(REP_REG_NAME);
        final Region pr = getCache().getRegion(PR_REG_NAME);
        assertEquals("newValue", r.get("clientNullKey"));
        assertEquals("newValue", pr.get("clientNullKey"));
        return null;
      }
    });
  }

  @Test
  public void testNullValueFromNonEmptyClients() {
    Host host = Host.getHost(0);
    VM server = host.getVM(0);
    VM client = host.getVM(2);
    int port1 = createRegionsAndStartServer(server);

    createClientRegion(client, port1, true, -1);
    server.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        final Region r = getCache().getRegion(REP_REG_NAME);
        final Region pr = getCache().getRegion(PR_REG_NAME);
        r.create("createKey", null);
        pr.create("createKey", null);
        assertNull(r.putIfAbsent("putAbsentKey", null));
        assertNull(pr.putIfAbsent("putAbsentKey", null));
        return null;
      }
    });
    client.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        final Region r = getCache().getRegion(REP_REG_NAME);
        final Region pr = getCache().getRegion(PR_REG_NAME);
        assertEquals(r.get("createKey"), r.get("putAbsentKey"));
        assertEquals(pr.get("createKey"), pr.get("putAbsentKey"));
        assertFalse(r.containsKey("createKey"));
        assertFalse(pr.containsKey("createKey"));
        assertEquals(r.containsKey("createKey"), r.containsKey("putAbsentKey"));
        assertEquals(pr.containsKey("createKey"), pr.containsKey("putAbsentKey"));
        return null;
      }
    });
  }

  @Test
  public void testPutIfAbsent() {
    doPutIfAbsentWork(false);
  }

  @Test
  public void testPutIfAbsentCS() {
    doPutIfAbsentWork(true);
  }

  private void doPutIfAbsentWork(final boolean cs) {
    Host host = Host.getHost(0);
    VM vm1 = host.getVM(0);
    VM vm2 = host.getVM(2);
    if (cs) {
      int port1 = createRegionsAndStartServer(vm1);
      createClientRegion(vm2, port1, false, -1);
    } else {
      createRegions(vm1);
      createRegions(vm2);
    }
    vm1.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        final Region r = getCache().getRegion(REP_REG_NAME);
        final Region pr = getCache().getRegion(PR_REG_NAME);
        assertNull(r.putIfAbsent("key0", "value"));
        assertNull(pr.putIfAbsent("key0", "value"));
        assertNull(r.putIfAbsent("keyForClient", "value"));
        assertNull(pr.putIfAbsent("keyForClient", "value"));
        assertEquals("value", r.putIfAbsent("key0", "value2"));
        assertEquals("value", pr.putIfAbsent("key0", "value2"));
        return null;
      }
    });
    vm2.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        final Region r = getCache().getRegion(REP_REG_NAME);
        final Region pr = getCache().getRegion(PR_REG_NAME);
        assertEquals("value", r.putIfAbsent("key0", "value2"));
        assertEquals("value", pr.putIfAbsent("key0", "value2"));
        if (cs) {
          r.get("key0");
          pr.get("key0");
        }
        assertTrue(r.containsKey("key0"));
        assertTrue(pr.containsKey("key0"));
        assertTrue(r.containsValueForKey("key0"));
        assertTrue(pr.containsValueForKey("key0"));
        return null;
      }
    });
  }

  @Test
  public void testRemove() {
    doRemoveWork(false);
  }

  @Test
  public void testRemoveCS() {
    doRemoveWork(true);
  }

  private void doRemoveWork(final boolean cs) {
    Host host = Host.getHost(0);
    VM vm1 = host.getVM(0);
    VM vm2 = host.getVM(2);
    if (cs) {
      int port1 = createRegionsAndStartServer(vm1);
      createClientRegion(vm2, port1, true, -1);
    } else {
      createRegions(vm1);
      createRegions(vm2);
    }
    vm1.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        final Region r = getCache().getRegion(REP_REG_NAME);
        final Region pr = getCache().getRegion(PR_REG_NAME);
        assertNull(r.putIfAbsent("key0", "value"));
        assertNull(pr.putIfAbsent("key0", "value"));
        assertNull(r.putIfAbsent("keyForClient", "value"));
        assertNull(pr.putIfAbsent("keyForClient", "value"));
        assertFalse(r.remove("nonExistentkey", "value"));
        assertFalse(pr.remove("nonExistentkey", "value"));
        assertFalse(r.remove("key0", "newValue"));
        assertFalse(pr.remove("key0", "newValue"));
        assertTrue(r.remove("key0", "value"));
        assertTrue(pr.remove("key0", "value"));
        assertFalse(r.containsKey("key0"));
        assertFalse(pr.containsKey("key0"));
        return null;
      }
    });
    vm2.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        final Region r = getCache().getRegion(REP_REG_NAME);
        final Region pr = getCache().getRegion(PR_REG_NAME);
        assertFalse(r.remove("nonExistentkey", "value"));
        assertFalse(pr.remove("nonExistentkey", "value"));
        assertFalse(r.remove("keyForClient", "newValue"));
        assertFalse(pr.remove("keyForClient", "newValue"));
        assertTrue(r.remove("keyForClient", "value"));
        assertTrue(pr.remove("keyForClient", "value"));
        return null;
      }
    });
    vm1.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        final Region r = getCache().getRegion(REP_REG_NAME);
        final Region pr = getCache().getRegion(PR_REG_NAME);
        assertFalse(r.containsKey("keyForClient"));
        assertFalse(pr.containsKey("keyForClient"));
        return null;
      }
    });
  }

  @Test
  public void testReplaceCS() {
    doReplaceWork(true);
  }

  @Test
  public void testReplace() {
    doReplaceWork(false);
  }

  private void doReplaceWork(final boolean cs) {
    Host host = Host.getHost(0);
    VM vm1 = host.getVM(0);
    VM vm2 = host.getVM(2);
    if (cs) {
      int port1 = createRegionsAndStartServer(vm1);
      createClientRegion(vm2, port1, true, -1);
    } else {
      createRegions(vm1);
      createRegions(vm2);
    }
    vm1.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        final Region r = getCache().getRegion(REP_REG_NAME);
        final Region pr = getCache().getRegion(PR_REG_NAME);
        assertNull(r.putIfAbsent("key0", "value"));
        assertNull(pr.putIfAbsent("key0", "value"));
        assertNull(r.putIfAbsent("keyForClient", "value"));
        assertNull(pr.putIfAbsent("keyForClient", "value"));
        assertNull(r.replace("nonExistentkey", "value"));
        assertNull(pr.replace("nonExistentkey", "value"));
        assertEquals("value", r.replace("key0", "value2"));
        assertEquals("value", pr.replace("key0", "value2"));
        assertFalse(r.replace("key0", "value", "newValue"));
        assertFalse(pr.replace("key0", "value", "newValue"));
        assertTrue(r.replace("key0", "value2", "newValue"));
        assertTrue(pr.replace("key0", "value2", "newValue"));
        return null;
      }
    });
    vm2.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        final Region r = getCache().getRegion(REP_REG_NAME);
        final Region pr = getCache().getRegion(PR_REG_NAME);
        assertNull(r.replace("nonExistentkey", "value"));
        assertNull(pr.replace("nonExistentkey", "value"));
        assertEquals("value", r.replace("keyForClient", "value2"));
        assertEquals("value", pr.replace("keyForClient", "value2"));
        assertFalse(r.replace("keyForClient", "value", "newValue"));
        assertFalse(pr.replace("keyForClient", "value", "newValue"));
        assertTrue(r.replace("keyForClient", "value2", "newValue"));
        assertTrue(pr.replace("keyForClient", "value2", "newValue"));
        return null;
      }
    });
    vm1.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        final Region r = getCache().getRegion(REP_REG_NAME);
        final Region pr = getCache().getRegion(PR_REG_NAME);
        assertFalse(r.containsKey("nonExistentkey"));
        assertFalse(pr.containsKey("nonExistentkey"));
        return null;
      }
    });
  }

  @Test
  public void testBug42167() {
    do42167Work(false, REP_REG_NAME);
  }

  @Test
  public void testBug42167PR() {
    do42167Work(false, PR_REG_NAME);
  }

  @Test
  public void testBug42167Empty() {
    do42167Work(true, REP_REG_NAME);
  }

  @Test
  public void testBug42167EmptyPR() {
    do42167Work(true, PR_REG_NAME);
  }

  private void do42167Work(final boolean emptyClient, final String regionName) {
    Host host = Host.getHost(0);
    VM server = host.getVM(0);
    VM client = host.getVM(2);
    int port1 = createRegionsAndStartServer(server);

    createClientRegion(client, port1, emptyClient, -1);
    server.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        Region r = getCache().getRegion(regionName);
        r.put("key0", "value");
        r.put("key2", "value2");
        return null;
      }
    });
    client.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        Region r = getCache().getRegion(regionName);
        assertEquals("value", r.get("key0"));
        if (!emptyClient) {
          r.localDestroy("key0");
          assertFalse(r.containsKey("key0"));
        }
        getCache().getLogger().fine("SWAP:doingRemove");
        assertTrue(r.remove("key0", "value"));

        DestroyOp.TEST_HOOK_ENTRY_NOT_FOUND = false;
        assertTrue(r.remove("key0") == null);
        assertTrue(DestroyOp.TEST_HOOK_ENTRY_NOT_FOUND);

        DestroyOp.TEST_HOOK_ENTRY_NOT_FOUND = false;
        assertFalse(r.remove("key0", "value"));
        assertTrue(DestroyOp.TEST_HOOK_ENTRY_NOT_FOUND);

        DestroyOp.TEST_HOOK_ENTRY_NOT_FOUND = false;
        assertTrue(r.destroy("key0") == null);
        assertTrue(DestroyOp.TEST_HOOK_ENTRY_NOT_FOUND);

        DestroyOp.TEST_HOOK_ENTRY_NOT_FOUND = false;
        assertTrue(r.remove("nonExistentKey1") == null);
        assertTrue(DestroyOp.TEST_HOOK_ENTRY_NOT_FOUND);

        DestroyOp.TEST_HOOK_ENTRY_NOT_FOUND = false;
        assertFalse(r.remove("nonExistentKey2", "value"));
        assertTrue(DestroyOp.TEST_HOOK_ENTRY_NOT_FOUND);

        DestroyOp.TEST_HOOK_ENTRY_NOT_FOUND = false;
        assertTrue(r.destroy("nonExistentKey3") == null);
        assertTrue(DestroyOp.TEST_HOOK_ENTRY_NOT_FOUND);

        getCache().getLogger().fine("SWAP:doingReplace");
        assertEquals("value2", r.replace("key2", "newValue2"));
        getCache().getLogger().fine("SWAP:doingReplace2");
        assertEquals(null, r.replace("key0", "newValue"));
        assertNull(r.putIfAbsent("key4", "value4"));
        return null;
      }
    });
    server.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        Region r = getCache().getRegion(regionName);
        assertFalse(r.containsKey("key0"));
        assertFalse(r.containsValueForKey("key0"));
        assertTrue(r.containsKey("key2"));
        assertEquals("newValue2", r.get("key2"));
        r.getCache().getLogger().fine("SWAP:doingGet");
        assertEquals("value4", r.get("key4"));
        return null;
      }
    });
  }

  @Test
  public void testBug42189() {
    doBug42189Work(REP_REG_NAME);
  }

  @Test
  public void testBug42189PR() {
    doBug42189Work(PR_REG_NAME);
  }

  private void doBug42189Work(final String regionName) {
    Host host = Host.getHost(0);
    VM server = host.getVM(0);
    VM client = host.getVM(2);
    int port1 = createRegionsAndStartServer(server);

    createClientRegion(client, port1, false, -1);
    server.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        final Region r = getCache().getRegion(regionName);
        r.create("key0", null);
        assertNull(r.putIfAbsent("key0", "value"));
        assertTrue(r.containsKey("key0"));
        Object v = r.get("key0");
        assertNull("expected null but was " + v, v);
        return null;
      }
    });
    client.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        final Region r = getCache().getRegion(regionName);
        assertNull(r.putIfAbsent("key0", "value"));
        assertTrue(r.containsKeyOnServer("key0"));
        Object v = r.get("key0");
        assertNull("expected null but was " + v, v);
        return null;
      }
    });
    server.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        final Region r = getCache().getRegion(regionName);
        assertTrue(r.containsKey("key0"));
        assertFalse(r.containsValueForKey("key0"));
        return null;
      }
    });
  }

  /**
   * Replicate Region test for bug #42195: putIfAbsent from client does not put old value in local
   * cache
   */
  @Ignore("TODO")
  @Test
  public void testBug42195() {
    doPutIfAbsentPutsKeyInLocalClientCacheWork(REP_REG_NAME);
  }

  /**
   * Partitioned Region test for bug #42195: putIfAbsent from client does not put old value in local
   * cache
   */
  @Ignore("TODO")
  @Test
  public void testBug42195PR() {
    doPutIfAbsentPutsKeyInLocalClientCacheWork(PR_REG_NAME);
  }

  private void doPutIfAbsentPutsKeyInLocalClientCacheWork(final String regionName) {
    Host host = Host.getHost(0);
    VM server = host.getVM(0);
    VM client = host.getVM(2);
    int port1 = createRegionsAndStartServer(server);

    createClientRegion(client, port1, false, -1);
    server.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        final Region r = getCache().getRegion(regionName);
        assertNull(r.putIfAbsent("key0", "value"));
        assertTrue(r.containsKey("key0"));
        return null;
      }
    });
    client.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        final Region r = getCache().getRegion(regionName);
        assertEquals("value", r.putIfAbsent("key0", "newValue"));
        assertTrue(r.containsKeyOnServer("key0"));
        assertTrue(r.containsKey("key0"));
        assertTrue(r.containsValueForKey("key0"));
        return null;
      }
    });
  }

  @Test
  public void testReplacePutsKeyInLocalClientCache() {
    doReplacePutsKeyInLocalClientCacheWork(REP_REG_NAME);
  }

  @Test
  public void testReplacePutsKeyInLocalClientCachePR() {
    doReplacePutsKeyInLocalClientCacheWork(PR_REG_NAME);
  }

  private void doReplacePutsKeyInLocalClientCacheWork(final String regionName) {
    Host host = Host.getHost(0);
    VM server = host.getVM(0);
    VM client = host.getVM(2);
    int port1 = createRegionsAndStartServer(server);

    createClientRegion(client, port1, false, -1);
    server.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        final Region r = getCache().getRegion(regionName);
        assertNull(r.putIfAbsent("key0", "value"));
        assertTrue(r.containsKey("key0"));
        assertNull(r.putIfAbsent("key2", "value2"));
        assertTrue(r.containsKey("key2"));
        return null;
      }
    });
    client.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        final Region r = getCache().getRegion(regionName);
        assertEquals("value", r.replace("key0", "newValue"));
        assertTrue(r.containsKeyOnServer("key0"));
        assertTrue(r.containsKey("key0"));
        assertTrue(r.containsValueForKey("key0"));

        assertFalse(r.replace("key2", "DontReplace", "newValue"));
        assertTrue(r.replace("key2", "value2", "newValu2"));
        assertTrue(r.containsKeyOnServer("key2"));
        assertTrue(r.containsKey("key2"));
        assertTrue(r.containsValueForKey("key2"));
        return null;
      }
    });
    // bug #42221 - replace does not put entry on client when server has invalid value
    client.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        Region r = getCache().getRegion(regionName);
        final String key = "bug42221";
        r.putIfAbsent(key, null);
        assertTrue(r.containsKey(key));
        Object result = r.replace(key, "not null");
        assertEquals(null, result);
        assertTrue(r.containsKey(key));
        assertEquals(r.get(key), "not null");
        r.remove(key); // cleanup
        return null;
      }
    });
    // bug #42242 - remove(K,null) doesn't work
    client.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        Region r = getCache().getRegion(regionName);
        final String key = "bug42242";
        r.putIfAbsent(key, null);
        assertTrue(r.containsKey(key));
        assertTrue(r.containsKeyOnServer(key));
        boolean result = r.remove(key, null);
        assertTrue(result);
        assertFalse(r.containsKey(key));
        assertFalse(r.containsKeyOnServer(key));
        return null;
      }
    });
    // bug #42242b - second scenario with a replace(K,V,V) that didn't work
    client.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        Region r = getCache().getRegion(regionName);
        final String key = "bug42242b";
        r.putIfAbsent(key, null);
        assertTrue(r.containsKey(key));
        assertTrue(r.containsKeyOnServer(key));
        boolean result = r.replace(key, null, "new value");
        assertTrue(result);
        result = r.remove(key, "new value");
        assertTrue(result);
        assertFalse(r.containsKey(key));
        assertFalse(r.containsKeyOnServer(key));
        return null;
      }
    });
    // bug #42242c - remove does not work for entry that's on the server but not on the client
    final String key = "bug42242c";
    client.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        Region r = getCache().getRegion(regionName);
        r.registerInterest("ALL_KEYS");
        return null;
      }
    });
    server.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        Region r = getCache().getRegion(regionName);
        r.putIfAbsent(key, null);
        assertTrue(r.containsKey(key));
        return null;
      }
    });
    client.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        final Region r = getCache().getRegion(regionName);
        WaitCriterion w = new WaitCriterion() {
          @Override
          public String description() {
            return "waiting for server operation to reach client";
          }

          @Override
          public boolean done() {
            return r.containsKey(key);
          }
        };
        GeodeAwaitility.await().untilAsserted(w);
        assertTrue(r.containsKeyOnServer(key));
        boolean result = r.remove(key, null);
        // if (!result) {
        // ((LocalRegion)r).dumpBackingMap();
        // }
        assertTrue(result);
        assertFalse(r.containsKey(key));
        assertFalse(r.containsKeyOnServer(key));
        return null;
      }
    });
  }

  @Test
  public void testWithDelta() {
    doTestWithDeltaWork(false, REP_REG_NAME);
  }

  @Test
  public void testWithDeltaPR() {
    doTestWithDeltaWork(false, PR_REG_NAME);
  }

  @Test
  public void testWithDeltaCS() {
    doTestWithDeltaWork(true, REP_REG_NAME);
  }

  @Test
  public void testWithDeltaPRCS() {
    doTestWithDeltaWork(true, PR_REG_NAME);
  }

  private void doTestWithDeltaWork(final boolean clientServer, final String regName) {
    Host host = Host.getHost(0);
    VM vm1 = host.getVM(0);
    VM vm2 = host.getVM(1);

    if (clientServer) {
      int port = createRegionsAndStartServer(vm1);
      createClientRegion(vm2, port, false, -1);
    } else {
      createRegions(vm1);
      createRegions(vm2);
    }

    vm2.invoke(new SerializableCallable() {
      @Override
      public Object call() throws Exception {
        Region r = getCache().getRegion(regName);
        CustomerDelta c = new CustomerDelta("cust1", "addr1");
        assertNull(r.putIfAbsent("k1", c));
        CustomerDelta newc = new CustomerDelta(c);
        newc.setAddress("updatedAddress");
        assertEquals(c, r.putIfAbsent("k1", c));
        assertEquals(c, r.replace("k1", newc));
        assertFalse(r.replace("k1", c, newc));
        assertTrue(r.replace("k1", newc, c));
        assertFalse(r.remove("k1", newc));
        assertTrue(r.remove("k1", c));
        return null;
      }
    });
  }

  /** test putIfAbsent with failover & retry. This is bugs 42559 and 43640 */
  @Test
  public void testRetriedPutIfAbsent() throws Exception {
    doRetriedOperation(Operation.PUT_IF_ABSENT, false);
  }

  @Test
  public void testRetriedReplace() throws Exception {
    doRetriedOperation(Operation.REPLACE, false);
  }

  @Test
  public void testRetriedRemove() throws Exception {
    doRetriedOperation(Operation.REMOVE, false);
  }

  @Test
  public void testRetriedPutIfAbsentPR() throws Exception {
    doRetriedOperation(Operation.PUT_IF_ABSENT, false);
  }

  @Test
  public void testRetriedReplacePR() throws Exception {
    doRetriedOperation(Operation.REPLACE, false);
  }

  @Test
  public void testRetriedRemovePR() throws Exception {
    doRetriedOperation(Operation.REMOVE, false);
  }

  private void doRetriedOperation(final Operation op, boolean usePR) {
    Host host = Host.getHost(0);
    final VM server1 = host.getVM(0);
    final VM server2 = host.getVM(1);
    final VM client = host.getVM(2);
    final int port1 = createRegionsAndStartServer(server1, true);
    final int port2 = createRegionsAndStartServer(server2, true);
    final String regionName = usePR ? PR_REG_NAME : REP_REG_NAME;

    IgnoredException.addIgnoredException("java.net.SocketException");

    createClientRegion(client, port1, false, port2);

    SerializableCallable getID = new SerializableCallable("get DM ID") {
      @Override
      public Object call() {
        return getSystem().getDistributedMember();
      }
    };

    final DistributedMember server1ID = (DistributedMember) server1.invoke(getID);
    final DistributedMember server2ID = (DistributedMember) server2.invoke(getID);

    Set<IgnoredException> exceptions = new HashSet<IgnoredException>();
    exceptions.add(IgnoredException.addIgnoredException("Membership: requesting removal", server1));
    exceptions.add(IgnoredException.addIgnoredException("Membership: requesting removal", server2));
    exceptions.add(IgnoredException.addIgnoredException("ForcedDisconnect", server1));
    exceptions.add(IgnoredException.addIgnoredException("ForcedDisconnect", server2));

    try {

      server1.invoke(new SerializableCallable("install crasher in server1") {
        @Override
        public Object call() throws Exception {
          Region r = getCache().getRegion(regionName);
          r.put("key0", "value");
          if (op == Operation.PUT_IF_ABSENT) {
            r.destroy("key0");
          }
          // force client to use server1 for now
          // getCache().getCacheServers().get(0).stop();
          r.getAttributesMutator().addCacheListener(new CacheListenerAdapter() {
            private void killSender(EntryEvent event) {
              if (event.isOriginRemote()) {
                MembershipManager mgr =
                    MembershipManagerHelper.getMembership(getSystem());
                mgr.requestMemberRemoval(server2ID, "removing for test");
                try {
                  mgr.waitForDeparture(server2ID);
                } catch (Exception e) {
                  fail("failed to stop the other server for this test:" + e.getMessage());
                }
              }
            }

            @Override
            public void afterCreate(EntryEvent event) {
              getCache().getLogger().info("afterCreate invoked with " + event);
              killSender(event);
            }

            @Override
            public void afterUpdate(EntryEvent event) {
              getCache().getLogger().info("afterUpdate invoked with " + event);
              killSender(event);
            }

            @Override
            public void afterDestroy(EntryEvent event) {
              getCache().getLogger().info("afterDestroy invoked with " + event);
              killSender(event);
            }
          });
          return null;
        }
      });

      server2.invoke(new SerializableCallable("install crasher in server2") {
        @Override
        public Object call() throws Exception {
          Region r = getCache().getRegion(regionName);
          // force client to use server1 for now
          // getCache().getCacheServers().get(0).stop();
          r.getAttributesMutator().addCacheListener(new CacheListenerAdapter() {
            private void killSender(EntryEvent event) {
              if (event.isOriginRemote()) {
                MembershipManager mgr =
                    MembershipManagerHelper.getMembership(getSystem());
                mgr.requestMemberRemoval(server1ID, "removing for test");
                try {
                  mgr.waitForDeparture(server1ID);
                } catch (Exception e) {
                  fail("failed to stop the other server for this test:" + e.getMessage());
                }
              }
            }

            @Override
            public void afterCreate(EntryEvent event) {
              getCache().getLogger().info("afterCreate invoked with " + event);
              killSender(event);
            }

            @Override
            public void afterUpdate(EntryEvent event) {
              getCache().getLogger().info("afterUpdate invoked with " + event);
              killSender(event);
            }

            @Override
            public void afterDestroy(EntryEvent event) {
              getCache().getLogger().info("afterDestroy invoked with " + event);
              killSender(event);
            }
          });
          return null;
        }
      });


      client.invoke(new SerializableRunnable() {
        @Override
        public void run() {
          GemFireCacheImpl cache = (GemFireCacheImpl) getCache();
          Region r = cache.getRegion(regionName);
          if (op == Operation.PUT_IF_ABSENT) {
            assertTrue("expected putIfAbsent to succeed and return null",
                r.putIfAbsent("key0", "newvalue") == null);
          } else if (op == Operation.REMOVE) {
            assertTrue("expected remove operation to succeed and return true",
                r.remove("key0", "value"));
          } else if (op == Operation.REPLACE) {
            assertTrue("expected replace operation to succeed and return true",
                r.replace("key0", "value", "newvalue"));
          }
        }
      });
    } finally {
      disconnectAllFromDS();
      for (IgnoredException ex : exceptions) {
        ex.remove();
      }
    }
  }

  private static class CustomerDelta implements DataSerializable, Delta {
    private String name;
    private String address;
    private boolean nameChanged;
    private boolean addressChanged;

    public CustomerDelta() {}

    public CustomerDelta(CustomerDelta o) {
      this.address = o.address;
      this.name = o.name;
    }

    public CustomerDelta(String name, String address) {
      this.name = name;
      this.address = address;
    }

    @Override
    public void toData(DataOutput out) throws IOException {
      out.writeUTF(name);
      out.writeUTF(address);
      out.writeBoolean(nameChanged);
      out.writeBoolean(addressChanged);
    }

    @Override
    public void fromData(DataInput in) throws IOException, ClassNotFoundException {
      name = in.readUTF();
      address = in.readUTF();
      nameChanged = in.readBoolean();
      addressChanged = in.readBoolean();
    }

    @Override
    public void fromDelta(DataInput in) throws IOException, InvalidDeltaException {
      boolean nameC = in.readBoolean();
      if (nameC) {
        this.name = in.readUTF();
      }
      boolean addressC = in.readBoolean();
      if (addressC) {
        this.address = in.readUTF();
      }
    }

    @Override
    public boolean hasDelta() {
      return nameChanged || addressChanged;
    }

    @Override
    public void toDelta(DataOutput out) throws IOException {
      if (this.nameChanged) {
        out.writeBoolean(nameChanged);
        out.writeUTF(name);
      }
      if (this.addressChanged) {
        out.writeBoolean(addressChanged);
        out.writeUTF(address);
      }
    }

    public void setName(String name) {
      this.nameChanged = true;
      this.name = name;
    }

    public String getName() {
      return name;
    }

    public void setAddress(String address) {
      this.addressChanged = true;
      this.address = address;
    }

    public String getAddress() {
      return address;
    }

    @Override
    public boolean equals(Object obj) {
      if (!(obj instanceof CustomerDelta))
        return false;
      CustomerDelta other = (CustomerDelta) obj;
      return this.name.equals(other.name) && this.address.equals(other.address);
    }

    @Override
    public int hashCode() {
      return this.address.hashCode() + this.name.hashCode();
    }

  }
}