fix(governance): validate Flowable pool connections to survive DB failover (#28835)

The runtime Flowable engine builds its own MyBatis PooledDataSource from raw
JDBC settings and does not validate pooled connections, so when the database
drops a connection (Aurora/RDS failover, maintenance restart, or idle reaper)
the async-executor poll threads (ResetExpiredJobsRunnable) keep borrowing the
dead connection and fail with "PSQLException: FATAL: terminating connection due
to administrator command".

Enable MyBatis pool ping (SELECT 1) on the runtime engine so any connection idle
past 30s is validated and transparently replaced before reuse. The migration
path uses setDataSource() and is unaffected.

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: 2 people

openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/WorkflowHandler.java

@@ -76,6 +76,13 @@ public class WorkflowHandler {
   @Getter private static volatile boolean initialized = false;
   private final boolean isMigrationContext;
 
+  private static final String CONNECTION_VALIDATION_QUERY = "SELECT 1";
+
+  // Validate any pooled connection idle longer than this before reuse. Kept below Flowable's
+  // 60s reset-expired-jobs interval so the periodic async-executor threads always re-validate,
+  // while connections in active sub-second use skip the check and pay no overhead.
+  private static final int CONNECTION_PING_NOT_USED_FOR_MILLIS = 30000;
+
   private WorkflowHandler(OpenMetadataApplicationConfig config, boolean isMigrationContext) {
     this.isMigrationContext = isMigrationContext;
     StandaloneProcessEngineConfiguration processEngineConfiguration =
@@ -179,6 +186,7 @@ public void initializeNewProcessEngine(
       processEngineConfiguration.setJdbcPassword(
           currentProcessEngineConfiguration.getJdbcPassword());
       processEngineConfiguration.setJdbcDriver(currentProcessEngineConfiguration.getJdbcDriver());
+      configureConnectionPoolHealthChecks(processEngineConfiguration);
     }
     processEngineConfiguration.setDatabaseType(currentProcessEngineConfiguration.getDatabaseType());
     processEngineConfiguration.setDatabaseSchemaUpdate(
@@ -239,6 +247,25 @@ public void initializeNewProcessEngine(
         .addMapper(SqlMapper.class);
   }
 
+  /**
+   * Enable connection-pool health checks on the runtime engine's MyBatis pool.
+   *
+   * <p>Unlike the application's HikariCP pool, the Flowable runtime engine builds its own MyBatis
+   * {@code PooledDataSource} from the raw JDBC settings and does not validate pooled connections.
+   * When the database drops a connection out from under the pool — an Aurora/RDS failover, a
+   * maintenance restart, or an idle-connection reaper, all of which surface as {@code
+   * PSQLException: terminating connection due to administrator command} — the async executor's
+   * polling threads (e.g. {@code ResetExpiredJobsRunnable}) keep borrowing the dead connection and
+   * failing until the pool happens to recycle it. Pool ping runs a lightweight validation query on
+   * any connection idle past the threshold and transparently replaces it before handing it out.
+   */
+  private static void configureConnectionPoolHealthChecks(
+      StandaloneProcessEngineConfiguration processEngineConfiguration) {
+    processEngineConfiguration.setJdbcPingEnabled(true);
+    processEngineConfiguration.setJdbcPingQuery(CONNECTION_VALIDATION_QUERY);
+    processEngineConfiguration.setJdbcPingConnectionNotUsedFor(CONNECTION_PING_NOT_USED_FOR_MILLIS);
+  }
+
   public static void initialize(OpenMetadataApplicationConfig config) {
     initialize(config, false);
   }

openmetadata-service/src/test/java/org/openmetadata/service/governance/workflows/WorkflowHandlerSchemaUpdateTest.java

@@ -18,11 +18,13 @@
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
 import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
 import static org.mockito.Mockito.lenient;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.mockConstruction;
 import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
@@ -165,6 +167,59 @@ void runtimeModeSetsDbSchemaUpdateFalse() {
     }
   }
 
+  @Test
+  void runtimeModeEnablesConnectionPoolHealthChecks() {
+    try (MockedConstruction<StandaloneProcessEngineConfiguration> engineMock =
+            mockConstruction(
+                StandaloneProcessEngineConfiguration.class,
+                (mock, ctx) ->
+                    when(mock.buildProcessEngine())
+                        .thenThrow(new FlowableWrongDbException("7.2.0.2", "7.1.0.0")));
+        MockedStatic<ProcessEngines> ignored = mockStatic(ProcessEngines.class);
+        MockedStatic<Entity> entityMock = mockStatic(Entity.class);
+        MockedStatic<PipelineServiceClientFactory> pscMock =
+            mockStatic(PipelineServiceClientFactory.class)) {
+
+      setupEntityMock(entityMock);
+      pscMock
+          .when(() -> PipelineServiceClientFactory.createPipelineServiceClient(any()))
+          .thenReturn(null);
+
+      assertThrows(
+          IllegalStateException.class, () -> WorkflowHandler.initialize(buildMockConfig(), false));
+
+      StandaloneProcessEngineConfiguration engineConfig = engineMock.constructed().getLast();
+      verify(engineConfig).setJdbcPingEnabled(true);
+      verify(engineConfig).setJdbcPingQuery("SELECT 1");
+      verify(engineConfig).setJdbcPingConnectionNotUsedFor(30000);
+    }
+  }
+
+  @Test
+  void migrationModeDoesNotEnableConnectionPoolPing() {
+    ProcessEngine mockEngine = mock(ProcessEngine.class, RETURNS_DEEP_STUBS);
+
+    try (MockedConstruction<StandaloneProcessEngineConfiguration> engineMock =
+            mockConstruction(
+                StandaloneProcessEngineConfiguration.class,
+                (mock, ctx) -> when(mock.buildProcessEngine()).thenReturn(mockEngine));
+        MockedStatic<ProcessEngines> ignored = mockStatic(ProcessEngines.class);
+        MockedStatic<Entity> entityMock = mockStatic(Entity.class);
+        MockedStatic<PipelineServiceClientFactory> pscMock =
+            mockStatic(PipelineServiceClientFactory.class)) {
+
+      setupEntityMock(entityMock);
+      pscMock
+          .when(() -> PipelineServiceClientFactory.createPipelineServiceClient(any()))
+          .thenReturn(null);
+
+      WorkflowHandler.initialize(buildMockConfig(), true);
+
+      StandaloneProcessEngineConfiguration engineConfig = engineMock.constructed().getLast();
+      verify(engineConfig, never()).setJdbcPingEnabled(anyBoolean());
+    }
+  }
+
   // ── Helpers ──────────────────────────────────────────────────────────────────
 
   private void setupEntityMock(MockedStatic<Entity> entityMock) {