Fixed JIRA issue SGF-197 involving bean ordering issues when GemFire is configured with a PDX persistent Disk Store and persistent Regions contain data (keys) based on PDX types.

2013-10-17 17:23:19 -07:00
parent e6e1573f0c
commit fd35006c5e
8 changed files with 736 additions and 2 deletions
--- a/src/main/java/org/springframework/data/gemfire/config/CacheParser.java
+++ b/src/main/java/org/springframework/data/gemfire/config/CacheParser.java
@@ -21,10 +21,12 @@ import java.util.List;
 import org.springframework.beans.factory.BeanDefinitionStoreException;
 import org.springframework.beans.factory.support.AbstractBeanDefinition;
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.support.BeanDefinitionRegistry;
 import org.springframework.beans.factory.support.ManagedList;
 import org.springframework.beans.factory.support.ManagedMap;
 import org.springframework.beans.factory.xml.AbstractSimpleBeanDefinitionParser;
 import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.data.gemfire.CacheFactoryBean;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.StringUtils;
@@ -55,7 +57,7 @@ class CacheParser extends AbstractSimpleBeanDefinitionParser {
 		ParsingUtils.setPropertyValue(element, builder, "cache-xml-location", "cacheXml");
 		ParsingUtils.setPropertyReference(element, builder, "properties-ref", "properties");
 		ParsingUtils.setPropertyReference(element, builder, "pdx-serializer-ref", "pdxSerializer");
-		ParsingUtils.setPropertyValue(element, builder, "pdx-disk-store", "pdxDiskStoreName");
+		parsePdxDiskStore(element, parserContext, builder);
 		ParsingUtils.setPropertyValue(element, builder, "pdx-persistent");
 		ParsingUtils.setPropertyValue(element, builder, "pdx-read-serialized");
 		ParsingUtils.setPropertyValue(element, builder, "pdx-ignore-unread-fields");
@@ -103,6 +105,21 @@ class CacheParser extends AbstractSimpleBeanDefinitionParser {
 		parseJndiBindings(element, builder);
 	}

+	private void parsePdxDiskStore(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
+		ParsingUtils.setPropertyValue(element, builder, "pdx-disk-store", "pdxDiskStoreName");
+
+		final String pdxDiskStoreName = element.getAttribute("pdx-disk-store");
+
+		if (!StringUtils.isEmpty(pdxDiskStoreName)) {
+			final BeanDefinitionRegistry registry = parserContext.getRegistry();
+
+			if (registry instanceof ConfigurableApplicationContext) {
+				((ConfigurableApplicationContext) registry).addBeanFactoryPostProcessor(
+					new PdxDiskStoreAwareBeanFactoryPostProcessor(pdxDiskStoreName));
+			}
+		}
+	}
+
 	private void parseDynamicRegionFactory(Element element, BeanDefinitionBuilder builder) {
 		Element dynamicRegionFactory = DomUtils.getChildElementByTagName(element, "dynamic-region-factory");
 		if (dynamicRegionFactory != null) {
@@ -188,4 +205,4 @@ class CacheParser extends AbstractSimpleBeanDefinitionParser {
 		return name;
 	}

-}
+}
--- a/src/main/java/org/springframework/data/gemfire/config/PdxDiskStoreAwareBeanFactoryPostProcessor.java
+++ b/src/main/java/org/springframework/data/gemfire/config/PdxDiskStoreAwareBeanFactoryPostProcessor.java
@@ -0,0 +1,127 @@
+/*
+ * Copyright 2010-2013 the original author or authors.
+ *
+ * Licensed 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.springframework.data.gemfire.config;
+
+import org.springframework.beans.BeansException;
+import org.springframework.beans.PropertyValue;
+import org.springframework.beans.factory.config.BeanDefinition;
+import org.springframework.beans.factory.config.BeanFactoryPostProcessor;
+import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
+import org.springframework.beans.factory.support.AbstractBeanDefinition;
+import org.springframework.data.gemfire.DiskStoreFactoryBean;
+import org.springframework.data.gemfire.RegionLookupFactoryBean;
+import org.springframework.data.gemfire.util.ArrayUtils;
+import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
+
+/**
+ * The PdxDiskStoreAwareBeanFactoryPostProcessor class is a BeanFactoryPostProcessor that modifies all Region bean
+ * definitions in the Spring BeanFactory to form a dependency on the Cache's PDX Disk Store bean.  A persistent Region
+ * may contain PDX typed data, in which case, the PDX type meta-data stored to disk needs to be loaded before the Region
+ * having PDX data is loaded from disk.
+ * <p/>
+ * @author John Blum
+ * @see org.springframework.beans.factory.config.BeanFactoryPostProcessor
+ * @see org.springframework.beans.factory.config.ConfigurableListableBeanFactory
+ * @since 1.3.3
+ */
+@SuppressWarnings("unused")
+public class PdxDiskStoreAwareBeanFactoryPostProcessor implements BeanFactoryPostProcessor {
+
+	private static final String[] EMPTY_STRING_ARRAY = new String[0];
+
+	private String pdxDiskStoreName;
+
+	public PdxDiskStoreAwareBeanFactoryPostProcessor() {
+	}
+
+	public PdxDiskStoreAwareBeanFactoryPostProcessor(final String pdxDiskStoreName) {
+		setPdxDiskStoreName(pdxDiskStoreName);
+	}
+
+	public String getPdxDiskStoreName() {
+		Assert.state(!StringUtils.isEmpty(pdxDiskStoreName), "The PDX Disk Store name was not properly initialized!");
+		return pdxDiskStoreName;
+	}
+
+	public final void setPdxDiskStoreName(final String pdxDiskStoreName) {
+		this.pdxDiskStoreName = pdxDiskStoreName;
+	}
+
+	@Override
+	public void postProcessBeanFactory(final ConfigurableListableBeanFactory beanFactory) throws BeansException {
+		for (String beanName : beanFactory.getBeanDefinitionNames()) {
+			// NOTE do not add the PDX Disk Store bean dependency to itself!
+			if (!beanName.equalsIgnoreCase(getPdxDiskStoreName())) {
+				BeanDefinition beanDefinition = beanFactory.getBeanDefinition(beanName);
+
+				// NOTE an optimization to only inspect Region bean definitions for persistent Regions (NOTE, the
+				// persistent Region might also define a Disk Store, or would depend on the DEFAULT Disk Store,
+				// which may depend on PDX type meta-data).
+				// NOTE add the dependency to the Disk Store bean definition in case the Disk Store bean is defined
+				// before the Region that uses it!
+				// TODO what else depends on a PDX Disk Store besides Regions?
+				if (isPersistentRegion(beanDefinition) || isDiskStore(beanDefinition)) {
+					addPdxDiskStoreDependency(beanDefinition);
+				}
+			}
+		}
+	}
+
+	private boolean isDiskStore(final BeanDefinition beanDefinition) {
+		return (beanDefinition instanceof AbstractBeanDefinition
+			&& isDiskStore((AbstractBeanDefinition) beanDefinition));
+	}
+
+	private boolean isDiskStore(final AbstractBeanDefinition beanDefinition) {
+		return (beanDefinition.hasBeanClass()
+			&& DiskStoreFactoryBean.class.isAssignableFrom(beanDefinition.getBeanClass()));
+	}
+
+	private boolean isPersistentRegion(final BeanDefinition beanDefinition) {
+		return (beanDefinition instanceof AbstractBeanDefinition
+			&& isPersistentRegion((AbstractBeanDefinition) beanDefinition));
+	}
+
+	private boolean isPersistentRegion(final AbstractBeanDefinition beanDefinition) {
+		return (isRegion(beanDefinition) && isPersistent(beanDefinition));
+	}
+
+	// NOTE Class.isAssignableFrom is not null-safe, hence the AbstractBeanDefinition.hasBeanClass call!
+	private boolean isRegion(final AbstractBeanDefinition beanDefinition) {
+		return (beanDefinition.hasBeanClass()
+			&& RegionLookupFactoryBean.class.isAssignableFrom(beanDefinition.getBeanClass()));
+	}
+
+	private boolean isPersistent(final AbstractBeanDefinition beanDefinition) {
+		PropertyValue persistentPropertyValue = beanDefinition.getPropertyValues().getPropertyValue("persistent");
+
+		return (persistentPropertyValue != null
+			&& Boolean.parseBoolean(String.valueOf(persistentPropertyValue.getValue())));
+	}
+
+	private void addPdxDiskStoreDependency(final BeanDefinition beanDefinition) {
+		String[] newDependsOn = (String[]) ArrayUtils.insert(getDependsOn(beanDefinition), 0, getPdxDiskStoreName());
+
+		beanDefinition.setDependsOn(newDependsOn);
+	}
+
+	private String[] getDependsOn(final BeanDefinition beanDefinition) {
+		return (beanDefinition.getDependsOn() != null ? beanDefinition.getDependsOn() : EMPTY_STRING_ARRAY);
+	}
+
+}
--- a/src/test/java/org/springframework/data/gemfire/PdxDiskStoreIntegrationTest.java
+++ b/src/test/java/org/springframework/data/gemfire/PdxDiskStoreIntegrationTest.java
@@ -0,0 +1,228 @@
+/*
+ * Copyright 2010-2013 the original author or authors.
+ *
+ * Licensed 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.springframework.data.gemfire;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import java.io.File;
+import java.io.Serializable;
+import javax.annotation.Resource;
+
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+import org.springframework.util.Assert;
+
+import com.gemstone.gemfire.cache.Region;
+
+/**
+ * The PdxDiskStoreTest class is a test suite containing tests to reproduce the issue in JIRA SGF-197.
+ * <p/>
+ * @author John Blum
+ * @link https://jira.springsource.org/browse/SGF-197
+ * @see org.junit.Test
+ * @see org.junit.runner.RunWith
+ * @see org.springframework.test.annotation.DirtiesContext
+ * @see org.springframework.test.context.ContextConfiguration
+ * @see org.springframework.test.context.junit4.SpringJUnit4ClassRunner
+ * @since 1.3.3
+ */
+@ContextConfiguration("/org/springframework/data/gemfire/pdxdiskstore-config.xml")
+@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_EACH_TEST_METHOD)
+@RunWith(SpringJUnit4ClassRunner.class)
+public class PdxDiskStoreIntegrationTest {
+
+	protected static final int NUMBER_OF_REGION_ENTRIES = 1000;
+
+	@Resource(name = "pdxDataRegion")
+	private Region<KeyHolder<String>, ValueHolder<Integer>> pdxDataRegion;
+
+	protected static void assertRegionExists(final String expectedRegionName, final String expectedRegionPath, final Region region) {
+		assertNotNull(region);
+		assertEquals(String.format("Expected Region with name %1$s; but was %2$s!",
+			expectedRegionName, region.getName()), expectedRegionName, region.getName());
+		assertEquals(String.format("Expected Region with path %1$s; but was %2$s!",
+			expectedRegionPath, region.getFullPath()), expectedRegionPath, region.getFullPath());
+	}
+
+	protected static boolean createDirectory(final File path) {
+		return (path != null && (path.isDirectory() || path.mkdirs()));
+	}
+
+	protected static File createFile(final String pathname) {
+		return new File(pathname);
+	}
+
+	protected static void deleteRecursive(final File path) {
+		if (path.isDirectory()) {
+			for (File file : path.listFiles()) {
+				deleteRecursive(file);
+			}
+		}
+
+		path.delete();
+	}
+
+	@BeforeClass
+	public static void setupBeforeClass() {
+		assertTrue(createDirectory(createFile("./gemfire/data-store")));
+		assertTrue(createDirectory(createFile("./gemfire/pdx-store")));
+	}
+
+	@AfterClass
+	public static void tearDownAfterClass() {
+		deleteRecursive(createFile("./gemfire"));
+	}
+
+	@Before
+	public void setup() {
+		assertNotNull("The PdxData GemFire Region was not created successfully!", pdxDataRegion);
+
+		if (pdxDataRegion.size() == 0) {
+			System.out.printf("Creating entries for Region (%1$s)...%n", pdxDataRegion.getName());
+			for (int index = 1; index <= NUMBER_OF_REGION_ENTRIES; index++) {
+				pdxDataRegion.put(new KeyHolder<String>("key" + index), new ValueHolder<Integer>(index));
+			}
+		}
+	}
+
+	@Test
+	public void testPersistentRegionWithDataCreation() {
+		assertRegionExists("PdxData", "/PdxData", pdxDataRegion);
+		assertEquals(NUMBER_OF_REGION_ENTRIES, pdxDataRegion.size());
+	}
+
+	@Test
+	public void testPersistentRegionWithDataRecovery() {
+		assertRegionExists("PdxData", "/PdxData", pdxDataRegion);
+		assertEquals(NUMBER_OF_REGION_ENTRIES, pdxDataRegion.size());
+	}
+
+	protected static class AbstractHolderSupport {
+
+		protected static boolean equals(final Object obj1, final Object obj2) {
+			return (obj1 != null && obj1.equals(obj2));
+		}
+
+		protected static int hashCode(final Object obj) {
+			return (obj == null ? 0 : obj.hashCode());
+		}
+	}
+
+	public static class KeyHolder<T extends Serializable> extends AbstractHolderSupport {
+
+		private T key;
+
+		public KeyHolder() {
+		}
+
+		public KeyHolder(final T key) {
+			Assert.notNull(key, "The key cannot be null!");
+			this.key = key;
+		}
+
+		public T getKey() {
+			return key;
+		}
+
+		public void setKey(final T key) {
+			this.key = key;
+		}
+
+		@Override
+		public boolean equals(final Object obj) {
+			if (obj == this) {
+				return true;
+			}
+
+			if (!(obj instanceof KeyHolder)) {
+				return false;
+			}
+
+			final KeyHolder that = (KeyHolder) obj;
+
+			return equals(this.getKey(), that.getKey());
+		}
+
+		@Override
+		public int hashCode() {
+			int hashValue = 17;
+			hashValue = 37 * hashValue + hashCode(this.getKey());
+			return hashValue;
+		}
+
+		@Override
+		public String toString() {
+			return String.valueOf(getKey());
+		}
+	}
+
+	public static class ValueHolder<T extends Serializable> extends AbstractHolderSupport {
+
+		private T value;
+
+		public ValueHolder() {
+		}
+
+		public ValueHolder(final T value) {
+			this.value = value;
+		}
+
+		public T getValue() {
+			return value;
+		}
+
+		public void setValue(final T value) {
+			this.value = value;
+		}
+
+		@Override
+		public boolean equals(final Object obj) {
+			if (obj == this) {
+				return true;
+			}
+
+			if (!(obj instanceof ValueHolder)) {
+				return false;
+			}
+
+			final ValueHolder that = (ValueHolder) obj;
+
+			return equals(this.getValue(), that.getValue());
+		}
+
+		@Override
+		public int hashCode() {
+			int hashValue = 17;
+			hashValue = 17 * hashValue + hashCode(this.getValue());
+			return hashValue;
+		}
+
+		@Override
+		public String toString() {
+			return String.valueOf(getValue());
+		}
+	}
+
+}
--- a/src/test/java/org/springframework/data/gemfire/config/CacheUsingPdxNamespaceTest.java
+++ b/src/test/java/org/springframework/data/gemfire/config/CacheUsingPdxNamespaceTest.java
@@ -0,0 +1,94 @@
+/*
+ * Copyright 2010-2013 the original author or authors.
+ *
+ * Licensed 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.springframework.data.gemfire.config;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assume.assumeTrue;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.config.BeanFactoryPostProcessor;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.support.AbstractApplicationContext;
+import org.springframework.data.gemfire.CacheFactoryBean;
+import org.springframework.data.gemfire.test.GemfireTestApplicationContextInitializer;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+import com.gemstone.gemfire.pdx.PdxSerializer;
+
+/**
+ * The CacheUsingPdxNamespaceTest class is a test suite of test case testing the Spring Data GemFire XML namespace
+ * when PDX is configured in GemFire.
+ * <p/>
+ * @author John Blum
+ * @see org.junit.Test
+ * @see org.junit.runner.RunWith
+ * @see org.springframework.test.context.ContextConfiguration
+ * @see org.springframework.test.context.junit4.SpringJUnit4ClassRunner
+ * @since 1.3.3
+ */
+@ContextConfiguration(locations = "/org/springframework/data/gemfire/config/cache-using-pdx-ns.xml",
+	initializers = GemfireTestApplicationContextInitializer.class)
+@RunWith(SpringJUnit4ClassRunner.class)
+@SuppressWarnings("unused")
+public class CacheUsingPdxNamespaceTest {
+
+	@Autowired
+	private ApplicationContext context;
+
+	protected PdxDiskStoreAwareBeanFactoryPostProcessor getPdxDiskStoreAwareBeanFactoryPostProcessor(AbstractApplicationContext context) {
+		for (BeanFactoryPostProcessor postProcessor : context.getBeanFactoryPostProcessors()) {
+			if (postProcessor instanceof PdxDiskStoreAwareBeanFactoryPostProcessor) {
+				return (PdxDiskStoreAwareBeanFactoryPostProcessor) postProcessor;
+			}
+		}
+
+		return null;
+	}
+
+	@Test
+	public void testApplicationContextHasPdxDiskStoreAwareBeanFactoryPostProcessor() {
+		assumeTrue(context instanceof AbstractApplicationContext);
+
+		final PdxDiskStoreAwareBeanFactoryPostProcessor postProcessor = getPdxDiskStoreAwareBeanFactoryPostProcessor(
+			(AbstractApplicationContext) context);
+
+		assertNotNull(postProcessor);
+		assertEquals("pdxStore", postProcessor.getPdxDiskStoreName());
+	}
+
+	@Test
+	public void testCachePdxConfiguration() {
+		final CacheFactoryBean cacheFactoryBean = context.getBean("&gemfireCache", CacheFactoryBean.class);
+
+		assertNotNull(cacheFactoryBean);
+		assertEquals("pdxStore", cacheFactoryBean.getPdxDiskStoreName());
+		assertTrue(Boolean.TRUE.equals(cacheFactoryBean.getPdxPersistent()));
+		assertTrue(Boolean.TRUE.equals(cacheFactoryBean.getPdxReadSerialized()));
+
+		final PdxSerializer autoSerializer = context.getBean("autoSerializer", PdxSerializer.class);
+
+		assertNotNull(autoSerializer);
+		assertSame(autoSerializer, cacheFactoryBean.getPdxSerializer());
+	}
+
+}
--- a/src/test/java/org/springframework/data/gemfire/config/PdxDiskStoreAwareBeanFactoryPostProcessorTest.java
+++ b/src/test/java/org/springframework/data/gemfire/config/PdxDiskStoreAwareBeanFactoryPostProcessorTest.java
@@ -0,0 +1,193 @@
+/*
+ * Copyright 2010-2013 the original author or authors.
+ *
+ * Licensed 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.springframework.data.gemfire.config;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Matchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.junit.Test;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+import org.springframework.beans.factory.config.BeanDefinition;
+import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.data.gemfire.CacheFactoryBean;
+import org.springframework.data.gemfire.DiskStoreFactoryBean;
+import org.springframework.data.gemfire.PartitionedRegionFactoryBean;
+import org.springframework.data.gemfire.ReplicatedRegionFactoryBean;
+
+/**
+ * The PdxDiskStoreAwareBeanFactoryPostProcessorTest class is a test suite of test cases testing the functionality
+ * of the PdxDiskStoreAwareBeanFactoryPostProcessor class.
+ * <p/>
+ * @author John Blum
+ * @see org.mockito.Mockito
+ * @see org.junit.Test
+ * @see org.springframework.data.gemfire.config.PdxDiskStoreAwareBeanFactoryPostProcessor
+ * @since 1.3.3
+ */
+public class PdxDiskStoreAwareBeanFactoryPostProcessorTest {
+
+	protected static final boolean PERSISTENT = true;
+	protected static final boolean NOT_PERSISTENT = false;
+
+	protected static String[] toStringArray(final Collection<String> collection) {
+		return collection.toArray(new String[collection.size()]);
+	}
+
+	protected static boolean isEmpty(final Object[] array) {
+		return (array == null || array.length == 0);
+	}
+
+	protected ConfigurableListableBeanFactory createMockBeanFactory(final Map<String, BeanDefinition> beanDefinitions) {
+		final ConfigurableListableBeanFactory mockBeanFactory = mock(ConfigurableListableBeanFactory.class);
+
+		when(mockBeanFactory.getBeanDefinitionNames()).thenReturn(toStringArray(beanDefinitions.keySet()));
+
+		when(mockBeanFactory.getBeanDefinition(anyString())).then(new Answer<BeanDefinition>() {
+			@Override
+			public BeanDefinition answer(final InvocationOnMock invocation) throws Throwable {
+				final Object[] arguments = invocation.getArguments();
+				assertNotNull(arguments);
+				assertTrue(arguments.length > 0);
+				return beanDefinitions.get(String.valueOf(arguments[0]));
+			}
+		});
+
+		return mockBeanFactory;
+	}
+
+	protected static BeanDefinitionBuilder createBeanDefinitionBuilder(Object beanClassObject, String... dependencies) {
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition();
+
+		if (beanClassObject instanceof Class) {
+			builder.getRawBeanDefinition().setBeanClass((Class) beanClassObject);
+		}
+		else {
+			builder.getRawBeanDefinition().setBeanClassName(String.valueOf(beanClassObject));
+		}
+
+		return addDependsOn(builder, dependencies);
+	}
+
+	protected static BeanDefinitionBuilder addDependsOn(BeanDefinitionBuilder builder, String... dependencies) {
+		for (String dependency : dependencies) {
+			builder.addDependsOn(dependency);
+		}
+
+		return builder;
+	}
+
+	protected static void assertDependencies(BeanDefinition beanDefinition, String... expectedDependencies) {
+		assertFalse(isEmpty(beanDefinition.getDependsOn()));
+		assertTrue(Arrays.asList(beanDefinition.getDependsOn()).equals(Arrays.asList(expectedDependencies)));
+	}
+
+	protected BeanDefinition defineBean(String beanClassName, String... dependencies) {
+		return createBeanDefinitionBuilder(beanClassName, dependencies).getBeanDefinition();
+	}
+
+	protected BeanDefinition defineCache() {
+		return createBeanDefinitionBuilder(CacheFactoryBean.class).getBeanDefinition();
+	}
+
+	protected BeanDefinition defineDiskStore(String... dependencies) {
+		return createBeanDefinitionBuilder(DiskStoreFactoryBean.class, dependencies).getBeanDefinition();
+	}
+
+	protected BeanDefinition defineRegion(Class beanClass, boolean persistent, String... dependencies) {
+		BeanDefinitionBuilder builder = createBeanDefinitionBuilder(beanClass, dependencies);
+		builder.addPropertyValue("persistent", persistent);
+		return builder.getBeanDefinition();
+	}
+
+	protected BeanDefinition definePartitionedRegion(boolean persistent, String... dependencies) {
+		return defineRegion(PartitionedRegionFactoryBean.class, persistent, dependencies);
+	}
+
+	protected BeanDefinition defineReplicatedRegion(boolean persistent, String... dependencies) {
+		return defineRegion(ReplicatedRegionFactoryBean.class, persistent, dependencies);
+	}
+
+	@Test
+	public void testInitializedPdxDiskStoreAwareBeanFactoryPostProcessor() {
+		final PdxDiskStoreAwareBeanFactoryPostProcessor postProcessor =
+			new PdxDiskStoreAwareBeanFactoryPostProcessor("testPdxDiskStoreName");
+
+		assertNotNull(postProcessor);
+		assertEquals("testPdxDiskStoreName", postProcessor.getPdxDiskStoreName());
+
+		postProcessor.setPdxDiskStoreName("mockPdxDiskStoreName");
+
+		assertEquals("mockPdxDiskStoreName", postProcessor.getPdxDiskStoreName());
+	}
+
+	@Test(expected = IllegalStateException.class)
+	public void testUninitializedPdxDiskStoreAwareBeanFactoryPostProcessor() {
+		new PdxDiskStoreAwareBeanFactoryPostProcessor().getPdxDiskStoreName();
+	}
+
+	@Test
+	public void testPostProcessBeanFactory() {
+		final Map<String, BeanDefinition> beanDefinitions = new HashMap<String, BeanDefinition>(13);
+
+		beanDefinitions.put("someBean", defineBean("org.company.app.domain.SomeBean", "someOtherBean"));
+		beanDefinitions.put("gemfireCache", defineCache());
+		beanDefinitions.put("pdxDiskStore", defineDiskStore());
+		beanDefinitions.put("someOtherBean", defineBean("org.company.app.domain.SomeOtherBean"));
+		beanDefinitions.put("overflowDiskStore", defineDiskStore());
+		beanDefinitions.put("region1", defineReplicatedRegion(NOT_PERSISTENT, "overflowDiskStore"));
+		beanDefinitions.put("region2DiskStore", defineDiskStore("someBean"));
+		beanDefinitions.put("region2", defineReplicatedRegion(PERSISTENT, "region2DiskStore"));
+		beanDefinitions.put("colocatedRegion", definePartitionedRegion(NOT_PERSISTENT, "residentRegion",
+			"overflowDiskStore"));
+		beanDefinitions.put("residentRegionDiskStore", defineDiskStore("someBean", "yetAnotherBean"));
+		beanDefinitions.put("residentRegion", definePartitionedRegion(PERSISTENT, "residentRegionDiskStore"));
+		beanDefinitions.put("yetAnotherBean", defineBean("org.company.app.domain.YetAnotherBean", "someBean"));
+
+		final ConfigurableListableBeanFactory mockBeanFactory = createMockBeanFactory(beanDefinitions);
+
+		final PdxDiskStoreAwareBeanFactoryPostProcessor postProcessor =
+			new PdxDiskStoreAwareBeanFactoryPostProcessor("pdxDiskStore");
+
+		postProcessor.postProcessBeanFactory(mockBeanFactory);
+
+		assertDependencies(beanDefinitions.get("someBean"), "someOtherBean");
+		assertTrue(isEmpty(beanDefinitions.get("gemfireCache").getDependsOn()));
+		assertTrue(isEmpty(beanDefinitions.get("pdxDiskStore").getDependsOn()));
+		assertTrue(isEmpty(beanDefinitions.get("someOtherBean").getDependsOn()));
+		assertDependencies(beanDefinitions.get("overflowDiskStore"), "pdxDiskStore");
+		assertDependencies(beanDefinitions.get("region1"), "overflowDiskStore");
+		assertDependencies(beanDefinitions.get("region2DiskStore"), "pdxDiskStore", "someBean");
+		assertDependencies(beanDefinitions.get("region2"), "pdxDiskStore", "region2DiskStore");
+		assertDependencies(beanDefinitions.get("colocatedRegion"), "residentRegion", "overflowDiskStore");
+		assertDependencies(beanDefinitions.get("residentRegionDiskStore"), "pdxDiskStore", "someBean", "yetAnotherBean");
+		assertDependencies(beanDefinitions.get("residentRegion"), "pdxDiskStore", "residentRegionDiskStore");
+		assertDependencies(beanDefinitions.get("yetAnotherBean"), "someBean");
+	}
+
+}
--- a/src/test/java/org/springframework/data/gemfire/test/MockCacheFactoryBean.java
+++ b/src/test/java/org/springframework/data/gemfire/test/MockCacheFactoryBean.java
@@ -53,6 +53,7 @@ public class MockCacheFactoryBean extends CacheFactoryBean {
 			this.pdxDiskStoreName = cacheFactoryBean.getPdxDiskStoreName();
 			this.pdxIgnoreUnreadFields = cacheFactoryBean.getPdxIgnoreUnreadFields();
 			this.pdxPersistent = cacheFactoryBean.getPdxPersistent();
+			this.pdxReadSerialized = cacheFactoryBean.getPdxReadSerialized();
 			this.pdxSerializer = cacheFactoryBean.getPdxSerializer();
 			this.properties = cacheFactoryBean.getProperties();
 			this.searchTimeout = cacheFactoryBean.getSearchTimeout();
--- a/src/test/resources/org/springframework/data/gemfire/config/cache-using-pdx-ns.xml
+++ b/src/test/resources/org/springframework/data/gemfire/config/cache-using-pdx-ns.xml
@@ -0,0 +1,15 @@
+<?xml version="1.0" encoding="utf-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	   xmlns:gfe="http://www.springframework.org/schema/gemfire"
+	   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	   xsi:schemaLocation="
+        http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
+        http://www.springframework.org/schema/gemfire http://www.springframework.org/schema/gemfire/spring-gemfire.xsd
+">
+
+	<bean id="autoSerializer" class="com.gemstone.gemfire.pdx.ReflectionBasedAutoSerializer"/>
+
+	<gfe:cache pdx-serializer-ref="autoSerializer" pdx-read-serialized="true" pdx-persistent="true" pdx-disk-store="pdxStore"/>
+
+</beans>
+
--- a/src/test/resources/org/springframework/data/gemfire/pdxdiskstore-config.xml
+++ b/src/test/resources/org/springframework/data/gemfire/pdxdiskstore-config.xml
@@ -0,0 +1,59 @@
+<?xml version="1.0" encoding="utf-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	   xmlns:gfe="http://www.springframework.org/schema/gemfire"
+	   xmlns:p="http://www.springframework.org/schema/p"
+	   xmlns:util="http://www.springframework.org/schema/util"
+	   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	   xsi:schemaLocation="
+        http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
+        http://www.springframework.org/schema/gemfire http://www.springframework.org/schema/gemfire/spring-gemfire.xsd
+        http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util-3.0.xsd
+  ">
+
+	<util:properties id="peerCacheConfigurationSettings">
+		<prop key="log-level">config</prop>
+		<prop key="name">pdxDiskStoreTest</prop>
+	</util:properties>
+	<!--
+	  <util:properties id="peerCacheConfigurationSettings" location="classpath:gemfire.properties"/>
+	  -->
+
+	<bean id="autoSerializer" class="com.gemstone.gemfire.pdx.ReflectionBasedAutoSerializer">
+		<constructor-arg type="boolean" value="true"/>
+		<constructor-arg>
+			<list>
+				<value>org.springframework.data.gemfire.PdxDiskStoreIntegrationTest\$KeyHolder</value>
+				<value>org.springframework.data.gemfire.PdxDiskStoreIntegrationTest\$ValueHolder</value>
+			</list>
+		</constructor-arg>
+	</bean>
+
+	<!--
+		NOTE if the GemFire Disk Stores are declared/defined in the right order inside this Spring application context
+		configuration file, then the test passes since the PDX Disk Store gets created before the Data Disk Store that
+		is used by the Region to store data with the PDX typed key and value objects.
+	  -->
+
+	<!--
+	  <gfe:disk-store id="pdxStore" auto-compact="true" compaction-threshold="50" queue-size="50" max-oplog-size="10" time-interval="60000">
+		<gfe:disk-dir location="./gemfire/pdx-store" max-size="50"/>
+	  </gfe:disk-store>
+	  -->
+
+	<gfe:cache properties-ref="peerCacheConfigurationSettings" pdx-serializer-ref="autoSerializer" pdx-persistent="true" pdx-disk-store="pdxStore"/>
+
+	<gfe:disk-store id="dataStore" auto-compact="true" compaction-threshold="75" queue-size="50" max-oplog-size="10" time-interval="600000">
+		<gfe:disk-dir location="./gemfire/data-store/" max-size="50"/>
+	</gfe:disk-store>
+
+	<gfe:partitioned-region id="pdxDataRegion" name="PdxData" persistent="true" disk-store-ref="dataStore" disk-synchronous="true"/>
+
+	<gfe:disk-store id="pdxStore" auto-compact="true" compaction-threshold="50" queue-size="50" max-oplog-size="10" time-interval="60000">
+		<gfe:disk-dir location="./gemfire/pdx-store" max-size="50"/>
+	</gfe:disk-store>
+
+	<!--
+	  <bean id="pdxDataTemplate" class="org.springframework.data.gemfire.GemfireTemplate" p:region-ref="pdxDataRegion"/>
+	  -->
+
+</beans>