zhangshenghang commented on PR #10349:
URL: https://github.com/apache/seatunnel/pull/10349#issuecomment-3756852568
<!-- code-pr-reviewer -->
Thanks for the PR! Here are the blocking issues:
**CRITICAL**
- `DatabendSinkAggregatedCommitter.java:238-255` (close()) - **Resource leak
risk**: If `performMerge()` throws an exception (line 243), the
`connection.close()` on line 246 will never execute, leaking database
connections. Consider wrapping `performMerge()` in its own try-catch within the
outer try-finally to ensure cleanup runs.
```java
try {
if (!aborted && isCdcMode && connection != null &&
!connection.isClosed()) {
try {
log.info("[Instance {}] Performing final merge before closing",
instanceId);
performMerge(new ArrayList<>());
} catch (Exception mergeEx) {
log.error("[Instance {}] Final merge failed, will still close
connection: {}",
instanceId, mergeEx.getMessage(), mergeEx);
}
}
} finally {
// connection.close() here
```
**HIGH**
- `DatabendSinkAggregatedCommitter.java:62` (aborted) - **Concurrency
hazard**: The `aborted` flag is a non-volatile boolean, but `abort()` and
`close()` may be called concurrently (e.g., checkpoint failure triggers
`abort()` while job end triggers `close()`). Use `volatile boolean` or
`AtomicBoolean` to ensure visibility.
- `DatabendSinkAggregatedCommitter.java:141-160` (performMerge) -
**Misleading signature**: The method takes a
`List<DatabendSinkAggregatedCommitInfo>` parameter but never uses it. This
confuses the semantic (should merge "this batch" vs "all stream data"). Add
null-check and JavaDoc clarifying the parameter is unused but retained for
compatibility.
--
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]
