[SPARK-16003] SerializationDebugger runs into infinite loop

## What changes were proposed in this pull request?

This fixes SerializationDebugger to not recurse forever when `writeReplace` returns an object of the same class, which is the case for at least the `SQLMetrics` class.

See also the OpenJDK unit tests on the behavior of recursive `writeReplace()`:
https://github.com/openjdk-mirror/jdk7u-jdk/blob/f4d80957e89a19a29bb9f9807d2a28351ed7f7df/test/java/io/Serializable/nestedReplace/NestedReplace.java

cc davies cloud-fan

## How was this patch tested?

Unit tests for SerializationDebugger.

Author: Eric Liang <ekl@databricks.com>

Closes #13814 from ericl/spark-16003.

core/src/main/scala/org/apache/spark/serializer/SerializationDebugger.scala

@@ -155,7 +155,7 @@ private[spark] object SerializationDebugger extends Logging {
 
       // If the object has been replaced using writeReplace(),
       // then call visit() on it again to test its type again.
-      if (!finalObj.eq(o)) {
+      if (finalObj.getClass != o.getClass) {
         return visit(finalObj, s"writeReplace data (class: ${finalObj.getClass.getName})" :: stack)
       }
 
@@ -265,11 +265,10 @@ private[spark] object SerializationDebugger extends Logging {
     if (!desc.hasWriteReplaceMethod) {
       (o, desc)
     } else {
-      // write place
       val replaced = desc.invokeWriteReplace(o)
-      // `writeReplace` may return the same object.
-      if (replaced eq o) {
-        (o, desc)
+      // `writeReplace` recursion stops when the returned object has the same class.
+      if (replaced.getClass == o.getClass) {
+        (replaced, desc)
       } else {
         findObjectAndDescriptor(replaced)
       }

core/src/test/scala/org/apache/spark/serializer/SerializationDebuggerSuite.scala

@@ -126,7 +126,11 @@ class SerializationDebuggerSuite extends SparkFunSuite with BeforeAndAfterEach {
     assert(find(new SerializableClassWithWriteReplace(new SerializableClass1)).isEmpty)
   }
 
-    test("object containing writeObject() and not serializable field") {
+  test("no infinite loop with writeReplace() which returns class of its own type") {
+    assert(find(new SerializableClassWithRecursiveWriteReplace).isEmpty)
+  }
+
+  test("object containing writeObject() and not serializable field") {
     val s = find(new SerializableClassWithWriteObject(new NotSerializable))
     assert(s.size === 3)
     assert(s(0).contains("NotSerializable"))
@@ -229,6 +233,13 @@ class SerializableClassWithWriteReplace(@(transient @param) replacementFieldObje
 }
 
 
+class SerializableClassWithRecursiveWriteReplace extends Serializable {
+  private def writeReplace(): Object = {
+    new SerializableClassWithRecursiveWriteReplace
+  }
+}
+
+
 class ExternalizableClass(objectField: Object) extends java.io.Externalizable {
   val serializableObjectField = new SerializableClass1