pjfanning opened a new pull request, #1930:
URL: https://github.com/apache/pekko-connectors/pull/1930
`DiskBuffer` writes each multipart chunk to a temp file and deletes it
through a countdown:
```scala
val deleteCounter = new AtomicInteger(maxMaterializations)
val src = FileIO.fromPath(path.toPath, 65536).mapMaterializedValue { f =>
if (deleteCounter.decrementAndGet() <= 0) f.onComplete(_ => path.delete())
NotUsed
}
```
`maxMaterializations` is `(maxRetriesPerChunk + 1) * 2` — every possible
upload attempt times two materializations per attempt. So the file is deleted
only when *every* retry has been used. A chunk that uploads successfully first
time materializes twice out of a budget of eight, the counter never reaches
zero, and the file stays on disk until the JVM exits via the `deleteOnExit`
registered at line 68.
For a long-running service uploading through the disk buffer that means
every 5MB+ chunk it has ever sent is still on disk, plus a permanent entry in
the JVM-global `DeleteOnExitHook` set, which is never pruned.
### Changes
* `Chunk` gains `dispose()`, a no-op by default; `DiskChunk` carries its
`File` and deletes it.
* `chunkAndRequest` calls `dispose()` in the branch of the `RetryFlow`
decision function that returns `None`. That is the point at which the chunk is
definitively finished with: the request has been sent and the result is not
going to be retried. The retry branch leaves the file alone, since a retry
re-materializes the source.
* `DiskBuffer` deletes its temp file in `postStop` when it never emitted a
chunk, so a failure or cancellation before emission cleans up immediately
instead of at JVM exit.
The existing countdown stays as-is. It now covers the one case disposal does
not: retries exhausted, where `RetryFlow` emits the final failure without
asking the decision function again — and that is exactly when the counter
reaches zero.
`deleteOnExit` is deliberately kept. It is the only remaining net for a
chunk that is emitted and then abandoned mid-flight, and dropping it would
trade a small unbounded registration for leaked multi-megabyte files on
cancelled uploads. With eager disposal the files themselves no longer
accumulate during a run.
`MemoryChunk` inherits the no-op, so the memory buffer path and
`chunkAndRequestWithContext` (memory-only) are unaffected. `buffer = "memory"`
is the default, so this only affects users who opt into `DiskBufferType` or
`MemoryAndDiskBufferType`.
### Testing
Two cases in `DiskBufferSpec`: disposal releases the file after a
materialization count well below the retry budget, and a failure before
emission leaves nothing behind. The existing "after N materializations" test
still passes unchanged.
`s3/test` passes — 169 tests, 20 suites. `MinioS3IntegrationSpec` aborts in
my environment because Docker is not available, not from this change.
`checkCodeStyle` is clean.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
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]
