[fix](editlog) Fix concurrency bugs and robustness issues in editlog subsystem

1. EditLog.txId data race: Convert `txId` from plain `long` to `AtomicLong`.
   The flusher thread and DDL caller threads both read/write `txId` without
   synchronization, causing torn reads on 64-bit values.

2. EditLog.logEdit(op, List<T>) missing roll check: This bulk-write path
   incremented txId but never checked against `edit_log_roll_num`, so the
   edit log would never roll when this path was used.

3. BDBEnvironment.close() not thread-safe: `close()` iterated
   `openedDatabases` without holding `lock.writeLock()`, while
   `openDatabase()` concurrently modifies the same list under the write
   lock. This causes ConcurrentModificationException when a RollbackException
   triggers close() on the replayer thread while HTTP/heartbeat/RPC threads
   call openDatabase() via getMaxJournalId().

4. EditLog.flushEditLog() silent hang on fatal error: When `System.exit(-1)`
   is called after a write failure, producer threads blocked on
   `req.lock.wait()` are never notified. If `System.exit` is delayed by
   shutdown hooks, those threads hang indefinitely.

5. BDBJEJournal.write() burns journal IDs on failure: Both single and batch
   write methods called `nextJournalId.getAndIncrement/getAndAdd` before
   the actual write, so a failed write permanently skips journal IDs.
   Now IDs are only advanced after successful commit.

6. BDBEnvironment.removeDatabase() index corruption: Manual index tracking
   with `openedDatabases.remove(index)` computed wrong index when the
   target was not the first element. Replaced with `iterator.remove()`.

7. JournalObservable.upperBound() off-by-one: The binary search used
   `right = size - 1` and `right = middle - 1`, which could miss notifying
   observers whose targetJournalVersion exactly equals the journalId.
   Fixed to standard upper_bound pattern.

8. Remove dead `editStream` field and `getEditLogSize()` method from
   EditLog. The field was never assigned (always null), so
   `getEditLogSize()` would always NPE.

Author: dataroaring

fe/fe-core/src/main/java/org/apache/doris/journal/bdbje/BDBEnvironment.java

@@ -312,21 +312,15 @@ public Database openDatabase(String dbName) {
     public void removeDatabase(String dbName) {
         lock.writeLock().lock();
         try {
-            String targetDbName = null;
-            int index = 0;
-            for (Database db : openedDatabases) {
+            for (java.util.Iterator<Database> iter = openedDatabases.iterator(); iter.hasNext();) {
+                Database db = iter.next();
                 String name = db.getDatabaseName();
                 if (dbName.equals(name)) {
                     db.close();
                     LOG.info("database {} has been closed", name);
-                    targetDbName = name;
+                    iter.remove();
                     break;
                 }
-                index++;
-            }
-            if (targetDbName != null) {
-                LOG.info("begin to remove database {} from openedDatabases", targetDbName);
-                openedDatabases.remove(index);
             }
             try {
                 LOG.info("begin to remove database {} from replicatedEnvironment", dbName);
@@ -481,32 +475,37 @@ public List<Long> getDatabaseNames() {
 
     // Close the store and environment
     public void close() {
-        for (Database db : openedDatabases) {
-            try {
-                db.close();
-            } catch (DatabaseException exception) {
-                LOG.error("Error closing db {} will exit", db.getDatabaseName(), exception);
+        lock.writeLock().lock();
+        try {
+            for (Database db : openedDatabases) {
+                try {
+                    db.close();
+                } catch (DatabaseException exception) {
+                    LOG.error("Error closing db {} will exit", db.getDatabaseName(), exception);
+                }
             }
-        }
-        openedDatabases.clear();
+            openedDatabases.clear();
 
-        if (epochDB != null) {
-            try {
-                epochDB.close();
-                epochDB = null;
-            } catch (DatabaseException exception) {
-                LOG.error("Error closing db {} will exit", epochDB.getDatabaseName(), exception);
+            if (epochDB != null) {
+                try {
+                    epochDB.close();
+                    epochDB = null;
+                } catch (DatabaseException exception) {
+                    LOG.error("Error closing db {} will exit", epochDB.getDatabaseName(), exception);
+                }
             }
-        }
 
-        if (replicatedEnvironment != null) {
-            try {
-                // Finally, close the store and environment.
-                replicatedEnvironment.close();
-                replicatedEnvironment = null;
-            } catch (DatabaseException exception) {
-                LOG.error("Error closing replicatedEnvironment", exception);
+            if (replicatedEnvironment != null) {
+                try {
+                    // Finally, close the store and environment.
+                    replicatedEnvironment.close();
+                    replicatedEnvironment = null;
+                } catch (DatabaseException exception) {
+                    LOG.error("Error closing replicatedEnvironment", exception);
+                }
             }
+        } finally {
+            lock.writeLock().unlock();
         }
     }
 

fe/fe-core/src/main/java/org/apache/doris/journal/bdbje/BDBJEJournal.java

@@ -131,7 +131,8 @@ public synchronized long write(JournalBatch batch) throws IOException {
         List<JournalBatch.Entity> entities = batch.getJournalEntities();
         int entitySize = entities.size();
         long dataSize = 0;
-        long firstId = nextJournalId.getAndAdd(entitySize);
+        // Reserve IDs only after successful commit to avoid burning IDs on write failure.
+        long firstId = nextJournalId.get();
 
         // Write the journals to bdb.
         for (int i = 0; i < RETRY_TIME; i++) {
@@ -155,6 +156,7 @@ public synchronized long write(JournalBatch batch) throws IOException {
 
                 txn.commit();
                 txn = null;
+                nextJournalId.addAndGet(entitySize);
 
                 if (MetricRepo.isInit) {
                     MetricRepo.COUNTER_EDIT_LOG_SIZE_BYTES.increase(dataSize);
@@ -237,8 +239,9 @@ public synchronized long write(short op, Writable writable) throws IOException {
         entity.setOpCode(op);
         entity.setData(writable);
 
-        // id is the key
-        long id = nextJournalId.getAndIncrement();
+        // id is the key. Reserve ID only after successful write to avoid burning IDs on failure.
+        // This is safe because the method is synchronized.
+        long id = nextJournalId.get();
         DatabaseEntry theKey = idToKey(id);
 
         // entity is the value
@@ -273,6 +276,7 @@ public synchronized long write(short op, Writable writable) throws IOException {
                 // Parameter null means auto commit
                 if (currentJournalDB.put(null, theKey, theData) == OperationStatus.SUCCESS) {
                     writeSucceed = true;
+                    nextJournalId.incrementAndGet();
                     if (LOG.isDebugEnabled()) {
                         LOG.debug("master write journal {} finished. db name {}, current time {}",
                                 id, currentJournalDB.getDatabaseName(), System.currentTimeMillis());

fe/fe-core/src/main/java/org/apache/doris/persist/EditLog.java

@@ -167,9 +167,7 @@ public long await() {
     private final BlockingQueue<EditLogItem> logEditQueue = new LinkedBlockingQueue<>();
     private final Thread flushThread;
 
-    private EditLogOutputStream editStream = null;
-
-    private long txId = 0;
+    private final AtomicLong txId = new AtomicLong(0);
 
 
     private AtomicLong numTransactions = new AtomicLong(0);
@@ -255,6 +253,14 @@ private void flushEditLog() {
                 }
             }
         } catch (Throwable t) {
+            // Notify all pending items so producer threads don't hang if System.exit is delayed
+            for (EditLogItem req : batch) {
+                synchronized (req.lock) {
+                    req.finished = true;
+                    req.logId = -1;
+                    req.lock.notifyAll();
+                }
+            }
             // Throwable contains all Exception and Error, such as IOException and
             // OutOfMemoryError
             if (journal instanceof BDBJEJournal) {
@@ -264,13 +270,13 @@ private void flushEditLog() {
             System.exit(-1);
         }
 
-        txId += batch.size();
+        txId.addAndGet(batch.size());
         // update statistics, etc. (optional, can be added as needed)
-        if (txId >= Config.edit_log_roll_num) {
-            LOG.info("txId {} is equal to or larger than edit_log_roll_num {}, will roll edit.", txId,
+        if (txId.get() >= Config.edit_log_roll_num) {
+            LOG.info("txId {} is equal to or larger than edit_log_roll_num {}, will roll edit.", txId.get(),
                     Config.edit_log_roll_num);
             rollEditLog();
-            txId = 0;
+            txId.set(0);
         }
         if (MetricRepo.isInit) {
             MetricRepo.COUNTER_EDIT_LOG_WRITE.increase(Long.valueOf(batch.size()));
@@ -1557,7 +1563,13 @@ private <T extends Writable> void logEdit(short op, List<T> entries) throws IOEx
         if (!batch.getJournalEntities().isEmpty()) {
             journal.write(batch);
         }
-        txId += entries.size();
+        long newTxId = txId.addAndGet(entries.size());
+        if (newTxId >= Config.edit_log_roll_num) {
+            LOG.info("txId {} is equal to or larger than edit_log_roll_num {}, will roll edit.",
+                    newTxId, Config.edit_log_roll_num);
+            rollEditLog();
+            txId.set(0);
+        }
     }
 
     /**
@@ -1654,13 +1666,13 @@ private synchronized long logEditDirectly(short op, Writable writable) {
         }
 
         // get a new transactionId
-        txId++;
+        long newTxId = txId.incrementAndGet();
 
-        if (txId >= Config.edit_log_roll_num) {
-            LOG.info("txId {} is equal to or larger than edit_log_roll_num {}, will roll edit.", txId,
+        if (newTxId >= Config.edit_log_roll_num) {
+            LOG.info("txId {} is equal to or larger than edit_log_roll_num {}, will roll edit.", newTxId,
                     Config.edit_log_roll_num);
             rollEditLog();
-            txId = 0;
+            txId.set(0);
         }
 
         return logId;
@@ -1709,19 +1721,12 @@ private long logEdit(short op, Writable writable) {
 
         if (LOG.isDebugEnabled()) {
             LOG.debug("nextId = {}, numTransactions = {}, totalTimeTransactions = {}, op = {} delta = {}",
-                    txId, numTransactions, totalTimeTransactions, op, end - start);
+                    txId.get(), numTransactions, totalTimeTransactions, op, end - start);
         }
 
         return logId;
     }
 
-    /**
-     * Return the size of the current EditLog
-     */
-    public synchronized long getEditLogSize() throws IOException {
-        return editStream.length();
-    }
-
     /**
      * Return the number of the current EditLog
      */

fe/fe-core/src/main/java/org/apache/doris/qe/JournalObservable.java

@@ -63,23 +63,20 @@ public void waitOn(Long expectedJournalVersion, int timeoutMs) throws DdlExcepti
         }
     }
 
-    // return min pos which is bigger than value
+    // Return the number of elements whose targetJournalVersion <= value.
+    // This is the standard upper_bound pattern: returns the index of the first element > value.
     public static int upperBound(Object[] array, int size, Long value) {
         int left = 0;
-        int right = size - 1;
+        int right = size;
         while (left < right) {
             int middle = left + ((right - left) >> 1);
-            if (value >= ((JournalObserver) array[middle]).getTargetJournalVersion()) {
+            if (((JournalObserver) array[middle]).getTargetJournalVersion() <= value) {
                 left = middle + 1;
             } else {
-                right = middle - 1;
+                right = middle;
             }
         }
-        if (right == -1) {
-            return 0;
-        }
-        Long rightValue = ((JournalObserver) array[right]).getTargetJournalVersion();
-        return value >= rightValue ? right + 1  : right;
+        return left;
     }
 
     public void notifyObservers(Long journalId) {