CAMEL-24195: camel-aws2-s3 - Fix StreamUploadProducer race condition and timeout task exception handling#24872
Conversation
…ption handling in StreamUploadProducer Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> Signed-off-by: Claus Ibsen <claus.ibsen@gmail.com>
|
🌟 Thank you for your contribution to the Apache Camel project! 🌟 🐫 Apache Camel Committers, please review the following items:
|
gnodet
left a comment
gnodet
left a comment
There was a problem hiding this comment.
Claude Code on behalf of gnodet
CAMEL-24195: StreamUploadProducer race condition and timeout task exception handling — LGTM ✅
Two well-targeted concurrency fixes:
1. Timeout task exception handling ✅
Wrapping uploadPart()/completeUpload() in try-catch inside StreamingUploadTimeoutTask.run() is the right call. ScheduledExecutorService.scheduleAtFixedRate silently cancels all future executions when the task throws an uncaught exception — so any transient S3 API error would permanently disable the timeout flush mechanism. Now it logs a warning and continues.
Design note: uploadAggregate = null runs even on failure, which means the current aggregate's data is discarded on error. This is a reasonable tradeoff — in the original code, the exception would kill the timeout task entirely, so the data would be lost anyway (stuck in memory forever with no future flushes). The new behavior is strictly better: current aggregate lost, but future timeout flushes keep working.
If retry semantics are ever needed, uploadAggregate = null could move inside the try (only null on success), but that would require careful handling of partial state (e.g., uploadPart succeeded, completeUpload failed → retrying uploadPart could duplicate parts).
2. Race condition fix ✅
The initial maxRead computation in processWithoutTimestampGrouping() now runs under the lock. I verified that subsequent uploadAggregate accesses inside the while loop are already protected by lock.lock(), so this closes the only unprotected window. There's a minor TOCTOU between releasing this lock and the first loop iteration (the timeout task could null uploadAggregate), but that only results in a slightly suboptimal maxRead on one iteration — not a correctness issue since the loop re-acquires the lock before accessing the aggregate.
Clean, minimal, well-scoped fix.
|
🧪 CI tested the following changed modules:
🔬 Scalpel shadow comparison — Scalpel: 10 tested, 29 compile-only — current: 10 all testedMaveniverse Scalpel detected 39 affected modules (current approach: 10).
|
2 participants
Summary
Claude Code on behalf of davsclaus
Fixes two concurrency bugs in
AWS2S3StreamUploadProducer:Race condition:
uploadAggregatewas read and mutated outside the lock inprocessWithoutTimestampGrouping()(lines 181-184). The timeout task acquires the lock and can setuploadAggregate = null, so the main thread could dereference a null reference (NPE). Fix: moved the reads inside a lock block.Timeout task exception handling:
StreamingUploadTimeoutTask.run()did not catch exceptions fromuploadPart()/completeUpload(). SinceScheduledExecutorService.scheduleAtFixedRatesilently cancels all future executions when the task throws an uncaught exception, any S3 API error would permanently disable the timeout flush mechanism. Fix: wrapped the upload calls in try-catch to log and continue.Test plan
mvn install -DskipTests)🤖 Generated with Claude Code