This is an automated email from the ASF dual-hosted git repository.
pjfanning pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko-connectors.git
The following commit(s) were added to refs/heads/main by this push:
new 18033ce10 HDFS: close the writer when the stream does not complete
normally (#1878)
18033ce10 is described below
commit 18033ce105f4a9ceb29cdf423361443cd690eabd
Author: PJ Fanning <[email protected]>
AuthorDate: Fri Sep 18 09:37:49 2026 +0100
HDFS: close the writer when the stream does not complete normally (#1878)
Motivation:
HdfsFlowLogic had no postStop, and the writer's output is closed only by
HdfsWriter.rotate, which runs on the normal completion path. On any other
stop the output stayed open:
- upstream fails: onUpstreamFailure calls failStage and nothing else
- downstream cancels: the default onDownstreamFinish completes the stage
- a write throws: supervision fails the stage
Each of those leaves an FSDataOutputStream or SequenceFile.Writer open on
an HDFS lease. The file stays "under construction", so the bytes written
before the failure are never committed and the NameNode reports length 0
until the lease expires.
CompressedDataWriter additionally took a Compressor from CodecPool and
never returned it. rotate() builds a new writer that takes another one,
so the pool drained on the normal path too, and each rotation allocated
fresh native memory.
Modification:
Add HdfsWriter.close, implemented by the three writers, and call it from
a new postStop. `output` is lazy, so HdfsWriter records whether it was
forced and close skips it otherwise — a stage that stopped before writing
must not create a file just to close it. This also makes the post-rotation
case a no-op, since rotate returns a fresh writer whose output is unset.
CompressedDataWriter.close and rotate now return the compressor to
CodecPool. Failures are logged rather than rethrown: the stage has already
stopped, so failing here would only mask why.
The temp file is deliberately left in place, as it already is when a
rotation fails.
Result:
Output streams and compressors are released when the stage stops, so the
lease is freed and the bytes written before a failure are committed.
Tests:
Added a test to HdfsWriterSpec that fails a stream after it has written,
then asserts the temp file has a committed non-zero length. Verified it
fails without the postStop ("0 was not greater than 0") and passes with
it. It filters temp files by modification time because other tests in the
suite leave their own behind. Full hdfs suite passes, 19 tests. Ran
hdfs/mimaReportBinaryIssues (clean) and the module scalafmt checks.
References:
Compare HdfsWriter.rotate, the only other place output is closed.
---
.../connectors/hdfs/impl/HdfsFlowStage.scala | 15 ++++++-
.../hdfs/impl/writer/CompressedDataWriter.scala | 9 ++++
.../connectors/hdfs/impl/writer/DataWriter.scala | 3 ++
.../connectors/hdfs/impl/writer/HdfsWriter.scala | 22 +++++++++-
.../hdfs/impl/writer/SequenceWriter.scala | 3 ++
.../test/scala/docs/scaladsl/HdfsWriterSpec.scala | 49 +++++++++++++++++++++-
6 files changed, 98 insertions(+), 3 deletions(-)
diff --git
a/hdfs/src/main/scala/org/apache/pekko/stream/connectors/hdfs/impl/HdfsFlowStage.scala
b/hdfs/src/main/scala/org/apache/pekko/stream/connectors/hdfs/impl/HdfsFlowStage.scala
index fcf8410b2..c48a1e2b7 100644
---
a/hdfs/src/main/scala/org/apache/pekko/stream/connectors/hdfs/impl/HdfsFlowStage.scala
+++
b/hdfs/src/main/scala/org/apache/pekko/stream/connectors/hdfs/impl/HdfsFlowStage.scala
@@ -25,6 +25,7 @@ import pekko.stream.stage._
import cats.data.State
import scala.concurrent.duration.FiniteDuration
+import scala.util.control.NonFatal
/**
* Internal API
@@ -61,7 +62,8 @@ private[hdfs] final class HdfsFlowLogic[W, I, C](
outlet: Outlet[OutgoingMessage[C]],
shape: FlowShape[HdfsWriteMessage[I, C], OutgoingMessage[C]]) extends
TimerGraphStageLogic(shape)
with InHandler
- with OutHandler {
+ with OutHandler
+ with StageLogging {
private var state = FlowState(initialHdfsWriter, initialRotationStrategy,
initialSyncStrategy)
@@ -92,6 +94,17 @@ private[hdfs] final class HdfsFlowLogic[W, I, C](
override def onUpstreamFailure(ex: Throwable): Unit =
failStage(ex)
+ // The writer's output is only closed by `rotate`, which runs on the normal
+ // completion path. On any other stop — upstream failure, downstream cancel,
+ // or a write throwing — the output would stay open on an HDFS lease, so it
+ // is released here. Closing twice is harmless; the temp file is deliberately
+ // left in place, as it is on a failed rotation.
+ override def postStop(): Unit =
+ try state.writer.close()
+ catch {
+ case NonFatal(ex) => log.error(ex, "Problem occurred during writer
close")
+ }
+
override def onUpstreamFinish(): Unit =
state.logicState match {
case LogicState.Writing =>
diff --git
a/hdfs/src/main/scala/org/apache/pekko/stream/connectors/hdfs/impl/writer/CompressedDataWriter.scala
b/hdfs/src/main/scala/org/apache/pekko/stream/connectors/hdfs/impl/writer/CompressedDataWriter.scala
index 66d8ec94c..ffd318006 100644
---
a/hdfs/src/main/scala/org/apache/pekko/stream/connectors/hdfs/impl/writer/CompressedDataWriter.scala
+++
b/hdfs/src/main/scala/org/apache/pekko/stream/connectors/hdfs/impl/writer/CompressedDataWriter.scala
@@ -52,9 +52,18 @@ private[writer] final case class CompressedDataWriter(
override def rotate(rotationCount: Long): CompressedDataWriter = {
cmpOutput.finish()
output.close()
+ // the compressor came from CodecPool and holds native memory; a rotation
+ // builds a new writer that takes another one, so this one has to go back
+ CodecPool.returnCompressor(compressor)
copy(maybeTargetPath = Some(outputFileWithExtension(rotationCount)))
}
+ override def close(): Unit = {
+ // `cmpOutput` wraps `output`, so creating it already forced the lazy value
+ if (isOutputOpened) output.close()
+ CodecPool.returnCompressor(compressor)
+ }
+
override protected def create(fs: FileSystem, file: Path):
FSDataOutputStream = fs.create(file, overwrite)
private def outputFileWithExtension(rotationCount: Long): Path = {
diff --git
a/hdfs/src/main/scala/org/apache/pekko/stream/connectors/hdfs/impl/writer/DataWriter.scala
b/hdfs/src/main/scala/org/apache/pekko/stream/connectors/hdfs/impl/writer/DataWriter.scala
index bc43884a3..db02b4277 100644
---
a/hdfs/src/main/scala/org/apache/pekko/stream/connectors/hdfs/impl/writer/DataWriter.scala
+++
b/hdfs/src/main/scala/org/apache/pekko/stream/connectors/hdfs/impl/writer/DataWriter.scala
@@ -47,6 +47,9 @@ private[writer] final case class DataWriter(
copy(maybeTargetPath = Some(createTargetPath(pathGenerator,
rotationCount)))
}
+ override def close(): Unit =
+ if (isOutputOpened) output.close()
+
override def create(fs: FileSystem, file: Path): FSDataOutputStream =
fs.create(file, overwrite)
}
diff --git
a/hdfs/src/main/scala/org/apache/pekko/stream/connectors/hdfs/impl/writer/HdfsWriter.scala
b/hdfs/src/main/scala/org/apache/pekko/stream/connectors/hdfs/impl/writer/HdfsWriter.scala
index 70e1e174c..5e5a74fb3 100644
---
a/hdfs/src/main/scala/org/apache/pekko/stream/connectors/hdfs/impl/writer/HdfsWriter.scala
+++
b/hdfs/src/main/scala/org/apache/pekko/stream/connectors/hdfs/impl/writer/HdfsWriter.scala
@@ -25,7 +25,17 @@ import org.apache.hadoop.fs.{ FileSystem, Path }
@InternalApi
private[hdfs] trait HdfsWriter[W, I] {
- protected lazy val output: W = create(fs, temp)
+ // `outputOpened` records whether the lazy `output` was forced, so `close`
+ // can skip it rather than creating a file only to close it.
+ private var outputOpened = false
+
+ protected lazy val output: W = {
+ outputOpened = true
+ create(fs, temp)
+ }
+
+ /** Whether `output` has been forced. */
+ protected def isOutputOpened: Boolean = outputOpened
protected lazy val temp: Path = tempFromTarget(pathGenerator, target)
@@ -45,6 +55,16 @@ private[hdfs] trait HdfsWriter[W, I] {
def rotate(rotationCount: Long): HdfsWriter[W, I]
+ /**
+ * Release the underlying output without moving anything to the target.
+ *
+ * Called when the stage stops without completing normally, where `rotate` is
+ * never reached and the output would otherwise stay open on an HDFS lease.
+ * Must be safe to call when nothing was ever written: `output` is lazy, so
+ * implementations have to avoid forcing it.
+ */
+ def close(): Unit
+
protected def target: Path
protected def fs: FileSystem
diff --git
a/hdfs/src/main/scala/org/apache/pekko/stream/connectors/hdfs/impl/writer/SequenceWriter.scala
b/hdfs/src/main/scala/org/apache/pekko/stream/connectors/hdfs/impl/writer/SequenceWriter.scala
index 07aec4662..7fe173d36 100644
---
a/hdfs/src/main/scala/org/apache/pekko/stream/connectors/hdfs/impl/writer/SequenceWriter.scala
+++
b/hdfs/src/main/scala/org/apache/pekko/stream/connectors/hdfs/impl/writer/SequenceWriter.scala
@@ -48,6 +48,9 @@ private[writer] final case class SequenceWriter[K <:
Writable, V <: Writable](
copy(maybeTargetPath = Some(createTargetPath(pathGenerator,
rotationCount)))
}
+ override def close(): Unit =
+ if (isOutputOpened) output.close()
+
override protected def create(fs: FileSystem, file: Path):
SequenceFile.Writer = {
val ops = SequenceFile.Writer.file(file) +: writerOptions
SequenceFile.createWriter(fs.getConf, ops: _*)
diff --git a/hdfs/src/test/scala/docs/scaladsl/HdfsWriterSpec.scala
b/hdfs/src/test/scala/docs/scaladsl/HdfsWriterSpec.scala
index 79aafa51c..78cf7d3fa 100644
--- a/hdfs/src/test/scala/docs/scaladsl/HdfsWriterSpec.scala
+++ b/hdfs/src/test/scala/docs/scaladsl/HdfsWriterSpec.scala
@@ -14,6 +14,7 @@
package docs.scaladsl
import org.apache.pekko
+import pekko.NotUsed
import pekko.actor.ActorSystem
import pekko.stream.connectors.hdfs._
import pekko.stream.connectors.hdfs.scaladsl.HdfsFlow
@@ -28,9 +29,10 @@ import org.apache.hadoop.io.Text
import org.apache.hadoop.io.compress._
import org.apache.hadoop.io.compress.zlib.ZlibCompressor.CompressionLevel
import org.scalatest._
+import org.scalatest.concurrent.Eventually
import scala.concurrent.duration.{ Duration, _ }
-import scala.concurrent.{ Await, ExecutionContextExecutor }
+import scala.concurrent.{ Await, ExecutionContextExecutor, Future }
import org.scalatest.matchers.should.Matchers
import org.scalatest.wordspec.AnyWordSpecLike
@@ -39,6 +41,7 @@ class HdfsWriterSpec
with Matchers
with BeforeAndAfterAll
with BeforeAndAfterEach
+ with Eventually
with LogCapturing {
private var hdfsCluster: MiniDFSCluster = null
@@ -493,6 +496,50 @@ class HdfsWriterSpec
}
}
+ "HdfsFlowStage" should {
+ "release the writer when the stream fails" in {
+ // Regression test: the output is only closed by `rotate`, which runs on
+ // the normal completion path. Without a postStop, a stream that fails
+ // mid-write leaves the temp file open on an HDFS lease, so the file
stays
+ // "under construction" and the bytes written are never committed.
+ val flow = HdfsFlow.data(
+ fs,
+ SyncStrategy.none,
+ RotationStrategy.size(1, FileUnit.GB),
+ settings)
+
+ val startedAt = System.currentTimeMillis()
+ val failure = new RuntimeException("boom")
+ val result = Source(books)
+ .map(HdfsWriteMessage(_))
+
.concat(Source.future(pekko.pattern.after(200.millis)(Future.failed(failure))))
+ .via(flow)
+ .runWith(Sink.ignore)
+
+ Await.ready(result, 10.seconds)
+ result.value.get.isFailure shouldBe true
+
+ // only the file this stage wrote: other tests in this suite leave their
+ // own temp files behind, and afterEach does not always clear them
+ val tempDir = new Path(settings.pathGenerator.tempDirectory)
+ val leftOver =
+ if (fs.exists(tempDir))
fs.listStatus(tempDir).toList.filter(_.getModificationTime >= startedAt)
+ else Nil
+ withClue("the stage should have written a temp file before failing: ") {
+ leftOver should not be empty
+ }
+
+ // A file still held open for write reports length 0 until the lease is
+ // released, because the last block is not committed. Closing the writer
+ // in postStop commits it, so the bytes written before the failure show
up.
+ leftOver.foreach { st =>
+ withClue(s"temp file ${st.getPath} was not closed, so its length is
not committed: ") {
+ fs.getFileStatus(st.getPath).getLen should be > 0L
+ }
+ }
+ }
+ }
+
override protected def afterEach(): Unit = {
fs.delete(new Path(destination), true)
fs.delete(settings.pathGenerator(0, 0).getParent, true)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
