This is an automated email from the ASF dual-hosted git repository.
ArafatKhan2198 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/master by this push:
new 6cc8f31d2fe HDDS-16074. Make Recon RECON_TASK_STATUS column upgrade
idempotent and fail on error. (#11060).
6cc8f31d2fe is described below
commit 6cc8f31d2fe7a75d044eef29f692da120e2af0ed
Author: Arafat2198 <[email protected]>
AuthorDate: Fri Aug 28 14:06:48 2026 +0530
HDDS-16074. Make Recon RECON_TASK_STATUS column upgrade idempotent and fail
on error. (#11060).
---
.../recon/upgrade/ReconLayoutVersionManager.java | 95 +++++++----
.../upgrade/ReconTaskStatusTableUpgradeAction.java | 87 +++++++---
.../upgrade/TestReconLayoutVersionManager.java | 23 ++-
.../TestReconTaskStatusTableUpgradeAction.java | 176 +++++++++++++++++++++
4 files changed, 328 insertions(+), 53 deletions(-)
diff --git
a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/upgrade/ReconLayoutVersionManager.java
b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/upgrade/ReconLayoutVersionManager.java
index cb12546771e..2d0915898f2 100644
---
a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/upgrade/ReconLayoutVersionManager.java
+++
b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/upgrade/ReconLayoutVersionManager.java
@@ -17,6 +17,7 @@
package org.apache.hadoop.ozone.recon.upgrade;
+import com.google.common.annotations.VisibleForTesting;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.Arrays;
@@ -42,6 +43,7 @@ public class ReconLayoutVersionManager {
private final ReconSchemaVersionTableManager schemaVersionTableManager;
private final ReconContext reconContext;
private final DataSource dataSource;
+ private final ReconUpgradeAction taskStatusSchemaRepairAction;
// Metadata Layout Version (MLV) of the Recon Metadata on disk
private int currentMLV;
@@ -49,10 +51,20 @@ public class ReconLayoutVersionManager {
public ReconLayoutVersionManager(ReconSchemaVersionTableManager
schemaVersionTableManager,
ReconContext reconContext, DataSource
dataSource)
throws SQLException {
+ this(schemaVersionTableManager, reconContext, dataSource,
+ new ReconTaskStatusTableUpgradeAction());
+ }
+
+ @VisibleForTesting
+ ReconLayoutVersionManager(
+ ReconSchemaVersionTableManager schemaVersionTableManager,
+ ReconContext reconContext, DataSource dataSource,
+ ReconUpgradeAction taskStatusSchemaRepairAction) throws SQLException {
this.schemaVersionTableManager = schemaVersionTableManager;
this.currentMLV = determineMLV();
this.reconContext = reconContext;
this.dataSource = dataSource;
+ this.taskStatusSchemaRepairAction = taskStatusSchemaRepairAction;
ReconLayoutFeature.registerUpgradeActions(); // Register actions via
annotation
}
@@ -80,38 +92,47 @@ private int determineSLV() {
* feature that is registered for finalization.
*/
public void finalizeLayoutFeatures() {
- // Get features that need finalization, sorted by version
- List<ReconLayoutFeature> featuresToFinalize = getRegisteredFeatures();
- LOG.debug("Starting finalization of {} features.",
featuresToFinalize.size());
-
- try (Connection connection = dataSource.getConnection()) {
- connection.setAutoCommit(false); // Turn off auto-commit for
transactional control
-
- for (ReconLayoutFeature feature : featuresToFinalize) {
- LOG.debug("Processing feature version: {}", feature.getVersion());
- try {
- // Fetch the action for the feature
- Optional<ReconUpgradeAction> action = feature.getAction();
- if (action.isPresent()) {
- LOG.debug("Finalize action found for feature version: {}",
feature.getVersion());
- // Update the schema version in the database
- updateSchemaVersion(feature.getVersion(), connection);
-
- // Execute the upgrade action
- action.get().execute(dataSource);
-
- // Commit the transaction only if both operations succeed
- connection.commit();
- LOG.info("Feature versioned {} finalized successfully.",
feature.getVersion());
- } else {
- LOG.info("No finalize action found for feature version: {}",
feature.getVersion());
+ try {
+ repairTaskStatusSchemaIfRequired();
+
+ // Get features that need finalization, sorted by version
+ List<ReconLayoutFeature> featuresToFinalize = getRegisteredFeatures();
+ LOG.debug("Starting finalization of {} features.",
+ featuresToFinalize.size());
+
+ try (Connection connection = dataSource.getConnection()) {
+ connection.setAutoCommit(false);
+
+ for (ReconLayoutFeature feature : featuresToFinalize) {
+ LOG.debug("Processing feature version: {}", feature.getVersion());
+ try {
+ // Fetch the action for the feature
+ Optional<ReconUpgradeAction> action = feature.getAction();
+ if (action.isPresent()) {
+ LOG.debug("Finalize action found for feature version: {}",
+ feature.getVersion());
+ // Update the schema version in the database
+ updateSchemaVersion(feature.getVersion(), connection);
+
+ // Execute the upgrade action
+ action.get().execute(dataSource);
+
+ // Commit the transaction only if both operations succeed
+ connection.commit();
+ LOG.info("Feature versioned {} finalized successfully.",
+ feature.getVersion());
+ } else {
+ LOG.info("No finalize action found for feature version: {}",
+ feature.getVersion());
+ }
+ } catch (Exception e) {
+ // Rollback pending changes for the current feature on failure
+ connection.rollback();
+ currentMLV = determineMLV();
+ LOG.error("Failed to finalize feature {}. Rolling back changes.",
+ feature.getVersion(), e);
+ throw e;
}
- } catch (Exception e) {
- // Rollback any pending changes for the current feature due to
failure
- connection.rollback();
- currentMLV = determineMLV(); // Rollback the MLV to the original
value
- LOG.error("Failed to finalize feature {}. Rolling back changes.",
feature.getVersion(), e);
- throw e;
}
}
} catch (Exception e) {
@@ -123,6 +144,18 @@ public void finalizeLayoutFeatures() {
}
}
+ /**
+ * Repairs the RECON_TASK_STATUS schema based on the table's actual column
+ * state, independent of the stored layout version. Running this before
feature
+ * finalization ensures the required columns exist and completes any
partially
+ * applied migration, so a table left in an inconsistent state is recovered
+ * regardless of what the layout version reports.
+ */
+ private void repairTaskStatusSchemaIfRequired() throws Exception {
+ LOG.info("Checking whether the Recon task status schema requires repair.");
+ taskStatusSchemaRepairAction.execute(dataSource);
+ }
+
/**
* Returns a list of ReconLayoutFeature objects that are registered for
finalization.
*/
diff --git
a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/upgrade/ReconTaskStatusTableUpgradeAction.java
b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/upgrade/ReconTaskStatusTableUpgradeAction.java
index 17d64abb9c5..f619e26ebd6 100644
---
a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/upgrade/ReconTaskStatusTableUpgradeAction.java
+++
b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/upgrade/ReconTaskStatusTableUpgradeAction.java
@@ -21,10 +21,13 @@
import static org.apache.ozone.recon.schema.SqlDbUtils.TABLE_EXISTS_CHECK;
import java.sql.Connection;
+import java.sql.DatabaseMetaData;
+import java.sql.ResultSet;
import java.sql.SQLException;
import javax.sql.DataSource;
import org.apache.ozone.recon.schema.ReconTaskSchemaDefinition;
import org.jooq.DSLContext;
+import org.jooq.Field;
import org.jooq.exception.DataAccessException;
import org.jooq.impl.DSL;
import org.jooq.impl.SQLDataType;
@@ -33,13 +36,20 @@
/**
* Upgrade action for TASK_STATUS_STATISTICS feature layout change, which adds
- * <code>last_task_run_status</code> and <code>current_task_run_status</code>
columns to
- * {@link ReconTaskSchemaDefinition} in case it is missing .
+ * <code>last_task_run_status</code> and
+ * <code>is_current_task_running</code> columns to
+ * {@link ReconTaskSchemaDefinition} when they are missing.
*/
@UpgradeActionRecon(feature = ReconLayoutFeature.TASK_STATUS_STATISTICS)
public class ReconTaskStatusTableUpgradeAction implements ReconUpgradeAction {
- private static final Logger LOG =
LoggerFactory.getLogger(ReconTaskStatusTableUpgradeAction.class);
+ private static final Logger LOG =
+ LoggerFactory.getLogger(ReconTaskStatusTableUpgradeAction.class);
+ private static final String LAST_TASK_RUN_STATUS =
+ "last_task_run_status";
+ private static final String IS_CURRENT_TASK_RUNNING =
+ "is_current_task_running";
+ private static final int COLUMN_MISSING = -1;
/**
* Utility function to add provided column to RECON_TASK_STATUS table as
INTEGER type.
@@ -63,33 +73,68 @@ private void setColumnAsNonNullable(DSLContext dslContext,
String columnName) {
.execute();
}
+ /**
+ * Returns the JDBC nullability value for a column, or
+ * {@link #COLUMN_MISSING} if it does not exist.
+ */
+ private int getColumnNullability(Connection connection, String columnName)
+ throws SQLException {
+ DatabaseMetaData metaData = connection.getMetaData();
+ try (ResultSet columns = metaData.getColumns(null, null, null, null)) {
+ while (columns.next()) {
+ String table = columns.getString("TABLE_NAME");
+ String column = columns.getString("COLUMN_NAME");
+ if (RECON_TASK_STATUS_TABLE_NAME.equalsIgnoreCase(table)
+ && columnName.equalsIgnoreCase(column)) {
+ return columns.getInt("NULLABLE");
+ }
+ }
+ }
+ return COLUMN_MISSING;
+ }
+
+ /**
+ * Adds a missing column and completes any partially applied migration.
+ */
+ private void repairColumn(Connection connection, DSLContext dslContext,
+ String columnName) throws SQLException {
+ int nullability = getColumnNullability(connection, columnName);
+ if (nullability == COLUMN_MISSING) {
+ LOG.info("Adding '{}' column to task status table.", columnName);
+ addColumnToTable(dslContext, columnName);
+ nullability = DatabaseMetaData.columnNullable;
+ }
+
+ if (nullability != DatabaseMetaData.columnNoNulls) {
+ Field<Integer> column =
+ DSL.field(DSL.name(columnName), SQLDataType.INTEGER);
+ int updatedRowCount = dslContext
+ .update(DSL.table(RECON_TASK_STATUS_TABLE_NAME))
+ .set(column, 0)
+ .where(column.isNull())
+ .execute();
+ LOG.info("Updated {} rows with a default value for '{}'.",
+ updatedRowCount, columnName);
+ setColumnAsNonNullable(dslContext, columnName);
+ }
+ }
+
@Override
- public void execute(DataSource dataSource) throws DataAccessException {
+ public void execute(DataSource dataSource) throws SQLException {
try (Connection conn = dataSource.getConnection()) {
if (!TABLE_EXISTS_CHECK.test(conn, RECON_TASK_STATUS_TABLE_NAME)) {
+ LOG.info("{} table does not exist; task status schema repair is not "
+ + "required.", RECON_TASK_STATUS_TABLE_NAME);
return;
}
DSLContext dslContext = DSL.using(conn);
- // JOOQ doesn't support Derby DB officially, there is no way to run 'ADD
COLUMN' command in single call
- // for multiple columns. Hence, we run it as two separate steps.
- LOG.info("Adding 'last_task_run_status' column to task status table");
- addColumnToTable(dslContext, "last_task_run_status");
- LOG.info("Adding 'is_current_task_running' column to task status table");
- addColumnToTable(dslContext, "is_current_task_running");
-
- //Handle previous table values with new columns default values
- int updatedRowCount =
dslContext.update(DSL.table(RECON_TASK_STATUS_TABLE_NAME))
- .set(DSL.field(DSL.name("last_task_run_status"),
SQLDataType.INTEGER), 0)
- .set(DSL.field(DSL.name("is_current_task_running"),
SQLDataType.INTEGER), 0)
- .execute();
- LOG.info("Updated {} rows with default value for new columns",
updatedRowCount);
-
- // Now we will set the column as not-null to enforce constraints
- setColumnAsNonNullable(dslContext, "last_task_run_status");
- setColumnAsNonNullable(dslContext, "is_current_task_running");
+ repairColumn(conn, dslContext, LAST_TASK_RUN_STATUS);
+ repairColumn(conn, dslContext, IS_CURRENT_TASK_RUNNING);
} catch (SQLException | DataAccessException ex) {
LOG.error("Error while upgrading RECON_TASK_STATUS table.", ex);
+ throw new SQLException(
+ "Failed to repair the RECON_TASK_STATUS table.", ex);
}
}
}
diff --git
a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/upgrade/TestReconLayoutVersionManager.java
b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/upgrade/TestReconLayoutVersionManager.java
index 5f2d451dd06..9234e1c541b 100644
---
a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/upgrade/TestReconLayoutVersionManager.java
+++
b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/upgrade/TestReconLayoutVersionManager.java
@@ -85,8 +85,10 @@ public void setUp() throws SQLException {
// Define the custom features to be returned
mockedEnum.when(ReconLayoutFeature::values).thenReturn(new
ReconLayoutFeature[]{feature1, feature2});
+ // Inject a no-op task-status schema repair action so these tests exercise
+ // only the layout finalization flow, independent of the real action's SQL.
layoutVersionManager = new
ReconLayoutVersionManager(schemaVersionTableManager, mock(ReconContext.class),
- mockDataSource);
+ mockDataSource, mock(ReconUpgradeAction.class));
when(scmFacadeMock.getDataSource()).thenReturn(mockDataSource);
when(mockDataSource.getConnection()).thenReturn(mockConnection);
@@ -294,6 +296,25 @@ public void testNoUpgradeActionsNeeded() throws
SQLException {
verify(schemaVersionTableManager, never()).updateSchemaVersion(anyInt(),
eq(mockConnection));
}
+ @Test
+ public void testTaskStatusRepairDoesNotChangeMlv() throws Exception {
+ ReconUpgradeAction repairAction = mock(ReconUpgradeAction.class);
+ when(schemaVersionTableManager.getCurrentSchemaVersion()).thenReturn(2);
+ mockedEnum.when(ReconLayoutFeature::values)
+ .thenReturn(new ReconLayoutFeature[]{});
+
+ ReconLayoutVersionManager manager = new ReconLayoutVersionManager(
+ schemaVersionTableManager, mock(ReconContext.class), mockDataSource,
+ repairAction);
+
+ manager.finalizeLayoutFeatures();
+
+ verify(repairAction).execute(mockDataSource);
+ verify(schemaVersionTableManager, never())
+ .updateSchemaVersion(anyInt(), eq(mockConnection));
+ assertEquals(2, manager.getCurrentMLV());
+ }
+
/**
* Tests the scenario where the first two features are finalized,
* and then a third feature is introduced. Ensures that only the
diff --git
a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/upgrade/TestReconTaskStatusTableUpgradeAction.java
b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/upgrade/TestReconTaskStatusTableUpgradeAction.java
new file mode 100644
index 00000000000..2e91375ef8f
--- /dev/null
+++
b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/upgrade/TestReconTaskStatusTableUpgradeAction.java
@@ -0,0 +1,176 @@
+/*
+ * 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.hadoop.ozone.recon.upgrade;
+
+import static
org.apache.ozone.recon.schema.ReconTaskSchemaDefinition.RECON_TASK_STATUS_TABLE_NAME;
+import static org.jooq.impl.DSL.field;
+import static org.jooq.impl.DSL.name;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.sql.Connection;
+import java.sql.DatabaseMetaData;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import javax.sql.DataSource;
+import org.apache.hadoop.ozone.recon.persistence.AbstractReconSqlDBTest;
+import org.jooq.DSLContext;
+import org.jooq.SQLDialect;
+import org.jooq.impl.DSL;
+import org.jooq.impl.SQLDataType;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests for ReconTaskStatusTableUpgradeAction.
+ */
+public class TestReconTaskStatusTableUpgradeAction
+ extends AbstractReconSqlDBTest {
+
+ private static final String LAST_TASK_RUN_STATUS =
+ "last_task_run_status";
+ private static final String IS_CURRENT_TASK_RUNNING =
+ "is_current_task_running";
+
+ private DataSource dataSource;
+ private DSLContext dslContext;
+ private ReconTaskStatusTableUpgradeAction upgradeAction;
+
+ @BeforeEach
+ public void setUp() throws SQLException {
+ dataSource = getDataSource();
+ // Use the Derby dialect explicitly. The shared DSLContext from the base
+ // class uses SQLDialect.DEFAULT, which renders Derby-incompatible DDL
+ // (e.g. "DROP TABLE IF EXISTS" and "bigint null"); a Derby-dialect context
+ // generates SQL that the embedded Derby database accepts.
+ dslContext = DSL.using(dataSource, SQLDialect.DERBY);
+ upgradeAction = new ReconTaskStatusTableUpgradeAction();
+ createLegacyTaskStatusTable();
+ }
+
+ @Test
+ public void testRepairsLegacyTaskStatusTable() throws Exception {
+ upgradeAction.execute(dataSource);
+
+ assertTrue(columnExists(LAST_TASK_RUN_STATUS));
+ assertTrue(columnExists(IS_CURRENT_TASK_RUNNING));
+ assertFalse(columnIsNullable(LAST_TASK_RUN_STATUS));
+ assertFalse(columnIsNullable(IS_CURRENT_TASK_RUNNING));
+ assertEquals(0, getStatusValue(LAST_TASK_RUN_STATUS));
+ assertEquals(0, getStatusValue(IS_CURRENT_TASK_RUNNING));
+ }
+
+ @Test
+ public void testRepairIsIdempotentAndPreservesValues() throws Exception {
+ upgradeAction.execute(dataSource);
+ setStatusValue(LAST_TASK_RUN_STATUS, 7);
+ setStatusValue(IS_CURRENT_TASK_RUNNING, 1);
+
+ assertDoesNotThrow(() -> upgradeAction.execute(dataSource));
+
+ assertEquals(7, getStatusValue(LAST_TASK_RUN_STATUS));
+ assertEquals(1, getStatusValue(IS_CURRENT_TASK_RUNNING));
+ }
+
+ @Test
+ public void testCompletesPartiallyAppliedRepair() throws Exception {
+ dslContext.alterTable(RECON_TASK_STATUS_TABLE_NAME)
+ .addColumn(LAST_TASK_RUN_STATUS,
+ SQLDataType.INTEGER.nullable(true))
+ .execute();
+ setStatusValue(LAST_TASK_RUN_STATUS, 7);
+
+ upgradeAction.execute(dataSource);
+
+ assertTrue(columnExists(IS_CURRENT_TASK_RUNNING));
+ assertFalse(columnIsNullable(LAST_TASK_RUN_STATUS));
+ assertFalse(columnIsNullable(IS_CURRENT_TASK_RUNNING));
+ assertEquals(7, getStatusValue(LAST_TASK_RUN_STATUS));
+ assertEquals(0, getStatusValue(IS_CURRENT_TASK_RUNNING));
+ }
+
+ @Test
+ public void testDatabaseFailureIsPropagated() throws SQLException {
+ DataSource failingDataSource = mock(DataSource.class);
+ when(failingDataSource.getConnection())
+ .thenThrow(new SQLException("Database unavailable"));
+
+ assertThrows(SQLException.class,
+ () -> upgradeAction.execute(failingDataSource));
+ }
+
+ private void createLegacyTaskStatusTable() throws SQLException {
+ // The base class always creates RECON_TASK_STATUS before this runs, so a
+ // plain DROP TABLE is safe (no IF EXISTS needed).
+ dslContext.dropTable(RECON_TASK_STATUS_TABLE_NAME).execute();
+ dslContext.createTable(RECON_TASK_STATUS_TABLE_NAME)
+ .column("task_name", SQLDataType.VARCHAR(766).nullable(false))
+ .column("last_updated_timestamp", SQLDataType.BIGINT)
+ .column("last_updated_seq_number", SQLDataType.BIGINT)
+ .constraint(DSL.constraint("pk_task_name").primaryKey("task_name"))
+ .execute();
+ dslContext.insertInto(DSL.table(RECON_TASK_STATUS_TABLE_NAME))
+ .columns(field(name("task_name")),
+ field(name("last_updated_timestamp")),
+ field(name("last_updated_seq_number")))
+ .values("OmDeltaRequest", 1L, 1L)
+ .execute();
+ }
+
+ private boolean columnExists(String columnName) throws SQLException {
+ return getColumnNullability(columnName) != -1;
+ }
+
+ private boolean columnIsNullable(String columnName) throws SQLException {
+ return getColumnNullability(columnName)
+ != DatabaseMetaData.columnNoNulls;
+ }
+
+ private int getColumnNullability(String columnName) throws SQLException {
+ try (Connection connection = dataSource.getConnection();
+ ResultSet columns = connection.getMetaData()
+ .getColumns(null, null, null, null)) {
+ while (columns.next()) {
+ if (RECON_TASK_STATUS_TABLE_NAME.equalsIgnoreCase(
+ columns.getString("TABLE_NAME"))
+ && columnName.equalsIgnoreCase(
+ columns.getString("COLUMN_NAME"))) {
+ return columns.getInt("NULLABLE");
+ }
+ }
+ }
+ return -1;
+ }
+
+ private void setStatusValue(String columnName, int value) {
+ dslContext.update(DSL.table(RECON_TASK_STATUS_TABLE_NAME))
+ .set(field(name(columnName), Integer.class), value)
+ .execute();
+ }
+
+ private int getStatusValue(String columnName) {
+ return dslContext.select(field(name(columnName), Integer.class))
+ .from(DSL.table(RECON_TASK_STATUS_TABLE_NAME))
+ .fetchOne(field(name(columnName), Integer.class));
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
