github-actions[bot] commented on code in PR #67114:
URL: https://github.com/apache/doris/pull/67114#discussion_r3860859907
##########
be/src/exec/sink/writer/paimon/jni_paimon_write_backend.cpp:
##########
@@ -135,15 +190,13 @@ Status JniPaimonWriteBackend::close() {
env->CallVoidMethod(_jni_writer_obj, _close_id);
close_status = _check_jni_exception(env, "close PaimonJniWriter");
Review Comment:
[P2] Bound JNI locals for the initial close too.
The new `PushLocalFrame` fixes the previously reported quarantine-retry
site, but this distinct first-close path still calls `GetJniExceptionMsg(...,
true)` through `_check_jni_exception` without a frame. That helper deletes the
throwable but not its `msg` and optional `stack` local jstrings. Because
successful recovery now admits later writers on reused TLS-attached native
workers, successive first-close failures can accumulate those locals over the
BE lifetime. Frame this call or release the helper's temporary references, and
cover repeated fail-recover-readmit cycles.
##########
be/src/exec/sink/writer/paimon/jni_paimon_write_backend.cpp:
##########
@@ -46,31 +46,93 @@ namespace doris {
namespace {
constexpr std::string_view PAIMON_JNI_WRITER_IO_TMP_DIR =
"paimon_jni_writer_io_tmp";
-std::atomic<bool>& paimon_jni_close_failed() {
- static std::atomic<bool> failed {false};
- return failed;
-}
+struct PaimonJniCloseState {
+ // The global writer reference keeps both the Java object and its defining
class alive. The
+ // cached method ID therefore remains valid until release_writer_ref()
deletes this reference.
+ jobject writer_obj = nullptr;
+ jmethodID recover_and_close_id = nullptr;
+ std::unique_ptr<PaimonJniMemoryManager> memory_manager;
-std::mutex& retained_memory_managers_mutex() {
+ Status retry_cleanup(JNIEnv* env) const {
+ if (writer_obj == nullptr) {
+ return Status::OK();
+ }
+ if (recover_and_close_id == nullptr) {
+ return Status::InternalError("PaimonJniWriter.recoverAndClose
method is unavailable");
+ }
+
+ if (env->PushLocalFrame(16) != JNI_OK) {
+ env->ExceptionClear();
+ return Status::InternalError(
+ "Failed to create a JNI local frame for Paimon cleanup
recovery");
+ }
+ Defer pop_local_frame {[&] { env->PopLocalFrame(nullptr); }};
+ env->CallVoidMethod(writer_obj, recover_and_close_id);
+ return Jni::Env::GetJniExceptionMsg(env, true, "JNI exception retrying
Paimon cleanup: ");
+ }
+
+ void release_writer_ref(JNIEnv* env) {
+ if (writer_obj != nullptr) {
+ env->DeleteGlobalRef(writer_obj);
+ writer_obj = nullptr;
+ }
+ }
+};
+
+std::mutex& quarantined_close_states_mutex() {
static auto* mutex = new std::mutex();
return *mutex;
}
-std::vector<std::unique_ptr<PaimonJniMemoryManager>>&
retained_memory_managers() {
- static auto* managers = new
std::vector<std::unique_ptr<PaimonJniMemoryManager>>();
- return *managers;
+std::vector<PaimonJniCloseState>& quarantined_close_states() {
+ // Deliberately process-lifetime storage: if cleanup never succeeds,
retaining the writer and
+ // native pages at process exit is safer than destroying memory still used
by Java tasks.
+ static auto* states = new std::vector<PaimonJniCloseState>();
+ return *states;
}
-void retain_memory_after_failed_close(std::unique_ptr<PaimonJniMemoryManager>
manager) {
+void quarantine_failed_close(jobject writer_obj, jmethodID
recover_and_close_id,
+ std::unique_ptr<PaimonJniMemoryManager>
memory_manager) {
// An unconfirmed Java close means a background Paimon task may still
reference this manager's
- // native pages. Quarantine the manager and stop admitting new writers so
repeated failures
- // cannot accumulate process-lifetime native memory without a bound.
- paimon_jni_close_failed().store(true, std::memory_order_release);
- if (manager == nullptr) {
- return;
+ // native pages. Keep the Java writer reachable as well, so a later open
can finish deferred
+ // abort and close cleanup before admitting more writers instead of
requiring a BE restart.
+ std::lock_guard<std::mutex> lock(quarantined_close_states_mutex());
+ quarantined_close_states().push_back(
Review Comment:
[P1] Make quarantine publication allocation-safe.
Both close-failure paths have already exchanged away the global writer and
moved `_memory_manager` into this by-value call. First initialization of the
heap-backed mutex/vector, or growth in this `push_back`, can throw
`std::bad_alloc` on the exact memory-pressure path. Unwinding then destroys the
manager and releases all native pages while the raw global Java writer was
neither queued nor deleted and may still have Paimon tasks using those pages;
destructor-driven close can also terminate if this escapes. Allocate/fence the
storage before transferring ownership, or provide a no-fail fallback that
deliberately retains both owners, and failure-inject initial allocation and
vector growth.
##########
fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/PaimonJniWriter.java:
##########
@@ -617,22 +718,63 @@ private void clearWriterState() {
}
private void abortWriter() throws Exception {
- try {
- List<CommitMessage> messages = preparedCommitMessages;
- if (messages.isEmpty() && writer != null) {
- messages = prepareCommitMessages();
+ if (writer == null && preparedCommitMessages.isEmpty()) {
+ abortPending = false;
+ deferAbortAfterOutOfMemory = false;
+ return;
+ }
+
+ abortPending = true;
+ if (deferAbortAfterOutOfMemory && preparedCommitMessages.isEmpty()) {
+ LOG.warn("Deferring Paimon prepare-and-abort cleanup after OOM");
+ return;
+ }
+ finishPendingAbort();
+ closeWriter();
+ }
+
+ private void finishPendingAbort() throws Exception {
+ if (!abortPending) {
+ return;
+ }
+
+ List<CommitMessage> messages = preparedCommitMessages;
+ if (messages.isEmpty() && writer != null) {
+ messages = prepareCommitMessages();
Review Comment:
[P1] Do not retry a non-resumable whole-writer prepare.
In Paimon 1.4.2, `AbstractFileStoreWrite.prepareCommit()` visits buckets
sequentially into a method-local result, while append/postpone bucket writers
destructively clear their pending increments. If a later bucket throws, earlier
messages never return; similarly, an OOM in Doris's post-return `new
ArrayList<>(messages)` loses the already-drained complete result. This retry
then sees those buckets empty and cannot pass their file paths to abort. Use an
atomic/resumable preparation contract or durably capture each increment before
advancing, and test a real multi-bucket partial failure with file deletion
assertions.
##########
fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/PaimonJniWriter.java:
##########
@@ -617,22 +718,63 @@ private void clearWriterState() {
}
private void abortWriter() throws Exception {
- try {
- List<CommitMessage> messages = preparedCommitMessages;
- if (messages.isEmpty() && writer != null) {
- messages = prepareCommitMessages();
+ if (writer == null && preparedCommitMessages.isEmpty()) {
+ abortPending = false;
+ deferAbortAfterOutOfMemory = false;
+ return;
+ }
+
+ abortPending = true;
+ if (deferAbortAfterOutOfMemory && preparedCommitMessages.isEmpty()) {
+ LOG.warn("Deferring Paimon prepare-and-abort cleanup after OOM");
+ return;
+ }
+ finishPendingAbort();
+ closeWriter();
+ }
+
+ private void finishPendingAbort() throws Exception {
+ if (!abortPending) {
+ return;
+ }
+
+ List<CommitMessage> messages = preparedCommitMessages;
+ if (messages.isEmpty() && writer != null) {
+ messages = prepareCommitMessages();
+ }
+ if (!messages.isEmpty()) {
+ InnerTableCommit committer = table.newCommit(commitUser);
+ try {
+ committer.abort(messages);
+ } finally {
+ committer.close();
}
- if (!messages.isEmpty()) {
- InnerTableCommit committer = table.newCommit(commitUser);
- try {
- committer.abort(messages);
- } finally {
- committer.close();
- }
+ }
+
+ preparedCommitMessages = Collections.emptyList();
+ abortPending = false;
+ deferAbortAfterOutOfMemory = false;
+ }
+
+ private void recordOutOfMemoryFailure(Throwable failure) {
+ if (containsOutOfMemory(failure)) {
+ deferAbortAfterOutOfMemory = true;
+ }
+ }
+
+ static boolean containsOutOfMemory(Throwable failure) {
+ Throwable current = failure;
+ for (int depth = 0; current != null && depth < 32; depth++) {
+ if (current instanceof OutOfMemoryError || current instanceof
OutOfMemoryException) {
Review Comment:
[P1] Preserve native OOM identity for the deferral decision.
The production page callback maps Doris `QUERY_MEMORY_EXCEEDED` and native
allocator failures to a cause-less Java `RuntimeException`, which this loop
cannot recognize; C++ Arrow conversion/export OOM can also fail before
`writeArrow` sets the Java flag. The common C++ abort path then immediately
invokes allocation-heavy prepare/abort under the same pressure this state
machine is intended to defer. Propagate a machine-detectable memory-exhaustion
status across both native paths and add callback/conversion-level tests proving
prepare is postponed until recovery.
##########
fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/PaimonJniWriter.java:
##########
@@ -617,22 +718,63 @@ private void clearWriterState() {
}
private void abortWriter() throws Exception {
- try {
- List<CommitMessage> messages = preparedCommitMessages;
- if (messages.isEmpty() && writer != null) {
- messages = prepareCommitMessages();
+ if (writer == null && preparedCommitMessages.isEmpty()) {
+ abortPending = false;
+ deferAbortAfterOutOfMemory = false;
+ return;
+ }
+
+ abortPending = true;
+ if (deferAbortAfterOutOfMemory && preparedCommitMessages.isEmpty()) {
+ LOG.warn("Deferring Paimon prepare-and-abort cleanup after OOM");
+ return;
+ }
+ finishPendingAbort();
+ closeWriter();
+ }
+
+ private void finishPendingAbort() throws Exception {
+ if (!abortPending) {
Review Comment:
[P1] Abort prepared output after a later close failure.
`PaimonTableWriter::close()` aborts only before backend close. If prepare
succeeds but writer/index/IO/allocator close then fails, C++ suppresses
publication and drops its local message copy, while Java never marked this
successful prepare as `abortPending`. Recovery therefore takes this return,
eventually clears `preparedCommitMessages`, and never calls
`InnerTableCommit.abort` for the unpublished files. Preserve an abort
obligation for prepared-but-unpublished output until abort and dependent
cleanup both succeed, and test with a non-empty message plus an injected close
failure.
--
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]
