Spring/spring-data-geode
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

DATAGEODE-179 - Polish Durable Client and Eviction/Expiration Annotation configuration support after Framework regressions.

Browse Source
This commit is contained in:
John Blum
2019-04-09 11:32:28 -07:00
parent e001901127
commit c1d6a710c0
7 changed files with 197 additions and 183 deletions
Download Patch File Download Diff File Expand all files Collapse all files

202
src/test/java/org/springframework/data/gemfire/client/DurableClientCacheIntegrationTest.java
View File

@@ -16,16 +16,11 @@
package org.springframework.data.gemfire.client;
import static org.hamcrest.CoreMatchers.equalTo;
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.CoreMatchers.notNullValue;
import static org.hamcrest.number.OrderingComparison.greaterThanOrEqualTo;
import static org.junit.Assert.assertThat;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.Assume.assumeTrue;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -89,24 +84,54 @@ public class DurableClientCacheIntegrationTest extends AbstractGemFireClientServ
private static int serverPort;
private static AtomicBoolean DIRTIES_CONTEXT = new AtomicBoolean(false);
private static AtomicBoolean dirtiesContext = new AtomicBoolean(false);
private static List<Integer> regionCacheListenerEventValues =
Collections.synchronizedList(new ArrayList<Integer>());
private static ProcessWrapper serverProcess;
private static final String CACHE_SERVER_PORT_SYSTEM_PROPERTY =
private static final String CACHE_SERVER_PORT =
DurableClientCacheIntegrationTest.class.getName().concat(".cache-server-port");
private static final String CLIENT_CACHE_INTERESTS_RESULT_POLICY_SYSTEM_PROPERTY =
private static final String CLIENT_CACHE_INTERESTS_RESULT_POLICY =
DurableClientCacheIntegrationTest.class.getName().concat(".interests-result-policy");
private static final String DURABLE_CLIENT_TIMEOUT_SYSTEM_PROPERTY =
private static final String DURABLE_CLIENT_TIMEOUT =
DurableClientCacheIntegrationTest.class.getName().concat(".durable-client-timeout");
private static final String SERVER_HOST = "localhost";
@BeforeClass
public static void startGemFireServer() throws IOException {
serverPort = setSystemProperty(CACHE_SERVER_PORT, SocketUtils.findAvailableTcpPort());
serverProcess = startGemFireServer(DurableClientCacheIntegrationTest.class);
}
@AfterClass
public static void stopGemFireServer() {
stopGemFireServer(serverProcess);
clearTestClassSystemProperties(DurableClientCacheIntegrationTest.class);
}
private static boolean isAfterDirtiesContext() {
return dirtiesContext.get();
}
private static boolean isBeforeDirtiesContext() {
return !isAfterDirtiesContext();
}
private boolean dirtiesContext() {
return !dirtiesContext.getAndSet(true);
}
private <T> T valueBeforeAndAfterDirtiesContext(T valueBefore, T valueAfter) {
return isBeforeDirtiesContext() ? valueBefore : valueAfter;
}
@Autowired
private ConfigurableApplicationContext applicationContext;
@@ -116,54 +141,26 @@ public class DurableClientCacheIntegrationTest extends AbstractGemFireClientServ
@Resource(name = "Example")
private Region<String, Integer> example;
@BeforeClass
public static void startGemFireServer() throws IOException {
serverPort = setSystemProperty(CACHE_SERVER_PORT_SYSTEM_PROPERTY, SocketUtils.findAvailableTcpPort());
serverProcess = startGemFireServer(DurableClientCacheIntegrationTest.class);
}
@AfterClass
public static void stopGemFireServer() {
stopGemFireServer(serverProcess);
clearSystemProperties(DurableClientCacheIntegrationTest.class);
}
private static boolean isAfterDirtiesContext() {
return DIRTIES_CONTEXT.get();
}
private static boolean isBeforeDirtiesContext() {
return !isAfterDirtiesContext();
}
private boolean dirtiesContext() {
return !DIRTIES_CONTEXT.getAndSet(true);
}
private <T> T valueBeforeAndAfterDirtiesContext(T before, T after) {
return (isBeforeDirtiesContext() ? before : after);
}
@Before
public void setup() {
Properties distributedSystemProperties = clientCache.getDistributedSystem().getProperties();
assertThat(distributedSystemProperties.getProperty(
DistributedSystemUtils.DURABLE_CLIENT_ID_PROPERTY_NAME),
is(equalTo(DurableClientCacheIntegrationTest.class.getSimpleName())));
Properties distributedSystemProperties = this.clientCache.getDistributedSystem().getProperties();
assertThat(clientCache.getDistributedSystem().getProperties().getProperty(
DistributedSystemUtils.DURABLE_CLIENT_TIMEOUT_PROPERTY_NAME),
is(equalTo(valueBeforeAndAfterDirtiesContext("300", "600"))));
assertThat(distributedSystemProperties.getProperty(DistributedSystemUtils.DURABLE_CLIENT_ID_PROPERTY_NAME))
.isEqualTo(DurableClientCacheIntegrationTest.class.getSimpleName());
assertRegion(example, "Example", DataPolicy.NORMAL);
assertThat(distributedSystemProperties.getProperty(DistributedSystemUtils.DURABLE_CLIENT_TIMEOUT_PROPERTY_NAME))
.isEqualTo(valueBeforeAndAfterDirtiesContext("300", "600"));
assertRegion(this.example, "Example", DataPolicy.NORMAL);
}
@After
public void tearDown() {
if (dirtiesContext()) {
closeApplicationContext();
//closeApplicationContext();
forceCloseClientCache(this.clientCache);
runClientCacheProducer();
setSystemProperties();
}
@@ -171,26 +168,47 @@ public class DurableClientCacheIntegrationTest extends AbstractGemFireClientServ
regionCacheListenerEventValues.clear();
}
protected void closeApplicationContext() {
private void closeApplicationContext() {
applicationContext.close();
this.applicationContext.close();
assertThat(applicationContext.isRunning(), is(false));
assertThat(applicationContext.isActive(), is(false));
assertThat(this.applicationContext.isRunning()).isFalse();
assertThat(this.applicationContext.isActive()).isFalse();
}
protected void runClientCacheProducer() {
private void forceCloseClientCache(ClientCache clientCache) {
if (clientCache != null) {
long timeout = System.currentTimeMillis() + 5000L;
while (timeout > System.currentTimeMillis() && !clientCache.isClosed()) {
clientCache.close(true);
try {
synchronized (clientCache) {
TimeUnit.MILLISECONDS.timedWait(clientCache, 1000L);
}
}
catch (InterruptedException ignore) { }
}
}
}
private void runClientCacheProducer() {
try {
ClientCache gemfireClientCache = new ClientCacheFactory()
ClientCache clientCache = new ClientCacheFactory()
.addPoolServer(SERVER_HOST, serverPort)
.set("name", "ClientCacheProducer")
.set("mcast-port", "0")
.set("log-level", "warning")
.set("log-level", "error")
.create();
Region<String, Integer> exampleRegion = gemfireClientCache.<String, Integer>createClientRegionFactory(
ClientRegionShortcut.PROXY).create("Example");
Region<String, Integer> exampleRegion =
clientCache.<String, Integer>createClientRegionFactory(ClientRegionShortcut.PROXY)
.create("Example");
exampleRegion.put("four", 4);
exampleRegion.put("five", 5);
@@ -201,42 +219,36 @@ public class DurableClientCacheIntegrationTest extends AbstractGemFireClientServ
}
private void setSystemProperties() {
System.setProperty(CLIENT_CACHE_INTERESTS_RESULT_POLICY_SYSTEM_PROPERTY, InterestResultPolicyType.NONE.name());
System.setProperty(DURABLE_CLIENT_TIMEOUT_SYSTEM_PROPERTY, "600");
System.setProperty(CLIENT_CACHE_INTERESTS_RESULT_POLICY, InterestResultPolicyType.NONE.name());
System.setProperty(DURABLE_CLIENT_TIMEOUT, "600");
}
private void assertRegion(Region<?, ?> region, String expectedName, DataPolicy expectedDataPolicy) {
assertRegion(region, expectedName, String.format("%1$s%2$s", Region.SEPARATOR, expectedName),
expectedDataPolicy);
assertRegion(region, expectedName, Region.SEPARATOR + expectedName, expectedDataPolicy);
}
private void assertRegion(Region<?, ?> region, String expectedName, String expectedPath,
DataPolicy expectedDataPolicy) {
assertThat(region, is(notNullValue()));
assertThat(region.getName(), is(equalTo(expectedName)));
assertThat(region.getFullPath(), is(equalTo(expectedPath)));
assertThat(region.getAttributes(), is(notNullValue()));
assertThat(region.getAttributes().getDataPolicy(), is(equalTo(expectedDataPolicy)));
assertThat(region).isNotNull();
assertThat(region.getName()).isEqualTo(expectedName);
assertThat(region.getFullPath()).isEqualTo(expectedPath);
assertThat(region.getAttributes()).isNotNull();
assertThat(region.getAttributes().getDataPolicy()).isEqualTo(expectedDataPolicy);
}
private void assertRegionValues(Region<?, ?> region, Object... values) {
assertThat(region.size(), is(equalTo(values.length)));
assertThat(region.size()).isEqualTo(values.length);
for (Object value : values) {
assertThat(region.containsValue(value), is(true));
assertThat(region.containsValue(value)).isTrue();
}
}
private void waitForRegionEntryEvents() {
ThreadUtils.timedWait(TimeUnit.SECONDS.toMillis(5), TimeUnit.MILLISECONDS.toMillis(500),
new ThreadUtils.WaitCondition() {
@Override public boolean waiting() {
return (regionCacheListenerEventValues.size() < 2);
}
}
);
ThreadUtils.timedWait(5000L, 500L, () -> regionCacheListenerEventValues.size() > 2);
}
@Test
@@ -244,30 +256,25 @@ public class DurableClientCacheIntegrationTest extends AbstractGemFireClientServ
public void durableClientGetsInitializedWithDataOnServer() {
assumeTrue(isBeforeDirtiesContext());
assertRegionValues(example, 1, 2, 3);
assertThat(regionCacheListenerEventValues.isEmpty(), is(true));
assertRegionValues(this.example, 1, 2, 3);
assertThat(regionCacheListenerEventValues.isEmpty()).isTrue();
}
@Test
public void durableClientGetsUpdatesFromServerWhileClientWasOffline() {
assumeTrue(isAfterDirtiesContext());
assertThat(example.isEmpty(), is(true));
assertThat(this.example.isEmpty()).isTrue();
waitForRegionEntryEvents();
assertThat(regionCacheListenerEventValues.size(), is(equalTo(2)));
assertThat(regionCacheListenerEventValues, is(equalTo(Arrays.asList(4, 5))));
assertThat(example.isEmpty(), is(true));
assertThat(regionCacheListenerEventValues.size()).isEqualTo(2);
assertThat(regionCacheListenerEventValues).containsExactly(4, 5);
assertThat(this.example.isEmpty()).isTrue();
}
public static class ClientCacheBeanPostProcessor implements BeanPostProcessor {
@Override
public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
return bean;
}
@Override
public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
@@ -276,13 +283,13 @@ public class DurableClientCacheIntegrationTest extends AbstractGemFireClientServ
Pool gemfireServerPool = (Pool) bean;
if (isBeforeDirtiesContext()) {
// NOTE: A value of -2 indicates the client connected to the server for the first time...
assertThat(gemfireServerPool.getPendingEventCount(), is(equalTo(-2)));
// NOTE: A value of -2 indicates the client connected to the server for the first time.
assertThat(gemfireServerPool.getPendingEventCount()).isEqualTo(-2);
}
else {
// NOTE: the pending event count could be 3 because it should minimally include the 2 puts
// from the client cache producer and possibly a "marker" as well...
assertThat(gemfireServerPool.getPendingEventCount(), is(greaterThanOrEqualTo(2)));
// NOTE: The pending event count could be 3 because it should minimally include the 2 puts
// from the ClientCAche producer and possibly a "marker" as well.
assertThat(gemfireServerPool.getPendingEventCount()).isGreaterThanOrEqualTo(2);
}
}
@@ -306,26 +313,23 @@ public class DurableClientCacheIntegrationTest extends AbstractGemFireClientServ
}
protected Map<K, V> getRegionData() {
Assert.state(regionData != null, "Region data was not properly initialized");
return regionData;
Assert.state(this.regionData != null, "Region data was not provided");
return this.regionData;
}
protected String getRegionName() {
return regionName;
return this.regionName;
}
protected void loadData(Region<K, V> region) {
region.putAll(getRegionData());
}
@Override
public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
return bean;
}
@Override
public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
if (bean instanceof Region) {
Region<K, V> region = (Region) bean;
if (getRegionName().equals(region.getName())) {

40
src/test/java/org/springframework/data/gemfire/config/annotation/EnableEvictionConfigurationUnitTests.java
View File

@@ -34,7 +34,6 @@ import org.apache.geode.cache.RegionFactory;
import org.apache.geode.cache.util.ObjectSizer;
import org.junit.After;
import org.junit.Test;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
@@ -177,28 +176,25 @@ public class EnableEvictionConfigurationUnitTests {
@Bean("mockCache")
@SuppressWarnings("unchecked")
Cache mockCache() {
Cache mockCache = mock(Cache.class);
RegionFactory mockRegionFactory = mock(RegionFactory.class);
final AtomicReference<EvictionAttributes> evictionAttributes =
new AtomicReference<>(null);
AtomicReference<EvictionAttributes> evictionAttributes = new AtomicReference<>(null);
when(mockCache.createRegionFactory()).thenReturn(mockRegionFactory);
when(mockRegionFactory.setEvictionAttributes(any(EvictionAttributes.class))).thenAnswer(
new Answer<RegionFactory>() {
@Override
public RegionFactory answer(InvocationOnMock invocation) throws Throwable {
evictionAttributes.set(invocation.getArgument(0));
return (RegionFactory) invocation.getMock();
}
when(mockRegionFactory.setEvictionAttributes(any(EvictionAttributes.class)))
.thenAnswer((Answer<RegionFactory>) invocation -> {
evictionAttributes.set(invocation.getArgument(0));
return (RegionFactory) invocation.getMock();
}
);
when(mockRegionFactory.create(anyString())).thenAnswer(new Answer<Region>() {
@Override
public Region answer(InvocationOnMock invocation) throws Throwable {
when(mockRegionFactory.create(anyString()))
.thenAnswer((Answer<Region>) invocation -> {
String regionName = invocation.getArgument(0);
Region mockRegion = mock(Region.class, regionName);
@@ -212,14 +208,14 @@ public class EnableEvictionConfigurationUnitTests {
when(mockRegionAttributes.getEvictionAttributes()).thenReturn(evictionAttributes.get());
return mockRegion;
}
});
});
return mockCache;
}
@Bean("PartitionRegion")
PartitionedRegionFactoryBean<Object, Object> mockPartitionRegion(Cache gemfireCache) {
PartitionedRegionFactoryBean<Object, Object> partitionRegion =
new PartitionedRegionFactoryBean<>();
@@ -232,6 +228,7 @@ public class EnableEvictionConfigurationUnitTests {
@Bean("ReplicateRegion")
ReplicatedRegionFactoryBean<Object, Object> mockReplicateRegion(Cache gemfireCache) {
ReplicatedRegionFactoryBean<Object, Object> replicateRegion =
new ReplicatedRegionFactoryBean<>();
@@ -249,13 +246,11 @@ public class EnableEvictionConfigurationUnitTests {
}
@EnableEviction
static class DefaultEvictionPolicyConfiguration extends CacheRegionConfiguration {
}
static class DefaultEvictionPolicyConfiguration extends CacheRegionConfiguration { }
@EnableEviction(policies = @EvictionPolicy(maximum = 65536, type = EvictionPolicyType.MEMORY_SIZE,
action = EvictionActionType.OVERFLOW_TO_DISK, objectSizerName = "mockObjectSizer"))
static class CustomEvictionPolicyConfiguration extends CacheRegionConfiguration {
}
static class CustomEvictionPolicyConfiguration extends CacheRegionConfiguration { }
@EnableEviction(policies = {
@EvictionPolicy(maximum = 85, type = EvictionPolicyType.HEAP_PERCENTAGE, action = EvictionActionType.OVERFLOW_TO_DISK,
@@ -263,14 +258,13 @@ public class EnableEvictionConfigurationUnitTests {
@EvictionPolicy(maximum = 10000, type = EvictionPolicyType.ENTRY_COUNT, action = EvictionActionType.LOCAL_DESTROY,
regionNames = "ReplicateRegion")
})
static class RegionSpecificEvictionPolicyConfiguration extends CacheRegionConfiguration {
}
static class RegionSpecificEvictionPolicyConfiguration extends CacheRegionConfiguration { }
@EnableEviction(policies = {
@EvictionPolicy(maximum = 1, type = EvictionPolicyType.ENTRY_COUNT, action = EvictionActionType.LOCAL_DESTROY,
objectSizerName = "mockObjectSizer", regionNames = "ReplicateRegion"),
@EvictionPolicy(maximum = 99, type = EvictionPolicyType.ENTRY_COUNT, action = EvictionActionType.OVERFLOW_TO_DISK)
})
static class LastMatchingWinsEvictionPolicyConfiguration extends CacheRegionConfiguration {
}
static class LastMatchingWinsEvictionPolicyConfiguration extends CacheRegionConfiguration { }
}

81
src/test/java/org/springframework/data/gemfire/test/support/AbstractGemFireClientServerIntegrationTest.java
View File

@@ -29,8 +29,8 @@ import org.springframework.data.gemfire.process.ProcessWrapper;
import org.springframework.util.Assert;
/**
* The AbstractGemFireClientServerIntegrationTest class is an abstract test suite base class encapsulating functionality
* common to all test classes implementing GemFire client/server test cases.
* The {@link AbstractGemFireClientServerIntegrationTest} class is an abstract test suite base class
* encapsulating functionality common to all test classes implementing client/server test cases.
*
* @author John Blum
* @see org.springframework.data.gemfire.fork.ServerProcess
@@ -49,17 +49,13 @@ public abstract class AbstractGemFireClientServerIntegrationTest {
protected static String CLEAN_PROCESS_WORKING_DIRECTORY_SYSTEM_PROPERTY = "spring.data.gemfire.force.clean";
protected static void pause(final long duration) {
ThreadUtils.timedWait(Math.max(duration, ONE_SECOND_IN_MILLISECONDS), ONE_SECOND_IN_MILLISECONDS,
new ThreadUtils.WaitCondition() {
@Override public boolean waiting() {
return true;
}
}
);
ThreadUtils.timedWait(Math.max(duration, ONE_SECOND_IN_MILLISECONDS), ONE_SECOND_IN_MILLISECONDS, () -> true);
}
protected static <T> T setSystemProperty(String propertyName, T propertyValue) {
System.setProperty(propertyName, String.valueOf(propertyValue));
return propertyValue;
}
@@ -70,61 +66,71 @@ public abstract class AbstractGemFireClientServerIntegrationTest {
protected static ProcessWrapper startGemFireServer(Class<?> testClass, long waitTimeInMilliseconds)
throws IOException {
Assert.notNull(testClass, "'testClass' must not be null");
Assert.notNull(testClass, "testClass must not be null");
String serverName = (testClass.getSimpleName() + "Server");
String serverName = testClass.getSimpleName() + "Server";
File serverWorkingDirectory = new File(FileSystemUtils.WORKING_DIRECTORY, serverName.toLowerCase());
Assert.isTrue(serverWorkingDirectory.isDirectory() || serverWorkingDirectory.mkdirs());
List<String> arguments = new ArrayList<String>();
List<String> arguments = new ArrayList<>();
addTestClassSystemProperties(testClass, arguments)
.add(String.format("-Dgemfire.name=%1$s", serverName));
addTestClassSystemProperties(testClass, arguments).add(String.format("-Dgemfire.name=%1$s", serverName));
arguments.add("/".concat(testClass.getName().replace(".", "/").concat("-server-context.xml")));
ProcessWrapper gemfireServerProcess = ProcessExecutor.launch(serverWorkingDirectory, ServerProcess.class,
arguments.toArray(new String[arguments.size()]));
ProcessWrapper gemfireServerProcess =
ProcessExecutor.launch(serverWorkingDirectory, ServerProcess.class, arguments.toArray(new String[0]));
waitForServerToStart(gemfireServerProcess, waitTimeInMilliseconds);
System.out.printf("The Spring-based, GemFire Cache Server process for %1$s should be running...%n",
testClass.getSimpleName());
return gemfireServerProcess;
}
static List<String> addTestClassSystemProperties(Class<?> testClass, List<String> arguments) {
String testClassName = testClass.getName();
for (String propertyName : System.getProperties().stringPropertyNames()) {
if (propertyName.startsWith(testClassName)) {
arguments.add(String.format("-D%1$s=%2$s", propertyName, System.getProperty(propertyName)));
}
}
System.getProperties().stringPropertyNames().stream()
.filter(propertyName -> propertyName.startsWith(testClassName))
.forEach(propertyName -> {
String argument = String.format("-D%1$s=%2$s", propertyName, System.getProperty(propertyName));
arguments.add(argument);
});
return arguments;
}
static void waitForServerToStart(final ProcessWrapper process, long duration) {
ThreadUtils.timedWait(Math.max(duration, FIVE_HUNDRED_MILLISECONDS), FIVE_HUNDRED_MILLISECONDS,
new ThreadUtils.WaitCondition() {
private File processPidControlFile = new File(process.getWorkingDirectory(),
ServerProcess.getServerProcessControlFilename());
@Override public boolean waiting() {
return !processPidControlFile.isFile();
ThreadUtils.timedWait(Math.max(duration, FIVE_HUNDRED_MILLISECONDS), FIVE_HUNDRED_MILLISECONDS,
new ThreadUtils.WaitCondition() {
private File processPidFile =
new File(process.getWorkingDirectory(), ServerProcess.getServerProcessControlFilename());
@Override
public boolean waiting() {
return !processPidFile.isFile();
}
}
);
}
protected static ProcessWrapper stopGemFireServer(ProcessWrapper process) {
try {
process.shutdown();
boolean springGemFireForceClean = Boolean.valueOf(System.getProperty(
CLEAN_PROCESS_WORKING_DIRECTORY_SYSTEM_PROPERTY, Boolean.TRUE.toString()));
boolean springGemFireForceClean =
Boolean.valueOf(System.getProperty(CLEAN_PROCESS_WORKING_DIRECTORY_SYSTEM_PROPERTY,
String.valueOf(true)));
if (springGemFireForceClean) {
org.springframework.util.FileSystemUtils.deleteRecursively(process.getWorkingDirectory());
@@ -132,18 +138,17 @@ public abstract class AbstractGemFireClientServerIntegrationTest {
return null;
}
catch (Exception e) {
catch (Exception ignore) {
return process;
}
}
protected static void clearSystemProperties(Class<?> testClass) {
protected static void clearTestClassSystemProperties(Class<?> testClass) {
String testClassName = testClass.getName();
for (String propertyName : System.getProperties().stringPropertyNames()) {
if (propertyName.startsWith(testClassName)) {
System.clearProperty(propertyName);
}
}
System.getProperties().stringPropertyNames().stream()
.filter(propertyName -> propertyName.startsWith(testClassName))
.forEach(propertyName -> System.clearProperty(propertyName));
}
}