GEODE-4295: convert loaded PdxInstance (apache#1453)

If a loader returns a PdxInstance, and if
the cache has pdx-read-serialized false,
then the PdxInstance is now deserialized
by calling PdxInstance.getObject.
So a "get" of a loaded PdxInstance will now
only be a PdxInstance if pdx-read-serialized
is true or if PdxInstance.getObject returns
a PdxInstance.

Author: dschneider-pivotal

geode-core/src/main/java/org/apache/geode/internal/cache/GemFireCacheImpl.java

@@ -5289,4 +5289,13 @@ public void invokeRegionEntrySynchronizationListenersAfterSynchronization(
       }
     }
   }
+
+  @Override
+  public Object convertPdxInstanceIfNeeded(Object obj) {
+    Object result = obj;
+    if (!this.getPdxReadSerialized() && obj instanceof PdxInstance) {
+      result = ((PdxInstance) obj).getObject();
+    }
+    return result;
+  }
 }

geode-core/src/main/java/org/apache/geode/internal/cache/HARegion.java

@@ -363,13 +363,7 @@ protected Object findObjectInSystem(KeyInfo keyInfo, boolean isCreate, TXStateIn
     Assert.assertTrue(!hasServerProxy());
     CacheLoader loader = basicGetLoader();
     if (loader != null) {
-      final LoaderHelper loaderHelper =
-          loaderHelperFactory.createLoaderHelper(key, aCallbackArgument,
-              false /* netSearchAllowed */, true /* netloadallowed */, null/* searcher */);
-      try {
-        value = loader.load(loaderHelper);
-      } finally {
-      }
+      value = callCacheLoader(loader, key, aCallbackArgument);
 
       if (value != null) {
         try {

geode-core/src/main/java/org/apache/geode/internal/cache/InternalCache.java

@@ -346,4 +346,13 @@ void invokeRegionEntrySynchronizationListenersAfterSynchronization(
   Set<AsyncEventQueue> getAsyncEventQueues(boolean visibleOnly);
 
   void closeDiskStores();
+
+  /**
+   * If obj is a PdxInstance and pdxReadSerialized is not true
+   * then convert obj by calling PdxInstance.getObject.
+   *
+   * @return either the original obj if no conversion was needed;
+   *         or the result of calling PdxInstance.getObject on obj.
+   */
+  Object convertPdxInstanceIfNeeded(Object obj);
 }

geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegion.java

@@ -2770,14 +2770,11 @@ protected Object findObjectInSystem(KeyInfo keyInfo, boolean isCreate, TXStateIn
       // copy into local var to prevent race condition
       CacheLoader loader = basicGetLoader();
       if (loader != null) {
-        final LoaderHelper loaderHelper =
-            this.loaderHelperFactory.createLoaderHelper(key, aCallbackArgument,
-                false /* netSearchAllowed */, true /* netloadAllowed */, null /* searcher */);
+        fromServer = false;
         CachePerfStats stats = getCachePerfStats();
         long statStart = stats.startLoad();
         try {
-          value = loader.load(loaderHelper);
-          fromServer = false;
+          value = callCacheLoader(loader, key, aCallbackArgument);
         } finally {
           stats.endLoad(statStart);
         }
@@ -2863,6 +2860,16 @@ protected Object findObjectInSystem(KeyInfo keyInfo, boolean isCreate, TXStateIn
     return value;
   }
 
+  @SuppressWarnings({"rawtypes", "unchecked"})
+  protected Object callCacheLoader(CacheLoader loader, final Object key,
+      final Object aCallbackArgument) {
+    LoaderHelper loaderHelper = this.loaderHelperFactory.createLoaderHelper(key, aCallbackArgument,
+        false /* netSearchAllowed */, true /* netloadAllowed */, null /* searcher */);
+    Object result = loader.load(loaderHelper);
+    result = this.getCache().convertPdxInstanceIfNeeded(result);
+    return result;
+  }
+
   protected boolean isMemoryThresholdReachedForLoad() {
     return this.memoryThresholdReached.get();
   }

geode-core/src/main/java/org/apache/geode/internal/cache/SearchLoadAndWriteProcessor.java

@@ -790,6 +790,7 @@ private Object doLocalLoad(CacheLoader loader, boolean netSearchAllowed)
       long statStart = stats.startLoad();
       try {
         obj = loader.load(loaderHelper);
+        obj = this.region.getCache().convertPdxInstanceIfNeeded(obj);
       } finally {
         stats.endLoad(statStart);
       }
@@ -2226,6 +2227,9 @@ private void doLoad(ClusterDistributionManager dm) {
             long start = stats.startLoad();
             try {
               Object o = loader.load(loaderHelper);
+              // no need to call convertPdxInstanceIfNeeded since we are serializing
+              // this into the NetLoadRequestMessage. The loaded object will be deserialized
+              // on the other side and have the correct form in that member.
               Assert.assertTrue(o != Token.INVALID && o != Token.LOCAL_INVALID);
               NetLoadReplyMessage.sendMessage(NetLoadRequestMessage.this.getSender(), processorId,
                   o, dm, loaderHelper.getArgument(), null, false, false);

geode-core/src/main/java/org/apache/geode/internal/cache/xmlcache/CacheCreation.java

@@ -2265,4 +2265,9 @@ public void invokeRegionEntrySynchronizationListenersAfterSynchronization(
       List<InitialImageOperation.Entry> entriesToSynchronize) {
     throw new UnsupportedOperationException(LocalizedStrings.SHOULDNT_INVOKE.toLocalizedString());
   }
+
+  @Override
+  public Object convertPdxInstanceIfNeeded(Object obj) {
+    throw new UnsupportedOperationException(LocalizedStrings.SHOULDNT_INVOKE.toLocalizedString());
+  }
 }

geode-core/src/test/java/org/apache/geode/pdx/PdxInstanceLoaderIntegrationTest.java

@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+package org.apache.geode.pdx;
+
+import static com.googlecode.catchexception.CatchException.catchException;
+import static com.googlecode.catchexception.CatchException.caughtException;
+import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import org.apache.geode.cache.Cache;
+import org.apache.geode.cache.CacheFactory;
+import org.apache.geode.cache.CacheLoader;
+import org.apache.geode.cache.CacheLoaderException;
+import org.apache.geode.cache.LoaderHelper;
+import org.apache.geode.cache.Region;
+import org.apache.geode.cache.RegionShortcut;
+import org.apache.geode.test.junit.categories.IntegrationTest;
+import org.apache.geode.test.junit.categories.SerializationTest;
+
+@Category({IntegrationTest.class, SerializationTest.class})
+public class PdxInstanceLoaderIntegrationTest {
+
+  private Cache cache;
+
+  @Before
+  public void setUp() {}
+
+  @After
+  public void tearDown() {
+    cache.close();
+  }
+
+  private class PdxInstanceLoader implements CacheLoader {
+    @Override
+    public void close() {}
+
+    @Override
+    public Object load(LoaderHelper helper) throws CacheLoaderException {
+      return cache.createPdxInstanceFactory("no class name").create();
+    }
+  }
+
+  @Test
+  public void loadOfPdxInstanceWithReadSerializedFalseAttemptsToDeserialize() {
+    cache = new CacheFactory().set(MCAST_PORT, "0").setPdxReadSerialized(false).create();
+    Region region = cache.createRegionFactory(RegionShortcut.LOCAL)
+        .setCacheLoader(new PdxInstanceLoader()).create("region");
+    catchException(region).get("key");
+    assertThat((Exception) caughtException()).isInstanceOf(PdxSerializationException.class);
+  }
+
+  @Test
+  public void loadOfPdxInstanceWithReadSerializedTrueReturnsPdxInstance() {
+    cache = new CacheFactory().set(MCAST_PORT, "0").setPdxReadSerialized(true).create();
+    Region region = cache.createRegionFactory(RegionShortcut.LOCAL)
+        .setCacheLoader(new PdxInstanceLoader()).create("region");
+    Object obj = region.get("key");
+    assertThat(obj).isInstanceOf(PdxInstance.class);
+  }
+}