geode-core/src/integrationTest/java/org/apache/geode/TXExpirationIntegrationTest.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;

 import static java.util.concurrent.TimeUnit.SECONDS;
 import static org.apache.geode.cache.ExpirationAction.DESTROY;
 import static org.apache.geode.cache.ExpirationAction.INVALIDATE;
 import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS;
 import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
 import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.atLeast;
 import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.verify;

 import java.util.Properties;
 import java.util.function.Consumer;

 import junitparams.JUnitParamsRunner;
 import junitparams.Parameters;
 import junitparams.naming.TestCaseName;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;

 import org.apache.geode.cache.AttributesMutator;
 import org.apache.geode.cache.CacheFactory;
 import org.apache.geode.cache.CacheTransactionManager;
 import org.apache.geode.cache.CommitConflictException;
 import org.apache.geode.cache.EntryEvent;
 import org.apache.geode.cache.ExpirationAttributes;
 import org.apache.geode.cache.Region;
 import org.apache.geode.cache.RegionEvent;
 import org.apache.geode.cache.RegionFactory;
 import org.apache.geode.cache.RegionShortcut;
 import org.apache.geode.cache.util.CacheListenerAdapter;
 import org.apache.geode.internal.cache.ExpiryTask;
 import org.apache.geode.internal.cache.InternalCache;
 import org.apache.geode.internal.cache.InternalRegion;
 import org.apache.geode.internal.cache.LocalRegion;
 import org.apache.geode.internal.cache.TXManagerImpl;
 import org.apache.geode.internal.cache.TXStateProxy;

 /**
  * Tests transaction expiration functionality
  *
  * @since GemFire 4.0
  */
 @RunWith(JUnitParamsRunner.class)
 public class TXExpirationIntegrationTest {

   private static final String REGION_NAME =
       TXExpirationIntegrationTest.class.getSimpleName() + "_region";
   private static final String KEY = "key";

   private InternalCache cache;
   private CacheTransactionManager transactionManager;

   @Before
   public void setUp() throws Exception {
     cache = (InternalCache) new CacheFactory(getConfig()).create();
     transactionManager = cache.getCacheTransactionManager();
   }

   protected Properties getConfig() {
     Properties config = new Properties();
     config.setProperty(LOCATORS, "");
     config.setProperty(MCAST_PORT, "0");
     return config;
   }

   @After
   public void tearDown() throws Exception {
     try {
       transactionManager.rollback();
     } catch (IllegalStateException ignore) {
     }
     cache.close();

     ExpiryTask.permitExpiration();
   }

   @Test
   @Parameters({"ENTRY_IDLE_DESTROY", "ENTRY_IDLE_INVALIDATE", "ENTRY_TTL_DESTROY",
       "ENTRY_TTL_INVALIDATE"})
   @TestCaseName("{method}({params})")
   public void entryExpirationDoesNotCauseConflict(ExpirationOperation operation) throws Exception {
     InternalRegion region = createRegion(REGION_NAME);
     AttributesMutator<String, String> mutator = region.getAttributesMutator();

     KeyCacheListener keyCacheListener = spy(new KeyCacheListener());
     mutator.addCacheListener(keyCacheListener);

     ExpiryTask.suspendExpiration();
     operation.mutate(mutator);

     region.put(KEY, "value1");

     transactionManager.begin();
     region.put(KEY, "value2");
     awaitEntryExpiration(region, KEY); // enables expiration
     assertThat(region.getEntry(KEY).getValue()).isEqualTo("value2");

     ExpiryTask.suspendExpiration();
     transactionManager.commit();

     assertThat(region.getEntry(KEY).getValue()).isEqualTo("value2");
     awaitEntryExpiration(region, KEY); // enables expiration
     operation.verifyInvoked(keyCacheListener);

     operation.verifyState(region);
   }

   @Test
   @Parameters({"ENTRY_IDLE_DESTROY", "ENTRY_IDLE_INVALIDATE", "ENTRY_TTL_DESTROY",
       "ENTRY_TTL_INVALIDATE"})
   @TestCaseName("{method}({params})")
   public void entryExpirationContinuesAfterCommitConflict(ExpirationOperation operation)
       throws Exception {
     Region<String, String> region = createRegion(REGION_NAME);

     AttributesMutator<String, String> mutator = region.getAttributesMutator();

     KeyCacheListener keyCacheListener = spy(new KeyCacheListener());
     mutator.addCacheListener(keyCacheListener);

     ExpiryTask.suspendExpiration();
     operation.mutate(mutator);

     ExpiryTask.suspendExpiration();
     region.put(KEY, "value1");

     transactionManager.begin();
     region.put(KEY, "value2");
     awaitEntryExpiration(region, KEY); // enables expiration
     assertThat(region.getEntry(KEY).getValue()).isEqualTo("value2");

     ExpiryTask.suspendExpiration();

     TXManagerImpl txManagerImpl = (TXManagerImpl) transactionManager;
     TXStateProxy txStateProxy = txManagerImpl.pauseTransaction();
     region.put(KEY, "conflict");
     txManagerImpl.unpauseTransaction(txStateProxy);

     assertThatThrownBy(() -> transactionManager.commit())
         .isInstanceOf(CommitConflictException.class);

     awaitEntryExpiration(region, KEY); // enables expiration
     operation.verifyInvoked(keyCacheListener);
     operation.verifyState(region);
   }

   @Test
   @Parameters({"ENTRY_IDLE_DESTROY", "ENTRY_IDLE_INVALIDATE", "ENTRY_TTL_DESTROY",
       "ENTRY_TTL_INVALIDATE"})
   @TestCaseName("{method}({params})")
   public void entryExpirationContinuesAfterRollback(ExpirationOperation operation)
       throws Exception {
     Region<String, String> region = createRegion(REGION_NAME);

     AttributesMutator<String, String> mutator = region.getAttributesMutator();

     KeyCacheListener keyCacheListener = spy(new KeyCacheListener());
     mutator.addCacheListener(keyCacheListener);

     ExpiryTask.suspendExpiration();
     operation.mutate(mutator);

     region.put(KEY, "value1");

     transactionManager.begin();
     region.put(KEY, "value2");
     awaitEntryExpiration(region, KEY); // enables expiration
     assertThat(region.getEntry(KEY).getValue()).isEqualTo("value2");

     ExpiryTask.suspendExpiration();

     transactionManager.rollback();

     awaitEntryExpiration(region, KEY); // enables expiration
     operation.verifyInvoked(keyCacheListener);
     operation.verifyState(region);
   }

   @Test
   @Parameters({"REGION_IDLE_DESTROY", "REGION_IDLE_INVALIDATE", "REGION_TTL_DESTROY",
       "REGION_TTL_INVALIDATE"})
   @TestCaseName("{method}({params})")
   public void regionExpirationDoesNotCauseConflict(ExpirationOperation operation) throws Exception {
     Region<String, String> region = createRegion(REGION_NAME);

     KeyCacheListener keyCacheListener = spy(new KeyCacheListener());

     AttributesMutator<String, String> mutator = region.getAttributesMutator();
     mutator.addCacheListener(keyCacheListener);

     ExpiryTask.suspendExpiration();
     operation.mutate(mutator);

     transactionManager.begin();
     region.put(KEY, "value1");

     awaitRegionExpiration(region); // enables expiration
     assertThat(region.getEntry(KEY).getValue()).isEqualTo("value1");

     ExpiryTask.suspendExpiration();
     transactionManager.commit();

     assertThat(region.getEntry(KEY).getValue()).isEqualTo("value1");

     awaitRegionExpiration(region); // enables expiration
     operation.verifyInvoked(keyCacheListener);
     operation.verifyState(region);
   }

   private enum ExpirationOperation {
     ENTRY_IDLE_DESTROY(m -> m.setEntryIdleTimeout(new ExpirationAttributes(1, DESTROY)),
         s -> verify(s, atLeast(1)).afterDestroyKey(eq(KEY)),
         r -> assertThat(r.containsKey(KEY)).isFalse()),
     ENTRY_IDLE_INVALIDATE(m -> m.setEntryIdleTimeout(new ExpirationAttributes(1, INVALIDATE)),
         s -> verify(s, atLeast(1)).afterInvalidateKey(eq(KEY)),
         r -> assertThat(r.get(KEY)).isNull()),
     ENTRY_TTL_DESTROY(m -> m.setEntryTimeToLive(new ExpirationAttributes(1, DESTROY)),
         s -> verify(s, atLeast(1)).afterDestroyKey(eq(KEY)),
         r -> assertThat(r.containsKey(KEY)).isFalse()),
     ENTRY_TTL_INVALIDATE(m -> m.setEntryTimeToLive(new ExpirationAttributes(1, INVALIDATE)),
         s -> verify(s, atLeast(1)).afterInvalidateKey(eq(KEY)),
         r -> assertThat(r.get(KEY)).isNull()),
     REGION_IDLE_DESTROY(m -> m.setRegionIdleTimeout(new ExpirationAttributes(1, DESTROY)),
         s -> verify(s, atLeast(1)).afterRegionDestroyName(eq(REGION_NAME)),
         r -> await().until(() -> r.isDestroyed())),
     REGION_IDLE_INVALIDATE(m -> m.setRegionIdleTimeout(new ExpirationAttributes(1, INVALIDATE)),
         s -> verify(s, atLeast(1)).afterRegionInvalidateName(eq(REGION_NAME)),
         r -> await().until(() -> r.get(KEY) == null)),
     REGION_TTL_DESTROY(m -> m.setRegionTimeToLive(new ExpirationAttributes(1, DESTROY)),
         s -> verify(s, atLeast(1)).afterRegionDestroyName(eq(REGION_NAME)),
         r -> await().until(() -> r.isDestroyed())),
     REGION_TTL_INVALIDATE(m -> m.setRegionTimeToLive(new ExpirationAttributes(1, INVALIDATE)),
         s -> verify(s, atLeast(1)).afterRegionInvalidateName(eq(REGION_NAME)),
         r -> await().until(() -> r.get(KEY) == null));

     private final Consumer<AttributesMutator> mutatorConsumer;
     private final Consumer<KeyCacheListener> listenerConsumer;
     private final Consumer<Region> regionConsumer;

     ExpirationOperation(Consumer<AttributesMutator> mutatorConsumer,
         Consumer<KeyCacheListener> listenerConsumer, Consumer<Region> regionConsumer) {
       this.mutatorConsumer = mutatorConsumer;
       this.listenerConsumer = listenerConsumer;
       this.regionConsumer = regionConsumer;
     }

     void mutate(AttributesMutator mutator) {
       mutatorConsumer.accept(mutator);
     }

     void verifyInvoked(KeyCacheListener keyCacheListener) {
       listenerConsumer.accept(keyCacheListener);
     }

     public void verifyState(Region region) {
       regionConsumer.accept(region);
     }
   }

   private InternalRegion createRegion(String name) {
     RegionFactory<String, String> rf = cache.createRegionFactory(RegionShortcut.REPLICATE);
     rf.setStatisticsEnabled(true);

     System.setProperty(LocalRegion.EXPIRY_MS_PROPERTY, "true");
     try {
       return (InternalRegion) rf.create(name);
     } finally {
       System.clearProperty(LocalRegion.EXPIRY_MS_PROPERTY);
     }
   }

   private void awaitEntryExpired(Region region, String key) {
     await().until(() -> region.getEntry(key) == null || region.get(key) == null);
   }

   private void awaitEntryExpiration(Region region, String key) {
     InternalRegion internalRegion = (InternalRegion) region;
     try {
       ExpirationDetector detector;
       do {
         detector = new ExpirationDetector(internalRegion.getEntryExpiryTask(key));
         ExpiryTask.expiryTaskListener = detector;
         ExpiryTask.permitExpiration();
         detector.awaitExecuted(30, SECONDS);
       } while (!detector.hasExpired() && detector.wasRescheduled());
     } finally {
       ExpiryTask.expiryTaskListener = null;
     }
   }

   private void awaitRegionExpiration(Region region) {
     InternalRegion internalRegion = (InternalRegion) region;
     try {
       ExpirationDetector detector;
       do {
         detector = new ExpirationDetector(internalRegion.getRegionTTLExpiryTask() != null
             ? internalRegion.getRegionTTLExpiryTask() : internalRegion.getRegionIdleExpiryTask());
         ExpiryTask.expiryTaskListener = detector;
         ExpiryTask.permitExpiration();
         detector.awaitExecuted(30, SECONDS);
       } while (!detector.hasExpired() && detector.wasRescheduled());
     } finally {
       ExpiryTask.expiryTaskListener = null;
     }
   }

   private static class KeyCacheListener extends CacheListenerAdapter<String, String> {

     @Override
     public void afterCreate(EntryEvent<String, String> event) {
       afterCreateKey(event.getKey());
     }

     @Override
     public void afterUpdate(EntryEvent<String, String> event) {
       afterUpdateKey(event.getKey());
     }

     @Override
     public void afterInvalidate(EntryEvent<String, String> event) {
       afterInvalidateKey(event.getKey());
     }

     @Override
     public void afterDestroy(EntryEvent<String, String> event) {
       afterDestroyKey(event.getKey());
     }

     @Override
     public void afterRegionInvalidate(RegionEvent<String, String> event) {
       afterRegionInvalidateName(event.getRegion().getName());
     }

     @Override
     public void afterRegionDestroy(RegionEvent<String, String> event) {
       afterRegionDestroyName(event.getRegion().getName());
     }

     public void afterCreateKey(Object key) {
       // nothing
     }

     public void afterUpdateKey(Object key) {
       // nothing
     }

     public void afterInvalidateKey(Object key) {
       // nothing
     }

     public void afterDestroyKey(Object key) {
       // nothing
     }

     public void afterRegionInvalidateName(Object key) {
       // nothing
     }

     public void afterRegionDestroyName(Object key) {
       // nothing
     }
   }
 }
	/*
	* 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;

	import static java.util.concurrent.TimeUnit.SECONDS;
	import static org.apache.geode.cache.ExpirationAction.DESTROY;
	import static org.apache.geode.cache.ExpirationAction.INVALIDATE;
	import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS;
	import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
	import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
	import static org.assertj.core.api.Assertions.assertThat;
	import static org.assertj.core.api.Assertions.assertThatThrownBy;
	import static org.mockito.ArgumentMatchers.eq;
	import static org.mockito.Mockito.atLeast;
	import static org.mockito.Mockito.spy;
	import static org.mockito.Mockito.verify;

	import java.util.Properties;
	import java.util.function.Consumer;

	import junitparams.JUnitParamsRunner;
	import junitparams.Parameters;
	import junitparams.naming.TestCaseName;
	import org.junit.After;
	import org.junit.Before;
	import org.junit.Test;
	import org.junit.runner.RunWith;

	import org.apache.geode.cache.AttributesMutator;
	import org.apache.geode.cache.CacheFactory;
	import org.apache.geode.cache.CacheTransactionManager;
	import org.apache.geode.cache.CommitConflictException;
	import org.apache.geode.cache.EntryEvent;
	import org.apache.geode.cache.ExpirationAttributes;
	import org.apache.geode.cache.Region;
	import org.apache.geode.cache.RegionEvent;
	import org.apache.geode.cache.RegionFactory;
	import org.apache.geode.cache.RegionShortcut;
	import org.apache.geode.cache.util.CacheListenerAdapter;
	import org.apache.geode.internal.cache.ExpiryTask;
	import org.apache.geode.internal.cache.InternalCache;
	import org.apache.geode.internal.cache.InternalRegion;
	import org.apache.geode.internal.cache.LocalRegion;
	import org.apache.geode.internal.cache.TXManagerImpl;
	import org.apache.geode.internal.cache.TXStateProxy;

	/**
	* Tests transaction expiration functionality
	*
	* @since GemFire 4.0
	*/
	@RunWith(JUnitParamsRunner.class)
	public class TXExpirationIntegrationTest {

	private static final String REGION_NAME =
	TXExpirationIntegrationTest.class.getSimpleName() + "_region";
	private static final String KEY = "key";

	private InternalCache cache;
	private CacheTransactionManager transactionManager;

	@Before
	public void setUp() throws Exception {
	cache = (InternalCache) new CacheFactory(getConfig()).create();
	transactionManager = cache.getCacheTransactionManager();
	}

	protected Properties getConfig() {
	Properties config = new Properties();
	config.setProperty(LOCATORS, "");
	config.setProperty(MCAST_PORT, "0");
	return config;
	}

	@After
	public void tearDown() throws Exception {
	try {
	transactionManager.rollback();
	} catch (IllegalStateException ignore) {
	}
	cache.close();

	ExpiryTask.permitExpiration();
	}

	@Test
	@Parameters({"ENTRY_IDLE_DESTROY", "ENTRY_IDLE_INVALIDATE", "ENTRY_TTL_DESTROY",
	"ENTRY_TTL_INVALIDATE"})
	@TestCaseName("{method}({params})")
	public void entryExpirationDoesNotCauseConflict(ExpirationOperation operation) throws Exception {
	InternalRegion region = createRegion(REGION_NAME);
	AttributesMutator<String, String> mutator = region.getAttributesMutator();

	KeyCacheListener keyCacheListener = spy(new KeyCacheListener());
	mutator.addCacheListener(keyCacheListener);

	ExpiryTask.suspendExpiration();
	operation.mutate(mutator);

	region.put(KEY, "value1");

	transactionManager.begin();
	region.put(KEY, "value2");
	awaitEntryExpiration(region, KEY); // enables expiration
	assertThat(region.getEntry(KEY).getValue()).isEqualTo("value2");

	ExpiryTask.suspendExpiration();
	transactionManager.commit();

	assertThat(region.getEntry(KEY).getValue()).isEqualTo("value2");
	awaitEntryExpiration(region, KEY); // enables expiration
	operation.verifyInvoked(keyCacheListener);

	operation.verifyState(region);
	}

	@Test
	@Parameters({"ENTRY_IDLE_DESTROY", "ENTRY_IDLE_INVALIDATE", "ENTRY_TTL_DESTROY",
	"ENTRY_TTL_INVALIDATE"})
	@TestCaseName("{method}({params})")
	public void entryExpirationContinuesAfterCommitConflict(ExpirationOperation operation)
	throws Exception {
	Region<String, String> region = createRegion(REGION_NAME);

	AttributesMutator<String, String> mutator = region.getAttributesMutator();

	KeyCacheListener keyCacheListener = spy(new KeyCacheListener());
	mutator.addCacheListener(keyCacheListener);

	ExpiryTask.suspendExpiration();
	operation.mutate(mutator);

	ExpiryTask.suspendExpiration();
	region.put(KEY, "value1");

	transactionManager.begin();
	region.put(KEY, "value2");
	awaitEntryExpiration(region, KEY); // enables expiration
	assertThat(region.getEntry(KEY).getValue()).isEqualTo("value2");

	ExpiryTask.suspendExpiration();

	TXManagerImpl txManagerImpl = (TXManagerImpl) transactionManager;
	TXStateProxy txStateProxy = txManagerImpl.pauseTransaction();
	region.put(KEY, "conflict");
	txManagerImpl.unpauseTransaction(txStateProxy);

	assertThatThrownBy(() -> transactionManager.commit())
	.isInstanceOf(CommitConflictException.class);

	awaitEntryExpiration(region, KEY); // enables expiration
	operation.verifyInvoked(keyCacheListener);
	operation.verifyState(region);
	}

	@Test
	@Parameters({"ENTRY_IDLE_DESTROY", "ENTRY_IDLE_INVALIDATE", "ENTRY_TTL_DESTROY",
	"ENTRY_TTL_INVALIDATE"})
	@TestCaseName("{method}({params})")
	public void entryExpirationContinuesAfterRollback(ExpirationOperation operation)
	throws Exception {
	Region<String, String> region = createRegion(REGION_NAME);

	AttributesMutator<String, String> mutator = region.getAttributesMutator();

	KeyCacheListener keyCacheListener = spy(new KeyCacheListener());
	mutator.addCacheListener(keyCacheListener);

	ExpiryTask.suspendExpiration();
	operation.mutate(mutator);

	region.put(KEY, "value1");

	transactionManager.begin();
	region.put(KEY, "value2");
	awaitEntryExpiration(region, KEY); // enables expiration
	assertThat(region.getEntry(KEY).getValue()).isEqualTo("value2");

	ExpiryTask.suspendExpiration();

	transactionManager.rollback();

	awaitEntryExpiration(region, KEY); // enables expiration
	operation.verifyInvoked(keyCacheListener);
	operation.verifyState(region);
	}

	@Test
	@Parameters({"REGION_IDLE_DESTROY", "REGION_IDLE_INVALIDATE", "REGION_TTL_DESTROY",
	"REGION_TTL_INVALIDATE"})
	@TestCaseName("{method}({params})")
	public void regionExpirationDoesNotCauseConflict(ExpirationOperation operation) throws Exception {
	Region<String, String> region = createRegion(REGION_NAME);

	KeyCacheListener keyCacheListener = spy(new KeyCacheListener());

	AttributesMutator<String, String> mutator = region.getAttributesMutator();
	mutator.addCacheListener(keyCacheListener);

	ExpiryTask.suspendExpiration();
	operation.mutate(mutator);

	transactionManager.begin();
	region.put(KEY, "value1");

	awaitRegionExpiration(region); // enables expiration
	assertThat(region.getEntry(KEY).getValue()).isEqualTo("value1");

	ExpiryTask.suspendExpiration();
	transactionManager.commit();

	assertThat(region.getEntry(KEY).getValue()).isEqualTo("value1");

	awaitRegionExpiration(region); // enables expiration
	operation.verifyInvoked(keyCacheListener);
	operation.verifyState(region);
	}

	private enum ExpirationOperation {
	ENTRY_IDLE_DESTROY(m -> m.setEntryIdleTimeout(new ExpirationAttributes(1, DESTROY)),
	s -> verify(s, atLeast(1)).afterDestroyKey(eq(KEY)),
	r -> assertThat(r.containsKey(KEY)).isFalse()),
	ENTRY_IDLE_INVALIDATE(m -> m.setEntryIdleTimeout(new ExpirationAttributes(1, INVALIDATE)),
	s -> verify(s, atLeast(1)).afterInvalidateKey(eq(KEY)),
	r -> assertThat(r.get(KEY)).isNull()),
	ENTRY_TTL_DESTROY(m -> m.setEntryTimeToLive(new ExpirationAttributes(1, DESTROY)),
	s -> verify(s, atLeast(1)).afterDestroyKey(eq(KEY)),
	r -> assertThat(r.containsKey(KEY)).isFalse()),
	ENTRY_TTL_INVALIDATE(m -> m.setEntryTimeToLive(new ExpirationAttributes(1, INVALIDATE)),
	s -> verify(s, atLeast(1)).afterInvalidateKey(eq(KEY)),
	r -> assertThat(r.get(KEY)).isNull()),
	REGION_IDLE_DESTROY(m -> m.setRegionIdleTimeout(new ExpirationAttributes(1, DESTROY)),
	s -> verify(s, atLeast(1)).afterRegionDestroyName(eq(REGION_NAME)),
	r -> await().until(() -> r.isDestroyed())),
	REGION_IDLE_INVALIDATE(m -> m.setRegionIdleTimeout(new ExpirationAttributes(1, INVALIDATE)),
	s -> verify(s, atLeast(1)).afterRegionInvalidateName(eq(REGION_NAME)),
	r -> await().until(() -> r.get(KEY) == null)),
	REGION_TTL_DESTROY(m -> m.setRegionTimeToLive(new ExpirationAttributes(1, DESTROY)),
	s -> verify(s, atLeast(1)).afterRegionDestroyName(eq(REGION_NAME)),
	r -> await().until(() -> r.isDestroyed())),
	REGION_TTL_INVALIDATE(m -> m.setRegionTimeToLive(new ExpirationAttributes(1, INVALIDATE)),
	s -> verify(s, atLeast(1)).afterRegionInvalidateName(eq(REGION_NAME)),
	r -> await().until(() -> r.get(KEY) == null));

	private final Consumer<AttributesMutator> mutatorConsumer;
	private final Consumer<KeyCacheListener> listenerConsumer;
	private final Consumer<Region> regionConsumer;

	ExpirationOperation(Consumer<AttributesMutator> mutatorConsumer,
	Consumer<KeyCacheListener> listenerConsumer, Consumer<Region> regionConsumer) {
	this.mutatorConsumer = mutatorConsumer;
	this.listenerConsumer = listenerConsumer;
	this.regionConsumer = regionConsumer;
	}

	void mutate(AttributesMutator mutator) {
	mutatorConsumer.accept(mutator);
	}

	void verifyInvoked(KeyCacheListener keyCacheListener) {
	listenerConsumer.accept(keyCacheListener);
	}

	public void verifyState(Region region) {
	regionConsumer.accept(region);
	}
	}

	private InternalRegion createRegion(String name) {
	RegionFactory<String, String> rf = cache.createRegionFactory(RegionShortcut.REPLICATE);
	rf.setStatisticsEnabled(true);

	System.setProperty(LocalRegion.EXPIRY_MS_PROPERTY, "true");
	try {
	return (InternalRegion) rf.create(name);
	} finally {
	System.clearProperty(LocalRegion.EXPIRY_MS_PROPERTY);
	}
	}

	private void awaitEntryExpired(Region region, String key) {
	await().until(() -> region.getEntry(key) == null \|\| region.get(key) == null);
	}

	private void awaitEntryExpiration(Region region, String key) {
	InternalRegion internalRegion = (InternalRegion) region;
	try {
	ExpirationDetector detector;
	do {
	detector = new ExpirationDetector(internalRegion.getEntryExpiryTask(key));
	ExpiryTask.expiryTaskListener = detector;
	ExpiryTask.permitExpiration();
	detector.awaitExecuted(30, SECONDS);
	} while (!detector.hasExpired() && detector.wasRescheduled());
	} finally {
	ExpiryTask.expiryTaskListener = null;
	}
	}

	private void awaitRegionExpiration(Region region) {
	InternalRegion internalRegion = (InternalRegion) region;
	try {
	ExpirationDetector detector;
	do {
	detector = new ExpirationDetector(internalRegion.getRegionTTLExpiryTask() != null
	? internalRegion.getRegionTTLExpiryTask() : internalRegion.getRegionIdleExpiryTask());
	ExpiryTask.expiryTaskListener = detector;
	ExpiryTask.permitExpiration();
	detector.awaitExecuted(30, SECONDS);
	} while (!detector.hasExpired() && detector.wasRescheduled());
	} finally {
	ExpiryTask.expiryTaskListener = null;
	}
	}

	private static class KeyCacheListener extends CacheListenerAdapter<String, String> {

	@Override
	public void afterCreate(EntryEvent<String, String> event) {
	afterCreateKey(event.getKey());
	}

	@Override
	public void afterUpdate(EntryEvent<String, String> event) {
	afterUpdateKey(event.getKey());
	}

	@Override
	public void afterInvalidate(EntryEvent<String, String> event) {
	afterInvalidateKey(event.getKey());
	}

	@Override
	public void afterDestroy(EntryEvent<String, String> event) {
	afterDestroyKey(event.getKey());
	}

	@Override
	public void afterRegionInvalidate(RegionEvent<String, String> event) {
	afterRegionInvalidateName(event.getRegion().getName());
	}

	@Override
	public void afterRegionDestroy(RegionEvent<String, String> event) {
	afterRegionDestroyName(event.getRegion().getName());
	}

	public void afterCreateKey(Object key) {
	// nothing
	}

	public void afterUpdateKey(Object key) {
	// nothing
	}

	public void afterInvalidateKey(Object key) {
	// nothing
	}

	public void afterDestroyKey(Object key) {
	// nothing
	}

	public void afterRegionInvalidateName(Object key) {
	// nothing
	}

	public void afterRegionDestroyName(Object key) {
	// nothing
	}
	}
	}