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 742dad0d3 perf(studio): consolidate query and audit persistence (#2966)
742dad0d3 is described below
commit 742dad0d32faff1d006ae035b98b3314533143c4
Author: aias00 <[email protected]>
AuthorDate: Wed Sep 2 15:15:59 2026 +0800
perf(studio): consolidate query and audit persistence (#2966)
* Improve query history access path stability
Add user-scoped composite indexes for message and trace query history in
the fresh Studio schema and backfill them through an idempotent startup
migration for existing databases.
The message history table keeps the main drawer path index on queried_by,
cluster_id, gmt_create, id and adds a separate query_type-aware index for
filtered requests.
Constraint: Studio does not currently have a merged Flyway migration path,
so this follows the existing schema.sql plus ApplicationRunner upgrade pattern.
Rejected: depend on #1374 | the Flyway work is not present on
origin/rocketmq-studio.
Tested: JAVA_HOME=$(/usr/libexec/java_home -v 21) mvn
-Dtest=QueryHistorySchemaIndexTest,QueryHistoryServiceTest,QueryHistoryServiceIntegrationTest,QueryHistoryControllerTest,DemoDataSqlCompatibilityTest
test
Tested: JAVA_HOME=$(/usr/libexec/java_home -v 21) mvn test
Not-tested: MySQL 8 EXPLAIN | local mysql service and Docker daemon are not
running.
Confidence: high
Scope-risk: narrow
Signed-off-by: liuhy <[email protected]>
* perf(studio): bound history cleanup deletes
Limit scheduled query-history and audit cleanup to indexed ID batches so
large installations avoid single massive delete transactions.
Constraint: keep message, trace, and audit cleanup behavior isolated and
reviewable.
Directive: do not reintroduce gmt_create-only unbounded cleanup deletes
without large-inventory tests.
Confidence: high
Scope-risk: narrow
Tested: mvn -q
-Dtest=QueryHistoryServiceTest,MybatisPlusAuditRepositoryTest,AuditServiceTest
test; mvn -q
-Dtest=QueryHistoryServiceIntegrationTest,QueryHistoryControllerTest,AuditControllerTest
test; mvn -q test
Signed-off-by: liuhy <[email protected]>
* fix: stabilize persisted audit queries
Signed-off-by: liuhy <[email protected]>
* test: align audit query assertion with schema
Signed-off-by: liuhy <[email protected]>
* test: align test method naming
Signed-off-by: liuhy <[email protected]>
* fix(message): own query history schema migration
Signed-off-by: liuhy <[email protected]>
---------
Signed-off-by: liuhy <[email protected]>
---
.../instance/message/QueryHistoryProperties.java | 10 ++
.../message/QueryHistorySchemaMigration.java | 117 +++++++++++++++++++++
.../instance/message/QueryHistoryService.java | 54 +++++++++-
.../studio/ops/alert/AlertSchemaMigration.java | 3 +-
.../rocketmq/studio/ops/audit/AuditRepository.java | 2 +-
.../rocketmq/studio/ops/audit/AuditService.java | 4 +-
.../ops/audit/MybatisPlusAuditRepository.java | 59 ++++++++++-
server/src/main/resources/db/schema.sql | 9 +-
.../message/QueryHistorySchemaIndexTest.java | 97 +++++++++++++++++
.../instance/message/QueryHistoryServiceTest.java | 79 ++++++++++++--
.../studio/ops/audit/AuditServiceTest.java | 10 ++
.../ops/audit/MybatisPlusAuditRepositoryTest.java | 66 +++++++++++-
12 files changed, 485 insertions(+), 25 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/QueryHistoryProperties.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/QueryHistoryProperties.java
index 0f081ba4a..62869a76e 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/QueryHistoryProperties.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/QueryHistoryProperties.java
@@ -31,4 +31,14 @@ public class QueryHistoryProperties {
* Number of days query records are retained. A non-positive value
disables cleanup.
*/
private int retentionDays = 90;
+
+ /**
+ * Maximum expired records deleted per database statement.
+ */
+ private int cleanupBatchSize = 500;
+
+ /**
+ * Maximum delete batches performed by one scheduled cleanup pass.
+ */
+ private int cleanupMaxBatches = 20;
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/QueryHistorySchemaMigration.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/QueryHistorySchemaMigration.java
new file mode 100644
index 000000000..fd13062b9
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/QueryHistorySchemaMigration.java
@@ -0,0 +1,117 @@
+/*
+ * 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.
+ */
+package org.apache.rocketmq.studio.instance.message;
+
+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;
+
+/** Adds query-history indexes to Studio databases created before the index
contract was added. */
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class QueryHistorySchemaMigration implements ApplicationRunner {
+ private static final List<Column> COLUMNS = List.of(
+ new Column("rmq_instance_message", "result_snapshot",
"MEDIUMTEXT"));
+ private static final List<Index> INDEXES = List.of(
+ new Index("rmq_instance_message", "idx_message_query_owner_lookup",
+ "queried_by, cluster_id, gmt_create, id"),
+ new Index("rmq_instance_message",
"idx_message_query_owner_type_lookup",
+ "queried_by, cluster_id, query_type, gmt_create, id"),
+ new Index("rmq_instance_trace", "idx_trace_query_owner_lookup",
+ "queried_by, cluster_id, gmt_create, id"));
+
+ private final DataSource dataSource;
+
+ @Override
+ public void run(ApplicationArguments args) throws Exception {
+ try (Connection connection = dataSource.getConnection(); Statement
statement = connection.createStatement()) {
+ DatabaseMetaData metadata = connection.getMetaData();
+ String catalog = connection.getCatalog();
+ for (Column column : COLUMNS) {
+ ensureColumn(metadata, catalog, statement, column);
+ }
+ for (Index index : INDEXES) {
+ ensureIndex(metadata, catalog, statement, index);
+ }
+ }
+ }
+
+ private static void ensureColumn(DatabaseMetaData metadata, String
catalog, Statement statement, Column column)
+ throws Exception {
+ if (!hasTable(metadata, catalog, column.table())
+ || hasColumn(metadata, catalog, column.table(),
column.name())) {
+ return;
+ }
+ try {
+ log.info("Adding query history column {}.{}", column.table(),
column.name());
+ statement.executeUpdate("ALTER TABLE " + column.table() + " ADD
COLUMN " + column.name()
+ + " " + column.definition());
+ } catch (SQLException failure) {
+ if (!hasColumn(metadata, catalog, column.table(), column.name())) {
+ throw failure;
+ }
+ }
+ }
+
+ private static void ensureIndex(DatabaseMetaData metadata, String catalog,
Statement statement, Index index)
+ throws Exception {
+ if (!hasTable(metadata, catalog, index.table()) || hasIndex(metadata,
catalog, index.table(), index.name())) {
+ return;
+ }
+ try {
+ log.info("Adding query history index {}.{}", index.table(),
index.name());
+ statement.executeUpdate("CREATE INDEX " + index.name() + " ON " +
index.table()
+ + " (" + index.columns() + ")");
+ } catch (SQLException failure) {
+ if (!hasIndex(metadata, catalog, index.table(), index.name())) {
+ throw failure;
+ }
+ }
+ }
+
+ private static boolean hasTable(DatabaseMetaData metadata, String catalog,
String table) throws Exception {
+ try (ResultSet tables = metadata.getTables(catalog, null, table, new
String[] {"TABLE"})) {
+ return tables.next();
+ }
+ }
+
+ private static boolean hasIndex(DatabaseMetaData metadata, String catalog,
String table, String index)
+ throws Exception {
+ try (ResultSet indexes = metadata.getIndexInfo(catalog, null, table,
false, false)) {
+ while (indexes.next()) {
+ if (index.equalsIgnoreCase(indexes.getString("INDEX_NAME"))) {
+ return true;
+ }
+ }
+ return false;
+ }
+ }
+
+ private static boolean hasColumn(DatabaseMetaData metadata, String
catalog, String table, String column)
+ throws Exception {
+ try (ResultSet columns = metadata.getColumns(catalog, null, table,
column)) {
+ return columns.next();
+ }
+ }
+
+ private record Column(String table, String name, String definition) {
+ }
+
+ private record Index(String table, String name, String columns) {
+ }
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/QueryHistoryService.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/QueryHistoryService.java
index 699d83797..b028084dc 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/QueryHistoryService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/QueryHistoryService.java
@@ -16,8 +16,8 @@
*/
package org.apache.rocketmq.studio.instance.message;
-import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
@@ -38,11 +38,18 @@ import java.time.LocalDateTime;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
+import java.util.function.Function;
@Slf4j
@Service
public class QueryHistoryService {
+ private static final int DEFAULT_CLEANUP_BATCH_SIZE = 500;
+ private static final int MAX_CLEANUP_BATCH_SIZE = 5_000;
+ private static final int DEFAULT_CLEANUP_MAX_BATCHES = 20;
+ private static final int MAX_CLEANUP_MAX_BATCHES = 100;
+
private final RmqMessageQueryMapper messageQueryMapper;
private final RmqTraceQueryMapper traceQueryMapper;
private final QueryHistoryProperties properties;
@@ -239,8 +246,7 @@ public class QueryHistoryService {
private void deleteExpiredMessageQueries(LocalDateTime cutoff) {
try {
- int deleted =
messageQueryMapper.delete(Wrappers.<RmqMessageQuery>query()
- .lt("gmt_create", cutoff));
+ int deleted = deleteExpiredInBatches(messageQueryMapper,
RmqMessageQuery::getId, cutoff);
log.debug("Purged {} expired message query records", deleted);
} catch (RuntimeException e) {
log.warn("Failed to purge expired message query records: {}",
e.getMessage());
@@ -249,14 +255,52 @@ public class QueryHistoryService {
private void deleteExpiredTraceQueries(LocalDateTime cutoff) {
try {
- int deleted =
traceQueryMapper.delete(Wrappers.<RmqTraceQuery>query()
- .lt("gmt_create", cutoff));
+ int deleted = deleteExpiredInBatches(traceQueryMapper,
RmqTraceQuery::getId, cutoff);
log.debug("Purged {} expired trace query records", deleted);
} catch (RuntimeException e) {
log.warn("Failed to purge expired trace query records: {}",
e.getMessage());
}
}
+ private <T> int deleteExpiredInBatches(BaseMapper<T> mapper, Function<T,
Long> idExtractor,
+ LocalDateTime cutoff) {
+ int batchSize = boundedPositive(properties.getCleanupBatchSize(),
+ DEFAULT_CLEANUP_BATCH_SIZE, MAX_CLEANUP_BATCH_SIZE);
+ int maxBatches = boundedPositive(properties.getCleanupMaxBatches(),
+ DEFAULT_CLEANUP_MAX_BATCHES, MAX_CLEANUP_MAX_BATCHES);
+ int totalDeleted = 0;
+ for (int batch = 0; batch < maxBatches; batch++) {
+ List<T> expired = mapper.selectList(new QueryWrapper<T>()
+ .select("id")
+ .lt("gmt_create", cutoff)
+ .orderByAsc("gmt_create", "id")
+ .last("LIMIT " + batchSize));
+ if (expired.isEmpty()) {
+ break;
+ }
+ List<Long> ids = expired.stream()
+ .map(idExtractor)
+ .filter(Objects::nonNull)
+ .toList();
+ if (ids.isEmpty()) {
+ break;
+ }
+ int deleted = mapper.deleteByIds(ids);
+ totalDeleted += deleted;
+ if (deleted < batchSize) {
+ break;
+ }
+ }
+ return totalDeleted;
+ }
+
+ private static int boundedPositive(int value, int defaultValue, int
maxValue) {
+ if (value <= 0) {
+ return defaultValue;
+ }
+ return Math.min(value, maxValue);
+ }
+
private static LocalDateTime latestOf(LocalDateTime left, LocalDateTime
right) {
if (left == null) return right;
if (right == null) return left;
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/alert/AlertSchemaMigration.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/alert/AlertSchemaMigration.java
index 9a773bb9b..b5391432c 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/alert/AlertSchemaMigration.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/alert/AlertSchemaMigration.java
@@ -107,8 +107,7 @@ public class AlertSchemaMigration implements
ApplicationRunner {
new Column("rmq_alert_notification_outbox", "delivered_at",
"DATETIME"),
new Column("rmq_alert_notification_outbox", "sending_started_at",
"DATETIME"),
new Column("rmq_alert_notification_outbox", "claim_token",
"VARCHAR(64)"),
- new Column("rmq_alert_notification_outbox", "message_content",
"TEXT"),
- new Column("rmq_instance_message", "result_snapshot",
"MEDIUMTEXT"));
+ new Column("rmq_alert_notification_outbox", "message_content",
"TEXT"));
private static final List<Index> INDEXES = List.of(
new Index("rmq_metric_snapshot", "idx_metric_snapshot_lookup",
"instance_id, metric_key, collected_at"),
new Index("rmq_metric_snapshot",
"idx_metric_snapshot_scope_cluster",
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/audit/AuditRepository.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/audit/AuditRepository.java
index 953c3bd0a..90ec12a38 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/audit/AuditRepository.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/audit/AuditRepository.java
@@ -29,5 +29,5 @@ public interface AuditRepository {
void save(AuditRecordVO record);
- int deleteBefore(LocalDateTime cutoff);
+ int deleteBefore(LocalDateTime cutoff, int batchSize, int maxBatches);
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/audit/AuditService.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/audit/AuditService.java
index b68bd5df8..a7ca5e243 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/audit/AuditService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/audit/AuditService.java
@@ -37,6 +37,8 @@ public class AuditService {
private static final int MAX_PAGE_SIZE = 100;
private static final int MAX_EXPORT_RECORDS = 10_000;
+ private static final int CLEANUP_BATCH_SIZE = 500;
+ private static final int CLEANUP_MAX_BATCHES = 20;
private static final String CSV_HEADER =
"timestamp,operator,operationType,resourceType,target,clusterId,detail,result,errorMessage\r\n";
@@ -118,7 +120,7 @@ public class AuditService {
}
log.info("Cleaning up audit logs older than {} days", beforeDays);
LocalDateTime cutoff = LocalDateTime.now().minusDays(beforeDays);
- return auditRepository.deleteBefore(cutoff);
+ return auditRepository.deleteBefore(cutoff, CLEANUP_BATCH_SIZE,
CLEANUP_MAX_BATCHES);
}
private void validatePagination(int page, int pageSize) {
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/audit/MybatisPlusAuditRepository.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/audit/MybatisPlusAuditRepository.java
index e422962e9..9e62d1036 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/audit/MybatisPlusAuditRepository.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/audit/MybatisPlusAuditRepository.java
@@ -29,6 +29,7 @@ import java.time.LocalDateTime;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
/** MySQL-backed audit repository (rmq_operation_audit). */
@@ -36,7 +37,10 @@ import java.util.stream.Collectors;
@Repository
public class MybatisPlusAuditRepository implements AuditRepository {
+ private static final long FILTER_OPTIONS_CACHE_TTL_NANOS =
TimeUnit.SECONDS.toNanos(30);
+
private final RmqOperationAuditMapper auditMapper;
+ private volatile CachedFilterOptions cachedFilterOptions;
@Override
public PageResult<AuditRecordVO> findPage(String search, String
operationType,
@@ -54,7 +58,7 @@ public class MybatisPlusAuditRepository implements
AuditRepository {
.ge(startDate != null, "gmt_create", startDate)
.le(endDate != null, "gmt_create", endDate)
.eq(StringUtils.hasText(result), "result", result)
- .orderByDesc("gmt_create");
+ .orderByDesc("gmt_create", "id");
Page<RmqOperationAudit> resultPage = auditMapper.selectPage(
new Page<>(page, pageSize), query);
List<AuditRecordVO> records = resultPage.getRecords().stream()
@@ -65,6 +69,23 @@ public class MybatisPlusAuditRepository implements
AuditRepository {
@Override
public AuditFilterOptionsVO findFilterOptions() {
+ long now = System.nanoTime();
+ CachedFilterOptions cached = cachedFilterOptions;
+ if (isCacheValid(cached, now)) {
+ return cached.options();
+ }
+ synchronized (this) {
+ cached = cachedFilterOptions;
+ if (isCacheValid(cached, now)) {
+ return cached.options();
+ }
+ AuditFilterOptionsVO options = loadFilterOptions();
+ cachedFilterOptions = new CachedFilterOptions(options, now);
+ return options;
+ }
+ }
+
+ private AuditFilterOptionsVO loadFilterOptions() {
List<Map<String, Object>> values = auditMapper.selectMaps(
new QueryWrapper<RmqOperationAudit>()
.select("operation", "resource_type", "cluster_id",
"result")
@@ -92,12 +113,42 @@ public class MybatisPlusAuditRepository implements
AuditRepository {
entity.setGmtCreate(timestamp);
entity.setGmtModified(timestamp);
auditMapper.insert(entity);
+ cachedFilterOptions = null;
+ }
+
+ private static boolean isCacheValid(CachedFilterOptions cached, long now) {
+ return cached != null && now - cached.createdAtNanos() <
FILTER_OPTIONS_CACHE_TTL_NANOS;
+ }
+
+ private record CachedFilterOptions(AuditFilterOptionsVO options, long
createdAtNanos) {
}
@Override
- public int deleteBefore(LocalDateTime cutoff) {
- return Math.toIntExact(auditMapper.delete(
- new QueryWrapper<RmqOperationAudit>().lt("gmt_create",
cutoff)));
+ public int deleteBefore(LocalDateTime cutoff, int batchSize, int
maxBatches) {
+ int totalDeleted = 0;
+ for (int batch = 0; batch < maxBatches; batch++) {
+ List<RmqOperationAudit> expired = auditMapper.selectList(new
QueryWrapper<RmqOperationAudit>()
+ .select("id")
+ .lt("gmt_create", cutoff)
+ .orderByAsc("gmt_create", "id")
+ .last("LIMIT " + batchSize));
+ if (expired.isEmpty()) {
+ break;
+ }
+ List<Long> ids = expired.stream()
+ .map(RmqOperationAudit::getId)
+ .filter(Objects::nonNull)
+ .toList();
+ if (ids.isEmpty()) {
+ break;
+ }
+ int deleted = auditMapper.deleteByIds(ids);
+ totalDeleted += deleted;
+ if (deleted < batchSize) {
+ break;
+ }
+ }
+ return totalDeleted;
}
private List<String> findDistinctValues(List<Map<String, Object>> rows,
String column) {
diff --git a/server/src/main/resources/db/schema.sql
b/server/src/main/resources/db/schema.sql
index d3e73952a..38b6cf44c 100644
--- a/server/src/main/resources/db/schema.sql
+++ b/server/src/main/resources/db/schema.sql
@@ -149,7 +149,9 @@ CREATE TABLE IF NOT EXISTS rmq_instance_message (
cluster_id VARCHAR(255),
queried_by VARCHAR(128),
PRIMARY KEY (`id`),
- INDEX idx_message_query_gmt_create (gmt_create),
+ INDEX idx_message_query_cleanup (gmt_create, id),
+ INDEX idx_message_query_owner_lookup (queried_by, cluster_id, gmt_create,
id),
+ INDEX idx_message_query_owner_type_lookup (queried_by, cluster_id,
query_type, gmt_create, id),
INDEX idx_topic (topic)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
@@ -166,7 +168,8 @@ CREATE TABLE IF NOT EXISTS rmq_instance_trace (
queried_by VARCHAR(128),
PRIMARY KEY (`id`),
INDEX idx_msg_id (msg_id),
- INDEX idx_trace_query_gmt_create (gmt_create)
+ INDEX idx_trace_query_cleanup (gmt_create, id),
+ INDEX idx_trace_query_owner_lookup (queried_by, cluster_id, gmt_create, id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
-- 8. 操作审计日志（所有写操作）
@@ -183,7 +186,7 @@ CREATE TABLE IF NOT EXISTS rmq_operation_audit (
error_message TEXT,
operator VARCHAR(128),
PRIMARY KEY (`id`),
- INDEX idx_gmt_create (gmt_create),
+ INDEX idx_operation_audit_cleanup (gmt_create, id),
INDEX idx_resource (resource_type, resource_name),
INDEX idx_operation (operation)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/message/QueryHistorySchemaIndexTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/message/QueryHistorySchemaIndexTest.java
new file mode 100644
index 000000000..9e869910e
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/message/QueryHistorySchemaIndexTest.java
@@ -0,0 +1,97 @@
+/*
+ * 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.
+ */
+package org.apache.rocketmq.studio.instance.message;
+
+import org.h2.jdbcx.JdbcDataSource;
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.DefaultApplicationArguments;
+import org.springframework.core.io.ClassPathResource;
+import org.springframework.jdbc.datasource.init.ScriptUtils;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class QueryHistorySchemaIndexTest {
+ @Test
+ void freshSchemaCreatesUserScopedHistoryIndexesTest() throws Exception {
+ JdbcDataSource dataSource = new JdbcDataSource();
+
dataSource.setURL("jdbc:h2:mem:query-history-schema-indexes;MODE=MySQL;DB_CLOSE_DELAY=-1;DATABASE_TO_LOWER=TRUE");
+ dataSource.setUser("sa");
+
+ try (Connection connection = dataSource.getConnection()) {
+ ScriptUtils.executeSqlScript(connection, new
ClassPathResource("db/schema.sql"));
+ }
+
+ try (Connection connection = dataSource.getConnection()) {
+ assertThat(indexColumns(connection, "rmq_instance_message",
"idx_message_query_owner_lookup"))
+ .containsExactly("queried_by", "cluster_id", "gmt_create",
"id");
+ assertThat(indexColumns(connection, "rmq_instance_message",
"idx_message_query_owner_type_lookup"))
+ .containsExactly("queried_by", "cluster_id", "query_type",
"gmt_create", "id");
+ assertThat(indexColumns(connection, "rmq_instance_trace",
"idx_trace_query_owner_lookup"))
+ .containsExactly("queried_by", "cluster_id", "gmt_create",
"id");
+ }
+ }
+
+ @Test
+ void migrationAddsUserScopedHistoryIndexesToExistingTablesTest() throws
Exception {
+ JdbcDataSource dataSource = new JdbcDataSource();
+
dataSource.setURL("jdbc:h2:mem:query-history-schema-index-migration;MODE=MySQL;DB_CLOSE_DELAY=-1;DATABASE_TO_LOWER=TRUE");
+ dataSource.setUser("sa");
+
+ try (Connection connection = dataSource.getConnection(); Statement
statement = connection.createStatement()) {
+ statement.execute("CREATE TABLE rmq_instance_message ("
+ + "id BIGINT AUTO_INCREMENT PRIMARY KEY, "
+ + "gmt_create DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
"
+ + "query_type VARCHAR(32) NOT NULL, "
+ + "cluster_id VARCHAR(255), "
+ + "queried_by VARCHAR(128))");
+ statement.execute("CREATE TABLE rmq_instance_trace ("
+ + "id BIGINT AUTO_INCREMENT PRIMARY KEY, "
+ + "gmt_create DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
"
+ + "cluster_id VARCHAR(255), "
+ + "queried_by VARCHAR(128))");
+ }
+
+ QueryHistorySchemaMigration migration = new
QueryHistorySchemaMigration(dataSource);
+ migration.run(new DefaultApplicationArguments());
+ migration.run(new DefaultApplicationArguments());
+
+ try (Connection connection = dataSource.getConnection()) {
+ assertThat(indexColumns(connection, "rmq_instance_message",
"idx_message_query_owner_lookup"))
+ .containsExactly("queried_by", "cluster_id", "gmt_create",
"id");
+ assertThat(indexColumns(connection, "rmq_instance_message",
"idx_message_query_owner_type_lookup"))
+ .containsExactly("queried_by", "cluster_id", "query_type",
"gmt_create", "id");
+ assertThat(indexColumns(connection, "rmq_instance_trace",
"idx_trace_query_owner_lookup"))
+ .containsExactly("queried_by", "cluster_id", "gmt_create",
"id");
+ assertThat(hasColumn(connection, "rmq_instance_message",
"result_snapshot")).isTrue();
+ }
+ }
+
+ private static boolean hasColumn(Connection connection, String tableName,
String columnName) throws Exception {
+ try (ResultSet columns = connection.getMetaData().getColumns(null,
null, tableName, columnName)) {
+ return columns.next();
+ }
+ }
+
+ private static List<String> indexColumns(Connection connection, String
tableName, String indexName) throws Exception {
+ try (Statement statement = connection.createStatement();
+ ResultSet result = statement.executeQuery("SELECT column_name
FROM information_schema.index_columns "
+ + "WHERE table_name = '" + tableName + "' AND
index_name = '" + indexName + "' "
+ + "ORDER BY ordinal_position")) {
+ List<String> columns = new ArrayList<>();
+ while (result.next()) {
+ columns.add(result.getString(1));
+ }
+ return columns;
+ }
+ }
+}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/message/QueryHistoryServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/message/QueryHistoryServiceTest.java
index 5ed0445ee..b5a03a738 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/message/QueryHistoryServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/message/QueryHistoryServiceTest.java
@@ -33,11 +33,13 @@ import java.time.Clock;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
+import java.util.List;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -83,17 +85,70 @@ class QueryHistoryServiceTest {
@Test
void purgesBothQueryHistoriesUsingConfiguredRetention() {
properties.setRetentionDays(7);
- when(messageQueryMapper.delete(any())).thenReturn(2);
- when(traceQueryMapper.delete(any())).thenReturn(3);
+ when(messageQueryMapper.selectList(any())).thenReturn(List.of());
+ when(traceQueryMapper.selectList(any())).thenReturn(List.of());
service.purgeExpiredQueries();
ArgumentCaptor<Wrapper<RmqMessageQuery>> messageCaptor =
ArgumentCaptor.forClass(Wrapper.class);
ArgumentCaptor<Wrapper<RmqTraceQuery>> traceCaptor =
ArgumentCaptor.forClass(Wrapper.class);
- verify(messageQueryMapper).delete(messageCaptor.capture());
- verify(traceQueryMapper).delete(traceCaptor.capture());
-
assertThat(messageCaptor.getValue().getCustomSqlSegment()).contains("gmt_create");
-
assertThat(traceCaptor.getValue().getCustomSqlSegment()).contains("gmt_create");
+ verify(messageQueryMapper).selectList(messageCaptor.capture());
+ verify(traceQueryMapper).selectList(traceCaptor.capture());
+ assertThat(messageCaptor.getValue().getCustomSqlSegment())
+ .contains("gmt_create", "ORDER BY gmt_create ASC,id ASC",
"LIMIT 500");
+ assertThat(traceCaptor.getValue().getCustomSqlSegment())
+ .contains("gmt_create", "ORDER BY gmt_create ASC,id ASC",
"LIMIT 500");
+ }
+
+ @Test
+ void
purgeExpiredQueriesDeletesMessageAndTraceHistoryInBoundedBatchesTest() {
+ properties.setRetentionDays(7);
+ properties.setCleanupBatchSize(2);
+ properties.setCleanupMaxBatches(3);
+ RmqMessageQuery message1 = messageQuery(1L);
+ RmqMessageQuery message2 = messageQuery(2L);
+ RmqMessageQuery message3 = messageQuery(3L);
+ RmqTraceQuery trace1 = traceQuery(11L);
+ RmqTraceQuery trace2 = traceQuery(12L);
+ when(messageQueryMapper.selectList(any()))
+ .thenReturn(List.of(message1, message2), List.of(message3));
+ when(traceQueryMapper.selectList(any()))
+ .thenReturn(List.of(trace1, trace2), List.of());
+ when(messageQueryMapper.deleteByIds(List.of(1L, 2L))).thenReturn(2);
+ when(messageQueryMapper.deleteByIds(List.of(3L))).thenReturn(1);
+ when(traceQueryMapper.deleteByIds(List.of(11L, 12L))).thenReturn(2);
+
+ service.purgeExpiredQueries();
+
+ ArgumentCaptor<Wrapper<RmqMessageQuery>> messageQueryCaptor =
ArgumentCaptor.forClass(Wrapper.class);
+ ArgumentCaptor<Wrapper<RmqTraceQuery>> traceQueryCaptor =
ArgumentCaptor.forClass(Wrapper.class);
+ verify(messageQueryMapper,
times(2)).selectList(messageQueryCaptor.capture());
+ verify(traceQueryMapper,
times(2)).selectList(traceQueryCaptor.capture());
+
assertThat(messageQueryCaptor.getAllValues().get(0).getCustomSqlSegment())
+ .contains("gmt_create", "ORDER BY gmt_create ASC,id ASC",
"LIMIT 2");
+
assertThat(traceQueryCaptor.getAllValues().get(0).getCustomSqlSegment())
+ .contains("gmt_create", "ORDER BY gmt_create ASC,id ASC",
"LIMIT 2");
+ verify(messageQueryMapper).deleteByIds(List.of(1L, 2L));
+ verify(messageQueryMapper).deleteByIds(List.of(3L));
+ verify(traceQueryMapper).deleteByIds(List.of(11L, 12L));
+ verify(messageQueryMapper, never()).delete(any());
+ verify(traceQueryMapper, never()).delete(any());
+ }
+
+ @Test
+ void purgeExpiredQueriesContinuesTraceCleanupWhenMessageCleanupFailsTest()
{
+ properties.setRetentionDays(7);
+ properties.setCleanupBatchSize(2);
+ properties.setCleanupMaxBatches(3);
+ when(messageQueryMapper.selectList(any())).thenThrow(new
IllegalStateException("message db down"));
+ RmqTraceQuery trace = traceQuery(21L);
+ when(traceQueryMapper.selectList(any())).thenReturn(List.of(trace));
+ when(traceQueryMapper.deleteByIds(List.of(21L))).thenReturn(1);
+
+ service.purgeExpiredQueries();
+
+ verify(traceQueryMapper).selectList(any());
+ verify(traceQueryMapper).deleteByIds(List.of(21L));
}
@Test
@@ -165,4 +220,16 @@ class QueryHistoryServiceTest {
assertThat(messageCountCaptor.getValue().getCustomSqlSegment()).contains("queried_by");
assertThat(traceCountCaptor.getValue().getCustomSqlSegment()).contains("queried_by");
}
+
+ private static RmqMessageQuery messageQuery(Long id) {
+ RmqMessageQuery query = new RmqMessageQuery();
+ query.setId(id);
+ return query;
+ }
+
+ private static RmqTraceQuery traceQuery(Long id) {
+ RmqTraceQuery query = new RmqTraceQuery();
+ query.setId(id);
+ return query;
+ }
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/audit/AuditServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/audit/AuditServiceTest.java
index 5a777bce0..43331f47f 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/audit/AuditServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/audit/AuditServiceTest.java
@@ -201,4 +201,14 @@ class AuditServiceTest {
.isInstanceOf(BusinessException.class)
.hasMessage("beforeDays must not exceed 365");
}
+
+ @Test
+ void cleanupLogsUsesBoundedRepositoryBatchesTest() {
+ when(auditRepository.deleteBefore(any(LocalDateTime.class), eq(500),
eq(20))).thenReturn(500);
+
+ int deleted = auditService.cleanupLogs(90);
+
+ assertThat(deleted).isEqualTo(500);
+ verify(auditRepository).deleteBefore(any(LocalDateTime.class),
eq(500), eq(20));
+ }
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/audit/MybatisPlusAuditRepositoryTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/audit/MybatisPlusAuditRepositoryTest.java
index 359302983..c518bccb2 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/audit/MybatisPlusAuditRepositoryTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/audit/MybatisPlusAuditRepositoryTest.java
@@ -36,6 +36,8 @@ import java.util.Map;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -49,7 +51,7 @@ class MybatisPlusAuditRepositoryTest {
private MybatisPlusAuditRepository repository;
@Test
- void findPageUsesMapperPaginationAndPreservesAuditContext() {
+ void findPageUsesMapperPaginationAndPreservesAuditContextTest() {
RmqOperationAudit entity = new RmqOperationAudit();
entity.setId(42L);
entity.setOperation("DELETE_TOPIC");
@@ -80,11 +82,11 @@ class MybatisPlusAuditRepositoryTest {
assertThat(record.getClusterId()).isEqualTo("prod-cn");
assertThat(record.getErrorMessage()).isEqualTo("denied");
assertThat(queryCaptor.getValue().getSqlSegment())
- .contains("operation", "resource_type", "cluster_id",
"result");
+ .contains("operation", "resource_type", "cluster_id",
"result", "gmt_create", "id");
}
@Test
- void findFilterOptionsPreservesPersistedValuesFromOneQuery() {
+ void findFilterOptionsPreservesPersistedValuesAndCachesTheResultTest() {
when(auditMapper.selectMaps(any(Wrapper.class))).thenReturn(List.of(
Map.of("operation", "DELETE_TOPIC", "resource_type", "TOPIC",
"cluster_id", "prod-cn", "result", "SUCCESS"),
@@ -94,11 +96,13 @@ class MybatisPlusAuditRepositoryTest {
"cluster_id", "", "result", "PARTIAL")));
AuditFilterOptionsVO options = repository.findFilterOptions();
+ AuditFilterOptionsVO cachedOptions = repository.findFilterOptions();
assertThat(options.getOperationTypes()).containsExactly(" CREATE_TOPIC
", "DELETE_TOPIC");
assertThat(options.getResourceTypes()).containsExactly("GROUP",
"TOPIC");
assertThat(options.getClusterIds()).containsExactly("prod-cn",
"prod-sh");
assertThat(options.getResults()).containsExactly("FAILED", "PARTIAL",
"SUCCESS");
+ assertThat(cachedOptions).isSameAs(options);
ArgumentCaptor<Wrapper<RmqOperationAudit>> queryCaptor =
ArgumentCaptor.forClass(Wrapper.class);
verify(auditMapper).selectMaps(queryCaptor.capture());
assertThat(((QueryWrapper<RmqOperationAudit>)
queryCaptor.getValue()).getSqlSelect())
@@ -107,4 +111,60 @@ class MybatisPlusAuditRepositoryTest {
.contains("GROUP BY
operation,resource_type,cluster_id,result");
}
+ @Test
+ void deleteBeforeShouldUseBoundedIdBatchesTest() {
+ RmqOperationAudit audit1 = auditRecord(1L);
+ RmqOperationAudit audit2 = auditRecord(2L);
+ RmqOperationAudit audit3 = auditRecord(3L);
+ when(auditMapper.selectList(any()))
+ .thenReturn(List.of(audit1, audit2), List.of(audit3));
+ when(auditMapper.deleteByIds(List.of(1L, 2L))).thenReturn(2);
+ when(auditMapper.deleteByIds(List.of(3L))).thenReturn(1);
+
+ int deleted = repository.deleteBefore(LocalDateTime.of(2026, 8, 1, 0,
0), 2, 5);
+
+ ArgumentCaptor<Wrapper<RmqOperationAudit>> queryCaptor =
ArgumentCaptor.forClass(Wrapper.class);
+ verify(auditMapper, times(2)).selectList(queryCaptor.capture());
+ assertThat(queryCaptor.getAllValues().get(0).getSqlSegment())
+ .contains("gmt_create", "ORDER BY gmt_create ASC,id ASC",
"LIMIT 2");
+ verify(auditMapper).deleteByIds(List.of(1L, 2L));
+ verify(auditMapper).deleteByIds(List.of(3L));
+ verify(auditMapper, never()).delete(any());
+ assertThat(deleted).isEqualTo(3);
+ }
+
+ @Test
+ void deleteBeforeShouldStopAfterConfiguredMaxBatchesTest() {
+ when(auditMapper.selectList(any()))
+ .thenReturn(List.of(auditRecord(1L), auditRecord(2L)))
+ .thenReturn(List.of(auditRecord(3L), auditRecord(4L)))
+ .thenReturn(List.of(auditRecord(5L), auditRecord(6L)));
+ when(auditMapper.deleteByIds(List.of(1L, 2L))).thenReturn(2);
+ when(auditMapper.deleteByIds(List.of(3L, 4L))).thenReturn(2);
+
+ int deleted = repository.deleteBefore(LocalDateTime.of(2026, 8, 1, 0,
0), 2, 2);
+
+ verify(auditMapper, times(2)).selectList(any());
+ verify(auditMapper).deleteByIds(List.of(1L, 2L));
+ verify(auditMapper).deleteByIds(List.of(3L, 4L));
+ assertThat(deleted).isEqualTo(4);
+ }
+
+ @Test
+ void saveInvalidatesCachedFilterOptionsTest() {
+ when(auditMapper.selectMaps(any(Wrapper.class))).thenReturn(List.of());
+
+ repository.findFilterOptions();
+
repository.save(AuditRecordVO.builder().operationType("CREATE_TOPIC").build());
+ repository.findFilterOptions();
+
+ verify(auditMapper,
org.mockito.Mockito.times(2)).selectMaps(any(Wrapper.class));
+ verify(auditMapper).insert(any(RmqOperationAudit.class));
+ }
+
+ private static RmqOperationAudit auditRecord(Long id) {
+ RmqOperationAudit audit = new RmqOperationAudit();
+ audit.setId(id);
+ return audit;
+ }
}
