Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ package org.apache.spark.scheduler

import java.io.{DataInputStream, DataOutputStream}
import java.nio.ByteBuffer
import java.nio.charset.StandardCharsets
import java.util.Properties

import scala.collection.JavaConverters._
Expand Down Expand Up @@ -86,7 +87,10 @@ private[spark] object TaskDescription {
dataOut.writeInt(taskDescription.properties.size())
taskDescription.properties.asScala.foreach { case (key, value) =>
dataOut.writeUTF(key)
dataOut.writeUTF(value)
// SPARK-19796 -- writeUTF doesn't work for long strings, which can happen for property values
val bytes = value.getBytes(StandardCharsets.UTF_8)
dataOut.writeInt(bytes.length)
dataOut.write(bytes)
}

// Write the task. The task is already serialized, so write it directly to the byte buffer.
Expand Down Expand Up @@ -124,7 +128,11 @@ private[spark] object TaskDescription {
val properties = new Properties()
val numProperties = dataIn.readInt()
for (i <- 0 until numProperties) {
properties.setProperty(dataIn.readUTF(), dataIn.readUTF())
val key = dataIn.readUTF()
val valueLength = dataIn.readInt()
val valueBytes = new Array[Byte](valueLength)
dataIn.readFully(valueBytes)
properties.setProperty(key, new String(valueBytes, StandardCharsets.UTF_8))
}

// Create a sub-buffer for the serialized task into its own buffer (to be deserialized later).
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@

package org.apache.spark.scheduler

import java.io.{ByteArrayOutputStream, DataOutputStream, UTFDataFormatException}
import java.nio.ByteBuffer
import java.util.Properties

Expand All @@ -36,6 +37,21 @@ class TaskDescriptionSuite extends SparkFunSuite {
val originalProperties = new Properties()
originalProperties.put("property1", "18")
originalProperties.put("property2", "test value")
// SPARK-19796 -- large property values (like a large job description for a long sql query)
// can cause problems for DataOutputStream, make sure we handle correctly
val sb = new StringBuilder()
(0 to 10000).foreach(_ => sb.append("1234567890"))
val largeString = sb.toString()
originalProperties.put("property3", largeString)
// make sure we've got a good test case
intercept[UTFDataFormatException] {
val out = new DataOutputStream(new ByteArrayOutputStream())
try {
out.writeUTF(largeString)
} finally {
out.close()
}
}

// Create a dummy byte buffer for the task.
val taskBuffer = ByteBuffer.wrap(Array[Byte](1, 2, 3, 4))
Expand Down