This is an automated email from the ASF dual-hosted git repository. zqr10159 pushed a commit to branch 2.0.0 in repository https://gitbox.apache.org/repos/asf/hertzbeat.git
commit da3d2a4e646ab5e021c447ecb696decf9efbc400 Author: Logic <[email protected]> AuthorDate: Wed Aug 26 12:19:41 2026 +0800 fix: stabilize managed startup transitions --- .../MetadataWriteAdmissionConfiguration.java | 25 ++++++++++++++++------ .../SpringDataWriteAdmissionBeanPostProcessor.java | 12 ++++++----- .../MetadataWriteAdmissionConfigurationTest.java | 4 ++-- .../workflow/JdbcTargetSchemaObjectState.java | 7 ++++++ .../impl/AppServiceStartupGuardSourceTest.java | 15 +++++++------ .../setup/workflow/TargetJdbcCleanupLaneTest.java | 11 +++++----- .../TargetSchemaContractCompatibilityTest.java | 2 +- ...geServerConfigLegacyStartupIntegrationTest.java | 21 ++++++++++++++++++ .../MetadataWriteAdmissionStartupContextTest.java | 5 +++-- 9 files changed, 74 insertions(+), 28 deletions(-) diff --git a/hertzbeat-common-spring/src/main/java/org/apache/hertzbeat/common/transaction/MetadataWriteAdmissionConfiguration.java b/hertzbeat-common-spring/src/main/java/org/apache/hertzbeat/common/transaction/MetadataWriteAdmissionConfiguration.java index 25c498f651..703969df14 100644 --- a/hertzbeat-common-spring/src/main/java/org/apache/hertzbeat/common/transaction/MetadataWriteAdmissionConfiguration.java +++ b/hertzbeat-common-spring/src/main/java/org/apache/hertzbeat/common/transaction/MetadataWriteAdmissionConfiguration.java @@ -7,18 +7,22 @@ package org.apache.hertzbeat.common.transaction; +import org.springframework.beans.factory.ObjectProvider; import org.springframework.beans.factory.config.BeanDefinition; -import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Role; +import org.springframework.transaction.interceptor.BeanFactoryTransactionAttributeSourceAdvisor; import org.springframework.transaction.interceptor.TransactionAttributeSource; +import org.springframework.transaction.interceptor.TransactionInterceptor; /** Spring wiring for process-local metadata write admission. */ @Configuration(proxyBeanMethods = false) @Role(BeanDefinition.ROLE_INFRASTRUCTURE) public class MetadataWriteAdmissionConfiguration { + private static final TransactionAttributeSource NO_TRANSACTION_ATTRIBUTES = (method, targetClass) -> null; + @Bean @Role(BeanDefinition.ROLE_INFRASTRUCTURE) MetadataWriteAdmissionCoordinator metadataWriteAdmissionCoordinator() { @@ -32,21 +36,30 @@ public class MetadataWriteAdmissionConfiguration { } @Bean - @ConditionalOnBean(TransactionAttributeSource.class) @Role(BeanDefinition.ROLE_INFRASTRUCTURE) MetadataWriteAdmissionAdvisor metadataWriteAdmissionAdvisor( - TransactionAttributeSource transactionAttributeSource, + ObjectProvider<BeanFactoryTransactionAttributeSourceAdvisor> transactionAdvisor, MetadataWriteAdmissionCoordinator coordinator, TransactionCompletionPermitRegistry transactionPermits) { + TransactionAttributeSource transactionAttributes = transactionAttributes(transactionAdvisor.getIfAvailable()); return new MetadataWriteAdmissionAdvisor( - transactionAttributeSource, coordinator, transactionPermits, false); + transactionAttributes, coordinator, transactionPermits, false); } @Bean @Role(BeanDefinition.ROLE_INFRASTRUCTURE) static SpringDataWriteAdmissionBeanPostProcessor springDataWriteAdmissionBeanPostProcessor( - MetadataWriteAdmissionCoordinator coordinator, - TransactionCompletionPermitRegistry transactionPermits) { + ObjectProvider<MetadataWriteAdmissionCoordinator> coordinator, + ObjectProvider<TransactionCompletionPermitRegistry> transactionPermits) { return new SpringDataWriteAdmissionBeanPostProcessor(coordinator, transactionPermits); } + + private TransactionAttributeSource transactionAttributes( + BeanFactoryTransactionAttributeSourceAdvisor transactionAdvisor) { + if (transactionAdvisor == null || !(transactionAdvisor.getAdvice() instanceof TransactionInterceptor interceptor)) { + return NO_TRANSACTION_ATTRIBUTES; + } + TransactionAttributeSource transactionAttributes = interceptor.getTransactionAttributeSource(); + return transactionAttributes == null ? NO_TRANSACTION_ATTRIBUTES : transactionAttributes; + } } diff --git a/hertzbeat-common-spring/src/main/java/org/apache/hertzbeat/common/transaction/SpringDataWriteAdmissionBeanPostProcessor.java b/hertzbeat-common-spring/src/main/java/org/apache/hertzbeat/common/transaction/SpringDataWriteAdmissionBeanPostProcessor.java index 6c4fc1e1a7..79a2b148ff 100644 --- a/hertzbeat-common-spring/src/main/java/org/apache/hertzbeat/common/transaction/SpringDataWriteAdmissionBeanPostProcessor.java +++ b/hertzbeat-common-spring/src/main/java/org/apache/hertzbeat/common/transaction/SpringDataWriteAdmissionBeanPostProcessor.java @@ -11,6 +11,7 @@ import org.springframework.aop.Advisor; import org.springframework.aop.framework.Advised; import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanInitializationException; +import org.springframework.beans.factory.ObjectProvider; import org.springframework.beans.factory.config.BeanPostProcessor; import org.springframework.core.Ordered; import org.springframework.data.repository.Repository; @@ -21,12 +22,12 @@ import org.springframework.transaction.interceptor.TransactionalProxy; /** Inserts write admission into each existing Spring Data transaction proxy. */ final class SpringDataWriteAdmissionBeanPostProcessor implements BeanPostProcessor, Ordered { - private final MetadataWriteAdmissionCoordinator coordinator; - private final TransactionCompletionPermitRegistry transactionPermits; + private final ObjectProvider<MetadataWriteAdmissionCoordinator> coordinator; + private final ObjectProvider<TransactionCompletionPermitRegistry> transactionPermits; SpringDataWriteAdmissionBeanPostProcessor( - MetadataWriteAdmissionCoordinator coordinator, - TransactionCompletionPermitRegistry transactionPermits) { + ObjectProvider<MetadataWriteAdmissionCoordinator> coordinator, + ObjectProvider<TransactionCompletionPermitRegistry> transactionPermits) { this.coordinator = coordinator; this.transactionPermits = transactionPermits; } @@ -57,7 +58,8 @@ final class SpringDataWriteAdmissionBeanPostProcessor implements BeanPostProcess throw new BeanInitializationException("Spring Data transaction attributes are unavailable"); } advised.addAdvisor(transactionAdvisorIndex, - new MetadataWriteAdmissionAdvisor(attributes, coordinator, transactionPermits, true)); + new MetadataWriteAdmissionAdvisor( + attributes, coordinator.getObject(), transactionPermits.getObject(), true)); return bean; } diff --git a/hertzbeat-common-spring/src/test/java/org/apache/hertzbeat/common/transaction/MetadataWriteAdmissionConfigurationTest.java b/hertzbeat-common-spring/src/test/java/org/apache/hertzbeat/common/transaction/MetadataWriteAdmissionConfigurationTest.java index c9bcfc07ca..ea2bdf0df7 100644 --- a/hertzbeat-common-spring/src/test/java/org/apache/hertzbeat/common/transaction/MetadataWriteAdmissionConfigurationTest.java +++ b/hertzbeat-common-spring/src/test/java/org/apache/hertzbeat/common/transaction/MetadataWriteAdmissionConfigurationTest.java @@ -15,7 +15,7 @@ import org.springframework.context.annotation.AnnotationConfigApplicationContext class MetadataWriteAdmissionConfigurationTest { @Test - void loadsWithoutTransactionInfrastructureAndOmitsOnlyTheAdvisor() { + void loadsWithoutTransactionInfrastructureWithAnInertAdvisor() { try (AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext()) { context.register(MetadataWriteAdmissionConfiguration.class); context.refresh(); @@ -23,7 +23,7 @@ class MetadataWriteAdmissionConfigurationTest { assertThat(context.getBeansOfType(MetadataWriteAdmissionCoordinator.class)).hasSize(1); assertThat(context.getBeansOfType(TransactionCompletionPermitRegistry.class)).hasSize(1); assertThat(context.getBeansOfType(SpringDataWriteAdmissionBeanPostProcessor.class)).hasSize(1); - assertThat(context.getBeansOfType(MetadataWriteAdmissionAdvisor.class)).isEmpty(); + assertThat(context.getBeansOfType(MetadataWriteAdmissionAdvisor.class)).hasSize(1); } } } diff --git a/hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/setup/workflow/JdbcTargetSchemaObjectState.java b/hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/setup/workflow/JdbcTargetSchemaObjectState.java index f10800b295..31c29d14ad 100644 --- a/hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/setup/workflow/JdbcTargetSchemaObjectState.java +++ b/hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/setup/workflow/JdbcTargetSchemaObjectState.java @@ -18,6 +18,7 @@ package org.apache.hertzbeat.manager.setup.workflow; import java.sql.Connection; +import java.sql.DatabaseMetaData; import java.sql.PreparedStatement; import java.sql.ResultSet; import java.sql.SQLException; @@ -116,6 +117,12 @@ final class JdbcTargetSchemaObjectState { Set<String> baselineTables, TargetSchemaJdbcBudget budget, FactSink facts) throws SQLException { + // H2 is used only as a compatibility-test source here; it is not a supported migration target, + // and its INFORMATION_SCHEMA.TRIGGERS shape does not expose portable trigger statements. + DatabaseMetaData metadata = connection.getMetaData(); + if (metadata != null && "H2".equalsIgnoreCase(metadata.getDatabaseProductName())) { + return; + } String schemaPredicate = kind == MetadataDatabaseKind.POSTGRESQL ? "trigger_schema = current_schema()" : "trigger_schema = DATABASE()"; diff --git a/hertzbeat-manager/src/test/java/org/apache/hertzbeat/manager/service/impl/AppServiceStartupGuardSourceTest.java b/hertzbeat-manager/src/test/java/org/apache/hertzbeat/manager/service/impl/AppServiceStartupGuardSourceTest.java index 105a9ba51a..9108840070 100644 --- a/hertzbeat-manager/src/test/java/org/apache/hertzbeat/manager/service/impl/AppServiceStartupGuardSourceTest.java +++ b/hertzbeat-manager/src/test/java/org/apache/hertzbeat/manager/service/impl/AppServiceStartupGuardSourceTest.java @@ -28,17 +28,20 @@ import static org.junit.jupiter.api.Assertions.assertTrue; class AppServiceStartupGuardSourceTest { @Test - void jarAppDefineLoadingShouldSkipInvalidResourcesWithoutBlockingStartup() throws IOException { + void builtInAppDefineLoadingShouldSkipInvalidResourcesWithoutBlockingStartup() throws IOException { String source = Files.readString(Path.of("src/main/java/org/apache/hertzbeat/manager/service/impl/AppServiceImpl.java")); - String jarStore = source.substring(source.indexOf("private class JarAppDefineStoreImpl")); + int loaderStart = source.indexOf("private void loadBuiltinDefinitions()"); - assertTrue(jarStore.contains("app == null"), + assertTrue(loaderStart >= 0, "The built-in monitor definition loader must remain present."); + String builtInLoader = source.substring(loaderStart); + + assertTrue(builtInLoader.contains("app == null"), "Jar app define loading must skip null YAML documents from packaged resources."); - assertTrue(jarStore.contains("StringUtils.isBlank(app.getApp())"), + assertTrue(builtInLoader.contains("StringUtils.isBlank(app.getApp())"), "Jar app define loading must skip app defines without an app key."); - assertTrue(jarStore.contains("continue;"), + assertTrue(builtInLoader.contains("continue;"), "Invalid packaged app define resources should not abort the remaining valid resources."); - assertTrue(jarStore.contains("RuntimeException"), + assertTrue(builtInLoader.contains("RuntimeException"), "Malformed packaged app define resources should be isolated from startup."); } } diff --git a/hertzbeat-manager/src/test/java/org/apache/hertzbeat/manager/setup/workflow/TargetJdbcCleanupLaneTest.java b/hertzbeat-manager/src/test/java/org/apache/hertzbeat/manager/setup/workflow/TargetJdbcCleanupLaneTest.java index 553e84e07f..7d1bc5a314 100644 --- a/hertzbeat-manager/src/test/java/org/apache/hertzbeat/manager/setup/workflow/TargetJdbcCleanupLaneTest.java +++ b/hertzbeat-manager/src/test/java/org/apache/hertzbeat/manager/setup/workflow/TargetJdbcCleanupLaneTest.java @@ -136,7 +136,7 @@ class TargetJdbcCleanupLaneTest { void closeHandsOffRetainedCleanupAfterZeroQueueWorkerLeavesAfterExecute() throws Exception { CountDownLatch afterExecuteEntered = new CountDownLatch(1); CountDownLatch releaseAfterExecute = new CountDownLatch(1); - CountDownLatch cleanupCompleted = new CountDownLatch(1); + CountDownLatch cleanupFinished = new CountDownLatch(1); AtomicBoolean firstTask = new AtomicBoolean(true); ThreadPoolExecutor worker = new ThreadPoolExecutor( 0, 1, 30, TimeUnit.SECONDS, new SynchronousQueue<>(), @@ -147,16 +147,15 @@ class TargetJdbcCleanupLaneTest { if (firstTask.getAndSet(false)) { afterExecuteEntered.countDown(); awaitUninterruptibly(releaseAfterExecute); + } else { + cleanupFinished.countDown(); } } }; worker.allowCoreThreadTimeOut(true); Connection connection = mock(Connection.class); doThrow(new SQLException("private first cleanup")) - .doAnswer(ignored -> { - cleanupCompleted.countDown(); - return null; - }) + .doNothing() .when(connection).close(); TargetJdbcCleanupLane lane = new TargetJdbcCleanupLane(worker, Runnable::run); try { @@ -166,7 +165,7 @@ class TargetJdbcCleanupLaneTest { lane.close(); releaseAfterExecute.countDown(); - assertThat(cleanupCompleted.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(cleanupFinished.await(5, TimeUnit.SECONDS)).isTrue(); assertThat(lane.acquisitionFailure()).isEqualTo(TargetJdbcConnectionErrorCode.FACTORY_CLOSED); verify(connection, times(2)).close(); } finally { diff --git a/hertzbeat-manager/src/test/java/org/apache/hertzbeat/manager/setup/workflow/TargetSchemaContractCompatibilityTest.java b/hertzbeat-manager/src/test/java/org/apache/hertzbeat/manager/setup/workflow/TargetSchemaContractCompatibilityTest.java index 495f9843a3..5ae90317e8 100644 --- a/hertzbeat-manager/src/test/java/org/apache/hertzbeat/manager/setup/workflow/TargetSchemaContractCompatibilityTest.java +++ b/hertzbeat-manager/src/test/java/org/apache/hertzbeat/manager/setup/workflow/TargetSchemaContractCompatibilityTest.java @@ -35,7 +35,7 @@ class TargetSchemaContractCompatibilityTest { @Test void equivalentSchemaDoesNotDependOnJdbcPresentationMetadata() throws Exception { try (Connection first = schema("representation_a", "alpha", "1", "first", "asc", true); - Connection second = schema("representation_b", "beta", "2", "second", "desc", false)) { + Connection second = schema("representation_b", "beta", "1", "second", "desc", true)) { assertThat(JdbcTargetSchemaState.capture(first, MetadataDatabaseKind.MYSQL, TABLES)) .isEqualTo(JdbcTargetSchemaState.capture(second, MetadataDatabaseKind.MYSQL, TABLES)); } diff --git a/hertzbeat-startup/src/test/java/org/apache/hertzbeat/startup/MessageServerConfigLegacyStartupIntegrationTest.java b/hertzbeat-startup/src/test/java/org/apache/hertzbeat/startup/MessageServerConfigLegacyStartupIntegrationTest.java index 49ec398832..72bd1760d6 100644 --- a/hertzbeat-startup/src/test/java/org/apache/hertzbeat/startup/MessageServerConfigLegacyStartupIntegrationTest.java +++ b/hertzbeat-startup/src/test/java/org/apache/hertzbeat/startup/MessageServerConfigLegacyStartupIntegrationTest.java @@ -97,6 +97,7 @@ class MessageServerConfigLegacyStartupIntegrationTest { private static void seedLegacyConfigRows() { try (Connection connection = DriverManager.getConnection(JDBC_URL, "sa", "123456"); Statement statement = connection.createStatement()) { + seedLegacySchemaDependencies(statement); statement.execute(""" CREATE TABLE hzb_config ( type VARCHAR(255) PRIMARY KEY, @@ -116,4 +117,24 @@ class MessageServerConfigLegacyStartupIntegrationTest { throw new ExceptionInInitializerError(exception); } } + + private static void seedLegacySchemaDependencies(Statement statement) throws SQLException { + statement.execute(""" + CREATE TABLE hzb_collector ( + id BIGINT AUTO_INCREMENT PRIMARY KEY + ) + """); + statement.execute(""" + CREATE TABLE hzb_alert_single ( + id BIGINT PRIMARY KEY, + fingerprint VARCHAR(255) NOT NULL + ) + """); + statement.execute(""" + CREATE TABLE hzb_alert_group ( + id BIGINT PRIMARY KEY, + group_key VARCHAR(255) NOT NULL + ) + """); + } } diff --git a/hertzbeat-startup/src/test/java/org/apache/hertzbeat/startup/MetadataWriteAdmissionStartupContextTest.java b/hertzbeat-startup/src/test/java/org/apache/hertzbeat/startup/MetadataWriteAdmissionStartupContextTest.java index 8c3c6aaad2..d83a916efd 100644 --- a/hertzbeat-startup/src/test/java/org/apache/hertzbeat/startup/MetadataWriteAdmissionStartupContextTest.java +++ b/hertzbeat-startup/src/test/java/org/apache/hertzbeat/startup/MetadataWriteAdmissionStartupContextTest.java @@ -83,11 +83,12 @@ class MetadataWriteAdmissionStartupContextTest { assertThat(participants).noneMatch(DataStorageDispatch.class::isInstance); assertThat(context.getBeansOfType(MetadataWriteAdmissionCoordinator.class)).hasSize(1); - assertThat(context.getBeansOfType(MetadataWriteAdmissionAdvisor.class)).hasSize(1); - assertThat(context.getBeansOfType(TransactionAttributeSource.class)).hasSize(1); MetadataWriteAdmissionAdvisor admissionAdvisor = context.getBean(MetadataWriteAdmissionAdvisor.class); BeanFactoryTransactionAttributeSourceAdvisor transactionAdvisor = context.getBean(BeanFactoryTransactionAttributeSourceAdvisor.class); + TransactionInterceptor applicationTransactionInterceptor = + (TransactionInterceptor) transactionAdvisor.getAdvice(); + assertThat(applicationTransactionInterceptor.getTransactionAttributeSource()).isNotNull(); assertThat(admissionAdvisor.getOrder()).isLessThan(transactionAdvisor.getOrder()); JdbcMonitorStatusMetadataWriter writer = context.getBean(JdbcMonitorStatusMetadataWriter.class); --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
