Reapply the target fix

Author: jiateoh

sql/core/src/main/scala/org/apache/spark/sql/execution/python/streaming/TransformWithStateInPySparkStateServer.scala

@@ -208,22 +208,7 @@ class TransformWithStateInPySparkStateServer(
   private def parseProtoMessage(): StateRequest = {
     val messageLen = inputStream.readInt()
     val messageBytes = new Array[Byte](messageLen)
-    // inputStream.readFully(messageBytes)
-    // inputStream.read(messageBytes)
-    // /*
-    val readBytes = inputStream.read(messageBytes)
-    if (readBytes != messageLen) {
-      throw new Exception(s"TESTING: Failed to read message bytes: expected $messageLen bytes, " +
-        s"but only read $readBytes bytes")
-    }
-    /*
-    else if (messageLen > 500) {
-      val debug = StateRequest.parseFrom(ByteString.copyFrom(messageBytes))
-      throw new Exception(s"Unexpectedly matched message length: $messageLen bytes, "
-        // + s"debug: ${debug.toString}")
-      )
-    }
-    */
+    inputStream.readFully(messageBytes)
     StateRequest.parseFrom(ByteString.copyFrom(messageBytes))
   }