[fix] make V181 migration idempotent and self-repair flyway history (#4326)

Co-authored-by: aias00 <liuhongyu@apache.org>

hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/config/FlywayConfiguration.java

@@ -17,7 +17,9 @@
 
 package org.apache.hertzbeat.manager.config;
 
+import lombok.extern.slf4j.Slf4j;
 import org.flywaydb.core.Flyway;
+import org.flywaydb.core.api.exception.FlywayValidateException;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.flyway.autoconfigure.FlywayMigrationInitializer;
 import org.springframework.boot.flyway.autoconfigure.FlywayProperties;
@@ -30,6 +32,7 @@ import org.springframework.context.annotation.DependsOn;
  * Delays Flyway execution until after Hibernate has created/updated the schema.
  */
 @Configuration
+@Slf4j
 @ConditionalOnProperty(prefix = "spring.flyway", name = "enabled", havingValue = "true")
 public class FlywayConfiguration {
 
@@ -52,7 +55,18 @@ public class FlywayConfiguration {
     @DependsOn("entityManagerFactory")
     Dummy delayedFlywayInitializer(Flyway flyway, FlywayProperties flywayProperties) {
         if (flywayProperties.isEnabled()) {
-            flyway.migrate();
+            try {
+                flyway.migrate();
+            } catch (FlywayValidateException e) {
+                if (e.getMessage() == null || !e.getMessage().contains("failed migration")) {
+                    // checksum mismatches and other validation problems need a human decision
+                    throw e;
+                }
+                // a recorded failed migration blocks every later start; repair + one retry un-bricks it
+                log.warn("Flyway history has a failed migration, repairing and retrying once: {}", e.getMessage());
+                flyway.repair();
+                flyway.migrate();
+            }
         }
         return new Dummy();
     }

hertzbeat-manager/src/test/java/org/apache/hertzbeat/manager/config/FlywayConfigurationTest.java

@@ -0,0 +1,71 @@
+/*
+ * 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.hertzbeat.manager.config;
+
+import org.flywaydb.core.Flyway;
+import org.flywaydb.core.api.CoreErrorCode;
+import org.flywaydb.core.api.ErrorDetails;
+import org.flywaydb.core.api.exception.FlywayValidateException;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.InOrder;
+import org.mockito.Mockito;
+import org.springframework.boot.flyway.autoconfigure.FlywayProperties;
+
+/**
+ * Test case for {@link FlywayConfiguration}
+ */
+class FlywayConfigurationTest {
+
+    private final Flyway flyway = Mockito.mock(Flyway.class);
+    private final FlywayConfiguration configuration = new FlywayConfiguration();
+
+    private FlywayProperties enabledProperties() {
+        FlywayProperties properties = new FlywayProperties();
+        properties.setEnabled(true);
+        return properties;
+    }
+
+    private static FlywayValidateException validateException(String message) {
+        return new FlywayValidateException(new ErrorDetails(CoreErrorCode.VALIDATE_ERROR, message), message);
+    }
+
+    @Test
+    void repairsAndRetriesWhenHistoryHasFailedMigration() {
+        Mockito.when(flyway.migrate())
+                .thenThrow(validateException("Detected failed migration to version 181 (update column)"))
+                .thenReturn(null);
+
+        configuration.delayedFlywayInitializer(flyway, enabledProperties());
+
+        InOrder inOrder = Mockito.inOrder(flyway);
+        inOrder.verify(flyway).migrate();
+        inOrder.verify(flyway).repair();
+        inOrder.verify(flyway).migrate();
+    }
+
+    @Test
+    void rethrowsOtherValidationErrorsWithoutRepair() {
+        Mockito.when(flyway.migrate())
+                .thenThrow(validateException("Migration checksum mismatch for migration version 180"));
+
+        Assertions.assertThrows(FlywayValidateException.class,
+                () -> configuration.delayedFlywayInitializer(flyway, enabledProperties()));
+        Mockito.verify(flyway, Mockito.never()).repair();
+    }
+}

hertzbeat-startup/src/main/resources/db/migration/mysql/V181__update_column.sql

@@ -33,7 +33,7 @@
 -- under the License.
 
 -- Scheduled SOP execution configurations
-CREATE TABLE hzb_sop_schedule (
+CREATE TABLE IF NOT EXISTS hzb_sop_schedule (
     id BIGINT AUTO_INCREMENT PRIMARY KEY,
     conversation_id BIGINT NOT NULL COMMENT 'Conversation ID to push results to',
     sop_name VARCHAR(64) NOT NULL COMMENT 'Name of the SOP skill to execute',

hertzbeat-startup/src/main/resources/db/migration/postgresql/V181__update_column.sql

@@ -33,7 +33,7 @@
 -- under the License.
 
 -- Scheduled SOP execution configurations
-CREATE TABLE hzb_sop_schedule (
+CREATE TABLE IF NOT EXISTS hzb_sop_schedule (
     id BIGSERIAL PRIMARY KEY,
     conversation_id BIGINT NOT NULL,
     sop_name VARCHAR(64) NOT NULL,
@@ -57,5 +57,5 @@ COMMENT ON COLUMN hzb_sop_schedule.enabled IS 'Whether the schedule is enabled';
 COMMENT ON COLUMN hzb_sop_schedule.last_run_time IS 'Last execution time';
 COMMENT ON COLUMN hzb_sop_schedule.next_run_time IS 'Next scheduled execution time';
 
-CREATE INDEX idx_schedule_conversation_id ON hzb_sop_schedule(conversation_id);
+CREATE INDEX IF NOT EXISTS idx_schedule_conversation_id ON hzb_sop_schedule(conversation_id);
-CREATE INDEX idx_schedule_enabled_next ON hzb_sop_schedule(enabled, next_run_time);
+CREATE INDEX IF NOT EXISTS idx_schedule_enabled_next ON hzb_sop_schedule(enabled, next_run_time);