pjfanning opened a new pull request, #1878: URL: https://github.com/apache/pekko-connectors/pull/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 — `CodecPool.returnCompressor` appears nowhere in the repo. `rotate()` builds a new writer that takes another one, so the pool drained on the **normal** path too, and each rotation allocated fresh native (zlib/snappy) 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 there 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`: ``` temp file hdfs://localhost:54310/tmp/pekko-connectors-hdfs/0 was not closed, so its length is not committed: 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 — without that it passes alone but fails in a full run. 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. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
