The GitHub Actions job "Update Dependency Graph" on pekko-connectors.git/main has succeeded. Run started by GitHub user pjfanning (triggered by pjfanning).
Head commit for run: 18033ce105f4a9ceb29cdf423361443cd690eabd / PJ Fanning <[email protected]> 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. Report URL: https://github.com/apache/pekko-connectors/actions/runs/35325350896 With regards, GitHub Actions via GitBox --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
