Don't use GiraphClassResolver when ZK is not initialized

Add boxed interface

giraph-core/src/main/java/org/apache/giraph/writable/kryo/Boxed.java

@@ -0,0 +1,36 @@
+/*
+ * 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.giraph.writable.kryo;
+
+/**
+ * Boxed interface
+ * @param <T>
+ */
+public interface Boxed<T> {
+  /**
+   * Gets the boxed value.
+   * @return Boxed object.
+   */
+  T get();
+
+  /**
+   * Sets the boxed value.
+   * @param value Value
+   */
+  void set(T value);
+}

giraph-core/src/main/java/org/apache/giraph/writable/kryo/GiraphClassResolver.java

@@ -95,6 +95,14 @@ public static void setZookeeperInfo(ZooKeeperExt zookeeperExt,
     KRYO_REGISTERED_CLASS_PATH = kryoClassPath;
   }
 
+  /**
+   * Return true of the zookeeper is initialized.
+   * @return True if the zookeeper is initialized.
+   */
+  public static boolean isInitialized() {
+    return ZK != null;
+  }
+
   /**
    * Creates a new node for the given class name.
    * Creation mode is persistent sequential, i.e.

giraph-core/src/main/java/org/apache/giraph/writable/kryo/HadoopKryo.java

@@ -26,6 +26,7 @@
 import java.util.Map.Entry;
 import java.util.Random;
 
+import com.esotericsoftware.kryo.util.DefaultClassResolver;
 import org.apache.giraph.conf.GiraphConfigurationSettable;
 import com.esotericsoftware.kryo.ClassResolver;
 import com.esotericsoftware.kryo.ReferenceResolver;
@@ -307,7 +308,12 @@ private static HadoopKryo createKryo(boolean trackReferences,
     if (trackReferences) {
       kryo = new HadoopKryo();
     } else {
-      kryo = new HadoopKryo(new GiraphClassResolver(),
+      // Only use GiraphClassResolver if it is properly initialized.
+      // This is to enable test cases which use KryoSimpleWrapper
+      // but don't start ZK.
+      kryo = new HadoopKryo(
+              GiraphClassResolver.isInitialized() ? new GiraphClassResolver() :
+                                                    new DefaultClassResolver(),
               new MapReferenceResolver());
     }
 
@@ -403,7 +409,12 @@ public void write(Kryo kryo, Output output, Object object) {
 
     if (!trackReferences) {
       kryo.setReferences(false);
-      kryo.setAutoReset(false);
+
+      // Auto reset can only be disabled if the GiraphClassResolver is
+      // properly initialized.
+      if (GiraphClassResolver.isInitialized()) {
+        kryo.setAutoReset(false);
+      }
     }
     return kryo;
   }

giraph-core/src/main/java/org/apache/giraph/writable/kryo/KryoSimpleWrapper.java

@@ -39,7 +39,7 @@
  *
  * @param <T> Object type
  */
-public class KryoSimpleWrapper<T> implements Writable {
+public class KryoSimpleWrapper<T> implements Writable, Boxed<T> {
 
   /** Wrapped object */
   private T object;