(rocketmq-dashboard) branch rocketmq-studio updated: fix(instance): tolerate H2's unsnapshotted index metadata during concurrent startup (#6736)

[lizhimin]

This is an automated email from the ASF dual-hosted git repository.

lizhimins pushed a commit to branch rocketmq-studio
in repository https://gitbox.apache.org/repos/asf/rocketmq-dashboard.git


The following commit(s) were added to refs/heads/rocketmq-studio by this push:
     new 94a90d7a1 fix(instance): tolerate H2's unsnapshotted index metadata 
during concurrent startup (#6736)
94a90d7a1 is described below

commit 94a90d7a1aebf7a9b5c5d347c70c7a6fd6b91851
Author: lizhimins <707364...@qq.com>
AuthorDate: Thu Oct 15 20:12:32 2026 +0800

    fix(instance): tolerate H2's unsnapshotted index metadata during concurrent 
startup (#6736)
    
    migrationConcurrentStartupIsIdempotentTest failed roughly one run in three 
with
    
        JdbcSQLNonTransientException: General error:
        "java.util.ConcurrentModificationException"
          at org.h2.jdbc.meta.DatabaseMetaLocal.getIndexInfo
          at ...ResourceOwnershipSchemaMigration.indexes
    
    H2 answers getIndexInfo from a mutable metadata cache without taking a
    snapshot, so an index committed by a concurrently starting node mutates the
    list being iterated. The read is idempotent and ensureIndex already 
tolerates a
    racing CREATE INDEX, so retry it a bounded number of times when the cause 
chain
    carries a ConcurrentModificationException and rethrow anything else 
unchanged.
    
    Deliberately not a lock: serializing the migration inside one JVM would make
    the concurrent-startup regression test assert nothing, and the production
    database is MySQL, whose metadata reads do not need this.
    
    Verified: eight consecutive runs of ResourceOwnershipGuardTest green (one in
    three failed before).
---
 .../instance/ResourceOwnershipSchemaMigration.java | 40 ++++++++++++++++++++++
 1 file changed, 40 insertions(+)

diff --git 
a/server/src/main/java/org/apache/rocketmq/studio/instance/ResourceOwnershipSchemaMigration.java
 
b/server/src/main/java/org/apache/rocketmq/studio/instance/ResourceOwnershipSchemaMigration.java
index 6cfbfab7c..9efdff712 100644
--- 
a/server/src/main/java/org/apache/rocketmq/studio/instance/ResourceOwnershipSchemaMigration.java
+++ 
b/server/src/main/java/org/apache/rocketmq/studio/instance/ResourceOwnershipSchemaMigration.java
@@ -29,6 +29,7 @@ import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.Statement;
 import java.util.ArrayList;
+import java.util.ConcurrentModificationException;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -38,6 +39,9 @@ import java.util.Map;
 @DependsOnDatabaseInitialization
 @RequiredArgsConstructor
 public class ResourceOwnershipSchemaMigration implements InitializingBean {
+    /** Bounded retries for the index-metadata read; see {@link 
#indexes(Connection, String)}. */
+    private static final int INDEX_METADATA_ATTEMPTS = 5;
+
     private final DataSource dataSource;
     private final RocketMQDefaultClusterResolver defaultClusters;
 
@@ -136,6 +140,27 @@ public class ResourceOwnershipSchemaMigration implements 
InitializingBean {
     }
 
     private Map<String, List<String>> indexes(Connection connection, String 
table) throws SQLException {
+        // H2 serves getIndexInfo from a mutable metadata cache without taking 
a snapshot, so an index
+        // created by a concurrently starting node surfaces here as a 
ConcurrentModificationException
+        // wrapped in a SQLException. The read is idempotent and ensureIndex 
already tolerates a racing
+        // CREATE INDEX, so retry it a bounded number of times instead of 
serializing every startup
+        // behind one lock — a lock would also turn the concurrent-startup 
regression test into a
+        // tautology. MySQL reads committed metadata and never takes this path.
+        SQLException transientFailure = null;
+        for (int attempt = 1; attempt <= INDEX_METADATA_ATTEMPTS; attempt++) {
+            try {
+                return readIndexes(connection, table);
+            } catch (SQLException failure) {
+                if (!isConcurrentMetadataRead(failure)) {
+                    throw failure;
+                }
+                transientFailure = failure;
+            }
+        }
+        throw transientFailure;
+    }
+
+    private Map<String, List<String>> readIndexes(Connection connection, 
String table) throws SQLException {
         Map<String, List<String>> indexes = new HashMap<>();
         try (ResultSet rows = 
connection.getMetaData().getIndexInfo(connection.getCatalog(), null, table, 
false, false)) {
             while (rows.next()) {
@@ -150,4 +175,19 @@ public class ResourceOwnershipSchemaMigration implements 
InitializingBean {
         }
         return indexes;
     }
+
+    private static boolean isConcurrentMetadataRead(SQLException failure) {
+        Throwable cause = failure;
+        while (cause != null) {
+            if (cause instanceof ConcurrentModificationException) {
+                return true;
+            }
+            Throwable next = cause.getCause();
+            if (next == cause) {
+                break;
+            }
+            cause = next;
+        }
+        return false;
+    }
 }