Merged PR #33 to fix JIRA issue SGF-203 concerning persistence.

src/main/java/org/springframework/data/gemfire/DataPolicyConverter.java

@@ -13,6 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package org.springframework.data.gemfire;
 
 import org.springframework.core.convert.converter.Converter;
@@ -24,54 +25,50 @@ import com.gemstone.gemfire.cache.DataPolicy;
  * 
  */
 public class DataPolicyConverter implements Converter<String, DataPolicy> {
-	private static enum Policy {
-		EMPTY, DEFAULT, NORMAL, PERSISTENT_PARTITION, PERSISTENT_REPLICATE, PRELOADED, REPLICATE;
-		public DataPolicy toDataPolicy() {
-			DataPolicy dataPolicy = null;
-			switch (this) {
-			case EMPTY:
-				dataPolicy = DataPolicy.EMPTY;
-				break;
-			case DEFAULT:
-				dataPolicy = DataPolicy.DEFAULT;
-				break;
-			case NORMAL:
-				dataPolicy = DataPolicy.NORMAL;
-				break;
-			case PERSISTENT_PARTITION:
-				dataPolicy = DataPolicy.PERSISTENT_PARTITION;
-				break;
-			case PERSISTENT_REPLICATE:
-				dataPolicy = DataPolicy.PERSISTENT_REPLICATE;
-				break;
-			case PRELOADED:
-				dataPolicy = DataPolicy.PRELOADED;
-				break;
-			case REPLICATE:
-				dataPolicy = DataPolicy.REPLICATE;
-				break;
-			}
-			return dataPolicy;
+
+	static enum Policy {
+		DEFAULT, EMPTY, NORMAL, PRELOADED, PARTITION, PERSISTENT_PARTITION, REPLICATE, PERSISTENT_REPLICATE;
+
+		private static String toUpperCase(String value) {
+			return (value == null ? null : value.toUpperCase());
 		}
 
 		public static Policy getValue(String value) {
-			Policy policy = null;
 			try {
-				policy = valueOf(value);
+				return valueOf(toUpperCase(value));
 			}
 			catch (Exception e) {
+				return null;
 			}
-			return policy;
 		}
-	};
+
+		public DataPolicy toDataPolicy() {
+			switch (this) {
+				case EMPTY:
+					return DataPolicy.EMPTY;
+				case NORMAL:
+					return DataPolicy.NORMAL;
+				case PRELOADED:
+					return DataPolicy.PRELOADED;
+				case PARTITION :
+					return DataPolicy.PARTITION;
+				case PERSISTENT_PARTITION:
+					return DataPolicy.PERSISTENT_PARTITION;
+				case REPLICATE:
+					return DataPolicy.REPLICATE;
+				case PERSISTENT_REPLICATE:
+					return DataPolicy.PERSISTENT_REPLICATE;
+				case DEFAULT:
+				default:
+					return DataPolicy.DEFAULT;
+			}
+		}
+	}
 
 	@Override
-	public DataPolicy convert(String source) {
-		if (source == null) {
-			return null;
-		}
-		source = source.toUpperCase();
-		return Policy.getValue(source) == null ? null : Policy.getValue(source).toDataPolicy();
+	public DataPolicy convert(String policyValue) {
+		Policy policy = Policy.getValue(policyValue);
+		return (policy == null ? null : policy.toDataPolicy());
 	}
 
 }

src/main/java/org/springframework/data/gemfire/GemfireUtils.java

@@ -0,0 +1,68 @@
+/*
+ * 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 java.util.concurrent.ConcurrentMap;
+
+import org.springframework.util.ClassUtils;
+
+import com.gemstone.gemfire.cache.CacheFactory;
+import com.gemstone.gemfire.cache.Region;
+
+/**
+ * The GemfireUtils class is a utility class encapsulating common functionality to access features and capabilities
+ * of GemFire based on version and other configuration meta-data.
+ * <p/>
+ * @author John Blum
+ * @since 1.3.3
+ */
+public abstract class GemfireUtils {
+
+	public final static String GEMFIRE_VERSION = CacheFactory.getVersion();
+
+	public static boolean isGemfireVersion65OrAbove() {
+		// expected 'major.minor'
+		try {
+			double version = Double.parseDouble(GEMFIRE_VERSION.substring(0, 3));
+			return version >= 6.5;
+		}
+		catch (NumberFormatException e) {
+			// NOTE based on logic from the PartitionedRegionFactoryBean class...
+			return ConcurrentMap.class.isAssignableFrom(Region.class);
+		}
+	}
+
+	public static boolean isGemfireVersion7OrAbove() {
+		try {
+			int version = Integer.parseInt(GEMFIRE_VERSION.substring(0, 1));
+			return version >= 7;
+		}
+		catch (NumberFormatException e) {
+			// NOTE the com.gemstone.gemfire.distributed.ServerLauncher class only exists in GemFire v 7.0.x or above...
+			return ClassUtils.isPresent("com.gemstone.gemfire.distributed.ServerLauncher",
+				Thread.currentThread().getContextClassLoader());
+		}
+
+	}
+
+	public static void main(final String... args) {
+		System.out.printf("GemFire Version %1$s%n", GEMFIRE_VERSION);
+		//System.out.printf("Is GemFire Version 6.5 of Above? %1$s%n", isGemfireVersion65OrAbove());
+		//System.out.printf("Is GemFire Version 7.0 of Above? %1$s%n", isGemfireVersion7OrAbove());
+	}
+
+}

src/main/java/org/springframework/data/gemfire/LocalRegionFactoryBean.java

@@ -16,6 +16,7 @@
 package org.springframework.data.gemfire;
 
 import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
 
 import com.gemstone.gemfire.cache.DataPolicy;
 import com.gemstone.gemfire.cache.RegionFactory;
@@ -23,12 +24,13 @@ import com.gemstone.gemfire.cache.Scope;
 
 /**
  * @author David Turanski
- * 
+ * @author John Blum
  */
 public class LocalRegionFactoryBean<K, V> extends RegionFactoryBean<K, V> {
+
 	@Override
 	public void setScope(Scope scope) {
-		throw new UnsupportedOperationException("setScope() is not allowed for Local Regions");
+		throw new UnsupportedOperationException("Setting the Scope on Local Regions is not allowed.");
 	}
 
 	@Override
@@ -37,24 +39,38 @@ public class LocalRegionFactoryBean<K, V> extends RegionFactoryBean<K, V> {
 		super.afterPropertiesSet();
 	}
 
+	/**
+	 * Resolves the Data Policy used by this "local" GemFire Region (i.e. locally Scoped; Scope.LOCAL) based on the
+	 * enumerated value from com.gemstone.gemfire.cache.RegionShortcuts (LOCAL, LOCAL_PERSISTENT, LOCAL_HEAP_LRU,
+	 * LOCAL_OVERFLOW, and LOCAL_PERSISTENT_OVERFLOW), but without consideration of the Eviction settings.
+	 * <p/>
+	 * @param regionFactory the GemFire RegionFactory used to created the Local Region.
+	 * @param persistent a boolean value indicating whether the Local Region should persist it's data.
+	 * @param dataPolicy requested Data Policy as set by the user in the Spring GemFire configuration meta-data.
+	 * @see com.gemstone.gemfire.cache.DataPolicy
+	 * @see com.gemstone.gemfire.cache.RegionFactory
+	 * @see com.gemstone.gemfire.cache.RegionShortcut
+	 */
 	@Override
 	protected void resolveDataPolicy(RegionFactory<K, V> regionFactory, Boolean persistent, String dataPolicy) {
-		if (dataPolicy != null) {
-			dataPolicy = dataPolicy.toUpperCase();
-			if ("NORMAL".equals(dataPolicy) || dataPolicy == null) {
-				regionFactory.setDataPolicy(DataPolicy.NORMAL);
-			}
-			else if ("PRELOADED".equals(dataPolicy)) {
-				regionFactory.setDataPolicy(DataPolicy.PRELOADED);
-			}
-			else if ("EMPTY".equals(dataPolicy)) {
-				Assert.isTrue(persistent == null || !persistent, "Cannot have persistence on an empty region");
-				regionFactory.setDataPolicy(DataPolicy.EMPTY);
-			}
-			else {
-				throw new IllegalArgumentException("Data policy '" + dataPolicy
-						+ "' is unsupported or invalid for local regions.");
-			}
+		DataPolicy resolvedDataPolicy = new DataPolicyConverter().convert(dataPolicy);
+
+		Assert.isTrue(resolvedDataPolicy != null || !StringUtils.hasText(dataPolicy),
+			String.format("Data Policy '%1$s' is invalid.", dataPolicy));
+
+		if (resolvedDataPolicy == null || DataPolicy.NORMAL.equals(resolvedDataPolicy)) {
+			// NOTE this is safe since a LOCAL Scoped NORMAL Region requiring persistence can be satisfied with
+			// PERSISTENT_REPLICATE, per the RegionShortcut.LOCAL_PERSISTENT
+			regionFactory.setDataPolicy(isPersistent() ? DataPolicy.PERSISTENT_REPLICATE : DataPolicy.NORMAL);
+		}
+		else if (DataPolicy.PRELOADED.equals(resolvedDataPolicy)) {
+			// NOTE this is safe since a LOCAL Scoped PRELOADED Region requiring persistence can be satisfied with
+			// PERSISTENT_REPLICATE, per the RegionShortcut.LOCAL_PERSISTENT
+			regionFactory.setDataPolicy(isPersistent() ? DataPolicy.PERSISTENT_REPLICATE : DataPolicy.PRELOADED);
+		}
+		else {
+			throw new IllegalArgumentException(String.format("Data Policy '%1$s' is not supported in Local Regions.",
+				dataPolicy));
 		}
 	}
 

src/main/java/org/springframework/data/gemfire/PartitionedRegionFactoryBean.java

@@ -15,48 +15,41 @@
  */
 package org.springframework.data.gemfire;
 
-import java.util.concurrent.ConcurrentMap;
-
 import org.springframework.util.Assert;
 
 import com.gemstone.gemfire.cache.CacheFactory;
 import com.gemstone.gemfire.cache.DataPolicy;
-import com.gemstone.gemfire.cache.Region;
 import com.gemstone.gemfire.cache.RegionFactory;
 
 /**
  * @author David Turanski
- * 
+ * @author John Blum
  */
 public class PartitionedRegionFactoryBean<K, V> extends RegionFactoryBean<K, V> {
 
 	@Override
 	protected void resolveDataPolicy(RegionFactory<K, V> regionFactory, Boolean persistent, String dataPolicy) {
 		if (dataPolicy != null) {
-			DataPolicy dp = new DataPolicyConverter().convert(dataPolicy);
-			Assert.notNull(dp, "Data policy " + dataPolicy + " is invalid");
-			Assert.isTrue(dp.withPartitioning(), "Data Policy " + dp.toString()
-					+ " is not supported in partitioned regions");
-			if (!isPersistent()) {
-				regionFactory.setDataPolicy(dp);
-			}
-			else {
-				Assert.isTrue(dp.withPersistence(), "Data Policy " + dp.toString()
-						+ "is invalid when persistent is false");
-				regionFactory.setDataPolicy(dp);
-			}
-			return;
+			DataPolicy resolvedDataPolicy = new DataPolicyConverter().convert(dataPolicy);
+
+			Assert.notNull(resolvedDataPolicy, String.format("Data Policy '%1$s' is invalid.", dataPolicy));
+
+			// Validate that the user-defined Data Policy matches the appropriate Spring GemFire XML namespace
+			// configuration meta-data element for Region (i.e. <gfe:partitioned-region .../>)!
+			Assert.isTrue(resolvedDataPolicy.withPartitioning(), String.format(
+				"Data Policy '%1$s' is not supported in Partitioned Regions.", resolvedDataPolicy));
+
+			// Validate that the data-policy and persistent attributes are compatible when specified!
+			assertDataPolicyAndPersistentAttributesAreCompatible(resolvedDataPolicy);
+
+			regionFactory.setDataPolicy(resolvedDataPolicy);
 		}
-		if (isPersistent()) {
-			// check first for GemFire 6.5
-			if (ConcurrentMap.class.isAssignableFrom(Region.class)) {
-				regionFactory.setDataPolicy(DataPolicy.PERSISTENT_PARTITION);
-			}
-			else {
-				throw new IllegalArgumentException(
-						"Can define persistent partitions only from GemFire 6.5 onwards - current version is ["
-								+ CacheFactory.getVersion() + "]");
-			}
+		else if (isPersistent()) {
+			// first, check the presence of GemFire 6.5 or Higher
+			Assert.isTrue(GemfireUtils.isGemfireVersion65OrAbove(), String.format(
+				"Can define Persistent Partitioned Regions only from GemFire 6.5 onwards; current version is [%1$s]",
+					CacheFactory.getVersion()));
+			regionFactory.setDataPolicy(DataPolicy.PERSISTENT_PARTITION);
 		}
 		else {
 			regionFactory.setDataPolicy(DataPolicy.PARTITION);

src/main/java/org/springframework/data/gemfire/RegionFactoryBean.java

@@ -24,14 +24,12 @@ import org.springframework.beans.factory.DisposableBean;
 import org.springframework.context.SmartLifecycle;
 import org.springframework.core.io.Resource;
 import org.springframework.data.gemfire.client.ClientRegionFactoryBean;
-import org.springframework.data.gemfire.wan.GatewaySenderWrapper;
 import org.springframework.util.Assert;
 import org.springframework.util.ObjectUtils;
 import org.springframework.util.ReflectionUtils;
 
 import com.gemstone.gemfire.cache.AttributesFactory;
 import com.gemstone.gemfire.cache.Cache;
-import com.gemstone.gemfire.cache.CacheClosedException;
 import com.gemstone.gemfire.cache.CacheListener;
 import com.gemstone.gemfire.cache.CacheLoader;
 import com.gemstone.gemfire.cache.CacheWriter;
@@ -56,20 +54,19 @@ import com.gemstone.gemfire.cache.wan.GatewaySender;
  *
  * @author Costin Leau
  * @author David Turanski
+ * @author John Blum
  */
 public class RegionFactoryBean<K, V> extends RegionLookupFactoryBean<K, V> implements DisposableBean, SmartLifecycle {
 
-	private boolean autoStartup = true;
-
-	private boolean running;
-
 	protected final Log log = LogFactory.getLog(getClass());
 
-	private boolean destroy = false;
-
+	private boolean autoStartup = true;
 	private boolean close = true;
+	private boolean destroy = false;
+	private boolean running;
 
-	private Resource snapshot;
+	private Boolean enableGateway;
+	private Boolean persistent;
 
 	private CacheListener<K, V> cacheListeners[];
 
@@ -77,31 +74,23 @@ public class RegionFactoryBean<K, V> extends RegionLookupFactoryBean<K, V> imple
 
 	private CacheWriter<K, V> cacheWriter;
 
-	private Object gatewaySenders[];
-
-	private Object asyncEventQueues[];
+	private Object[] asyncEventQueues;
+	private Object[] gatewaySenders;
 
 	private RegionAttributes<K, V> attributes;
 
+	private Resource snapshot;
+
 	private Scope scope;
 
-	private Boolean persistent;
-
-	private Boolean enableGateway;
-
-	private String hubId;
-
-	private String diskStoreName;
-
 	private String dataPolicy;
-
-	private Region<K, V> region;
+	private String diskStoreName;
+	private String hubId;
 
 	@Override
 	public void afterPropertiesSet() throws Exception {
 		super.afterPropertiesSet();
-		region = getRegion();
-		postProcess(region);
+		postProcess(getRegion());
 	}
 
 	@Override
@@ -110,11 +99,12 @@ public class RegionFactoryBean<K, V> extends RegionLookupFactoryBean<K, V> imple
 
 		Cache c = (Cache) cache;
 
-		if (attributes != null)
+		if (attributes != null) {
 			AttributesFactory.validateAttributes(attributes);
+		}
 
-		final RegionFactory<K, V> regionFactory = (attributes != null ? c.createRegionFactory(attributes) : c
-				.<K, V> createRegionFactory());
+		final RegionFactory<K, V> regionFactory = (attributes != null ? c.createRegionFactory(attributes) :
+			c.<K, V> createRegionFactory());
 
 		if (hubId != null) {
 			enableGateway = enableGateway == null ? true : enableGateway;
@@ -122,12 +112,14 @@ public class RegionFactoryBean<K, V> extends RegionLookupFactoryBean<K, V> imple
 
 			regionFactory.setGatewayHubId(hubId);
 		}
+
 		if (enableGateway != null) {
 			if (enableGateway) {
 				Assert.notNull(hubId, "enableGateway requires the hubId property to be true");
 			}
 			regionFactory.setEnableGateway(enableGateway);
 		}
+
 		if (!ObjectUtils.isEmpty(cacheListeners)) {
 			for (CacheListener<K, V> listener : cacheListeners) {
 				regionFactory.addCacheListener(listener);
@@ -135,9 +127,8 @@ public class RegionFactoryBean<K, V> extends RegionLookupFactoryBean<K, V> imple
 		}
 
 		if (!ObjectUtils.isEmpty(gatewaySenders)) {
-			Assert.isTrue(
-					hubId == null,
-					"It is invalid to configure a region with both a hubId and gatewaySenders. Note that the enableGateway and hubId properties are deprecated since Gemfire 7.0");
+			Assert.isTrue(hubId == null, "It is invalid to configure a region with both a hubId and gatewaySenders."
+				+ " Note that the enableGateway and hubId properties are deprecated since Gemfire 7.0");
 			for (Object gatewaySender : gatewaySenders) {
 				regionFactory.addGatewaySenderId(((GatewaySender) gatewaySender).getId());
 			}
@@ -169,54 +160,56 @@ public class RegionFactoryBean<K, V> extends RegionLookupFactoryBean<K, V> imple
 
 		if (attributes != null) {
 			Assert.state(!attributes.isLockGrantor() || scope.isGlobal(),
-					"Lock grantor only applies to a global scoped region");
+				"Lock grantor only applies to a global scoped region");
 		}
+
 		// get underlying AttributesFactory
 		postProcess(findAttrFactory(regionFactory));
 
-		Region<K, V> reg = regionFactory.create(regionName);
+		Region<K, V> region = regionFactory.create(regionName);
 		log.info("Created new cache region [" + regionName + "]");
+
 		if (snapshot != null) {
-			reg.loadSnapshot(snapshot.getInputStream());
+			region.loadSnapshot(snapshot.getInputStream());
 		}
 
 		if (attributes != null && attributes.isLockGrantor()) {
-			reg.becomeLockGrantor();
+			region.becomeLockGrantor();
 		}
 
-		return reg;
+		return region;
 	}
 
 	/**
-	 * This validates the configured data policy and may override it, taking
-	 * into account the persistent attribute and constraints for the region type
-	 * @param regionFactory
-	 * @param persistent
-	 * @param dataPolicy requested data policy
+	 * Validates the configured Data Policy and may override it, taking into account the 'persistent' attribute
+	 * and constraints for the Region type.
+	 * <p/>
+	 * @param regionFactory the GemFire RegionFactory used to created the Local Region.
+	 * @param persistent a boolean value indicating whether the Local Region should persist it's data.
+	 * @param dataPolicy requested Data Policy as set by the user in the Spring GemFire configuration meta-data.
+	 * @see com.gemstone.gemfire.cache.DataPolicy
+	 * @see com.gemstone.gemfire.cache.RegionFactory
 	 */
 	protected void resolveDataPolicy(RegionFactory<K, V> regionFactory, Boolean persistent, String dataPolicy) {
-		if (dataPolicy == null) {
-			if (isPersistent()) {
-				regionFactory.setDataPolicy(DataPolicy.PERSISTENT_REPLICATE);
-			} else {
-				regionFactory.setDataPolicy(DataPolicy.DEFAULT);
-			}
-			return;
-		}
+		if (dataPolicy != null) {
+			DataPolicy resolvedDataPolicy = new DataPolicyConverter().convert(dataPolicy);
 
-		DataPolicy dp = new DataPolicyConverter().convert(dataPolicy);
-		Assert.notNull(dp, "Data policy " + dataPolicy + " is invalid");
-		if (dp.withPersistence()) {
-			Assert.isTrue(!isNotPersistent(), "Data policy " + dataPolicy + " is invalid when persistent is false");
+			Assert.notNull(resolvedDataPolicy, String.format("Data Policy '%1$s' is invalid.", dataPolicy));
+			assertDataPolicyAndPersistentAttributesAreCompatible(resolvedDataPolicy);
+
+			regionFactory.setDataPolicy(resolvedDataPolicy);
+		}
+		else {
+			regionFactory.setDataPolicy(isPersistent() ? DataPolicy.PERSISTENT_REPLICATE : DataPolicy.DEFAULT);
 		}
-		regionFactory.setDataPolicy(dp);
 	}
 
 	@SuppressWarnings("unchecked")
 	private AttributesFactory<K, V> findAttrFactory(RegionFactory<K, V> regionFactory) {
-		Field attrField = ReflectionUtils.findField(RegionFactory.class, "attrsFactory", AttributesFactory.class);
-		ReflectionUtils.makeAccessible(attrField);
-		return (AttributesFactory<K, V>) ReflectionUtils.getField(attrField, regionFactory);
+		Field attrsFactoryField = ReflectionUtils.findField(RegionFactory.class, "attrsFactory",
+			AttributesFactory.class);
+		ReflectionUtils.makeAccessible(attrsFactoryField);
+		return (AttributesFactory<K, V>) ReflectionUtils.getField(attrsFactoryField, regionFactory);
 	}
 
 	/**
@@ -246,32 +239,22 @@ public class RegionFactoryBean<K, V> extends RegionLookupFactoryBean<K, V> imple
 
 	@Override
 	public void destroy() throws Exception {
-		if (region != null) {
+		if (getRegion() != null) {
 			if (close) {
-				if (!region.getRegionService().isClosed()) {
+				if (!getRegion().getRegionService().isClosed()) {
 					try {
-						region.close();
-						region = null;
+						getRegion().close();
 					} catch (Exception cce) {
 						// nothing to see folks, move on.
 					}
 				}
 
 			} if (destroy) {
-				region.destroyRegion();
-				region = null;
+				getRegion().destroyRegion();
 			}
 		}
 	}
 
-	/**
-	 * Indicates whether the region referred by this factory bean, will be
-	 * destroyed on shutdown (default false).
-	 */
-	public void setDestroy(boolean destroy) {
-		this.destroy = destroy;
-	}
-
 	/**
 	 * Indicates whether the region referred by this factory bean, will be
 	 * closed on shutdown (default true).
@@ -280,6 +263,14 @@ public class RegionFactoryBean<K, V> extends RegionLookupFactoryBean<K, V> imple
 		this.close = close;
 	}
 
+	/**
+	 * Indicates whether the region referred by this factory bean, will be
+	 * destroyed on shutdown (default false).
+	 */
+	public void setDestroy(boolean destroy) {
+		this.destroy = destroy;
+	}
+
 	/**
 	 * Sets the snapshots used for loading a newly <i>created</i> region. That
 	 * is, the snapshot will be used <i>only</i> when a new region is created -
@@ -343,7 +334,7 @@ public class RegionFactoryBean<K, V> extends RegionLookupFactoryBean<K, V> imple
 	/**
 	 * Sets the dataPolicy as a String. Required to support property
 	 * placeholders
-	 * @param dataPolicy the dataPolicy name (NORMAL, PRELOADED, etc)
+	 * @param dataPolicyName the dataPolicy name (NORMAL, PRELOADED, etc)
 	 */
 	public void setDataPolicy(String dataPolicyName) {
 		this.dataPolicy = dataPolicyName;
@@ -394,12 +385,54 @@ public class RegionFactoryBean<K, V> extends RegionLookupFactoryBean<K, V> imple
 		this.attributes = attributes;
 	}
 
-	protected boolean isPersistent() {
-		return persistent != null && persistent;
+	/**
+	 * Validates that the settings for Data Policy and the 'persistent' attribute in <gfe:*-region/> elements
+	 * are compatible.
+	 * <p/>
+	 * @param resolvedDataPolicy the GemFire Data Policy resolved form the Spring GemFire XML namespace configuration
+	 * meta-data.
+	 * @see #isPersistent()
+	 * @see #isNotPersistent()
+	 * @see com.gemstone.gemfire.cache.DataPolicy
+	 */
+	protected void assertDataPolicyAndPersistentAttributesAreCompatible(final DataPolicy resolvedDataPolicy) {
+		final boolean persistentNotSpecified = (this.persistent == null);
+
+		if (resolvedDataPolicy.withPersistence()) {
+			Assert.isTrue(persistentNotSpecified || isPersistent(), String.format(
+				"Data Policy '%1$s' is invalid when persistent is false.", resolvedDataPolicy));
+		}
+		else {
+			// NOTE otherwise, the Data Policy is with persistence, so...
+			Assert.isTrue(persistentNotSpecified || isNotPersistent(), String.format(
+				"Data Policy '%1$s' is invalid when persistent is true.", resolvedDataPolicy));
+		}
 	}
 
+	/**
+	 * Returns true when the user explicitly specified a value for the persistent attribute and it is true.  If the
+	 * persistent attribute was not explicitly specified, then the persistence setting is implicitly undefined
+	 * and will be determined by the Data Policy.
+	 * <p/>
+	 * @return true when the user specified an explicit value for the persistent attribute and it is true;
+	 * false otherwise.
+	 * @see #isNotPersistent()
+	 */
+	protected boolean isPersistent() {
+		return Boolean.TRUE.equals(persistent);
+	}
+
+	/**
+	 * Returns true when the user explicitly specified a value for the persistent attribute and it is false.  If the
+	 * persistent attribute was not explicitly specified, then the persistence setting is implicitly undefined
+	 * and will be determined by the Data Policy.
+	 * <p/>
+	 * @return true when the user specified an explicit value for the persistent attribute and it is false;
+	 * false otherwise.
+	 * @see #isPersistent()
+	 */
 	protected boolean isNotPersistent() {
-		return persistent != null && !persistent;
+		return Boolean.FALSE.equals(persistent);
 	}
 
 	/* (non-Javadoc)
@@ -407,12 +440,11 @@ public class RegionFactoryBean<K, V> extends RegionLookupFactoryBean<K, V> imple
 	 */
 	@Override
 	public void start() {
-
 		if (!ObjectUtils.isEmpty(gatewaySenders)) {
 			synchronized (gatewaySenders) {
 				for (Object obj : gatewaySenders) {
 					GatewaySender gws = (GatewaySender) obj;
-					if (!gws.isManualStart() && !gws.isRunning()) {
+					if (!(gws.isManualStart() || gws.isRunning())) {
 						gws.start();
 					}
 				}
@@ -429,9 +461,8 @@ public class RegionFactoryBean<K, V> extends RegionLookupFactoryBean<K, V> imple
 		if (!ObjectUtils.isEmpty(gatewaySenders)) {
 			synchronized (gatewaySenders) {
 				for (Object obj : gatewaySenders) {
-				 GatewaySender gws = (GatewaySender) obj;
-					gws.stop();
-				 }
+					((GatewaySender) obj).stop();
+				}
 			}
 		}
 		this.running = false;
@@ -469,4 +500,5 @@ public class RegionFactoryBean<K, V> extends RegionLookupFactoryBean<K, V> imple
 		stop();
 		callback.run();
 	}
+
 }

src/main/java/org/springframework/data/gemfire/RegionLookupFactoryBean.java

@@ -39,16 +39,19 @@ public class RegionLookupFactoryBean<K, V> implements FactoryBean<Region<K, V>>,
 
 	protected final Log log = LogFactory.getLog(getClass());
 
-	private String beanName;
 	private GemFireCache cache;
-	private String name;
 
 	private Region<K, V> region;
 
+	private String beanName;
+	private String name;
+
 	public void afterPropertiesSet() throws Exception {
 		Assert.notNull(cache, "Cache property must be set");
+
 		name = (!StringUtils.hasText(name) ? beanName : name);
 		Assert.hasText(name, "Name (or beanName) property must be set");
+
 		synchronized (cache) {
 			region = cache.getRegion(name);
 			if (region != null) {
@@ -116,4 +119,5 @@ public class RegionLookupFactoryBean<K, V> implements FactoryBean<Region<K, V>>,
 	protected Region<K, V> getRegion() {
 		return region;
 	}
-}
+
+}

src/main/java/org/springframework/data/gemfire/ReplicatedRegionFactoryBean.java

@@ -22,32 +22,35 @@ import com.gemstone.gemfire.cache.RegionFactory;
 
 /**
  * @author David Turanski
- * 
+ * @author John Blum
  */
 public class ReplicatedRegionFactoryBean<K, V> extends RegionFactoryBean<K, V> {
+
 	@Override
 	protected void resolveDataPolicy(RegionFactory<K, V> regionFactory, Boolean persistent, String dataPolicy) {
-		DataPolicy dp = null;
 		if (dataPolicy != null) {
-			dataPolicy = dataPolicy.toUpperCase();
-			if ("EMPTY".equals(dataPolicy)) {
-				Assert.isTrue(persistent == null || !persistent, "Cannot have persistence on an empty region");
-				dp = DataPolicy.EMPTY;
+			DataPolicy resolvedDataPolicy = new DataPolicyConverter().convert(dataPolicy);
+
+			Assert.notNull(resolvedDataPolicy, String.format("Data Policy '%1$s' is invalid.", dataPolicy));
+
+			// Validate that the data-policy and persistent attributes are compatible when specified!
+			assertDataPolicyAndPersistentAttributesAreCompatible(resolvedDataPolicy);
+
+			if (DataPolicy.EMPTY.equals(resolvedDataPolicy)) {
+				resolvedDataPolicy = DataPolicy.EMPTY;
 			}
 			else {
-				dp = new DataPolicyConverter().convert(dataPolicy);
-				Assert.notNull(dp, "Data policy " + dataPolicy + " is invalid");
-				Assert.isTrue(dp.withReplication(), "Data policy " + dataPolicy
-						+ " is invalid or unsupported in replicated regions");
-				if (isPersistent()) {
-					Assert.isTrue(dp.withPersistence(), "Data policy " + dataPolicy
-							+ " is invalid when persistent is false");
-				}
+				// Validate that the user-defined Data Policy matches the appropriate Spring GemFire XML namespace
+				// configuration meta-data element for Region (i.e. <gfe:replicated-region .../>)!
+				Assert.isTrue(resolvedDataPolicy.withReplication(), String.format("" +
+					"Data Policy '%1$s' is not supported in Replicated Regions.", resolvedDataPolicy));
 			}
+
+			regionFactory.setDataPolicy(resolvedDataPolicy);
 		}
 		else {
-			dp = (isPersistent()) ? DataPolicy.PERSISTENT_REPLICATE : DataPolicy.REPLICATE;
+			regionFactory.setDataPolicy(isPersistent() ? DataPolicy.PERSISTENT_REPLICATE : DataPolicy.REPLICATE);
 		}
-		regionFactory.setDataPolicy(dp);
 	}
+
 }

src/main/java/org/springframework/data/gemfire/client/ClientRegionFactoryBean.java

@@ -60,6 +60,8 @@ public class ClientRegionFactoryBean<K, V> extends RegionLookupFactoryBean<K, V>
 
 	private BeanFactory beanFactory;
 
+	private Boolean persistent;
+
 	private CacheListener<K, V>[] cacheListeners;
 
 	private CacheLoader<K, V> cacheLoader;
@@ -76,7 +78,6 @@ public class ClientRegionFactoryBean<K, V> extends RegionLookupFactoryBean<K, V>
 
 	private Resource snapshot;
 
-	private String dataPolicyName;
 	private String diskStoreName;
 	private String poolName;
 
@@ -92,43 +93,14 @@ public class ClientRegionFactoryBean<K, V> extends RegionLookupFactoryBean<K, V>
 
 		// TODO reference to an internal GemFire class!
 		if (cache instanceof GemFireCacheImpl) {
-			Assert.isTrue(((GemFireCacheImpl) cache).isClient(), "A client-cache instance is required");
+			Assert.isTrue(((GemFireCacheImpl) cache).isClient(), "A client-cache instance is required.");
 		}
 
-		Assert.isTrue(!(StringUtils.hasText(dataPolicyName) && dataPolicy != null),
-			"Only one of 'dataPolicy' or 'dataPolicyName' can be set");
+		final ClientCache clientCache = (ClientCache) cache;
 
-		if (StringUtils.hasText(dataPolicyName)) {
-			dataPolicy = new DataPolicyConverter().convert(dataPolicyName);
-			Assert.notNull(dataPolicy, "Data policy " + dataPolicyName + " is invalid");
-		}
+		ClientRegionFactory<K, V> factory = clientCache.createClientRegionFactory(resolveClientRegionShortcut());
 
-		// first look at shortcut
-		ClientRegionShortcut shortcut = this.shortcut;
-
-		if (shortcut == null) {
-			if (dataPolicy != null) {
-				if (DataPolicy.EMPTY.equals(dataPolicy)) {
-					shortcut = ClientRegionShortcut.PROXY;
-				}
-				else if (DataPolicy.NORMAL.equals(this.dataPolicy)) {
-					shortcut = ClientRegionShortcut.CACHING_PROXY;
-				}
-				else if (DataPolicy.PERSISTENT_REPLICATE.equals(dataPolicy)) {
-					shortcut = ClientRegionShortcut.LOCAL_PERSISTENT;
-				}
-				else {
-					shortcut = ClientRegionShortcut.LOCAL;
-				}
-			}
-			else {
-				shortcut = ClientRegionShortcut.LOCAL;
-			}
-		}
-
-		ClientRegionFactory<K, V> factory = ((ClientCache) cache).createClientRegionFactory(shortcut);
-
-		// map the attributes onto the client
+		// map region attributes onto the client region factory
 		if (attributes != null) {
 			factory.setCloningEnabled(attributes.getCloningEnabled());
 			factory.setConcurrencyLevel(attributes.getConcurrencyLevel());
@@ -155,11 +127,10 @@ public class ClientRegionFactoryBean<K, V> extends RegionLookupFactoryBean<K, V>
 			// try to eagerly initialize the pool name, if defined as a bean
 			if (beanFactory.isTypeMatch(poolName, Pool.class)) {
 				if (log.isDebugEnabled()) {
-					log.debug("Found bean definition for pool '" + poolName + "'. Eagerly initializing it...");
+					log.debug(String.format("Found bean definition for pool '%1$s'. Eagerly initializing...", poolName));
 				}
 				beanFactory.getBean(poolName, Pool.class);
 			}
-
 			factory.setPoolName(poolName);
 		}
 		else {
@@ -181,25 +152,100 @@ public class ClientRegionFactoryBean<K, V> extends RegionLookupFactoryBean<K, V>
 		return clientRegion;
 	}
 
+	protected ClientRegionShortcut resolveClientRegionShortcut() {
+		ClientRegionShortcut resolvedShortcut = this.shortcut;
+
+		if (resolvedShortcut == null) {
+			if (this.dataPolicy != null) {
+				assertDataPolicyAndPersistentAttributeAreCompatible(this.dataPolicy);
+
+				if (DataPolicy.EMPTY.equals(this.dataPolicy)) {
+					resolvedShortcut = ClientRegionShortcut.PROXY;
+				}
+				else if (DataPolicy.NORMAL.equals(this.dataPolicy)) {
+					resolvedShortcut = ClientRegionShortcut.CACHING_PROXY;
+				}
+				else if (DataPolicy.PERSISTENT_REPLICATE.equals(this.dataPolicy)) {
+					resolvedShortcut = ClientRegionShortcut.LOCAL_PERSISTENT;
+				}
+				else {
+					// NOTE the Data Policy validation is based on the ClientRegionShortcut initialization logic
+					// in com.gemstone.gemfire.internal.cache.GemFireCacheImpl.initializeClientRegionShortcuts
+					throw new IllegalArgumentException(String.format(
+						"Data Policy '%1$s' is invalid for Client Regions.", this.dataPolicy));
+				}
+			}
+			else {
+				resolvedShortcut = (isPersistent() ? ClientRegionShortcut.LOCAL_PERSISTENT
+					: ClientRegionShortcut.LOCAL);
+			}
+
+		}
+
+		// NOTE the ClientRegionShortcut and Persistent attribute will be compatible if the shortcut was derived from
+		// the Data Policy.
+		assertClientRegionShortcutAndPersistentAttributeAreCompatible(resolvedShortcut);
+
+		return resolvedShortcut;
+	}
+
+	protected void assertClientRegionShortcutAndPersistentAttributeAreCompatible(final ClientRegionShortcut resolvedShortcut) {
+		final boolean persistentNotSpecified = (this.persistent == null);
+
+		if (ClientRegionShortcut.LOCAL_PERSISTENT.equals(resolvedShortcut)
+				|| ClientRegionShortcut.LOCAL_PERSISTENT_OVERFLOW.equals(resolvedShortcut)) {
+			Assert.isTrue(persistentNotSpecified || isPersistent(), String.format(
+				"Client Region Shortcut '%1$s' is invalid when persistent is false.", resolvedShortcut));
+		}
+		else {
+			Assert.isTrue(persistentNotSpecified || isNotPersistent(), String.format(
+				"Client Region Shortcut '%1$s' is invalid when persistent is true.", resolvedShortcut));
+		}
+	}
+
+	/**
+	 * Validates that the settings for Data Policy and the 'persistent' attribute in <gfe:*-region/> elements
+	 * are compatible.
+	 * <p/>
+	 * @param resolvedDataPolicy the GemFire Data Policy resolved form the Spring GemFire XML namespace configuration
+	 * meta-data.
+	 * @see #isPersistent()
+	 * @see #isNotPersistent()
+	 * @see com.gemstone.gemfire.cache.DataPolicy
+	 */
+	protected void assertDataPolicyAndPersistentAttributeAreCompatible(final DataPolicy resolvedDataPolicy) {
+		final boolean persistentNotSpecified = (this.persistent == null);
+
+		if (resolvedDataPolicy.withPersistence()) {
+			Assert.isTrue(persistentNotSpecified || isPersistent(), String.format(
+				"Data Policy '%1$s' is invalid when persistent is false.", resolvedDataPolicy));
+		}
+		else {
+			// NOTE otherwise, the Data Policy is without persistence, so...
+			Assert.isTrue(persistentNotSpecified || isNotPersistent(), String.format(
+				"Data Policy '%1$s' is invalid when persistent is true.", resolvedDataPolicy));
+		}
+	}
+
 	private void addCacheListeners(ClientRegionFactory<K, V> factory) {
 		if (attributes != null) {
-			CacheListener<K, V>[] listeners = attributes.getCacheListeners();
+			CacheListener<K, V>[] cacheListeners = attributes.getCacheListeners();
 
-			if (!ObjectUtils.isEmpty(listeners)) {
-				for (CacheListener<K, V> listener : listeners) {
-					factory.addCacheListener(listener);
+			if (!ObjectUtils.isEmpty(cacheListeners)) {
+				for (CacheListener<K, V> cacheListener : cacheListeners) {
+					factory.addCacheListener(cacheListener);
 				}
 			}
 		}
 
 		if (!ObjectUtils.isEmpty(cacheListeners)) {
-			for (CacheListener<K, V> listener : cacheListeners) {
-				factory.addCacheListener(listener);
+			for (CacheListener<K, V> cacheListener : cacheListeners) {
+				factory.addCacheListener(cacheListener);
 			}
 		}
 	}
 
-	protected void postProcess(Region<K, V> region) {
+	protected void postProcess(final Region<K, V> region) {
 		registerInterests(region);
 		setCacheLoader(region);
 		setCacheWriter(region);
@@ -220,13 +266,13 @@ public class ClientRegionFactoryBean<K, V> extends RegionLookupFactoryBean<K, V>
 		}
 	}
 
-	private void setCacheLoader(Region<K, V> region) {
+	private void setCacheLoader(final Region<K, V> region) {
 		if (cacheLoader != null) {
 			region.getAttributesMutator().setCacheLoader(this.cacheLoader);
 		}
 	}
 
-	private void setCacheWriter(Region<K, V> region) {
+	private void setCacheWriter(final Region<K, V> region) {
 		if (cacheWriter != null) {
 			region.getAttributesMutator().setCacheWriter(this.cacheWriter);
 		}
@@ -272,6 +318,20 @@ public class ClientRegionFactoryBean<K, V> extends RegionLookupFactoryBean<K, V>
 		}
 	}
 
+	/**
+	 * Sets the region attributes used for the region used by this factory.
+	 * Allows maximum control in specifying the region settings. Used only when
+	 * a new region is created. Note that using this method allows for advanced
+	 * customization of the region - while it provides a lot of flexibility,
+	 * note that it's quite easy to create misconfigured regions (especially in
+	 * a client/server scenario).
+	 *
+	 * @param attributes the attributes to set on a newly created region
+	 */
+	public void setAttributes(RegionAttributes<K, V> attributes) {
+		this.attributes = attributes;
+	}
+
 	@Override
 	public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
 		this.beanFactory = beanFactory;
@@ -294,19 +354,9 @@ public class ClientRegionFactoryBean<K, V> extends RegionLookupFactoryBean<K, V>
 		return interests;
 	}
 
-	/**
-	 * Sets the pool name used by this client.
-	 * 
-	 * @param poolName
-	 */
-	public void setPoolName(String poolName) {
-		Assert.hasText(poolName, "pool name is required");
-		this.poolName = poolName;
-	}
-
 	/**
 	 * Sets the pool used by this client.
-	 * 
+	 *
 	 * @param pool
 	 */
 	public void setPool(Pool pool) {
@@ -315,31 +365,13 @@ public class ClientRegionFactoryBean<K, V> extends RegionLookupFactoryBean<K, V>
 	}
 
 	/**
-	 * Initializes the client using a GemFire {@link ClientRegionShortcut}. The
-	 * recommended way for creating clients since it covers all the major
-	 * scenarios with minimal configuration.
-	 * 
-	 * @param shortcut the ClientRegionShortcut to use.
+	 * Sets the pool name used by this client.
+	 *
+	 * @param poolName
 	 */
-	public void setShortcut(ClientRegionShortcut shortcut) {
-		this.shortcut = shortcut;
-	}
-
-	final boolean isDestroy() {
-		return destroy;
-	}
-
-	/**
-	 * Indicates whether the region referred by this factory bean will be
-	 * destroyed on shutdown (default false). Note: destroy and close are
-	 * mutually exclusive. Enabling one will automatically disable the other.
-	 * <p/>
-	 * @param destroy whether or not to destroy the region
-	 * @see #setClose(boolean)
-	 */
-	public void setDestroy(boolean destroy) {
-		this.destroy = destroy;
-		this.close = (this.close && !destroy); // retain previous value iff destroy is false;
+	public void setPoolName(String poolName) {
+		Assert.hasText(poolName, "pool name is required");
+		this.poolName = poolName;
 	}
 
 	final boolean isClose() {
@@ -359,23 +391,28 @@ public class ClientRegionFactoryBean<K, V> extends RegionLookupFactoryBean<K, V>
 		this.destroy = (this.destroy && !close); // retain previous value iff close is false.
 	}
 
+	final boolean isDestroy() {
+		return destroy;
+	}
+
 	/**
-	 * Sets the snapshots used for loading a newly <i>created</i> region. That
-	 * is, the snapshot will be used <i>only</i> when a new region is created -
-	 * if the region already exists, no loading will be performed.
-	 * 
-	 * @see #setName(String)
-	 * @param snapshot the snapshot to set
+	 * Indicates whether the region referred by this factory bean will be
+	 * destroyed on shutdown (default false). Note: destroy and close are
+	 * mutually exclusive. Enabling one will automatically disable the other.
+	 * <p/>
+	 * @param destroy whether or not to destroy the region
+	 * @see #setClose(boolean)
 	 */
-	public void setSnapshot(Resource snapshot) {
-		this.snapshot = snapshot;
+	public void setDestroy(boolean destroy) {
+		this.destroy = destroy;
+		this.close = (this.close && !destroy); // retain previous value iff destroy is false;
 	}
 
 	/**
 	 * Sets the cache listeners used for the region used by this factory. Used
 	 * only when a new region is created.Overrides the settings specified
 	 * through {@link #setAttributes(RegionAttributes)}.
-	 * 
+	 *
 	 * @param cacheListeners the cacheListeners to set on a newly created region
 	 */
 	public void setCacheListeners(CacheListener<K, V>[] cacheListeners) {
@@ -403,23 +440,14 @@ public class ClientRegionFactoryBean<K, V> extends RegionLookupFactoryBean<K, V>
 	}
 
 	/**
-	 * Sets the data policy. Used only when a new region is created.
-	 * 
-	 * @param dataPolicy the region data policy
+	 * Sets the Data Policy. Used only when a new Region is created.
+	 * <p/>
+	 * @param dataPolicy the client Region's Data Policy.
+	 * @see com.gemstone.gemfire.cache.DataPolicy
 	 */
 	public void setDataPolicy(DataPolicy dataPolicy) {
 		this.dataPolicy = dataPolicy;
 	}
-	
-	/**
-	 * An alternative way to set the data policy as a string. Useful for 
-	 * property placeholders, etc.
-	 * 
-	 * @param dataPolicyName
-	 */
-	public void setDataPolicyName(String dataPolicyName) {
-		this.dataPolicyName = dataPolicyName;
-	}
 
 	/**
 	 * Sets the name of disk store to use for overflow and persistence
@@ -430,17 +458,52 @@ public class ClientRegionFactoryBean<K, V> extends RegionLookupFactoryBean<K, V>
 	}
 
 	/**
-	 * Sets the region attributes used for the region used by this factory.
-	 * Allows maximum control in specifying the region settings. Used only when
-	 * a new region is created. Note that using this method allows for advanced
-	 * customization of the region - while it provides a lot of flexibility,
-	 * note that it's quite easy to create misconfigured regions (especially in
-	 * a client/server scenario).
-	 * 
-	 * @param attributes the attributes to set on a newly created region
+	 * An alternate way to set the Data Policy, using the String name of the enumerated value.
+	 * <p/>
+	 * @param dataPolicyName the enumerated value String name of the Data Policy.
+	 * @see com.gemstone.gemfire.cache.DataPolicy
+	 * @see #setDataPolicy(com.gemstone.gemfire.cache.DataPolicy)
+	 * @deprecated use setDataPolicy(:DataPolicy) instead.
 	 */
-	public void setAttributes(RegionAttributes<K, V> attributes) {
-		this.attributes = attributes;
+	public void setDataPolicyName(String dataPolicyName) {
+		final DataPolicy resolvedDataPolicy = new DataPolicyConverter().convert(dataPolicyName);
+		Assert.notNull(resolvedDataPolicy, String.format("Data Policy '%1$s' is invalid.", dataPolicyName));
+		setDataPolicy(resolvedDataPolicy);
+	}
+
+	protected boolean isPersistent() {
+		return Boolean.TRUE.equals(persistent);
+	}
+
+	protected boolean isNotPersistent() {
+		return Boolean.FALSE.equals(persistent);
+	}
+
+	public void setPersistent(final boolean persistent) {
+		this.persistent = persistent;
+	}
+
+	/**
+	 * Initializes the client using a GemFire {@link ClientRegionShortcut}. The
+	 * recommended way for creating clients since it covers all the major
+	 * scenarios with minimal configuration.
+	 *
+	 * @param shortcut the ClientRegionShortcut to use.
+	 */
+	public void setShortcut(ClientRegionShortcut shortcut) {
+		this.shortcut = shortcut;
+	}
+
+	/**
+	 * Sets the snapshots used for loading a newly <i>created</i> region. That
+	 * is, the snapshot will be used <i>only</i> when a new region is created -
+	 * if the region already exists, no loading will be performed.
+	 * <p/>
+	 * @param snapshot the snapshot to set
+	 * @see #setName(String)
+	 */
+	public void setSnapshot(Resource snapshot) {
+		this.snapshot = snapshot;
 	}
 
 }

src/main/java/org/springframework/data/gemfire/config/AbstractRegionParser.java

@@ -28,6 +28,7 @@ import org.springframework.beans.factory.support.BeanDefinitionBuilder;
 import org.springframework.beans.factory.support.ManagedArray;
 import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
 import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.data.gemfire.GemfireUtils;
 import org.springframework.data.gemfire.SubRegionFactoryBean;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.StringUtils;
@@ -100,14 +101,14 @@ abstract class AbstractRegionParser extends AbstractSingleBeanDefinitionParser {
 		String hubId = element.getAttribute("hub-id");
 		// Factory will enable gateway if it is not set and hub-id is set.
 		if (StringUtils.hasText(enableGateway)) {
-			if (ParsingUtils.isGemfireV7OrAbove()) {
+			if (GemfireUtils.isGemfireVersion7OrAbove()) {
 				log.warn("'enable-gateway' is deprecated since Gemfire 7.0");
 			}
 		}
 		ParsingUtils.setPropertyValue(element, builder, "enable-gateway");
 
 		if (StringUtils.hasText(hubId)) {
-			if (ParsingUtils.isGemfireV7OrAbove()) {
+			if (GemfireUtils.isGemfireVersion7OrAbove()) {
 				log.warn("'hub-id' is deprecated since Gemfire 7.0");
 			}
 			if (!CollectionUtils.isEmpty(DomUtils.getChildElementsByTagName(element, "gateway-sender"))) {

src/main/java/org/springframework/data/gemfire/config/AsyncEventQueueParser.java

@@ -18,6 +18,7 @@ package org.springframework.data.gemfire.config;
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
 import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
 import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.data.gemfire.GemfireUtils;
 import org.springframework.data.gemfire.wan.AsyncEventQueueFactoryBean;
 import org.springframework.util.StringUtils;
 import org.springframework.util.xml.DomUtils;
@@ -58,7 +59,7 @@ public class AsyncEventQueueParser extends AbstractSingleBeanDefinitionParser {
 		ParsingUtils.setPropertyValue(element, builder, "persistent");
 		ParsingUtils.setPropertyValue(element, builder, "parallel");
 		
-		if (ParsingUtils.GEMFIRE_VERSION.compareTo("7.0.1") >= 0 ) {
+		if (GemfireUtils.GEMFIRE_VERSION.compareTo("7.0.1") >= 0 ) {
 			ParsingUtils.setPropertyValue(element, builder, "batch-conflation-enabled");
 			ParsingUtils.setPropertyValue(element, builder, "disk-synchronous");
 			ParsingUtils.setPropertyValue(element, builder, "batch-time-interval");

src/main/java/org/springframework/data/gemfire/config/ClientRegionParser.java

@@ -29,10 +29,8 @@ import org.springframework.util.StringUtils;
 import org.springframework.util.xml.DomUtils;
 import org.w3c.dom.Element;
 
-import com.gemstone.gemfire.cache.DataPolicy;
-
 /**
- * Parser for <client-region;gt; definitions.
+ * Parser for <client-region;gt; bean definitions.
  * <p/>
  * To avoid eager evaluations, the region interests are declared as nested definition.
  * <p/>
@@ -51,54 +49,37 @@ class ClientRegionParser extends AbstractRegionParser {
 	protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
 		super.doParse(element, builder);
 
-		String cacheRefAttribute = element.getAttribute("cache-ref");
+		validateDataPolicyShortcutMutualExclusion(element, parserContext);
 
-		builder.addPropertyReference("cache", (StringUtils.hasText(cacheRefAttribute) ? cacheRefAttribute
+		String cacheRefAttributeValue = element.getAttribute("cache-ref");
+
+		builder.addPropertyReference("cache", (StringUtils.hasText(cacheRefAttributeValue) ? cacheRefAttributeValue
 			: GemfireConstants.DEFAULT_GEMFIRE_CACHE_NAME));
 
 		ParsingUtils.setPropertyValue(element, builder, "close");
 		ParsingUtils.setPropertyValue(element, builder, "destroy");
 		ParsingUtils.setPropertyValue(element, builder, "data-policy", "dataPolicyName");
 		ParsingUtils.setPropertyValue(element, builder, "name");
+		ParsingUtils.setPropertyValue(element, builder, "persistent");
 		ParsingUtils.setPropertyValue(element, builder, "pool-name");
 		ParsingUtils.setPropertyValue(element, builder, "shortcut");
 
 		parseDiskStoreAttribute(element, builder);
 
-		boolean dataPolicyFrozen = false;
-
-		// TODO why is the DataPolicy determined in the ClientRegionParser and not in the ClientRegionFactoryBean when evaluating the configuration settings?
-		// set the persistent policy
-		if (Boolean.parseBoolean(element.getAttribute("persistent"))) {
-			builder.addPropertyValue("dataPolicy", DataPolicy.PERSISTENT_REPLICATE);
-			dataPolicyFrozen = true;
-		}
-
-		// eviction + expiration + overflow + optional client region attributes
+		// Client RegionAttributes for overflow/eviction, expiration and statistics
 		BeanDefinitionBuilder regionAttributesBuilder = BeanDefinitionBuilder.genericBeanDefinition(
 			RegionAttributesFactoryBean.class);
 
-		boolean overwriteDataPolicy = ParsingUtils.parseEviction(parserContext, element, regionAttributesBuilder);
-
 		ParsingUtils.parseOptionalRegionAttributes(parserContext, element, regionAttributesBuilder);
 		ParsingUtils.parseStatistics(element, regionAttributesBuilder);
-		// NOTE parsing 'expiration' attributes must happen after parsing the user-defined setting for 'statistics'
-		// in GemFire this attribute corresponds to the RegionAttributes 'statistics-enabled' setting).  If the user
-		// configured expiration settings (any?), then statistics must be enabled, regardless if the user explicitly
-		// disabled them.
+		ParsingUtils.parseEviction(parserContext, element, regionAttributesBuilder);
 		ParsingUtils.parseExpiration(parserContext, element, regionAttributesBuilder);
 
-		// TODO understand why this determination is not made in the FactoryBean?
-		if (!dataPolicyFrozen && overwriteDataPolicy) {
-			builder.addPropertyValue("dataPolicy", DataPolicy.NORMAL);
-		}
-
 		builder.addPropertyValue("attributes", regionAttributesBuilder.getBeanDefinition());
 
 		List<Element> subElements = DomUtils.getChildElements(element);
 		ManagedList<Object> interests = new ManagedList<Object>();
 
-		// parse nested elements
 		for (Element subElement : subElements) {
 			String subElementLocalName = subElement.getLocalName();
 
@@ -115,20 +96,19 @@ class ClientRegionParser extends AbstractRegionParser {
 					parserContext, subElement, builder));
 			}
 			else if ("key-interest".equals(subElementLocalName)) {
-				interests.add(parseKeyInterest(parserContext, subElement));
+				interests.add(parseKeyInterest(subElement, parserContext));
 			}
 			else if ("regex-interest".equals(subElementLocalName)) {
-				interests.add(parseRegexInterest(parserContext, subElement));
+				interests.add(parseRegexInterest(subElement));
 			}
 		}
 
-		// TODO is adding an 'interests' property really based on whether there are "sub-elements", or should it be based on whether there are "interests" (as in 'key-interest' and 'regex-interest')?
-		if (!subElements.isEmpty()) {
+		if (!interests.isEmpty()) {
 			builder.addPropertyValue("interests", interests);
 		}
 	}
 
-	private void parseDiskStoreAttribute(Element element, BeanDefinitionBuilder builder) {
+	private void parseDiskStoreAttribute(final Element element, final BeanDefinitionBuilder builder) {
 		String diskStoreRefAttribute = element.getAttribute("disk-store-ref");
 
 		if (StringUtils.hasText(diskStoreRefAttribute)) {
@@ -137,32 +117,37 @@ class ClientRegionParser extends AbstractRegionParser {
 		}
 	}
 
+	private void validateDataPolicyShortcutMutualExclusion(final Element element, final ParserContext parserContext) {
+		if (element.hasAttribute("data-policy") && element.hasAttribute("shortcut")) {
+			parserContext.getReaderContext().error(String.format(
+				"Only one of [data-policy, shortcut] may be specified with element '%1$s'.", element.getTagName()),
+					element);
+		}
+	}
+
 	@Override
 	protected void doParseRegion(Element element, ParserContext parserContext, BeanDefinitionBuilder builder,
 			boolean subRegion) {
 		throw new UnsupportedOperationException(String.format(
-			"doParseRegion(:Element, :ParserContext, :BeanDefinitionBuilder, subRegion:boolean) is not supported on %1$s",
-			getClass().getName()));
+			"doParseRegion(:Element, :ParserContext, :BeanDefinitionBuilder, :boolean) is not supported on %1$s",
+				getClass().getName()));
 	}
 
-	private Object parseKeyInterest(ParserContext parserContext, Element subElement) {
+	private Object parseKeyInterest(Element keyInterestElement, ParserContext parserContext) {
 		BeanDefinitionBuilder keyInterestBuilder = BeanDefinitionBuilder.genericBeanDefinition(Interest.class);
 
-		parseCommonInterestAttributes(subElement, keyInterestBuilder);
-
-		Object key = ParsingUtils.parseRefOrNestedBeanDeclaration(parserContext, subElement, keyInterestBuilder,
-			"key-ref");
-
-		keyInterestBuilder.addConstructorArgValue(key);
+		parseCommonInterestAttributes(keyInterestElement, keyInterestBuilder);
+		keyInterestBuilder.addConstructorArgValue(ParsingUtils.parseRefOrNestedBeanDeclaration(parserContext,
+			keyInterestElement, keyInterestBuilder, "key-ref"));
 
 		return keyInterestBuilder.getBeanDefinition();
 	}
 
-	private Object parseRegexInterest(ParserContext parserContext, Element subElement) {
+	private Object parseRegexInterest(Element regexInterestElement) {
 		BeanDefinitionBuilder regexInterestBuilder = BeanDefinitionBuilder.genericBeanDefinition(RegexInterest.class);
 
-		parseCommonInterestAttributes(subElement, regexInterestBuilder);
-		ParsingUtils.setPropertyValue(subElement, regexInterestBuilder, "pattern", "key");
+		parseCommonInterestAttributes(regexInterestElement, regexInterestBuilder);
+		ParsingUtils.setPropertyValue(regexInterestElement, regexInterestBuilder, "pattern", "key");
 
 		return regexInterestBuilder.getBeanDefinition();
 	}

src/main/java/org/springframework/data/gemfire/config/ParsingUtils.java

@@ -26,11 +26,11 @@ import org.springframework.beans.factory.support.ManagedArray;
 import org.springframework.beans.factory.support.ManagedList;
 import org.springframework.beans.factory.xml.ParserContext;
 import org.springframework.core.Conventions;
+import org.springframework.data.gemfire.GemfireUtils;
 import org.springframework.util.StringUtils;
 import org.springframework.util.xml.DomUtils;
 import org.w3c.dom.Element;
 
-import com.gemstone.gemfire.cache.CacheFactory;
 import com.gemstone.gemfire.cache.ExpirationAction;
 import com.gemstone.gemfire.cache.ExpirationAttributes;
 import com.gemstone.gemfire.cache.LossAction;
@@ -48,9 +48,7 @@ import com.gemstone.gemfire.cache.Scope;
  */
 abstract class ParsingUtils {
 
-	private static Log log = LogFactory.getLog(ParsingUtils.class);
-
-	static final String GEMFIRE_VERSION = CacheFactory.getVersion();
+	private static final Log log = LogFactory.getLog(ParsingUtils.class);
 
 	static void setPropertyValue(Element element, BeanDefinitionBuilder builder, String attributeName,
 			String propertyName, Object defaultValue) {
@@ -310,7 +308,7 @@ abstract class ParsingUtils {
 		
 		String concurrencyChecksEnabled = element.getAttribute("concurrency-checks-enabled");
 		if (StringUtils.hasText(concurrencyChecksEnabled)) {
-			if (!ParsingUtils.isGemfireV7OrAbove()) {
+			if (!GemfireUtils.isGemfireVersion7OrAbove()) {
 				log.warn("'concurrency-checks-enabled' is only available in Gemfire 7.0 or above");
 			} else {
 				ParsingUtils.setPropertyValue(element, attrBuilder, "concurrency-checks-enabled");
@@ -348,21 +346,14 @@ abstract class ParsingUtils {
 	}
 
 	static void throwExceptionIfNotGemfireV7(String elementName, String attributeName, ParserContext parserContext) {
-		if (!isGemfireV7OrAbove()) {
+		if (!GemfireUtils.isGemfireVersion7OrAbove()) {
 			String messagePrefix = (attributeName == null) ? "element '" + elementName + "'" : "attribute '"
 					+ attributeName + " of element '" + elementName + "'";
 			parserContext.getReaderContext().error(
-					messagePrefix + " requires Gemfire version 7 or later. The current version is " + GEMFIRE_VERSION,
+					messagePrefix + " requires Gemfire version 7 or later. The current version is " + GemfireUtils.GEMFIRE_VERSION,
 					null);
 		}
 	}
-	
-	static boolean isGemfireV7OrAbove() {
-		
-		int version = Integer.parseInt(GEMFIRE_VERSION.substring(0,1));
-		return version >= 7;
-		
-	}
 
 	static void parseScope(Element element, BeanDefinitionBuilder builder) {
 		String scope = element.getAttribute("scope");