geode-core/src/distributedTest/java/org/apache/geode/cache30/PRBucketSynchronizationDUnitTest.java - geode - Git at Google

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

 import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
 import static org.junit.Assert.assertTrue;

 import java.util.HashSet;
 import java.util.Properties;
 import java.util.Set;

 import org.junit.After;
 import org.junit.Test;

 import org.apache.geode.cache.AttributesFactory;
 import org.apache.geode.cache.DataPolicy;
 import org.apache.geode.cache.EvictionAction;
 import org.apache.geode.cache.EvictionAttributes;
 import org.apache.geode.cache.Operation;
 import org.apache.geode.cache.PartitionAttributesFactory;
 import org.apache.geode.distributed.ConfigurationProperties;
 import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
 import org.apache.geode.internal.cache.BucketRegion;
 import org.apache.geode.internal.cache.EntryEventImpl;
 import org.apache.geode.internal.cache.LocalRegion;
 import org.apache.geode.internal.cache.PartitionedRegion;
 import org.apache.geode.internal.cache.RegionEntry;
 import org.apache.geode.internal.cache.Token;
 import org.apache.geode.internal.cache.VMCachedDeserializable;
 import org.apache.geode.internal.cache.versions.VMVersionTag;
 import org.apache.geode.internal.cache.versions.VersionSource;
 import org.apache.geode.internal.cache.versions.VersionTag;
 import org.apache.geode.test.dunit.DistributedTestUtils;
 import org.apache.geode.test.dunit.Host;
 import org.apache.geode.test.dunit.IgnoredException;
 import org.apache.geode.test.dunit.VM;
 import org.apache.geode.test.dunit.cache.CacheTestCase;

 /**
  * concurrency-control tests for client/server
  */

 @SuppressWarnings("serial")
 public class PRBucketSynchronizationDUnitTest extends CacheTestCase {

   private static LocalRegion testRegion;

   @After
   public void tearDown() {
     disconnectAllFromDS();
   }

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

   @Test
   public void testThatBucketSyncOnPrimaryLoss() {
     doBucketsSyncOnPrimaryLoss(TestType.IN_MEMORY);
   }

   @Test
   public void testThatBucketsSyncOnPrimaryLossWithPersistence() {
     doBucketsSyncOnPrimaryLoss(TestType.PERSISTENT);
   }

   @Test
   public void testThatBucketsSyncOnPrimaryLossWithOverflow() {
     doBucketsSyncOnPrimaryLoss(TestType.OVERFLOW);
   }

   /**
    * We hit this problem in bug #45669. A primary was lost and we did not see secondary buckets
    * perform a delta-GII.
    */
   private void doBucketsSyncOnPrimaryLoss(TestType typeOfTest) {
     IgnoredException.addIgnoredException("killing member's ds");
     IgnoredException.addIgnoredException("killing member's ds");
     Host host = Host.getHost(0);
     VM vm0 = host.getVM(0);
     VM vm1 = host.getVM(1);
     VM vm2 = host.getVM(2);
     Set<VM> verifyVMs = new HashSet<>();

     final String name = getUniqueName() + "Region";

     verifyVMs.add(vm0);
     verifyVMs.add(vm1);
     verifyVMs.add(vm2);

     createRegion(vm0, name, typeOfTest);
     createRegion(vm1, name, typeOfTest);
     createRegion(vm2, name, typeOfTest);

     createEntry1(vm0);

     VM primaryOwner;
     if (isPrimaryForBucket0(vm0)) {
       primaryOwner = vm0;
     } else if (isPrimaryForBucket0(vm1)) {
       primaryOwner = vm1;
     } else {
       primaryOwner = vm2;
     }

     verifyVMs.remove(primaryOwner);

     // cause one of the VMs to throw away the next operation
     VM creatorVM = null;
     InternalDistributedMember primaryID = getId(primaryOwner);
     VersionSource primaryVersionID = getVersionID(primaryOwner);
     for (VM vm : verifyVMs) {
       creatorVM = vm;
       createEntry2(creatorVM, primaryID, primaryVersionID);
       break;
     }

     verifyVMs.remove(creatorVM);

     // Now we crash the primary bucket owner simulating death during distribution.
     // The backup buckets should perform a delta-GII for the lost member and
     // get back in sync
     DistributedTestUtils.crashDistributedSystem(primaryOwner);

     for (VM vm : verifyVMs) {
       verifySynchronized(vm, primaryID);
     }
   }

   private void createEntry1(VM vm) {
     vm.invoke("create entry1", () -> {
       testRegion.create("Object1", 1);
     });
   }

   private boolean isPrimaryForBucket0(VM vm) {
     return vm.invoke("is primary?", () -> {
       PartitionedRegion pr = (PartitionedRegion) testRegion;
       return pr.getDataStore().getLocalBucketById(0).getBucketAdvisor().isPrimary();
     });
   }

   private InternalDistributedMember getId(VM vm) {
     return vm.invoke("get dmID", () -> {
       return testRegion.getCache().getMyId();
     });
   }

   private VersionSource getVersionID(VM vm) {
     return vm.invoke("get versionID", () -> {
       return testRegion.getVersionMember();
     });
   }

   private void createEntry2(VM vm, final InternalDistributedMember primary,
       final VersionSource primaryVersionID) {
     vm.invoke("create entry2", () -> {
       // create a fake event that looks like it came from the primary and apply it to this cache
       PartitionedRegion pr = (PartitionedRegion) testRegion;
       BucketRegion bucket = pr.getDataStore().getLocalBucketById(0);
       VersionTag tag = new VMVersionTag();
       tag.setMemberID(primaryVersionID);
       tag.setRegionVersion(2);
       tag.setEntryVersion(1);
       tag.setIsRemoteForTesting();
       EntryEventImpl event =
           EntryEventImpl.create(bucket, Operation.CREATE, "Object3", true, primary, true, false);
       event.setNewValue(new VMCachedDeserializable("value3", 12));
       event.setVersionTag(tag);
       bucket.getRegionMap().basicPut(event, System.currentTimeMillis(), true, false, null, false,
           false);
       event.release();

       // now create a tombstone so we can be sure these are transferred in delta-GII
       tag = new VMVersionTag();
       tag.setMemberID(primaryVersionID);
       tag.setRegionVersion(3);
       tag.setEntryVersion(1);
       tag.setIsRemoteForTesting();
       event =
           EntryEventImpl.create(bucket, Operation.CREATE, "Object5", true, primary, true, false);
       event.setNewValue(Token.TOMBSTONE);
       event.setVersionTag(tag);
       bucket.getRegionMap().basicPut(event, System.currentTimeMillis(), true, false, null, false,
           false);
       event.release();

       bucket.dumpBackingMap();
       assertTrue("bucket should hold entry Object3 now", bucket.containsKey("Object3"));
     });
   }

   private void verifySynchronized(VM vm, final InternalDistributedMember crashedMember) {
     vm.invoke("check that synchronization happened", () -> {
       PartitionedRegion pr = (PartitionedRegion) testRegion;
       final BucketRegion bucket = pr.getDataStore().getLocalBucketById(0);

       await().until(() -> {
         if (testRegion.getCache().getDistributionManager().isCurrentMember(crashedMember)) {
           return false;
         }
         if (!testRegion.containsKey("Object3")) {
           return false;
         }
         RegionEntry re = bucket.getRegionMap().getEntry("Object5");
         if (re == null) {
           return false;
         }
         return re.isTombstone();
       });
     });
   }

   private void createRegion(VM vm, final String regionName, final TestType typeOfTest) {
     vm.invoke(() -> {
       AttributesFactory af = new AttributesFactory();
       af.setDataPolicy(DataPolicy.PARTITION);
       af.setPartitionAttributes(
           (new PartitionAttributesFactory()).setTotalNumBuckets(2).setRedundantCopies(3).create());
       switch (typeOfTest) {
         case IN_MEMORY:
           break;
         case PERSISTENT:
           af.setDataPolicy(DataPolicy.PERSISTENT_PARTITION);
           break;
         case OVERFLOW:
           af.setEvictionAttributes(
               EvictionAttributes.createLRUEntryAttributes(5, EvictionAction.OVERFLOW_TO_DISK));
           break;
       }
       testRegion = (LocalRegion) createRootRegion(regionName, af.create());
     });
   }

   enum TestType {
     IN_MEMORY, OVERFLOW, PERSISTENT
   }
 }
	/*
	* 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.cache30;

	import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
	import static org.junit.Assert.assertTrue;

	import java.util.HashSet;
	import java.util.Properties;
	import java.util.Set;

	import org.junit.After;
	import org.junit.Test;

	import org.apache.geode.cache.AttributesFactory;
	import org.apache.geode.cache.DataPolicy;
	import org.apache.geode.cache.EvictionAction;
	import org.apache.geode.cache.EvictionAttributes;
	import org.apache.geode.cache.Operation;
	import org.apache.geode.cache.PartitionAttributesFactory;
	import org.apache.geode.distributed.ConfigurationProperties;
	import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
	import org.apache.geode.internal.cache.BucketRegion;
	import org.apache.geode.internal.cache.EntryEventImpl;
	import org.apache.geode.internal.cache.LocalRegion;
	import org.apache.geode.internal.cache.PartitionedRegion;
	import org.apache.geode.internal.cache.RegionEntry;
	import org.apache.geode.internal.cache.Token;
	import org.apache.geode.internal.cache.VMCachedDeserializable;
	import org.apache.geode.internal.cache.versions.VMVersionTag;
	import org.apache.geode.internal.cache.versions.VersionSource;
	import org.apache.geode.internal.cache.versions.VersionTag;
	import org.apache.geode.test.dunit.DistributedTestUtils;
	import org.apache.geode.test.dunit.Host;
	import org.apache.geode.test.dunit.IgnoredException;
	import org.apache.geode.test.dunit.VM;
	import org.apache.geode.test.dunit.cache.CacheTestCase;

	/**
	* concurrency-control tests for client/server
	*/

	@SuppressWarnings("serial")
	public class PRBucketSynchronizationDUnitTest extends CacheTestCase {

	private static LocalRegion testRegion;

	@After
	public void tearDown() {
	disconnectAllFromDS();
	}

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

	@Test
	public void testThatBucketSyncOnPrimaryLoss() {
	doBucketsSyncOnPrimaryLoss(TestType.IN_MEMORY);
	}

	@Test
	public void testThatBucketsSyncOnPrimaryLossWithPersistence() {
	doBucketsSyncOnPrimaryLoss(TestType.PERSISTENT);
	}

	@Test
	public void testThatBucketsSyncOnPrimaryLossWithOverflow() {
	doBucketsSyncOnPrimaryLoss(TestType.OVERFLOW);
	}

	/**
	* We hit this problem in bug #45669. A primary was lost and we did not see secondary buckets
	* perform a delta-GII.
	*/
	private void doBucketsSyncOnPrimaryLoss(TestType typeOfTest) {
	IgnoredException.addIgnoredException("killing member's ds");
	IgnoredException.addIgnoredException("killing member's ds");
	Host host = Host.getHost(0);
	VM vm0 = host.getVM(0);
	VM vm1 = host.getVM(1);
	VM vm2 = host.getVM(2);
	Set<VM> verifyVMs = new HashSet<>();

	final String name = getUniqueName() + "Region";

	verifyVMs.add(vm0);
	verifyVMs.add(vm1);
	verifyVMs.add(vm2);

	createRegion(vm0, name, typeOfTest);
	createRegion(vm1, name, typeOfTest);
	createRegion(vm2, name, typeOfTest);

	createEntry1(vm0);

	VM primaryOwner;
	if (isPrimaryForBucket0(vm0)) {
	primaryOwner = vm0;
	} else if (isPrimaryForBucket0(vm1)) {
	primaryOwner = vm1;
	} else {
	primaryOwner = vm2;
	}

	verifyVMs.remove(primaryOwner);

	// cause one of the VMs to throw away the next operation
	VM creatorVM = null;
	InternalDistributedMember primaryID = getId(primaryOwner);
	VersionSource primaryVersionID = getVersionID(primaryOwner);
	for (VM vm : verifyVMs) {
	creatorVM = vm;
	createEntry2(creatorVM, primaryID, primaryVersionID);
	break;
	}

	verifyVMs.remove(creatorVM);

	// Now we crash the primary bucket owner simulating death during distribution.
	// The backup buckets should perform a delta-GII for the lost member and
	// get back in sync
	DistributedTestUtils.crashDistributedSystem(primaryOwner);

	for (VM vm : verifyVMs) {
	verifySynchronized(vm, primaryID);
	}
	}

	private void createEntry1(VM vm) {
	vm.invoke("create entry1", () -> {
	testRegion.create("Object1", 1);
	});
	}

	private boolean isPrimaryForBucket0(VM vm) {
	return vm.invoke("is primary?", () -> {
	PartitionedRegion pr = (PartitionedRegion) testRegion;
	return pr.getDataStore().getLocalBucketById(0).getBucketAdvisor().isPrimary();
	});
	}

	private InternalDistributedMember getId(VM vm) {
	return vm.invoke("get dmID", () -> {
	return testRegion.getCache().getMyId();
	});
	}

	private VersionSource getVersionID(VM vm) {
	return vm.invoke("get versionID", () -> {
	return testRegion.getVersionMember();
	});
	}

	private void createEntry2(VM vm, final InternalDistributedMember primary,
	final VersionSource primaryVersionID) {
	vm.invoke("create entry2", () -> {
	// create a fake event that looks like it came from the primary and apply it to this cache
	PartitionedRegion pr = (PartitionedRegion) testRegion;
	BucketRegion bucket = pr.getDataStore().getLocalBucketById(0);
	VersionTag tag = new VMVersionTag();
	tag.setMemberID(primaryVersionID);
	tag.setRegionVersion(2);
	tag.setEntryVersion(1);
	tag.setIsRemoteForTesting();
	EntryEventImpl event =
	EntryEventImpl.create(bucket, Operation.CREATE, "Object3", true, primary, true, false);
	event.setNewValue(new VMCachedDeserializable("value3", 12));
	event.setVersionTag(tag);
	bucket.getRegionMap().basicPut(event, System.currentTimeMillis(), true, false, null, false,
	false);
	event.release();

	// now create a tombstone so we can be sure these are transferred in delta-GII
	tag = new VMVersionTag();
	tag.setMemberID(primaryVersionID);
	tag.setRegionVersion(3);
	tag.setEntryVersion(1);
	tag.setIsRemoteForTesting();
	event =
	EntryEventImpl.create(bucket, Operation.CREATE, "Object5", true, primary, true, false);
	event.setNewValue(Token.TOMBSTONE);
	event.setVersionTag(tag);
	bucket.getRegionMap().basicPut(event, System.currentTimeMillis(), true, false, null, false,
	false);
	event.release();

	bucket.dumpBackingMap();
	assertTrue("bucket should hold entry Object3 now", bucket.containsKey("Object3"));
	});
	}

	private void verifySynchronized(VM vm, final InternalDistributedMember crashedMember) {
	vm.invoke("check that synchronization happened", () -> {
	PartitionedRegion pr = (PartitionedRegion) testRegion;
	final BucketRegion bucket = pr.getDataStore().getLocalBucketById(0);

	await().until(() -> {
	if (testRegion.getCache().getDistributionManager().isCurrentMember(crashedMember)) {
	return false;
	}
	if (!testRegion.containsKey("Object3")) {
	return false;
	}
	RegionEntry re = bucket.getRegionMap().getEntry("Object5");
	if (re == null) {
	return false;
	}
	return re.isTombstone();
	});
	});
	}

	private void createRegion(VM vm, final String regionName, final TestType typeOfTest) {
	vm.invoke(() -> {
	AttributesFactory af = new AttributesFactory();
	af.setDataPolicy(DataPolicy.PARTITION);
	af.setPartitionAttributes(
	(new PartitionAttributesFactory()).setTotalNumBuckets(2).setRedundantCopies(3).create());
	switch (typeOfTest) {
	case IN_MEMORY:
	break;
	case PERSISTENT:
	af.setDataPolicy(DataPolicy.PERSISTENT_PARTITION);
	break;
	case OVERFLOW:
	af.setEvictionAttributes(
	EvictionAttributes.createLRUEntryAttributes(5, EvictionAction.OVERFLOW_TO_DISK));
	break;
	}
	testRegion = (LocalRegion) createRootRegion(regionName, af.create());
	});
	}

	enum TestType {
	IN_MEMORY, OVERFLOW, PERSISTENT
	}
	}