Skip to content
Snippets Groups Projects
Commit 21b4ba2d authored by Shixiong Zhu's avatar Shixiong Zhu
Browse files

[SPARK-19599][SS] Clean up HDFSMetadataLog

## What changes were proposed in this pull request?

SPARK-19464 removed support for Hadoop 2.5 and earlier, so we can do some cleanup for HDFSMetadataLog.

This PR includes the following changes:
- ~~Remove the workaround codes for HADOOP-10622.~~ Unfortunately, there is another issue [HADOOP-14084](https://issues.apache.org/jira/browse/HADOOP-14084) that prevents us from removing the workaround codes.
- Remove unnecessary `writer: (T, OutputStream) => Unit` and just call `serialize` directly.
- Remove catching FileNotFoundException.

## How was this patch tested?

Jenkins

Author: Shixiong Zhu <shixiong@databricks.com>

Closes #16932 from zsxwing/metadata-cleanup.
parent f6c3bba2
No related branches found
No related tags found
No related merge requests found
...@@ -114,15 +114,18 @@ class HDFSMetadataLog[T <: AnyRef : ClassTag](sparkSession: SparkSession, path: ...@@ -114,15 +114,18 @@ class HDFSMetadataLog[T <: AnyRef : ClassTag](sparkSession: SparkSession, path:
case ut: UninterruptibleThread => case ut: UninterruptibleThread =>
// When using a local file system, "writeBatch" must be called on a // When using a local file system, "writeBatch" must be called on a
// [[org.apache.spark.util.UninterruptibleThread]] so that interrupts can be disabled // [[org.apache.spark.util.UninterruptibleThread]] so that interrupts can be disabled
// while writing the batch file. This is because there is a potential dead-lock in // while writing the batch file.
// Hadoop "Shell.runCommand" before 2.5.0 (HADOOP-10622). If the thread running //
// "Shell.runCommand" is interrupted, then the thread can get deadlocked. In our case, // This is because Hadoop "Shell.runCommand" swallows InterruptException (HADOOP-14084).
// `writeBatch` creates a file using HDFS API and will call "Shell.runCommand" to set // If the user tries to stop a query, and the thread running "Shell.runCommand" is
// the file permission if using the local file system, and can get deadlocked if the // interrupted, then InterruptException will be dropped and the query will be still
// stream execution thread is stopped by interrupt. Hence, we make sure that // running. (Note: `writeBatch` creates a file using HDFS APIs and will call
// "writeBatch" is called on [[UninterruptibleThread]] which allows us to disable // "Shell.runCommand" to set the file permission if using the local file system)
// interrupts here. Also see SPARK-14131. //
ut.runUninterruptibly { writeBatch(batchId, metadata, serialize) } // Hence, we make sure that "writeBatch" is called on [[UninterruptibleThread]] which
// allows us to disable interrupts here, in order to propagate the interrupt state
// correctly. Also see SPARK-19599.
ut.runUninterruptibly { writeBatch(batchId, metadata) }
case _ => case _ =>
throw new IllegalStateException( throw new IllegalStateException(
"HDFSMetadataLog.add() on a local file system must be executed on " + "HDFSMetadataLog.add() on a local file system must be executed on " +
...@@ -132,20 +135,19 @@ class HDFSMetadataLog[T <: AnyRef : ClassTag](sparkSession: SparkSession, path: ...@@ -132,20 +135,19 @@ class HDFSMetadataLog[T <: AnyRef : ClassTag](sparkSession: SparkSession, path:
// For a distributed file system, such as HDFS or S3, if the network is broken, write // For a distributed file system, such as HDFS or S3, if the network is broken, write
// operations may just hang until timeout. We should enable interrupts to allow stopping // operations may just hang until timeout. We should enable interrupts to allow stopping
// the query fast. // the query fast.
writeBatch(batchId, metadata, serialize) writeBatch(batchId, metadata)
} }
true true
} }
} }
def writeTempBatch(metadata: T, writer: (T, OutputStream) => Unit = serialize): Option[Path] = { def writeTempBatch(metadata: T): Option[Path] = {
var nextId = 0
while (true) { while (true) {
val tempPath = new Path(metadataPath, s".${UUID.randomUUID.toString}.tmp") val tempPath = new Path(metadataPath, s".${UUID.randomUUID.toString}.tmp")
try { try {
val output = fileManager.create(tempPath) val output = fileManager.create(tempPath)
try { try {
writer(metadata, output) serialize(metadata, output)
return Some(tempPath) return Some(tempPath)
} finally { } finally {
IOUtils.closeQuietly(output) IOUtils.closeQuietly(output)
...@@ -164,7 +166,6 @@ class HDFSMetadataLog[T <: AnyRef : ClassTag](sparkSession: SparkSession, path: ...@@ -164,7 +166,6 @@ class HDFSMetadataLog[T <: AnyRef : ClassTag](sparkSession: SparkSession, path:
// big problem because it requires the attacker must have the permission to write the // big problem because it requires the attacker must have the permission to write the
// metadata path. In addition, the old Streaming also have this issue, people can create // metadata path. In addition, the old Streaming also have this issue, people can create
// malicious checkpoint files to crash a Streaming application too. // malicious checkpoint files to crash a Streaming application too.
nextId += 1
} }
} }
None None
...@@ -176,8 +177,8 @@ class HDFSMetadataLog[T <: AnyRef : ClassTag](sparkSession: SparkSession, path: ...@@ -176,8 +177,8 @@ class HDFSMetadataLog[T <: AnyRef : ClassTag](sparkSession: SparkSession, path:
* There may be multiple [[HDFSMetadataLog]] using the same metadata path. Although it is not a * There may be multiple [[HDFSMetadataLog]] using the same metadata path. Although it is not a
* valid behavior, we still need to prevent it from destroying the files. * valid behavior, we still need to prevent it from destroying the files.
*/ */
private def writeBatch(batchId: Long, metadata: T, writer: (T, OutputStream) => Unit): Unit = { private def writeBatch(batchId: Long, metadata: T): Unit = {
val tempPath = writeTempBatch(metadata, writer).getOrElse( val tempPath = writeTempBatch(metadata).getOrElse(
throw new IllegalStateException(s"Unable to create temp batch file $batchId")) throw new IllegalStateException(s"Unable to create temp batch file $batchId"))
try { try {
// Try to commit the batch // Try to commit the batch
...@@ -195,12 +196,6 @@ class HDFSMetadataLog[T <: AnyRef : ClassTag](sparkSession: SparkSession, path: ...@@ -195,12 +196,6 @@ class HDFSMetadataLog[T <: AnyRef : ClassTag](sparkSession: SparkSession, path:
// So throw an exception to tell the user this is not a valid behavior. // So throw an exception to tell the user this is not a valid behavior.
throw new ConcurrentModificationException( throw new ConcurrentModificationException(
s"Multiple HDFSMetadataLog are using $path", e) s"Multiple HDFSMetadataLog are using $path", e)
case e: FileNotFoundException =>
// Sometimes, "create" will succeed when multiple writers are calling it at the same
// time. However, only one writer can call "rename" successfully, others will get
// FileNotFoundException because the first writer has removed it.
throw new ConcurrentModificationException(
s"Multiple HDFSMetadataLog are using $path", e)
} finally { } finally {
fileManager.delete(tempPath) fileManager.delete(tempPath)
} }
......
...@@ -179,8 +179,8 @@ class StreamExecution( ...@@ -179,8 +179,8 @@ class StreamExecution(
/** /**
* The thread that runs the micro-batches of this stream. Note that this thread must be * The thread that runs the micro-batches of this stream. Note that this thread must be
* [[org.apache.spark.util.UninterruptibleThread]] to avoid potential deadlocks in using * [[org.apache.spark.util.UninterruptibleThread]] to avoid swallowing `InterruptException` when
* [[HDFSMetadataLog]]. See SPARK-14131 for more details. * using [[HDFSMetadataLog]]. See SPARK-19599 for more details.
*/ */
val microBatchThread = val microBatchThread =
new StreamExecutionThread(s"stream execution thread for $prettyIdString") { new StreamExecutionThread(s"stream execution thread for $prettyIdString") {
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment