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 2cb82a81d fix(persistence): widen username columns on existing
databases (#6915)
2cb82a81d is described below
commit 2cb82a81df9d7c8f8d0d599563687b8fcc8ec8fd
Author: Qingyuan Wang <[email protected]>
AuthorDate: Mon Oct 19 20:09:11 2026 +0800
fix(persistence): widen username columns on existing databases (#6915)
Verified against `rocketmq-studio` at bd1dafabf before merging: the defect
is still present on the
current trunk and the change merges cleanly.
Maintainer follow-ups on top of the contributor commit:
- fix(persistence): report an unreadable column width as an unconfirmed
widening
Each follow-up commit message records the review finding it addresses.
Locally verified: targeted suites green on the integration branch; a full
backend and frontend
build plus test pass runs on the trunk once this batch lands.
---
deploy/mysql/upgrade-username-column-width.sql | 18 ++
.../UsernameColumnWideningSchemaMigration.java | 131 +++++++++++
.../UsernameColumnWideningSchemaMigrationTest.java | 242 +++++++++++++++++++++
3 files changed, 391 insertions(+)
diff --git a/deploy/mysql/upgrade-username-column-width.sql
b/deploy/mysql/upgrade-username-column-width.sql
new file mode 100644
index 000000000..3d72398b4
--- /dev/null
+++ b/deploy/mysql/upgrade-username-column-width.sql
@@ -0,0 +1,18 @@
+-- deploy/mysql/upgrade-username-column-width.sql
+-- Existing MySQL volumes: widen the username columns to VARCHAR(128).
+-- Fresh volumes already receive these widths from
server/src/main/resources/db/schema.sql, and
+-- UsernameColumnWideningSchemaMigration re-applies them on every boot, so
this script is only
+-- needed when the application's database user is not allowed to run ALTER
(the runner then just
+-- logs a warning and leaves the columns narrow).
+-- Usernames may be up to 128 characters (see AuthService), so the
query-history owner columns and
+-- the audit operator must match.
+-- 幂等：可重复执行（MODIFY 只增长列宽，已是 128 时为空操作）。
+--
+-- 用法（远程容器内执行）：
+-- docker exec -i rocketmq-studio-mysql mysql -uroot -pstudio123 rocketmq <
upgrade-username-column-width.sql
+
+SET NAMES utf8mb4;
+
+ALTER TABLE rmq_instance_message MODIFY queried_by VARCHAR(128);
+ALTER TABLE rmq_instance_trace MODIFY queried_by VARCHAR(128);
+ALTER TABLE rmq_operation_audit MODIFY operator VARCHAR(128);
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/persistence/UsernameColumnWideningSchemaMigration.java
b/server/src/main/java/org/apache/rocketmq/studio/persistence/UsernameColumnWideningSchemaMigration.java
new file mode 100644
index 000000000..49bc074e2
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/persistence/UsernameColumnWideningSchemaMigration.java
@@ -0,0 +1,131 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.studio.persistence;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.boot.ApplicationArguments;
+import org.springframework.boot.ApplicationRunner;
+import org.springframework.stereotype.Component;
+
+import javax.sql.DataSource;
+import java.sql.Connection;
+import java.sql.DatabaseMetaData;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.List;
+
+/**
+ * Widens the username columns to VARCHAR(128) on databases created before
schema.sql widened them.
+ *
+ * <p>The trailing {@code ALTER TABLE ... MODIFY} statements in db/schema.sql
only run against a
+ * fresh MySQL data directory or the dev H2 profile; existing deployments are
upgraded by the
+ * application schema-migration runners, so this runner closes the gap for
volumes initialised
+ * while the columns were still VARCHAR(64).
+ */
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class UsernameColumnWideningSchemaMigration implements
ApplicationRunner {
+ private static final int REQUIRED_WIDTH = 128;
+ /**
+ * Width reported when the metadata lookup cannot answer, so the column is
left untouched. Used
+ * before an ALTER is considered: an unverifiable schema must not be
modified.
+ */
+ private static final int WIDTH_UNKNOWN_ASSUME_WIDE = REQUIRED_WIDTH;
+ /**
+ * Width reported when the metadata lookup cannot answer after an ALTER
was attempted, so the
+ * caller treats the widening as unconfirmed and warns.
+ */
+ private static final int WIDTH_UNKNOWN_ASSUME_NARROW = 0;
+ private static final List<Widening> WIDENINGS = List.of(
+ new Widening("rmq_instance_message", "queried_by"),
+ new Widening("rmq_instance_trace", "queried_by"),
+ new Widening("rmq_operation_audit", "operator"));
+
+ private final DataSource dataSource;
+
+ @Override
+ public void run(ApplicationArguments args) {
+ try (Connection connection = dataSource.getConnection()) {
+ DatabaseMetaData metadata = connection.getMetaData();
+ String catalog = connection.getCatalog();
+ for (Widening widening : WIDENINGS) {
+ ensureWidth(connection, metadata, catalog, widening);
+ }
+ } catch (SQLException failure) {
+ log.warn("Skipped username column widening: {}",
failure.getMessage());
+ }
+ }
+
+ private static void ensureWidth(Connection connection, DatabaseMetaData
metadata, String catalog,
+ Widening widening) {
+ if (!hasTable(metadata, catalog, widening.table())) {
+ return;
+ }
+ if (columnWidth(metadata, catalog, widening.table(), widening.column(),
+ WIDTH_UNKNOWN_ASSUME_WIDE) >= REQUIRED_WIDTH) {
+ return;
+ }
+ try (Statement statement = connection.createStatement()) {
+ log.info("Widening username column {}.{} to VARCHAR({})",
+ widening.table(), widening.column(), REQUIRED_WIDTH);
+ statement.executeUpdate("ALTER TABLE " + widening.table() + "
MODIFY " + widening.column()
+ + " VARCHAR(" + REQUIRED_WIDTH + ")");
+ } catch (SQLException failure) {
+ // A server that refuses the widening must not stop the
application from booting; the
+ // column stays narrow until the deployment grants the ALTER
(schema.sql stays idempotent).
+ // The re-check reports an unreadable width as "not widened" so
the warning below still
+ // reaches the log instead of being swallowed by a second metadata
failure.
+ if (columnWidth(metadata, catalog, widening.table(),
widening.column(),
+ WIDTH_UNKNOWN_ASSUME_NARROW) < REQUIRED_WIDTH) {
+ log.warn("Could not widen {}.{} to VARCHAR({}): {}",
+ widening.table(), widening.column(), REQUIRED_WIDTH,
failure.getMessage());
+ }
+ }
+ }
+
+ private static boolean hasTable(DatabaseMetaData metadata, String catalog,
String table) {
+ try (ResultSet tables = metadata.getTables(catalog, null, table, new
String[] {"TABLE"})) {
+ return tables.next();
+ } catch (SQLException failure) {
+ return false;
+ }
+ }
+
+ /**
+ * Returns the declared width of the column, or {@code undeterminable}
when the column or its size
+ * cannot be read. Callers pick the fallback: {@link
#WIDTH_UNKNOWN_ASSUME_WIDE} leaves an
+ * unverifiable schema untouched, {@link #WIDTH_UNKNOWN_ASSUME_NARROW}
reports a widening that
+ * could not be confirmed.
+ */
+ private static int columnWidth(DatabaseMetaData metadata, String catalog,
String table, String column,
+ int undeterminable) {
+ try (ResultSet columns = metadata.getColumns(catalog, null, table,
column)) {
+ if (!columns.next()) {
+ return undeterminable;
+ }
+ return columns.getInt("COLUMN_SIZE");
+ } catch (SQLException failure) {
+ return undeterminable;
+ }
+ }
+
+ private record Widening(String table, String column) {
+ }
+}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/persistence/UsernameColumnWideningSchemaMigrationTest.java
b/server/src/test/java/org/apache/rocketmq/studio/persistence/UsernameColumnWideningSchemaMigrationTest.java
new file mode 100644
index 000000000..364619d1e
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/persistence/UsernameColumnWideningSchemaMigrationTest.java
@@ -0,0 +1,242 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.studio.persistence;
+
+import org.h2.jdbcx.JdbcDataSource;
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.DefaultApplicationArguments;
+
+import javax.sql.DataSource;
+import java.io.PrintWriter;
+import java.lang.reflect.Proxy;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.SQLFeatureNotSupportedException;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.logging.Logger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class UsernameColumnWideningSchemaMigrationTest {
+
+ @Test
+ void widensNarrowUsernameColumnsTo128WithoutDroppingRowsAndIsIdempotent()
throws Exception {
+ JdbcDataSource dataSource = newDataSource("username-widening");
+ try (Connection connection = dataSource.getConnection(); Statement
statement = connection.createStatement()) {
+ statement.execute("CREATE TABLE rmq_instance_message "
+ + "(id BIGINT PRIMARY KEY, queried_by VARCHAR(64), msg_id
VARCHAR(128))");
+ statement.execute("CREATE TABLE rmq_instance_trace "
+ + "(id BIGINT PRIMARY KEY, queried_by VARCHAR(64), msg_id
VARCHAR(128))");
+ statement.execute("CREATE TABLE rmq_operation_audit "
+ + "(id BIGINT PRIMARY KEY, operator VARCHAR(64), action
VARCHAR(255))");
+ statement.execute("INSERT INTO rmq_instance_message (id,
queried_by, msg_id) "
+ + "VALUES (1, 'old-user', 'm1')");
+ statement.execute("INSERT INTO rmq_instance_trace (id, queried_by,
msg_id) "
+ + "VALUES (1, 'old-user', 't1')");
+ statement.execute("INSERT INTO rmq_operation_audit (id, operator,
action) "
+ + "VALUES (1, 'old-user', 'login')");
+ }
+
+ UsernameColumnWideningSchemaMigration migration = new
UsernameColumnWideningSchemaMigration(dataSource);
+ migration.run(new DefaultApplicationArguments());
+ migration.run(new DefaultApplicationArguments());
+
+ assertWidth(dataSource, "rmq_instance_message", "queried_by", 128);
+ assertWidth(dataSource, "rmq_instance_trace", "queried_by", 128);
+ assertWidth(dataSource, "rmq_operation_audit", "operator", 128);
+ try (Connection connection = dataSource.getConnection(); Statement
statement = connection.createStatement()) {
+ assertStoredValue(statement, "SELECT queried_by FROM
rmq_instance_message WHERE id = 1", "old-user");
+ assertStoredValue(statement, "SELECT queried_by FROM
rmq_instance_trace WHERE id = 1", "old-user");
+ assertStoredValue(statement, "SELECT operator FROM
rmq_operation_audit WHERE id = 1", "old-user");
+ }
+ }
+
+ @Test
+ void leavesAlreadyWidenedColumnsUntouched() throws Exception {
+ JdbcDataSource delegate = newDataSource("username-widening-done");
+ try (Connection connection = delegate.getConnection(); Statement
statement = connection.createStatement()) {
+ statement.execute("CREATE TABLE rmq_instance_message "
+ + "(id BIGINT PRIMARY KEY, queried_by VARCHAR(128))");
+ statement.execute("INSERT INTO rmq_instance_message (id,
queried_by) VALUES (1, 'wide-user')");
+ }
+
+ List<String> updates = new ArrayList<>();
+ UsernameColumnWideningSchemaMigration migration =
+ new
UsernameColumnWideningSchemaMigration(interceptingUpdates(delegate,
updates::add));
+ migration.run(new DefaultApplicationArguments());
+
+ assertWidth(delegate, "rmq_instance_message", "queried_by", 128);
+ try (Connection connection = delegate.getConnection(); Statement
statement = connection.createStatement()) {
+ assertStoredValue(statement, "SELECT queried_by FROM
rmq_instance_message WHERE id = 1", "wide-user");
+ }
+ // MODIFY to the width a column already has succeeds on H2, so the
width assertion above would
+ // still pass without the columnWidth() guard; only the statement
count proves it was skipped.
+ assertThat(updates).isEmpty();
+ }
+
+ @Test
+ void skipsMissingTablesSoCustomSchemaCanCreateThemLater() throws Exception
{
+ JdbcDataSource delegate = newDataSource("username-widening-missing");
+
+ List<String> updates = new ArrayList<>();
+ UsernameColumnWideningSchemaMigration migration =
+ new
UsernameColumnWideningSchemaMigration(interceptingUpdates(delegate,
updates::add));
+ migration.run(new DefaultApplicationArguments());
+
+ try (Connection connection = delegate.getConnection(); Statement
statement = connection.createStatement()) {
+ assertThat(tableExists(statement,
"rmq_instance_message")).isFalse();
+ assertThat(tableExists(statement, "rmq_instance_trace")).isFalse();
+ assertThat(tableExists(statement,
"rmq_operation_audit")).isFalse();
+ }
+ assertThat(updates).isEmpty();
+ }
+
+ @Test
+ void keepsBootingWhenTheServerRefusesTheWidening() throws Exception {
+ JdbcDataSource delegate = newDataSource("username-widening-refused");
+ try (Connection connection = delegate.getConnection(); Statement
statement = connection.createStatement()) {
+ statement.execute("CREATE TABLE rmq_instance_message "
+ + "(id BIGINT PRIMARY KEY, queried_by VARCHAR(64))");
+ }
+
+ List<String> updates = new ArrayList<>();
+ UsernameColumnWideningSchemaMigration migration = new
UsernameColumnWideningSchemaMigration(
+ interceptingUpdates(delegate, sql -> {
+ updates.add(sql);
+ throw new SQLException("widening refused by server");
+ }));
+ migration.run(new DefaultApplicationArguments());
+
+ assertWidth(delegate, "rmq_instance_message", "queried_by", 64);
+ // The ALTER has to be attempted for the failure branch (and its
re-check of the width) to run.
+ assertThat(updates).containsExactly("ALTER TABLE rmq_instance_message
MODIFY queried_by VARCHAR(128)");
+ }
+
+ private static JdbcDataSource newDataSource(String name) {
+ JdbcDataSource dataSource = new JdbcDataSource();
+ dataSource.setURL("jdbc:h2:mem:" + name +
";MODE=MySQL;DB_CLOSE_DELAY=-1;DATABASE_TO_LOWER=TRUE");
+ dataSource.setUser("sa");
+ return dataSource;
+ }
+
+ private static void assertWidth(DataSource dataSource, String table,
String column, int width)
+ throws Exception {
+ try (Connection connection = dataSource.getConnection();
+ ResultSet columns = connection.getMetaData()
+ .getColumns(connection.getCatalog(), null, table,
column)) {
+ assertThat(columns.next()).isTrue();
+ assertThat(columns.getInt("COLUMN_SIZE")).isEqualTo(width);
+ }
+ }
+
+ private static void assertStoredValue(Statement statement, String query,
String value) throws SQLException {
+ try (ResultSet rows = statement.executeQuery(query)) {
+ assertThat(rows.next()).isTrue();
+ assertThat(rows.getString(1)).isEqualTo(value);
+ }
+ }
+
+ private static boolean tableExists(Statement statement, String table)
throws SQLException {
+ try (ResultSet tables = statement.executeQuery("SELECT COUNT(*) FROM
information_schema.tables "
+ + "WHERE table_name = '" + table + "'")) {
+ tables.next();
+ return tables.getInt(1) > 0;
+ }
+ }
+
+ /** DataSource that hands every {@code executeUpdate} SQL to {@code
interceptor} before running it. */
+ private static DataSource interceptingUpdates(DataSource delegate,
UpdateInterceptor interceptor) {
+ return new DataSource() {
+ @Override
+ public Connection getConnection() throws SQLException {
+ return interceptedConnection(delegate.getConnection(),
interceptor);
+ }
+
+ @Override
+ public Connection getConnection(String username, String password)
throws SQLException {
+ return interceptedConnection(delegate.getConnection(username,
password), interceptor);
+ }
+
+ @Override
+ public PrintWriter getLogWriter() throws SQLException {
+ return delegate.getLogWriter();
+ }
+
+ @Override
+ public void setLogWriter(PrintWriter out) throws SQLException {
+ delegate.setLogWriter(out);
+ }
+
+ @Override
+ public void setLoginTimeout(int seconds) throws SQLException {
+ delegate.setLoginTimeout(seconds);
+ }
+
+ @Override
+ public int getLoginTimeout() throws SQLException {
+ return delegate.getLoginTimeout();
+ }
+
+ @Override
+ public Logger getParentLogger() throws
SQLFeatureNotSupportedException {
+ return delegate.getParentLogger();
+ }
+
+ @Override
+ public <T> T unwrap(Class<T> iface) throws SQLException {
+ return delegate.unwrap(iface);
+ }
+
+ @Override
+ public boolean isWrapperFor(Class<?> iface) throws SQLException {
+ return delegate.isWrapperFor(iface);
+ }
+ };
+ }
+
+ /** Records or rejects one {@code executeUpdate} statement before it
reaches the database. */
+ private interface UpdateInterceptor {
+ void intercept(String sql) throws SQLException;
+ }
+
+ private static Connection interceptedConnection(Connection delegate,
UpdateInterceptor interceptor) {
+ return (Connection) Proxy.newProxyInstance(
+
UsernameColumnWideningSchemaMigrationTest.class.getClassLoader(),
+ new Class<?>[] {Connection.class},
+ (proxy, method, args) -> {
+ if (method.getName().startsWith("createStatement")) {
+ return
interceptedStatement(delegate.createStatement(), interceptor);
+ }
+ return method.invoke(delegate, args);
+ });
+ }
+
+ private static Statement interceptedStatement(Statement delegate,
UpdateInterceptor interceptor) {
+ return (Statement) Proxy.newProxyInstance(
+
UsernameColumnWideningSchemaMigrationTest.class.getClassLoader(),
+ new Class<?>[] {Statement.class},
+ (proxy, method, args) -> {
+ if (method.getName().equals("executeUpdate")) {
+ interceptor.intercept(String.valueOf(args[0]));
+ }
+ return method.invoke(delegate, args);
+ });
+ }
+}
