Skip to content

Commit 1765c8d

Browse files
Colin McCabepwendell
authored andcommitted
SPARK-1518: FileLogger: Fix compile against Hadoop trunk
In Hadoop trunk (currently Hadoop 3.0.0), the deprecated FSDataOutputStream#sync() method has been removed. Instead, we should call FSDataOutputStream#hflush, which does the same thing as the deprecated method used to do. Author: Colin McCabe <[email protected]> Closes apache#898 from cmccabe/SPARK-1518 and squashes the following commits: 752b9d7 [Colin McCabe] FileLogger: Fix compile against Hadoop trunk
1 parent 189df16 commit 1765c8d

File tree

1 file changed

+12
-4
lines changed

1 file changed

+12
-4
lines changed

core/src/main/scala/org/apache/spark/util/FileLogger.scala

Lines changed: 12 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -61,6 +61,14 @@ private[spark] class FileLogger(
6161
// Only defined if the file system scheme is not local
6262
private var hadoopDataStream: Option[FSDataOutputStream] = None
6363

64+
// The Hadoop APIs have changed over time, so we use reflection to figure out
65+
// the correct method to use to flush a hadoop data stream. See SPARK-1518
66+
// for details.
67+
private val hadoopFlushMethod = {
68+
val cls = classOf[FSDataOutputStream]
69+
scala.util.Try(cls.getMethod("hflush")).getOrElse(cls.getMethod("sync"))
70+
}
71+
6472
private var writer: Option[PrintWriter] = None
6573

6674
/**
@@ -149,13 +157,13 @@ private[spark] class FileLogger(
149157
/**
150158
* Flush the writer to disk manually.
151159
*
152-
* If the Hadoop FileSystem is used, the underlying FSDataOutputStream (r1.0.4) must be
153-
* sync()'ed manually as it does not support flush(), which is invoked by when higher
154-
* level streams are flushed.
160+
* When using a Hadoop filesystem, we need to invoke the hflush or sync
161+
* method. In HDFS, hflush guarantees that the data gets to all the
162+
* DataNodes.
155163
*/
156164
def flush() {
157165
writer.foreach(_.flush())
158-
hadoopDataStream.foreach(_.sync())
166+
hadoopDataStream.foreach(hadoopFlushMethod.invoke(_))
159167
}
160168

161169
/**

0 commit comments

Comments
 (0)