This is an automated email from the ASF dual-hosted git repository.
devmadhuu 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 deb56f03629 HDDS-16939. Recon drops replica history for departed
datanodes on OM full reprocess (#11494)
deb56f03629 is described below
commit deb56f036298ffbc7ae26cf2fec0908254eeab6e
Author: Devesh Kumar Singh <[email protected]>
AuthorDate: Wed Oct 14 12:10:08 2026 +0530
HDDS-16939. Recon drops replica history for departed datanodes on OM full
reprocess (#11494)
---
hadoop-hdds/docs/content/concept/Recon.md | 14 +-
.../hadoop/ozone/debug/DBDefinitionFactory.java | 3 +-
.../ozone/debug/TestDBDefinitionFactory.java | 26 ++
hadoop-ozone/integration-test-recon/pom.xml | 5 +
.../ozone/recon/TestReconWithOzoneManager.java | 125 +++++++++-
.../apache/hadoop/ozone/recon/ReconConstants.java | 2 +
.../hadoop/ozone/recon/ReconControllerModule.java | 2 +
.../org/apache/hadoop/ozone/recon/ReconServer.java | 13 +
.../ozone/recon/scm/ReconContainerManager.java | 26 +-
.../impl/ReconContainerMetadataManagerImpl.java | 32 ++-
.../ozone/recon/spi/impl/ReconDBDefinition.java | 2 +-
.../spi/impl/ReconReplicaHistoryDBDefinition.java | 50 ++++
.../spi/impl/ReconReplicaHistoryDBProvider.java | 98 ++++++++
.../hadoop/ozone/recon/upgrade/ReconVersion.java | 1 +
.../upgrade/ReplicaHistoryDBUpgradeAction.java | 40 ++++
.../hadoop/ozone/recon/ReconTestInjector.java | 2 +
.../ozone/recon/api/TestContainerEndpoint.java | 54 +++++
.../scm/AbstractReconContainerManagerTest.java | 8 +-
.../ozone/recon/scm/TestReconContainerManager.java | 90 +++++++
.../TestReconStorageContainerManagerFacade.java | 8 +
.../impl/TestReconReplicaHistoryDBProvider.java | 262 +++++++++++++++++++++
.../recon/upgrade/TestReconVersionManager.java | 124 ++++++++++
22 files changed, 954 insertions(+), 33 deletions(-)
diff --git a/hadoop-hdds/docs/content/concept/Recon.md
b/hadoop-hdds/docs/content/concept/Recon.md
index 608a37c39b4..e84fb81b591 100644
--- a/hadoop-hdds/docs/content/concept/Recon.md
+++ b/hadoop-hdds/docs/content/concept/Recon.md
@@ -124,6 +124,10 @@ to the Prometheus endpoint like
`ozone.recon.prometheus.http.endpoint=http://pro
* Stores the mapping (container, key) -> count
* ContainerKeyCount table
* Stores containerID -> no. of keys count within the container
+ * Container replica history is stored in `recon-container-replica-history.db`
+ under `ozone.recon.db.dir`. This database is independent of the databases
+ replaced during OM reprocess and SCM snapshot synchronization, so history
for
+ departed datanodes survives those operations.
* The following data is stored in the configured SQL database (default is
Derby):
* GlobalStats table
@@ -134,14 +138,18 @@ to the Prometheus endpoint like
`ozone.recon.prometheus.http.endpoint=http://pro
* ReconTaskStatus table
* Keeps track of the status and last run timestamp of the registered OM
and SCM
db tasks in the [Recon Task Framework](#task-framework)
- * ContainerHistory table
- * Stores ContainerReplica -> Datanode mapping with last known
timestamp. This
- is used to determine the last known datanodes when a container is
reported missing
* UnhealthyContainers table
* Keeps track of all the Unhealthy Containers (MISSING,
UNDER_REPLICATED,
OVER_REPLICATED, MIS_REPLICATED) in the cluster at any given time
+On upgrade, Recon's versioned upgrade action migrates existing replica history
from
+`recon-container-key.db` before finalizing the new version and starting its
services.
+The migration streams the history through a temporary
+SST file and retains the source table. A durable completion marker prevents
migration
+on subsequent starts; an incomplete migration is retried on the next start.
Allow
+disk space for the new history database and the temporary SST file during
migration.
+
## Notable configurations
key | default | <div style="width: 300px;">description</div>
diff --git
a/hadoop-ozone/cli-debug/src/main/java/org/apache/hadoop/ozone/debug/DBDefinitionFactory.java
b/hadoop-ozone/cli-debug/src/main/java/org/apache/hadoop/ozone/debug/DBDefinitionFactory.java
index ad366b02421..d22f18e3b8b 100644
---
a/hadoop-ozone/cli-debug/src/main/java/org/apache/hadoop/ozone/debug/DBDefinitionFactory.java
+++
b/hadoop-ozone/cli-debug/src/main/java/org/apache/hadoop/ozone/debug/DBDefinitionFactory.java
@@ -39,6 +39,7 @@
import org.apache.hadoop.ozone.om.codec.OMDBDefinition;
import org.apache.hadoop.ozone.recon.scm.ReconSCMDBDefinition;
import org.apache.hadoop.ozone.recon.spi.impl.ReconDBDefinition;
+import org.apache.hadoop.ozone.recon.spi.impl.ReconReplicaHistoryDBDefinition;
/**
* Utility class to get appropriate DBDefinition.
@@ -51,7 +52,7 @@ public final class DBDefinitionFactory {
static {
final Map<String, DBDefinition> map = new HashMap<>();
Arrays.asList(SCMDBDefinition.get(), OMDBDefinition.get(),
ReconSCMDBDefinition.get(),
- WitnessedContainerDBDefinition.get())
+ WitnessedContainerDBDefinition.get(), new
ReconReplicaHistoryDBDefinition())
.forEach(dbDefinition -> map.put(dbDefinition.getName(),
dbDefinition));
DB_MAP = Collections.unmodifiableMap(map);
}
diff --git
a/hadoop-ozone/cli-debug/src/test/java/org/apache/hadoop/ozone/debug/TestDBDefinitionFactory.java
b/hadoop-ozone/cli-debug/src/test/java/org/apache/hadoop/ozone/debug/TestDBDefinitionFactory.java
index d12ea50123c..a4537024a23 100644
---
a/hadoop-ozone/cli-debug/src/test/java/org/apache/hadoop/ozone/debug/TestDBDefinitionFactory.java
+++
b/hadoop-ozone/cli-debug/src/test/java/org/apache/hadoop/ozone/debug/TestDBDefinitionFactory.java
@@ -19,19 +19,26 @@
import static
org.apache.hadoop.ozone.recon.ReconConstants.RECON_CONTAINER_KEY_DB;
import static
org.apache.hadoop.ozone.recon.ReconConstants.RECON_OM_SNAPSHOT_DB;
+import static
org.apache.hadoop.ozone.recon.ReconConstants.RECON_REPLICA_HISTORY_DB;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
import java.nio.file.Path;
import java.nio.file.Paths;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.scm.metadata.SCMDBDefinition;
+import org.apache.hadoop.hdds.utils.db.DBColumnFamilyDefinition;
import org.apache.hadoop.hdds.utils.db.DBDefinition;
import
org.apache.hadoop.ozone.container.metadata.DatanodeSchemaOneDBDefinition;
import
org.apache.hadoop.ozone.container.metadata.DatanodeSchemaThreeDBDefinition;
import
org.apache.hadoop.ozone.container.metadata.DatanodeSchemaTwoDBDefinition;
import org.apache.hadoop.ozone.om.codec.OMDBDefinition;
+import org.apache.hadoop.ozone.recon.scm.ContainerReplicaHistoryList;
import org.apache.hadoop.ozone.recon.scm.ReconSCMDBDefinition;
import org.apache.hadoop.ozone.recon.spi.impl.ReconDBDefinition;
+import org.apache.hadoop.ozone.recon.spi.impl.ReconReplicaHistoryDBDefinition;
import org.junit.jupiter.api.Test;
/**
@@ -39,6 +46,25 @@
*/
public class TestDBDefinitionFactory {
+ @Test
+ public void testGetReconReplicaHistoryDefinition() {
+ DBDefinition definition =
DBDefinitionFactory.getDefinition(RECON_REPLICA_HISTORY_DB);
+ assertInstanceOf(ReconReplicaHistoryDBDefinition.class, definition);
+ assertEquals(RECON_REPLICA_HISTORY_DB, definition.getName());
+ assertSame(definition, DBDefinitionFactory.getDefinition(
+ Paths.get("/tmp", RECON_REPLICA_HISTORY_DB), new
OzoneConfiguration()));
+
+ DBColumnFamilyDefinition<?, ?> history =
definition.getColumnFamily("replica_history_v2");
+ assertNotNull(history);
+ assertEquals(Long.class, history.getKeyType());
+ assertEquals(ContainerReplicaHistoryList.class, history.getValueType());
+
+ DBColumnFamilyDefinition<?, ?> metadata =
definition.getColumnFamily("metadata");
+ assertNotNull(metadata);
+ assertEquals(String.class, metadata.getKeyType());
+ assertEquals(String.class, metadata.getValueType());
+ }
+
@Test
public void testGetDefinition() {
DBDefinition definition =
DBDefinitionFactory.getDefinition(OMDBDefinition.get().getName());
diff --git a/hadoop-ozone/integration-test-recon/pom.xml
b/hadoop-ozone/integration-test-recon/pom.xml
index ca2fae43bf0..33e1b93493d 100644
--- a/hadoop-ozone/integration-test-recon/pom.xml
+++ b/hadoop-ozone/integration-test-recon/pom.xml
@@ -41,6 +41,11 @@
<artifactId>guava</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>com.google.inject</groupId>
+ <artifactId>guice</artifactId>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>
diff --git
a/hadoop-ozone/integration-test-recon/src/test/java/org/apache/hadoop/ozone/recon/TestReconWithOzoneManager.java
b/hadoop-ozone/integration-test-recon/src/test/java/org/apache/hadoop/ozone/recon/TestReconWithOzoneManager.java
index 8896f74c415..44e8c8f2296 100644
---
a/hadoop-ozone/integration-test-recon/src/test/java/org/apache/hadoop/ozone/recon/TestReconWithOzoneManager.java
+++
b/hadoop-ozone/integration-test-recon/src/test/java/org/apache/hadoop/ozone/recon/TestReconWithOzoneManager.java
@@ -19,7 +19,6 @@
import static java.net.HttpURLConnection.HTTP_CREATED;
import static java.net.HttpURLConnection.HTTP_OK;
-import static
org.apache.hadoop.hdds.recon.ReconConfigKeys.OZONE_RECON_HTTP_ADDRESS_KEY;
import static
org.apache.hadoop.ozone.recon.ReconServerConfigKeys.OZONE_RECON_OM_CONNECTION_REQUEST_TIMEOUT;
import static
org.apache.hadoop.ozone.recon.ReconServerConfigKeys.OZONE_RECON_OM_CONNECTION_REQUEST_TIMEOUT_DEFAULT;
import static
org.apache.hadoop.ozone.recon.ReconServerConfigKeys.OZONE_RECON_OM_CONNECTION_TIMEOUT;
@@ -29,14 +28,26 @@
import static
org.apache.hadoop.ozone.recon.ReconServerConfigKeys.RECON_OM_DELTA_UPDATE_LIMIT;
import static
org.apache.hadoop.ozone.recon.spi.impl.OzoneManagerServiceProviderImpl.OmSnapshotTaskName.OmDeltaRequest;
import static
org.apache.hadoop.ozone.recon.spi.impl.OzoneManagerServiceProviderImpl.OmSnapshotTaskName.OmSnapshotRequest;
+import static
org.apache.hadoop.ozone.recon.spi.impl.ReconDBDefinition.REPLICA_HISTORY_V2;
+import static
org.apache.hadoop.ozone.recon.spi.impl.ReconReplicaHistoryDBDefinition.METADATA;
+import static
org.apache.ozone.recon.codegen.JooqCodeGenerator.RECON_SCHEMA_NAME;
+import static
org.apache.ozone.recon.schema.SchemaVersionTableDefinition.SCHEMA_VERSION_TABLE_NAME;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.slf4j.event.Level.INFO;
import java.io.IOException;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -46,9 +57,13 @@
import org.apache.hadoop.hdds.client.BlockID;
import org.apache.hadoop.hdds.client.StandaloneReplicationConfig;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.protocol.DatanodeID;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
import org.apache.hadoop.hdds.scm.HddsTestUtils;
+import org.apache.hadoop.hdds.scm.container.ContainerChecksums;
import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
+import org.apache.hadoop.hdds.utils.db.DBStore;
+import org.apache.hadoop.hdds.utils.db.DBStoreBuilder;
import org.apache.hadoop.hdds.utils.db.RDBStore;
import org.apache.hadoop.ozone.MiniOzoneCluster;
import org.apache.hadoop.ozone.om.OMMetadataManager;
@@ -59,7 +74,13 @@
import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfoGroup;
import org.apache.hadoop.ozone.om.helpers.OmVolumeArgs;
import org.apache.hadoop.ozone.recon.metrics.OzoneManagerSyncMetrics;
+import org.apache.hadoop.ozone.recon.persistence.DataSourceConfiguration;
+import org.apache.hadoop.ozone.recon.scm.ContainerReplicaHistory;
+import org.apache.hadoop.ozone.recon.scm.ContainerReplicaHistoryList;
import org.apache.hadoop.ozone.recon.spi.impl.OzoneManagerServiceProviderImpl;
+import org.apache.hadoop.ozone.recon.spi.impl.ReconDBDefinition;
+import org.apache.hadoop.ozone.recon.spi.impl.ReconReplicaHistoryDBDefinition;
+import org.apache.hadoop.ozone.recon.upgrade.ReconVersion;
import org.apache.http.HttpEntity;
import org.apache.http.HttpResponse;
import org.apache.http.client.config.RequestConfig;
@@ -72,6 +93,7 @@
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
+import org.slf4j.LoggerFactory;
/**
* Test Ozone Recon.
@@ -82,6 +104,7 @@ public class TestReconWithOzoneManager {
private static MiniOzoneCluster cluster = null;
private static OMMetadataManager metadataManager;
private static CloseableHttpClient httpClient;
+ private static String reconHttpURL;
private static String taskStatusURL;
private static ReconService recon;
@@ -124,8 +147,9 @@ public static void init() throws Exception {
cluster.getStorageContainerManager().exitSafeMode();
- String reconHTTPAddress = conf.get(OZONE_RECON_HTTP_ADDRESS_KEY);
- taskStatusURL = "http://"; + reconHTTPAddress + "/api/v1/task/status";
+ // Keep requests on loopback when the HTTP server advertises the machine
hostname.
+ reconHttpURL = "http://127.0.0.1:"; +
recon.getReconServer().getHttpServer().getHttpAddress().getPort();
+ taskStatusURL = reconHttpURL + "/api/v1/task/status";
// initialize HTTPClient
httpClient = HttpClientBuilder
@@ -433,8 +457,7 @@ public void testManualOMDBRebuild() throws Exception {
assertThat(reconLoadedSeqNumber).isGreaterThanOrEqualTo(omLatestSeqNumber);
// 6. POST reinit
- String triggerUrl = "http://"; +
cluster.getConf().get(OZONE_RECON_HTTP_ADDRESS_KEY) +
- "/api/v1/triggerdbsync/om/reinit";
+ String triggerUrl = reconHttpURL + "/api/v1/triggerdbsync/om/reinit";
org.apache.http.client.methods.HttpPost httpPost = new
org.apache.http.client.methods.HttpPost(triggerUrl);
HttpResponse response = httpClient.execute(httpPost);
assertEquals(202, response.getStatusLine().getStatusCode());
@@ -481,6 +504,98 @@ public void testManualOMDBRebuild() throws Exception {
}, 1000, 30000);
}
+ @Test
+ public void testReplicaHistoryMigrationOnUpgradeAndRestart() throws
Exception {
+ OzoneConfiguration conf = cluster.getConf();
+ String jdbcUrl = ReconGuiceServletContextListener.getGlobalInjector()
+ .getInstance(DataSourceConfiguration.class).getJdbcUrl();
+ long containerID = 987654321L;
+ Map<DatanodeID, ContainerReplicaHistory> expected = new HashMap<>();
+ for (int i = 0; i < 3; i++) {
+ DatanodeID id = DatanodeID.randomID();
+ expected.put(id, new ContainerReplicaHistory(id, 100L + i, 200L + i,
300L + i,
+ "CLOSED", ContainerChecksums.of(1234L + i, 0L)));
+ }
+
+ try (LogCapturer logs =
LogCapturer.captureLogs(LoggerFactory.getLogger("org.apache.hadoop.ozone.recon")))
{
+ recon.stop();
+ // Seed unmigrated history while all Recon writers are stopped,
retaining other test data.
+ String migrationMessage;
+ try (DBStore legacy = DBStoreBuilder.createDBStore(conf,
+ new ReconDBDefinition(ReconConstants.RECON_CONTAINER_KEY_DB));
+ DBStore history = DBStoreBuilder.createDBStore(conf, new
ReconReplicaHistoryDBDefinition())) {
+ migrationMessage = "Migrated container replica history to " +
history.getDbLocation();
+ REPLICA_HISTORY_V2.getTable(legacy).put(containerID,
+ new ContainerReplicaHistoryList(new
ArrayList<>(expected.values())));
+ assertNull(REPLICA_HISTORY_V2.getTable(history).get(containerID));
+ METADATA.getTable(history).delete("replicaHistoryMigrated");
+ legacy.flushLog(true);
+ history.flushLog(true);
+ }
+ try (Connection connection = DriverManager.getConnection(jdbcUrl,
RECON_SCHEMA_NAME, "");
+ PreparedStatement statement = connection.prepareStatement(
+ "UPDATE " + SCHEMA_VERSION_TABLE_NAME + " SET
\"version_number\" = ?")) {
+ statement.setInt(1,
ReconVersion.UNHEALTHY_CONTAINERS_STATE_CONTAINER_ID_INDEX.serialize());
+ assertEquals(1, statement.executeUpdate());
+ }
+
assertEquals(ReconVersion.UNHEALTHY_CONTAINERS_STATE_CONTAINER_ID_INDEX.serialize(),
+ getReconSchemaVersion(jdbcUrl));
+
+ logs.clearOutput();
+ recon.start(conf);
+ assertEquals(ReconVersion.SOFTWARE_VERSION.serialize(),
getReconSchemaVersion(jdbcUrl));
+ assertReplicaHistory(containerID, expected);
+ String startupLogs = logs.getOutput();
+ int migration = startupLogs.indexOf(migrationMessage);
+ int versionPersisted = startupLogs.indexOf(
+ "Updated apparent version to '" +
ReconVersion.REPLICA_HISTORY_DB.serialize() + "'.");
+ assertThat(migration).isGreaterThanOrEqualTo(0);
+ assertThat(versionPersisted).isGreaterThan(migration);
+
+ recon.stop();
+ // A second copy would overwrite the migrated record, making repeated
migration observable.
+ try (DBStore legacy = DBStoreBuilder.createDBStore(conf,
+ new ReconDBDefinition(ReconConstants.RECON_CONTAINER_KEY_DB));
+ DBStore history = DBStoreBuilder.createDBStore(conf, new
ReconReplicaHistoryDBDefinition())) {
+ REPLICA_HISTORY_V2.getTable(legacy).put(containerID, new
ContainerReplicaHistoryList(Collections.emptyList()));
+ assertEquals("true",
METADATA.getTable(history).get("replicaHistoryMigrated"));
+ }
+ logs.clearOutput();
+ recon.start(conf);
+ assertEquals(ReconVersion.SOFTWARE_VERSION.serialize(),
getReconSchemaVersion(jdbcUrl));
+ assertReplicaHistory(containerID, expected);
+ assertThat(logs.getOutput()).doesNotContain(migrationMessage);
+ } finally {
+ if (recon.getReconServer() == null) {
+ recon.start(conf);
+ }
+ }
+ }
+
+ private int getReconSchemaVersion(String jdbcUrl) throws SQLException {
+ try (Connection connection = DriverManager.getConnection(jdbcUrl,
RECON_SCHEMA_NAME, "");
+ PreparedStatement statement = connection.prepareStatement(
+ "SELECT \"version_number\" FROM " + SCHEMA_VERSION_TABLE_NAME);
+ ResultSet result = statement.executeQuery()) {
+ assertTrue(result.next());
+ return result.getInt(1);
+ }
+ }
+
+ private void assertReplicaHistory(long containerID, Map<DatanodeID,
ContainerReplicaHistory> expected)
+ throws IOException {
+ Map<DatanodeID, ContainerReplicaHistory> actual =
+
recon.getReconServer().getReconContainerMetadataManager().getContainerReplicaHistory(containerID);
+ assertEquals(expected.size(), actual.size());
+ expected.forEach((id, history) -> {
+ assertNotNull(actual.get(id));
+ assertEquals(history.toProto(), actual.get(id).toProto());
+ });
+ List<Map<String, Object>> response =
JsonTestUtils.readTreeAsListOfMaps(makeHttpCall(
+ reconHttpURL + "/api/v1/containers/" + containerID +
"/replicaHistory"));
+ assertEquals(expected.size(), response.size());
+ }
+
private static void refreshOmMetadataManager() {
metadataManager = cluster.getOzoneManager().getMetadataManager();
}
diff --git
a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/ReconConstants.java
b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/ReconConstants.java
index 0320a8c8b2a..c96723c1011 100644
---
a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/ReconConstants.java
+++
b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/ReconConstants.java
@@ -26,6 +26,8 @@ public final class ReconConstants {
public static final String RECON_CONTAINER_KEY_DB = "recon-container-key.db";
+ public static final String RECON_REPLICA_HISTORY_DB =
"recon-container-replica-history.db";
+
public static final String CONTAINER_COUNT_KEY = "containerCount";
public static final String RECON_OM_SNAPSHOT_DB = "om.snapshot.db";
diff --git
a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/ReconControllerModule.java
b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/ReconControllerModule.java
index cdfdf416561..f5d9186306d 100644
---
a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/ReconControllerModule.java
+++
b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/ReconControllerModule.java
@@ -64,6 +64,7 @@
import org.apache.hadoop.ozone.recon.spi.impl.ReconFileMetadataManagerImpl;
import org.apache.hadoop.ozone.recon.spi.impl.ReconGlobalStatsManagerImpl;
import org.apache.hadoop.ozone.recon.spi.impl.ReconNamespaceSummaryManagerImpl;
+import org.apache.hadoop.ozone.recon.spi.impl.ReconReplicaHistoryDBProvider;
import
org.apache.hadoop.ozone.recon.spi.impl.StorageContainerServiceProviderImpl;
import org.apache.hadoop.ozone.recon.tasks.ContainerKeyMapperTaskFSO;
import org.apache.hadoop.ozone.recon.tasks.ContainerKeyMapperTaskOBS;
@@ -108,6 +109,7 @@ protected void configure() {
bind(ReconHttpServer.class).in(Singleton.class);
bind(ReconStorageConfig.class).in(Singleton.class);
bind(ReconDBProvider.class).in(Singleton.class);
+ bind(ReconReplicaHistoryDBProvider.class).in(Singleton.class);
bind(ReconOMMetadataManager.class)
.to(ReconOmMetadataManagerImpl.class);
bind(OMMetadataManager.class).to(ReconOmMetadataManagerImpl.class);
diff --git
a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/ReconServer.java
b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/ReconServer.java
index 28b0e60609a..deb9791b152 100644
---
a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/ReconServer.java
+++
b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/ReconServer.java
@@ -58,6 +58,7 @@
import org.apache.hadoop.ozone.recon.spi.ReconNamespaceSummaryManager;
import org.apache.hadoop.ozone.recon.spi.StorageContainerServiceProvider;
import org.apache.hadoop.ozone.recon.spi.impl.ReconDBProvider;
+import org.apache.hadoop.ozone.recon.spi.impl.ReconReplicaHistoryDBProvider;
import org.apache.hadoop.ozone.recon.tasks.ReconTaskController;
import org.apache.hadoop.ozone.recon.upgrade.ReconVersionManager;
import org.apache.hadoop.ozone.upgrade.UpgradeException;
@@ -85,6 +86,7 @@ public class ReconServer extends GenericCli implements
Callable<Void> {
private ReconContainerMetadataManager reconContainerMetadataManager;
private OzoneManagerServiceProvider ozoneManagerServiceProvider;
private ReconDBProvider reconDBProvider;
+ private ReconReplicaHistoryDBProvider replicaHistoryDBProvider;
private ReconNamespaceSummaryManager reconNamespaceSummaryManager;
private ReconStorageContainerManagerFacade reconStorageContainerManager;
private OzoneConfiguration configuration;
@@ -153,6 +155,7 @@ public Void call() throws Exception {
}
jvmPauseMonitor = newJvmPauseMonitor("Recon");
this.reconDBProvider = injector.getInstance(ReconDBProvider.class);
+ this.replicaHistoryDBProvider =
injector.getInstance(ReconReplicaHistoryDBProvider.class);
this.reconContainerMetadataManager =
injector.getInstance(ReconContainerMetadataManager.class);
this.reconNamespaceSummaryManager =
@@ -183,6 +186,9 @@ public Void call() throws Exception {
finalizeUpgrade();
+ // Fresh SQL installations are initialized at the software version and
do not run upgrade actions.
+
replicaHistoryDBProvider.migrateReplicaHistory(reconDBProvider.getDbStore());
+
LOG.debug("Now starting all services of Recon...");
// Start all services
start();
@@ -372,6 +378,13 @@ public void stop() {
}
}
}
+ if (replicaHistoryDBProvider != null) {
+ try {
+ replicaHistoryDBProvider.close();
+ } catch (IOException ex) {
+ LOG.error("Recon replica history DB close failed", ex);
+ }
+ }
if (certClient != null) {
try {
certClient.close();
diff --git
a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/scm/ReconContainerManager.java
b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/scm/ReconContainerManager.java
index 230ed8a5f7d..85bcad55da0 100644
---
a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/scm/ReconContainerManager.java
+++
b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/scm/ReconContainerManager.java
@@ -375,9 +375,9 @@ public void removeContainerReplica(ContainerID containerID,
final ContainerReplicaHistory ts = replicaLastSeenMap.get(id);
if (ts != null) {
// Flush to DB, then remove from in-memory map
- upsertContainerHistory(cid, id, ts.getLastSeenTime(), ts.getBcsId(),
- state, ts.getChecksums());
- replicaLastSeenMap.remove(id);
+ if (upsertContainerHistory(cid, id, ts.getLastSeenTime(),
ts.getBcsId(), state, ts.getChecksums())) {
+ replicaLastSeenMap.remove(id);
+ }
}
}
}
@@ -462,18 +462,20 @@ public void flushReplicaHistoryMapToDB(boolean clearMap) {
synchronized (replicaHistoryMap) {
try {
cdbServiceProvider.batchStoreContainerReplicaHistory(replicaHistoryMap);
+ if (clearMap) {
+ replicaHistoryMap.clear();
+ }
} catch (IOException e) {
- LOG.debug("Error flushing container replica history to DB. {}",
- e.getMessage());
- }
- if (clearMap) {
- replicaHistoryMap.clear();
+ LOG.warn("Error flushing container replica history to DB.", e);
}
}
}
- public void upsertContainerHistory(long containerID, DatanodeID id, long
time,
- long bcsId, String state,
ContainerChecksums checksums) {
+ /**
+ * Persist replica history and report whether it is safe to discard the
in-memory entry.
+ */
+ public boolean upsertContainerHistory(long containerID, DatanodeID id, long
time,
+ long bcsId, String state,
ContainerChecksums checksums) {
Map<DatanodeID, ContainerReplicaHistory> tsMap;
try {
tsMap = cdbServiceProvider.getContainerReplicaHistory(containerID);
@@ -489,8 +491,10 @@ public void upsertContainerHistory(long containerID,
DatanodeID id, long time,
ts.setChecksums(checksums);
}
cdbServiceProvider.storeContainerReplicaHistory(containerID, tsMap);
+ return true;
} catch (IOException e) {
- LOG.debug("Error on DB operations. {}", e.getMessage());
+ LOG.warn("Unable to persist replica history for container {} on datanode
{}", containerID, id, e);
+ return false;
}
}
diff --git
a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/ReconContainerMetadataManagerImpl.java
b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/ReconContainerMetadataManagerImpl.java
index 0e7510d5777..610541f8c9c 100644
---
a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/ReconContainerMetadataManagerImpl.java
+++
b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/ReconContainerMetadataManagerImpl.java
@@ -21,7 +21,6 @@
import static
org.apache.hadoop.ozone.recon.spi.impl.ReconDBDefinition.CONTAINER_KEY;
import static
org.apache.hadoop.ozone.recon.spi.impl.ReconDBDefinition.CONTAINER_KEY_COUNT;
import static
org.apache.hadoop.ozone.recon.spi.impl.ReconDBDefinition.KEY_CONTAINER;
-import static
org.apache.hadoop.ozone.recon.spi.impl.ReconDBDefinition.REPLICA_HISTORY_V2;
import jakarta.annotation.Nonnull;
import java.io.IOException;
@@ -76,11 +75,12 @@ public class ReconContainerMetadataManagerImpl
private Table<ContainerKeyPrefix, Integer> containerKeyTable;
private Table<KeyPrefixContainer, Integer> keyContainerTable;
private Table<Long, Long> containerKeyCountTable;
- private Table<Long, ContainerReplicaHistoryList>
+ private final Table<Long, ContainerReplicaHistoryList>
containerReplicaHistoryTable;
private GlobalStatsDao globalStatsDao;
private DBStore containerDbStore;
+ private final ReconReplicaHistoryDBProvider replicaHistoryDBProvider;
private Configuration sqlConfiguration;
@@ -88,13 +88,17 @@ public class ReconContainerMetadataManagerImpl
@Inject
public ReconContainerMetadataManagerImpl(
- ReconDBProvider reconDBProvider, Configuration sqlConfiguration,
ReconOMMetadataManager omMetadataManager) {
- this(reconDBProvider.getDbStore(), sqlConfiguration, omMetadataManager);
+ ReconDBProvider reconDBProvider, ReconReplicaHistoryDBProvider
replicaHistoryDBProvider,
+ Configuration sqlConfiguration, ReconOMMetadataManager
omMetadataManager) {
+ this(reconDBProvider.getDbStore(), replicaHistoryDBProvider,
sqlConfiguration, omMetadataManager);
}
private ReconContainerMetadataManagerImpl(
- DBStore reconDBStore, Configuration sqlConfiguration,
ReconOMMetadataManager omMetadataManager) {
+ DBStore reconDBStore, ReconReplicaHistoryDBProvider
replicaHistoryDBProvider,
+ Configuration sqlConfiguration, ReconOMMetadataManager
omMetadataManager) {
containerDbStore = reconDBStore;
+ this.replicaHistoryDBProvider = replicaHistoryDBProvider;
+ containerReplicaHistoryTable =
replicaHistoryDBProvider.getReplicaHistoryTable();
globalStatsDao = new GlobalStatsDao(sqlConfiguration);
this.sqlConfiguration = sqlConfiguration;
this.omMetadataManager = omMetadataManager;
@@ -104,7 +108,8 @@ private ReconContainerMetadataManagerImpl(
@Override
public ReconContainerMetadataManager getStagedReconContainerMetadataManager(
DBStore stagedReconDbStore) {
- return new ReconContainerMetadataManagerImpl(stagedReconDbStore,
sqlConfiguration, omMetadataManager);
+ return new ReconContainerMetadataManagerImpl(
+ stagedReconDbStore, replicaHistoryDBProvider, sqlConfiguration,
omMetadataManager);
}
@Override
@@ -167,8 +172,6 @@ private void initializeTables() {
}
this.containerKeyCountTable =
CONTAINER_KEY_COUNT.getTable(containerDbStore);
- this.containerReplicaHistoryTable =
- REPLICA_HISTORY_V2.getTable(containerDbStore);
} catch (IOException e) {
LOG.error("Unable to create Container Key tables.", e);
}
@@ -241,11 +244,18 @@ public void batchStoreContainerReplicaHistory(
Map<Long, Map<DatanodeID, ContainerReplicaHistory>> replicaHistoryMap)
throws IOException {
try (BatchOperation batchOperation =
- containerDbStore.initBatchOperation()) {
+ replicaHistoryDBProvider.getDbStore().initBatchOperation()) {
for (Map.Entry<Long, Map<DatanodeID, ContainerReplicaHistory>> entry :
replicaHistoryMap.entrySet()) {
final long containerId = entry.getKey();
- final Map<DatanodeID, ContainerReplicaHistory> tsMap =
entry.getValue();
+ // The in-memory map contains only current replicas; preserve departed
datanodes on shutdown.
+ final Map<DatanodeID, ContainerReplicaHistory> tsMap =
getContainerReplicaHistory(containerId);
+ entry.getValue().forEach((id, inMemory) -> tsMap.merge(id, inMemory,
(stored, current) -> {
+ // Prefer the in-memory record on equal timestamps, as before.
+ ContainerReplicaHistory latest = stored.getLastSeenTime() >
current.getLastSeenTime() ? stored : current;
+ return new ContainerReplicaHistory(id,
Math.min(stored.getFirstSeenTime(), current.getFirstSeenTime()),
+ latest.getLastSeenTime(), latest.getBcsId(), latest.getState(),
latest.getChecksums());
+ }));
List<ContainerReplicaHistory> tsList = new ArrayList<>();
for (Map.Entry<DatanodeID, ContainerReplicaHistory> e :
tsMap.entrySet()) {
@@ -256,7 +266,7 @@ public void batchStoreContainerReplicaHistory(
new ContainerReplicaHistoryList(tsList));
}
- containerDbStore.commitBatchOperation(batchOperation);
+
replicaHistoryDBProvider.getDbStore().commitBatchOperation(batchOperation);
}
}
diff --git
a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/ReconDBDefinition.java
b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/ReconDBDefinition.java
index 3fe489da79d..92482a71db9 100644
---
a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/ReconDBDefinition.java
+++
b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/ReconDBDefinition.java
@@ -73,7 +73,7 @@ public class ReconDBDefinition extends DBDefinition.WithMap {
LongCodec.get(),
NSSummaryCodec.get());
- // Container Replica History with bcsId tracking.
+ // Retained in the legacy DB for migration to
ReconReplicaHistoryDBDefinition.
public static final DBColumnFamilyDefinition
<Long, ContainerReplicaHistoryList> REPLICA_HISTORY_V2 =
new DBColumnFamilyDefinition<Long, ContainerReplicaHistoryList>(
diff --git
a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/ReconReplicaHistoryDBDefinition.java
b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/ReconReplicaHistoryDBDefinition.java
new file mode 100644
index 00000000000..f8c0f36e7be
--- /dev/null
+++
b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/ReconReplicaHistoryDBDefinition.java
@@ -0,0 +1,50 @@
+/*
+ * 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.spi.impl;
+
+import static
org.apache.hadoop.ozone.recon.ReconConstants.RECON_REPLICA_HISTORY_DB;
+import static
org.apache.hadoop.ozone.recon.spi.impl.ReconDBDefinition.REPLICA_HISTORY_V2;
+
+import org.apache.hadoop.hdds.utils.db.DBColumnFamilyDefinition;
+import org.apache.hadoop.hdds.utils.db.DBDefinition;
+import org.apache.hadoop.hdds.utils.db.StringCodec;
+import org.apache.hadoop.ozone.recon.ReconServerConfigKeys;
+
+/**
+ * Replica history is independent of the OM-derived tables rebuilt during
reprocess.
+ */
+public class ReconReplicaHistoryDBDefinition extends DBDefinition.WithMap {
+ // Stores the migration completion marker in the destination DB so a retry
can skip copying history
+ // if the copy succeeded but persisting Recon's upgrade version failed.
+ public static final DBColumnFamilyDefinition<String, String> METADATA =
+ new DBColumnFamilyDefinition<>("metadata", StringCodec.get(),
StringCodec.get());
+
+ public ReconReplicaHistoryDBDefinition() {
+ super(DBColumnFamilyDefinition.newUnmodifiableMap(REPLICA_HISTORY_V2,
METADATA));
+ }
+
+ @Override
+ public String getName() {
+ return RECON_REPLICA_HISTORY_DB;
+ }
+
+ @Override
+ public String getLocationConfigKey() {
+ return ReconServerConfigKeys.OZONE_RECON_DB_DIR;
+ }
+}
diff --git
a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/ReconReplicaHistoryDBProvider.java
b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/ReconReplicaHistoryDBProvider.java
new file mode 100644
index 00000000000..a6328d00011
--- /dev/null
+++
b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/ReconReplicaHistoryDBProvider.java
@@ -0,0 +1,98 @@
+/*
+ * 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.spi.impl;
+
+import static
org.apache.hadoop.ozone.recon.spi.impl.ReconDBDefinition.REPLICA_HISTORY_V2;
+import static
org.apache.hadoop.ozone.recon.spi.impl.ReconReplicaHistoryDBDefinition.METADATA;
+
+import java.io.Closeable;
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import javax.inject.Inject;
+import javax.inject.Singleton;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.utils.db.DBStore;
+import org.apache.hadoop.hdds.utils.db.DBStoreBuilder;
+import org.apache.hadoop.hdds.utils.db.Table;
+import org.apache.hadoop.ozone.recon.scm.ContainerReplicaHistoryList;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Owns the replica history DB, which is never replaced by an OM or SCM
snapshot.
+ */
+@Singleton
+public class ReconReplicaHistoryDBProvider implements Closeable {
+ private static final Logger LOG =
LoggerFactory.getLogger(ReconReplicaHistoryDBProvider.class);
+ private static final String MIGRATION_COMPLETE = "replicaHistoryMigrated";
+ private final DBStore dbStore;
+ private final Table<Long, ContainerReplicaHistoryList> replicaHistoryTable;
+
+ @Inject
+ public ReconReplicaHistoryDBProvider(OzoneConfiguration configuration)
throws IOException {
+ dbStore = DBStoreBuilder.createDBStore(configuration, new
ReconReplicaHistoryDBDefinition());
+ try {
+ replicaHistoryTable = REPLICA_HISTORY_V2.getTable(dbStore);
+ } catch (IOException | RuntimeException e) {
+ try {
+ dbStore.close();
+ } catch (RuntimeException closeException) {
+ e.addSuppressed(closeException);
+ }
+ throw e;
+ }
+ }
+
+ /**
+ * Migrate legacy history before services start. Safe to retry if version
persistence fails after the copy.
+ */
+ public void migrateReplicaHistory(DBStore legacyStore) throws IOException {
+ Table<String, String> metadata = METADATA.getTable(dbStore);
+ File migrationFile = new File(dbStore.getDbLocation(),
"replica-history-migration.sst");
+ try {
+ if (metadata.isExist(MIGRATION_COMPLETE)) {
+ return;
+ }
+ // Retry incomplete migrations from the unchanged source before any
services start.
+ Files.deleteIfExists(migrationFile.toPath());
+
legacyStore.getTable(REPLICA_HISTORY_V2.getName()).dumpToFileWithPrefix(migrationFile,
null);
+
dbStore.getTable(REPLICA_HISTORY_V2.getName()).loadFromFile(migrationFile);
+ metadata.put(MIGRATION_COMPLETE, "true");
+ dbStore.flushLog(true);
+ LOG.info("Migrated container replica history to {}",
dbStore.getDbLocation());
+ } finally {
+ if (migrationFile.exists() && !migrationFile.delete()) {
+ LOG.warn("Unable to delete replica history migration file {}",
migrationFile);
+ }
+ }
+ }
+
+ public DBStore getDbStore() {
+ return dbStore;
+ }
+
+ public Table<Long, ContainerReplicaHistoryList> getReplicaHistoryTable() {
+ return replicaHistoryTable;
+ }
+
+ @Override
+ public void close() throws IOException {
+ dbStore.close();
+ }
+}
diff --git
a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/upgrade/ReconVersion.java
b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/upgrade/ReconVersion.java
index 0e79f2a2a86..f85d835e303 100644
---
a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/upgrade/ReconVersion.java
+++
b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/upgrade/ReconVersion.java
@@ -43,6 +43,7 @@ public enum ReconVersion implements ComponentVersion {
REPLICATED_SIZE_OF_FILES(4, "Adds replicatedSizeOfFiles to NSSummary"),
UNHEALTHY_CONTAINERS_STATE_CONTAINER_ID_INDEX(5,
"Adds idx_state_container_id index on UNHEALTHY_CONTAINERS for
upgrades"),
+ REPLICA_HISTORY_DB(6, "Stores container replica history in separate
recon-container-replica-history.db"),
UNKNOWN_VERSION(-1, "Used when a version cannot be deserialized to any
version recognized by this"
+ " component, which may indicate it came from a component in a newer
version");
diff --git
a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/upgrade/ReplicaHistoryDBUpgradeAction.java
b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/upgrade/ReplicaHistoryDBUpgradeAction.java
new file mode 100644
index 00000000000..bed07dfaa66
--- /dev/null
+++
b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/upgrade/ReplicaHistoryDBUpgradeAction.java
@@ -0,0 +1,40 @@
+/*
+ * 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 com.google.inject.Injector;
+import javax.sql.DataSource;
+import org.apache.hadoop.ozone.recon.ReconGuiceServletContextListener;
+import org.apache.hadoop.ozone.recon.spi.impl.ReconDBProvider;
+import org.apache.hadoop.ozone.recon.spi.impl.ReconReplicaHistoryDBProvider;
+
+/**
+ * Preserve legacy container replica history before finalizing the separate
history DB layout.
+ */
+@ReconUpgradeActionForVersion(version = ReconVersion.REPLICA_HISTORY_DB)
+public class ReplicaHistoryDBUpgradeAction implements ReconUpgradeAction {
+ @Override
+ public void execute(DataSource source) throws Exception {
+ Injector injector = ReconGuiceServletContextListener.getGlobalInjector();
+ if (injector == null) {
+ throw new IllegalStateException("Guice injector not initialized. Replica
history migration cannot proceed.");
+ }
+ injector.getInstance(ReconReplicaHistoryDBProvider.class)
+
.migrateReplicaHistory(injector.getInstance(ReconDBProvider.class).getDbStore());
+ }
+}
diff --git
a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/ReconTestInjector.java
b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/ReconTestInjector.java
index e93e4d71f52..c634ebeb388 100644
---
a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/ReconTestInjector.java
+++
b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/ReconTestInjector.java
@@ -51,6 +51,7 @@
import org.apache.hadoop.ozone.recon.spi.impl.ReconFileMetadataManagerImpl;
import org.apache.hadoop.ozone.recon.spi.impl.ReconGlobalStatsManagerImpl;
import org.apache.hadoop.ozone.recon.spi.impl.ReconNamespaceSummaryManagerImpl;
+import org.apache.hadoop.ozone.recon.spi.impl.ReconReplicaHistoryDBProvider;
import org.apache.ratis.util.Preconditions;
/**
@@ -187,6 +188,7 @@ protected void configure() {
.to(ReconNamespaceSummaryManagerImpl.class)
.in(Singleton.class);
bind(ReconDBProvider.class).in(Singleton.class);
+ bind(ReconReplicaHistoryDBProvider.class).in(Singleton.class);
}
for (Map.Entry<Class, Object> entry :
diff --git
a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/api/TestContainerEndpoint.java
b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/api/TestContainerEndpoint.java
index 505f22c7c8e..68485d05ab9 100644
---
a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/api/TestContainerEndpoint.java
+++
b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/api/TestContainerEndpoint.java
@@ -65,9 +65,11 @@
import org.apache.hadoop.hdds.protocol.DatanodeID;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor;
+import
org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerReplicaProto;
import org.apache.hadoop.hdds.scm.container.ContainerChecksums;
import org.apache.hadoop.hdds.scm.container.ContainerID;
import org.apache.hadoop.hdds.scm.container.ContainerInfo;
+import org.apache.hadoop.hdds.scm.container.ContainerReplica;
import org.apache.hadoop.hdds.scm.container.ContainerStateManager;
import
org.apache.hadoop.hdds.scm.container.common.helpers.ContainerWithPipeline;
import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
@@ -100,6 +102,7 @@
import org.apache.hadoop.ozone.recon.persistence.ContainerHealthSchemaManager;
import org.apache.hadoop.ozone.recon.persistence.ContainerHistory;
import org.apache.hadoop.ozone.recon.recovery.ReconOMMetadataManager;
+import org.apache.hadoop.ozone.recon.scm.ContainerReplicaHistory;
import org.apache.hadoop.ozone.recon.scm.ReconContainerManager;
import org.apache.hadoop.ozone.recon.scm.ReconPipelineManager;
import org.apache.hadoop.ozone.recon.scm.ReconStorageContainerManagerFacade;
@@ -107,6 +110,7 @@
import org.apache.hadoop.ozone.recon.spi.ReconNamespaceSummaryManager;
import org.apache.hadoop.ozone.recon.spi.StorageContainerServiceProvider;
import org.apache.hadoop.ozone.recon.spi.impl.OzoneManagerServiceProviderImpl;
+import org.apache.hadoop.ozone.recon.spi.impl.ReconDBProvider;
import
org.apache.hadoop.ozone.recon.spi.impl.StorageContainerServiceProviderImpl;
import org.apache.hadoop.ozone.recon.tasks.ContainerKeyMapperHelper;
import org.apache.hadoop.ozone.recon.tasks.ContainerKeyMapperTaskFSO;
@@ -138,6 +142,7 @@ public class TestContainerEndpoint {
private ContainerStateManager containerStateManager;
private ReconPipelineManager reconPipelineManager;
private ReconContainerMetadataManager reconContainerMetadataManager;
+ private ReconDBProvider reconDBProvider;
private ContainerEndpoint containerEndpoint;
private boolean isSetupDone = false;
private ContainerHealthSchemaManager containerHealthSchemaManager;
@@ -208,6 +213,7 @@ private void initializeInjector() throws Exception {
ozoneStorageContainerManager.getPipelineManager();
reconContainerMetadataManager =
reconTestInjector.getInstance(ReconContainerMetadataManager.class);
+ reconDBProvider = reconTestInjector.getInstance(ReconDBProvider.class);
containerEndpoint = reconTestInjector.getInstance(ContainerEndpoint.class);
containerHealthSchemaManager =
reconTestInjector.getInstance(ContainerHealthSchemaManager.class);
@@ -1069,6 +1075,54 @@ public void testUnhealthyContainersPaging()
assertEquals(6L, records.get(2).getContainerID());
}
+ @Test
+ public void testReplicaHistorySurvivesOmReprocess() throws Exception {
+
reconContainerManager.addNewContainer(getTestContainer(HddsProtos.LifeCycleState.CLOSED,
1L));
+ List<ContainerReplica> replicas = new ArrayList<>();
+ for (int i = 1; i <= 4; i++) {
+ DatanodeID id = newDatanode("host" + i, "127.0.0." + i);
+ ContainerReplica replica = ContainerReplica.newBuilder()
+ .setContainerID(containerID)
+ .setContainerState(ContainerReplicaProto.State.CLOSED)
+ .setDatanodeDetails(reconContainerManager.getNodeDB().get(id))
+ .setSequenceId(100L)
+ .setChecksums(ContainerChecksums.of(1234L, 0L))
+ .build();
+ reconContainerManager.updateContainerReplica(containerID, replica);
+ replicas.add(replica);
+ }
+ reconContainerManager.removeContainerReplica(containerID, replicas.get(3));
+ assertEquals(3,
reconContainerManager.getReplicaHistoryMap().get(1L).size());
+ Map<DatanodeID, ContainerReplicaHistory> expected =
+ reconContainerMetadataManager.getContainerReplicaHistory(1L);
+ assertEquals(4, expected.size());
+ long expectedKeyCount =
reconContainerMetadataManager.getKeyCountForContainer(1L);
+
+ for (int i = 0; i < 2; i++) {
+ ReconConstants.resetTableTruncatedFlags();
+ ReconDBProvider stagedProvider =
reconDBProvider.getStagedReconDBProvider();
+ ReconContainerMetadataManager stagedManager =
+
reconContainerMetadataManager.getStagedReconContainerMetadataManager(stagedProvider.getDbStore());
+ assertTrue(new ContainerKeyMapperTaskOBS(stagedManager, omConfiguration)
+ .reprocess(reconOMMetadataManager).isTaskSuccess());
+ reconDBProvider.replaceStagedDb(stagedProvider);
+ reconContainerMetadataManager.reinitialize(reconDBProvider);
+ assertEquals(expectedKeyCount,
reconContainerMetadataManager.getKeyCountForContainer(1L));
+
+ List<ContainerHistory> histories = (List<ContainerHistory>)
+ containerEndpoint.getReplicaHistoryForContainer(1L).getEntity();
+ assertEquals(4, histories.size());
+ assertTrue(histories.stream().anyMatch(history ->
+
history.getDatanodeUuid().equals(replicas.get(3).getDatanodeDetails().getID().toString())));
+ Map<DatanodeID, ContainerReplicaHistory> actual =
reconContainerMetadataManager.getContainerReplicaHistory(1L);
+ expected.forEach((id, history) -> assertEquals(history.toProto(),
actual.get(id).toProto()));
+ }
+
+ // A shutdown flush must preserve the departed datanode after reprocess as
well.
+ reconContainerManager.flushReplicaHistoryMapToDB(true);
+ assertEquals(4, reconContainerManager.getAllContainerHistory(1L).size());
+ }
+
@Test
public void testGetReplicaHistoryForContainer() throws IOException {
// Add container history for container id 1
diff --git
a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/scm/AbstractReconContainerManagerTest.java
b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/scm/AbstractReconContainerManagerTest.java
index a395d4a3c6b..d02c9a821c4 100644
---
a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/scm/AbstractReconContainerManagerTest.java
+++
b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/scm/AbstractReconContainerManagerTest.java
@@ -73,6 +73,7 @@ public class AbstractReconContainerManagerTest {
private OzoneConfiguration conf;
private ReconPipelineManager pipelineManager;
private ReconContainerManager containerManager;
+ private ReconContainerMetadataManager reconContainerMetadataManager;
private DBStore store;
@BeforeEach
@@ -102,6 +103,7 @@ public void setUp(@TempDir File tempDir) throws Exception {
ContainerReplicaPendingOps pendingOps = new ContainerReplicaPendingOps(
Clock.system(ZoneId.systemDefault()), null);
+ reconContainerMetadataManager = mock(ReconContainerMetadataManager.class);
containerManager = new ReconContainerManager(
conf,
store,
@@ -109,7 +111,7 @@ public void setUp(@TempDir File tempDir) throws Exception {
pipelineManager,
getScmServiceProvider(),
mock(ContainerHealthSchemaManager.class),
- mock(ReconContainerMetadataManager.class),
+ reconContainerMetadataManager,
scmhaManager,
sequenceIdGen,
pendingOps);
@@ -133,6 +135,10 @@ protected ReconContainerManager getContainerManager() {
return containerManager;
}
+ protected ReconContainerMetadataManager getReconContainerMetadataManager() {
+ return reconContainerMetadataManager;
+ }
+
private StorageContainerServiceProvider getScmServiceProvider()
throws IOException, TimeoutException {
Pipeline pipeline = getRandomPipeline();
diff --git
a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/scm/TestReconContainerManager.java
b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/scm/TestReconContainerManager.java
index bd59c52a003..365fa3bcf19 100644
---
a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/scm/TestReconContainerManager.java
+++
b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/scm/TestReconContainerManager.java
@@ -25,8 +25,14 @@
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyMap;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
@@ -51,8 +57,11 @@
import
org.apache.hadoop.hdds.scm.container.common.helpers.ContainerWithPipeline;
import org.apache.hadoop.hdds.scm.ha.SCMHAManager;
import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
+import org.apache.hadoop.ozone.recon.spi.ReconContainerMetadataManager;
import org.apache.ozone.test.GenericTestUtils;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
/**
* Test Recon Container Manager.
@@ -384,6 +393,87 @@ public void testUpdateAndRemoveContainerReplica()
repHistMap.get(cIDlong1).keySet().iterator().next());
}
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testRemoveContainerReplicaRetainsHistoryOnPersistenceFailure(boolean
readFailure) throws Exception {
+ ContainerReplica replica = addContainerReplicaWithHistory();
+ ReconContainerManager manager = getContainerManager();
+ ContainerID containerID = replica.getContainerID();
+ DatanodeID id = replica.getDatanodeDetails().getID();
+ ContainerReplicaHistory history =
manager.getReplicaHistoryMap().get(containerID.getId()).get(id);
+ ReconContainerMetadataManager metadataManager =
getReconContainerMetadataManager();
+ if (readFailure) {
+ doThrow(new IOException("History read
failed")).when(metadataManager).getContainerReplicaHistory(anyLong());
+ } else {
+ doThrow(new IOException("History write failed"))
+ .when(metadataManager).storeContainerReplicaHistory(anyLong(),
anyMap());
+ }
+
+ manager.removeContainerReplica(containerID, replica);
+
+ assertTrue(manager.getContainerReplicas(containerID).isEmpty());
+ assertSame(history,
manager.getReplicaHistoryMap().get(containerID.getId()).get(id));
+ assertEquals(1,
manager.getAllContainerHistory(containerID.getId()).size());
+ doAnswer(invocation -> {
+ Map<Long, Map<DatanodeID, ContainerReplicaHistory>> pending =
invocation.getArgument(0);
+ assertSame(history, pending.get(containerID.getId()).get(id));
+ return null;
+ }).when(metadataManager).batchStoreContainerReplicaHistory(anyMap());
+ manager.flushReplicaHistoryMapToDB(true);
+ assertTrue(manager.getReplicaHistoryMap().isEmpty());
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testFailedShutdownFlushRetainsHistoryForRetry(boolean clearMap) throws
Exception {
+ ContainerReplica replica = addContainerReplicaWithHistory();
+ ReconContainerManager manager = getContainerManager();
+ long containerID = replica.getContainerID().getId();
+ DatanodeID id = replica.getDatanodeDetails().getID();
+ ContainerReplicaHistory history =
manager.getReplicaHistoryMap().get(containerID).get(id);
+ ReconContainerMetadataManager metadataManager =
getReconContainerMetadataManager();
+ doThrow(new IOException("History batch write failed"))
+ .when(metadataManager).batchStoreContainerReplicaHistory(anyMap());
+
+ manager.flushReplicaHistoryMapToDB(clearMap);
+
+ assertSame(history,
manager.getReplicaHistoryMap().get(containerID).get(id));
+
doNothing().when(metadataManager).batchStoreContainerReplicaHistory(anyMap());
+ manager.flushReplicaHistoryMapToDB(clearMap);
+ if (clearMap) {
+ assertTrue(manager.getReplicaHistoryMap().isEmpty());
+ } else {
+ assertSame(history,
manager.getReplicaHistoryMap().get(containerID).get(id));
+ }
+ }
+
+ @Test
+ void testUpsertContainerHistoryReportsPersistenceStatus() throws Exception {
+ ReconContainerManager manager = getContainerManager();
+ DatanodeID id = DatanodeID.randomID();
+ ContainerChecksums checksums = ContainerChecksums.of(1234L, 0L);
+ assertTrue(manager.upsertContainerHistory(1L, id, 10L, 20L, "OPEN",
checksums));
+ ReconContainerMetadataManager metadataManager =
getReconContainerMetadataManager();
+ doThrow(new IOException("History write failed")).doNothing()
+ .when(metadataManager).storeContainerReplicaHistory(anyLong(),
anyMap());
+ assertFalse(manager.upsertContainerHistory(1L, id, 10L, 20L, "OPEN",
checksums));
+ assertTrue(manager.upsertContainerHistory(1L, id, 10L, 20L, "OPEN",
checksums));
+ }
+
+ private ContainerReplica addContainerReplicaWithHistory() throws
IOException, TimeoutException {
+ ContainerWithPipeline container = getTestContainer(LifeCycleState.OPEN);
+ getContainerManager().addNewContainer(container);
+ ContainerReplica replica = ContainerReplica.newBuilder()
+ .setContainerID(container.getContainerInfo().containerID())
+ .setContainerState(State.OPEN)
+ .setDatanodeDetails(randomDatanodeDetails())
+ .setSequenceId(1001L)
+ .setChecksums(ContainerChecksums.of(1234L, 0L))
+ .build();
+ getContainerManager().updateContainerReplica(replica.getContainerID(),
replica);
+ return replica;
+ }
+
@Test
public void testAddNewContainerWithMissingPipeline()
throws IOException, TimeoutException {
diff --git
a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/scm/TestReconStorageContainerManagerFacade.java
b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/scm/TestReconStorageContainerManagerFacade.java
index 42fd25b36c7..b8bfe929a05 100644
---
a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/scm/TestReconStorageContainerManagerFacade.java
+++
b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/scm/TestReconStorageContainerManagerFacade.java
@@ -34,6 +34,8 @@
import java.util.ArrayList;
import java.util.List;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.protocol.DatanodeID;
+import org.apache.hadoop.hdds.scm.container.ContainerChecksums;
import org.apache.hadoop.hdds.scm.server.OzoneStorageContainerManager;
import org.apache.hadoop.hdds.utils.db.DBCheckpoint;
import org.apache.hadoop.hdds.utils.db.DBStore;
@@ -90,6 +92,10 @@ void testScmSnapshotDbIsOpenedAtCanonicalReconPath() throws
Exception {
ReconStorageContainerManagerFacade reconScm =
injector.getInstance(ReconStorageContainerManagerFacade.class);
+ ReconContainerManager containerManager = (ReconContainerManager)
reconScm.getContainerManager();
+ DatanodeID historyDatanode = DatanodeID.randomID();
+ containerManager.upsertContainerHistory(1L, historyDatanode, 10L, 20L,
+ "CLOSED", ContainerChecksums.of(1234L, 0L));
metricsToUnregister.add(getContainerSyncMetrics(reconScm));
OzoneConfiguration conf = injector.getInstance(OzoneConfiguration.class);
File checkpointDir =
@@ -113,6 +119,8 @@ void testScmSnapshotDbIsOpenedAtCanonicalReconPath() throws
Exception {
reconScm.getScmDBStore().getDbLocation().getCanonicalFile());
assertTrue(canonicalReconScmDb.exists());
assertFalse(checkpointDir.exists());
+ assertEquals(1, containerManager.getAllContainerHistory(1L).size());
+ assertEquals(historyDatanode.toString(),
containerManager.getAllContainerHistory(1L).get(0).getDatanodeUuid());
}
@Test
diff --git
a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/spi/impl/TestReconReplicaHistoryDBProvider.java
b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/spi/impl/TestReconReplicaHistoryDBProvider.java
new file mode 100644
index 00000000000..7bbe3e62b71
--- /dev/null
+++
b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/spi/impl/TestReconReplicaHistoryDBProvider.java
@@ -0,0 +1,262 @@
+/*
+ * 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.spi.impl;
+
+import static
org.apache.hadoop.ozone.recon.ReconServerConfigKeys.OZONE_RECON_DB_DIR;
+import static
org.apache.hadoop.ozone.recon.spi.impl.ReconDBDefinition.REPLICA_HISTORY_V2;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+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.anyString;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Map;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.protocol.DatanodeID;
+import org.apache.hadoop.hdds.scm.container.ContainerChecksums;
+import org.apache.hadoop.hdds.utils.db.DBStore;
+import org.apache.hadoop.hdds.utils.db.DBStoreBuilder;
+import org.apache.hadoop.hdds.utils.db.RDBBatchOperation;
+import org.apache.hadoop.hdds.utils.db.Table;
+import org.apache.hadoop.ozone.recon.ReconUtils;
+import org.apache.hadoop.ozone.recon.recovery.ReconOMMetadataManager;
+import org.apache.hadoop.ozone.recon.scm.ContainerReplicaHistory;
+import org.apache.hadoop.ozone.recon.scm.ContainerReplicaHistoryList;
+import org.apache.hadoop.ozone.recon.spi.ReconContainerMetadataManager;
+import org.jooq.impl.DefaultConfiguration;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+
+/**
+ * Tests startup migration and the independent lifetime of container replica
history.
+ */
+class TestReconReplicaHistoryDBProvider {
+ @TempDir
+ private Path temporaryFolder;
+
+ private OzoneConfiguration configuration;
+ private ReconDBProvider reconDBProvider;
+ private ReconReplicaHistoryDBProvider historyDBProvider;
+
+ @BeforeEach
+ void setUp() {
+ configuration = new OzoneConfiguration();
+ configuration.set(OZONE_RECON_DB_DIR, temporaryFolder.toString());
+ reconDBProvider = new ReconDBProvider(configuration, new ReconUtils());
+ }
+
+ @AfterEach
+ void tearDown() throws Exception {
+ if (historyDBProvider != null) {
+ historyDBProvider.close();
+ }
+ reconDBProvider.close();
+ }
+
+ @Test
+ void testMigratesLegacyHistoryWithoutChangingSource() throws Exception {
+ Table<Long, ContainerReplicaHistoryList> legacyTable =
REPLICA_HISTORY_V2.getTable(reconDBProvider.getDbStore());
+ ContainerReplicaHistoryList history = new
ContainerReplicaHistoryList(Arrays.asList(newHistory(), newHistory()));
+ for (long id = 1; id <= 2000; id++) {
+ legacyTable.put(id, history);
+ }
+
+ openHistoryDB();
+ for (long id = 1; id <= 2000; id++) {
+ assertEquals(history.toProto(),
historyDBProvider.getReplicaHistoryTable().get(id).toProto());
+ assertEquals(history.toProto(), legacyTable.get(id).toProto());
+ }
+ assertFalse(historyDBProvider.getDbStore().getTableNames()
+ .containsValue(ReconDBDefinition.CONTAINER_KEY.getName()));
+ }
+
+ @Test
+ void testFreshDatabaseAndRestartSkipMigration() throws Exception {
+ openHistoryDB();
+ assertTrue(historyDBProvider.getReplicaHistoryTable().isEmpty());
+ ContainerReplicaHistoryList history = new
ContainerReplicaHistoryList(Collections.singletonList(newHistory()));
+ historyDBProvider.getReplicaHistoryTable().put(1L, history);
+ historyDBProvider.close();
+
+ DBStore sourceStore = mock(DBStore.class);
+ historyDBProvider = new ReconReplicaHistoryDBProvider(configuration);
+ historyDBProvider.migrateReplicaHistory(sourceStore);
+ verifyNoInteractions(sourceStore);
+ assertEquals(history.toProto(),
historyDBProvider.getReplicaHistoryTable().get(1L).toProto());
+ }
+
+ @Test
+ void testRetriesPartiallyMigratedDatabase() throws Exception {
+ ContainerReplicaHistoryList history = new
ContainerReplicaHistoryList(Collections.singletonList(newHistory()));
+ Table<Long, ContainerReplicaHistoryList> source =
REPLICA_HISTORY_V2.getTable(reconDBProvider.getDbStore());
+ source.put(1L, history);
+ source.put(2L, history);
+ try (DBStore partialStore = DBStoreBuilder.createDBStore(configuration,
new ReconReplicaHistoryDBDefinition())) {
+ REPLICA_HISTORY_V2.getTable(partialStore).put(1L,
+ new
ContainerReplicaHistoryList(Collections.singletonList(newHistory())));
+ }
+
+ openHistoryDB();
+ assertEquals(history.toProto(),
historyDBProvider.getReplicaHistoryTable().get(1L).toProto());
+ assertEquals(history.toProto(),
historyDBProvider.getReplicaHistoryTable().get(2L).toProto());
+ }
+
+ @Test
+ void testFailedMigrationCanBeRetriedAfterRestart() throws Exception {
+ ContainerReplicaHistoryList history = new
ContainerReplicaHistoryList(Collections.singletonList(newHistory()));
+ REPLICA_HISTORY_V2.getTable(reconDBProvider.getDbStore()).put(1L, history);
+ Table<byte[], byte[]> sourceTable =
reconDBProvider.getDbStore().getTable(REPLICA_HISTORY_V2.getName());
+ Table<byte[], byte[]> failingTable = mock(Table.class);
+ doAnswer(invocation -> {
+ sourceTable.dumpToFileWithPrefix(invocation.getArgument(0), null);
+ throw new IOException("Interrupted migration");
+ }).when(failingTable).dumpToFileWithPrefix(any(), isNull());
+ DBStore failingSource = mock(DBStore.class);
+ when(failingSource.getTable(anyString())).thenReturn(failingTable);
+ historyDBProvider = new ReconReplicaHistoryDBProvider(configuration);
+ assertThrows(IOException.class, () ->
historyDBProvider.migrateReplicaHistory(failingSource));
+ historyDBProvider.close();
+ try (DBStore destination = DBStoreBuilder.createDBStore(configuration, new
ReconReplicaHistoryDBDefinition())) {
+
assertTrue(ReconReplicaHistoryDBDefinition.METADATA.getTable(destination).isEmpty());
+ assertTrue(REPLICA_HISTORY_V2.getTable(destination).isEmpty());
+ }
+ assertEquals(history.toProto(),
REPLICA_HISTORY_V2.getTable(reconDBProvider.getDbStore()).get(1L).toProto());
+
+ openHistoryDB();
+ assertEquals(history.toProto(),
historyDBProvider.getReplicaHistoryTable().get(1L).toProto());
+ }
+
+ @Test
+ void testHistorySurvivesRepeatedStagedDbReplacement() throws Exception {
+ openHistoryDB();
+ ReconContainerMetadataManager manager = newManager();
+ ContainerReplicaHistory history = newHistory();
+ manager.storeContainerReplicaHistory(1L,
Collections.singletonMap(history.getId(), history));
+ DBStore historyStore = historyDBProvider.getDbStore();
+ Table<Long, ContainerReplicaHistoryList> historyTable =
historyDBProvider.getReplicaHistoryTable();
+
+ for (int i = 0; i < 3; i++) {
+ ReconDBProvider stagedProvider =
reconDBProvider.getStagedReconDBProvider();
+ ReconContainerMetadataManager stagedManager =
manager.getStagedReconContainerMetadataManager(
+ stagedProvider.getDbStore());
+ try (RDBBatchOperation batch = RDBBatchOperation.newAtomicOperation()) {
+ stagedManager.batchStoreContainerKeyCounts(batch, 1L, (long) i + 1);
+ stagedManager.commitBatchOperation(batch);
+ }
+ // History writes continue while OM-derived data is being rebuilt in
staging.
+ ContainerReplicaHistory anotherHistory = newHistory();
+ Map<DatanodeID, ContainerReplicaHistory> histories =
manager.getContainerReplicaHistory(1L);
+ histories.put(anotherHistory.getId(), anotherHistory);
+ manager.storeContainerReplicaHistory(1L, histories);
+
assertTrue(REPLICA_HISTORY_V2.getTable(stagedProvider.getDbStore()).isEmpty());
+ reconDBProvider.replaceStagedDb(stagedProvider);
+ manager.reinitialize(reconDBProvider);
+
+ assertSame(historyStore, historyDBProvider.getDbStore());
+ assertSame(historyTable, historyDBProvider.getReplicaHistoryTable());
+ assertFalse(historyStore.isClosed());
+ assertEquals(i + 1, manager.getKeyCountForContainer(1L));
+ assertEquals(i + 2, manager.getContainerReplicaHistory(1L).size());
+ assertEquals(history.toProto(),
manager.getContainerReplicaHistory(1L).get(history.getId()).toProto());
+ }
+ historyDBProvider.close();
+ openHistoryDB();
+ assertEquals(4, newManager().getContainerReplicaHistory(1L).size());
+ }
+
+ @Test
+ void testShutdownBatchPreservesDepartedReplicas() throws Exception {
+ openHistoryDB();
+ ReconContainerMetadataManager manager = newManager();
+ ContainerReplicaHistory departed = newHistory();
+ ContainerReplicaHistory current = newHistory();
+ manager.storeContainerReplicaHistory(1L,
Collections.singletonMap(departed.getId(), departed));
+ manager.batchStoreContainerReplicaHistory(Collections.singletonMap(
+ 1L, Collections.singletonMap(current.getId(), current)));
+ historyDBProvider.close();
+ openHistoryDB();
+
+ Map<DatanodeID, ContainerReplicaHistory> history =
newManager().getContainerReplicaHistory(1L);
+ assertEquals(2, history.size());
+ assertEquals(departed.toProto(), history.get(departed.getId()).toProto());
+ assertEquals(current.toProto(), history.get(current.getId()).toProto());
+ assertNull(historyDBProvider.getReplicaHistoryTable().get(2L));
+ }
+
+ @ParameterizedTest
+ @CsvSource({
+ "10, 20, 30, 40, 10, 40, true",
+ "30, 40, 10, 20, 10, 40, false",
+ "10, 40, 30, 40, 10, 40, true"
+ })
+ void testShutdownBatchPreservesFirstSeenAndLatestReplicaInfo(long
storedFirstSeen, long storedLastSeen,
+ long currentFirstSeen, long currentLastSeen, long expectedFirstSeen,
long expectedLastSeen,
+ boolean currentFieldsExpected) throws Exception {
+ openHistoryDB();
+ ReconContainerMetadataManager manager = newManager();
+ DatanodeID id = DatanodeID.randomID();
+ ContainerReplicaHistory stored = new ContainerReplicaHistory(id,
storedFirstSeen, storedLastSeen, 100L,
+ "OPEN", ContainerChecksums.of(1111L, 0L));
+ ContainerReplicaHistory current = new ContainerReplicaHistory(id,
currentFirstSeen, currentLastSeen, 200L,
+ "CLOSED", ContainerChecksums.of(2222L, 0L));
+ manager.storeContainerReplicaHistory(1L, Collections.singletonMap(id,
stored));
+ manager.batchStoreContainerReplicaHistory(Collections.singletonMap(1L,
Collections.singletonMap(id, current)));
+ assertEquals(currentFirstSeen, current.getFirstSeenTime());
+ assertEquals(currentLastSeen, current.getLastSeenTime());
+ historyDBProvider.close();
+ openHistoryDB();
+
+ ContainerReplicaHistory expected = new ContainerReplicaHistory(id,
expectedFirstSeen, expectedLastSeen,
+ currentFieldsExpected ? 200L : 100L, currentFieldsExpected ? "CLOSED"
: "OPEN",
+ ContainerChecksums.of(currentFieldsExpected ? 2222L : 1111L, 0L));
+ Map<DatanodeID, ContainerReplicaHistory> history =
newManager().getContainerReplicaHistory(1L);
+ assertEquals(1, history.size());
+ assertEquals(expected.toProto(), history.get(id).toProto());
+ }
+
+ private void openHistoryDB() throws IOException {
+ historyDBProvider = new ReconReplicaHistoryDBProvider(configuration);
+ historyDBProvider.migrateReplicaHistory(reconDBProvider.getDbStore());
+ }
+
+ private ReconContainerMetadataManager newManager() {
+ return new ReconContainerMetadataManagerImpl(reconDBProvider,
historyDBProvider,
+ new DefaultConfiguration(), mock(ReconOMMetadataManager.class));
+ }
+
+ private static ContainerReplicaHistory newHistory() {
+ return new ContainerReplicaHistory(DatanodeID.randomID(), 10L, 20L, 30L,
+ "CLOSED", ContainerChecksums.of(1234L, 0L));
+ }
+}
diff --git
a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/upgrade/TestReconVersionManager.java
b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/upgrade/TestReconVersionManager.java
index 4017c09b24c..bd9e1715aa9 100644
---
a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/upgrade/TestReconVersionManager.java
+++
b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/upgrade/TestReconVersionManager.java
@@ -17,6 +17,8 @@
package org.apache.hadoop.ozone.recon.upgrade;
+import static
org.apache.hadoop.ozone.recon.ReconServerConfigKeys.OZONE_RECON_DB_DIR;
+import static
org.apache.hadoop.ozone.recon.spi.impl.ReconDBDefinition.REPLICA_HISTORY_V2;
import static
org.apache.ozone.recon.schema.SchemaVersionTableDefinition.SCHEMA_VERSION_TABLE_NAME;
import static org.jooq.impl.DSL.name;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -26,22 +28,29 @@
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.atLeastOnce;
+import static org.mockito.Mockito.clearInvocations;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
import com.google.inject.AbstractModule;
import com.google.inject.Guice;
import com.google.inject.Injector;
import com.google.inject.Provider;
+import com.google.inject.Singleton;
import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.sql.SQLException;
import java.util.Arrays;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -49,9 +58,18 @@
import java.util.stream.Stream;
import javax.sql.DataSource;
import org.apache.hadoop.hdds.ComponentVersion;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.protocol.DatanodeID;
+import org.apache.hadoop.hdds.scm.container.ContainerChecksums;
+import org.apache.hadoop.hdds.utils.db.DBStore;
+import org.apache.hadoop.ozone.recon.ReconGuiceServletContextListener;
import org.apache.hadoop.ozone.recon.persistence.AbstractReconSqlDBTest;
import org.apache.hadoop.ozone.recon.persistence.DataSourceConfiguration;
import org.apache.hadoop.ozone.recon.persistence.JooqPersistenceModule;
+import org.apache.hadoop.ozone.recon.scm.ContainerReplicaHistory;
+import org.apache.hadoop.ozone.recon.scm.ContainerReplicaHistoryList;
+import org.apache.hadoop.ozone.recon.spi.impl.ReconDBProvider;
+import org.apache.hadoop.ozone.recon.spi.impl.ReconReplicaHistoryDBProvider;
import org.apache.hadoop.ozone.upgrade.AbstractComponentVersionManagerTest;
import org.apache.hadoop.ozone.upgrade.ComponentUpgradeActionProvider;
import org.apache.hadoop.ozone.upgrade.ComponentVersionManager;
@@ -63,6 +81,7 @@
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.provider.Arguments;
+import org.mockito.MockedStatic;
/**
* Tests for {@link ReconVersionManager}.
@@ -150,6 +169,8 @@ public void testClasspathScanDiscoversUpgradeActions()
throws Exception {
ReconUpgradeAction taskStatusAction =
versionManager.getUpgradeActionsForTesting()
.get(ReconVersion.TASK_STATUS_STATISTICS);
assertInstanceOf(ReconTaskStatusTableUpgradeAction.class,
taskStatusAction);
+ assertInstanceOf(ReplicaHistoryDBUpgradeAction.class,
+
versionManager.getUpgradeActionsForTesting().get(ReconVersion.REPLICA_HISTORY_DB));
}
try (ReconVersionManager versionManager = createManager(
@@ -158,6 +179,8 @@ public void testClasspathScanDiscoversUpgradeActions()
throws Exception {
ReconUpgradeAction taskStatusAction =
versionManager.getUpgradeActionsForTesting()
.get(ReconVersion.TASK_STATUS_STATISTICS);
assertInstanceOf(ReconTaskStatusTableUpgradeAction.class,
taskStatusAction);
+ assertInstanceOf(ReplicaHistoryDBUpgradeAction.class,
+
versionManager.getUpgradeActionsForTesting().get(ReconVersion.REPLICA_HISTORY_DB));
}
}
@@ -188,6 +211,107 @@ public void testFinalizeRunsSuppliedUpgradeAction()
throws Exception {
}
}
+ @Test
+ void testReplicaHistoryMigrationBeforeVersionFinalization() throws Exception
{
+ Injector reconInjector = createReplicaHistoryInjector();
+ try (DBStore legacyStore =
reconInjector.getInstance(ReconDBProvider.class).getDbStore();
+ ReconReplicaHistoryDBProvider historyProvider =
reconInjector.getInstance(ReconReplicaHistoryDBProvider.class);
+ ReconVersionManager versionManager = createManager(
+
ReconVersion.UNHEALTHY_CONTAINERS_STATE_CONTAINER_ID_INDEX.serialize(), new
ReconUpgradeActionProvider());
+ MockedStatic<ReconGuiceServletContextListener> listener =
mockStatic(ReconGuiceServletContextListener.class)) {
+ ContainerReplicaHistoryList history = legacyReplicaHistory();
+ REPLICA_HISTORY_V2.getTable(legacyStore).put(1L, history);
+ assertTrue(historyProvider.getReplicaHistoryTable().isEmpty());
+
listener.when(ReconGuiceServletContextListener::getGlobalInjector).thenReturn(reconInjector);
+
+ versionManager.finalizeUpgrade();
+
+ assertEquals(history.toProto(),
historyProvider.getReplicaHistoryTable().get(1L).toProto());
+ assertEquals(history.toProto(),
REPLICA_HISTORY_V2.getTable(legacyStore).get(1L).toProto());
+ assertEquals(ReconVersion.REPLICA_HISTORY_DB,
versionManager.getApparentVersion());
+ assertEquals(ReconVersion.REPLICA_HISTORY_DB.serialize(),
versionManager.getPersistedApparentVersion());
+ }
+ }
+
+ @Test
+ void testReplicaHistoryMigrationFailureDoesNotAdvanceVersion() throws
Exception {
+ Injector reconInjector = mock(Injector.class);
+ ReconDBProvider sourceProvider = mock(ReconDBProvider.class);
+ when(sourceProvider.getDbStore()).thenReturn(mock(DBStore.class));
+ ReconReplicaHistoryDBProvider historyProvider =
mock(ReconReplicaHistoryDBProvider.class);
+
when(reconInjector.getInstance(ReconDBProvider.class)).thenReturn(sourceProvider);
+
when(reconInjector.getInstance(ReconReplicaHistoryDBProvider.class)).thenReturn(historyProvider);
+ doThrow(new IOException("Interrupted replica history
migration")).doNothing()
+ .when(historyProvider).migrateReplicaHistory(any());
+
+ try (ReconVersionManager versionManager = createManager(
+
ReconVersion.UNHEALTHY_CONTAINERS_STATE_CONTAINER_ID_INDEX.serialize(), new
ReconUpgradeActionProvider());
+ MockedStatic<ReconGuiceServletContextListener> listener =
mockStatic(ReconGuiceServletContextListener.class)) {
+
listener.when(ReconGuiceServletContextListener::getGlobalInjector).thenReturn(reconInjector);
+ UpgradeException thrown = assertThrows(UpgradeException.class,
versionManager::finalizeUpgrade);
+
assertEquals(UpgradeException.ResultCodes.FINALIZE_UPGRADE_ACTION_FAILED,
thrown.getResult());
+ assertEquals(ReconVersion.UNHEALTHY_CONTAINERS_STATE_CONTAINER_ID_INDEX,
versionManager.getApparentVersion());
+
assertEquals(ReconVersion.UNHEALTHY_CONTAINERS_STATE_CONTAINER_ID_INDEX.serialize(),
+ versionManager.getPersistedApparentVersion());
+
+ versionManager.finalizeUpgrade();
+ assertEquals(ReconVersion.REPLICA_HISTORY_DB.serialize(),
versionManager.getPersistedApparentVersion());
+ verify(historyProvider, times(2)).migrateReplicaHistory(any());
+ }
+ }
+
+ @Test
+ void testReplicaHistoryMigrationNotRepeatedAfterVersionPersistenceFailure()
throws Exception {
+ Injector reconInjector = createReplicaHistoryInjector();
+ ReconDBProvider sourceProvider =
spy(reconInjector.getInstance(ReconDBProvider.class));
+ DBStore sourceStore = spy(sourceProvider.getDbStore());
+ when(sourceProvider.getDbStore()).thenReturn(sourceStore);
+ Injector migrationInjector = mock(Injector.class);
+
when(migrationInjector.getInstance(ReconDBProvider.class)).thenReturn(sourceProvider);
+
+ try (DBStore legacyStore = sourceStore;
+ ReconReplicaHistoryDBProvider historyProvider =
reconInjector.getInstance(ReconReplicaHistoryDBProvider.class);
+ ReconVersionManager versionManager = spy(createManager(
+
ReconVersion.UNHEALTHY_CONTAINERS_STATE_CONTAINER_ID_INDEX.serialize(), new
ReconUpgradeActionProvider()));
+ MockedStatic<ReconGuiceServletContextListener> listener =
mockStatic(ReconGuiceServletContextListener.class)) {
+
when(migrationInjector.getInstance(ReconReplicaHistoryDBProvider.class)).thenReturn(historyProvider);
+
listener.when(ReconGuiceServletContextListener::getGlobalInjector).thenReturn(migrationInjector);
+ ContainerReplicaHistoryList history = legacyReplicaHistory();
+ REPLICA_HISTORY_V2.getTable(legacyStore).put(1L, history);
+ doThrow(new IOException("Version persistence failed")).doCallRealMethod()
+
.when(versionManager).persistApparentVersion(ReconVersion.REPLICA_HISTORY_DB);
+
+ UpgradeException thrown = assertThrows(UpgradeException.class,
versionManager::finalizeUpgrade);
+
assertEquals(UpgradeException.ResultCodes.APPARENT_VERSION_UPDATE_FAILED,
thrown.getResult());
+ assertEquals(history.toProto(),
historyProvider.getReplicaHistoryTable().get(1L).toProto());
+
assertEquals(ReconVersion.UNHEALTHY_CONTAINERS_STATE_CONTAINER_ID_INDEX.serialize(),
+ versionManager.getPersistedApparentVersion());
+
+ clearInvocations(sourceStore);
+ versionManager.finalizeUpgrade();
+ verifyNoInteractions(sourceStore);
+ assertEquals(ReconVersion.REPLICA_HISTORY_DB.serialize(),
versionManager.getPersistedApparentVersion());
+ }
+ }
+
+ private Injector createReplicaHistoryInjector() throws IOException {
+ OzoneConfiguration configuration = new OzoneConfiguration();
+ configuration.set(OZONE_RECON_DB_DIR,
Files.createDirectory(tempDir.resolve("replica-history")).toString());
+ return Guice.createInjector(new AbstractModule() {
+ @Override
+ protected void configure() {
+ bind(OzoneConfiguration.class).toInstance(configuration);
+ bind(ReconDBProvider.class).in(Singleton.class);
+ bind(ReconReplicaHistoryDBProvider.class).in(Singleton.class);
+ }
+ });
+ }
+
+ private static ContainerReplicaHistoryList legacyReplicaHistory() {
+ return new ContainerReplicaHistoryList(Collections.singletonList(new
ContainerReplicaHistory(
+ DatanodeID.randomID(), 10L, 20L, 30L, "CLOSED",
ContainerChecksums.of(1234L, 0L))));
+ }
+
@Override
@Test
public void testUpgradeActionFailureAbortsFinalize() throws Exception {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
