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 c31553bcded HDDS-15766. Make Recon OM DB large tarball transfer
reliable (#10696)
c31553bcded is described below
commit c31553bcdedbaf72d35de066b539f07dad1a8f64
Author: Devesh Kumar Singh <[email protected]>
AuthorDate: Tue Jul 21 10:00:59 2026 +0530
HDDS-15766. Make Recon OM DB large tarball transfer reliable (#10696)
Generated-by: Cursor (Opus 4.8)
Co-authored-by: Devesh Kumar Singh <[email protected]>
---
.../hadoop/hdds/utils/RDBSnapshotProvider.java | 2 +-
.../ozone/recon/TestReconWithOzoneManagerHA.java | 9 -
.../spi/impl/OzoneManagerServiceProviderImpl.java | 202 +++++++--------
.../recon/spi/impl/ReconRDBSnapshotProvider.java | 245 ++++++++++++++++++
.../ozone/recon/api/TestTriggerDBSyncEndpoint.java | 1 +
.../impl/TestOzoneManagerServiceProviderImpl.java | 281 +++++++++++----------
.../spi/impl/TestReconRDBSnapshotProvider.java | 223 ++++++++++++++++
7 files changed, 712 insertions(+), 251 deletions(-)
diff --git
a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/RDBSnapshotProvider.java
b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/RDBSnapshotProvider.java
index 1b00571f72e..8253f7bb112 100644
---
a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/RDBSnapshotProvider.java
+++
b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/RDBSnapshotProvider.java
@@ -126,7 +126,7 @@ public DBCheckpoint downloadDBSnapshotFromLeader(String
leaderNodeID)
LOG.info("Successfully untar the downloaded snapshot {} at {}.",
targetFile, unTarredDb.toAbsolutePath());
if (ratisSnapshotComplete(unTarredDb)) {
- LOG.info("Ratis snapshot transfer is complete.");
+ LOG.info("DB snapshot transfer is complete.");
return getCheckpointFromUntarredDb(unTarredDb);
}
}
diff --git
a/hadoop-ozone/integration-test-recon/src/test/java/org/apache/hadoop/ozone/recon/TestReconWithOzoneManagerHA.java
b/hadoop-ozone/integration-test-recon/src/test/java/org/apache/hadoop/ozone/recon/TestReconWithOzoneManagerHA.java
index 6533f482c8f..c2fca580fda 100644
---
a/hadoop-ozone/integration-test-recon/src/test/java/org/apache/hadoop/ozone/recon/TestReconWithOzoneManagerHA.java
+++
b/hadoop-ozone/integration-test-recon/src/test/java/org/apache/hadoop/ozone/recon/TestReconWithOzoneManagerHA.java
@@ -18,7 +18,6 @@
package org.apache.hadoop.ozone.recon;
import static java.nio.charset.StandardCharsets.UTF_8;
-import static
org.apache.hadoop.ozone.OzoneConsts.OZONE_DB_CHECKPOINT_HTTP_ENDPOINT;
import static
org.apache.hadoop.ozone.recon.ReconOmMetaManagerTestUtils.waitForEventBufferEmpty;
import static
org.apache.hadoop.ozone.recon.ReconOmMetaManagerTestUtils.waitUntilReconKeyCounts;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -120,14 +119,6 @@ public void testReconGetsSnapshotFromLeader() throws
Exception {
OzoneManagerServiceProviderImpl impl = (OzoneManagerServiceProviderImpl)
recon.getReconServer().getOzoneManagerServiceProvider();
- String hostname =
- ozoneManager.get().getHttpServer().getHttpAddress().getHostName();
- String expectedUrl = "http://"; +
- (hostname.equals("0.0.0.0") ? "localhost" : hostname) + ":" +
- ozoneManager.get().getHttpServer().getHttpAddress().getPort() +
- OZONE_DB_CHECKPOINT_HTTP_ENDPOINT;
- String snapshotUrl = impl.getOzoneManagerSnapshotUrl();
- assertEquals(expectedUrl, snapshotUrl);
// Write some data
String keyPrefix = "ratis";
OzoneOutputStream key = objectStore.getVolume(VOL_NAME)
diff --git
a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/OzoneManagerServiceProviderImpl.java
b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/OzoneManagerServiceProviderImpl.java
index f5d7e4e0acf..bec49088d27 100644
---
a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/OzoneManagerServiceProviderImpl.java
+++
b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/OzoneManagerServiceProviderImpl.java
@@ -18,8 +18,8 @@
package org.apache.hadoop.ozone.recon.spi.impl;
import static
org.apache.hadoop.hdds.recon.ReconConfigKeys.OZONE_RECON_DB_DIRS_PERMISSIONS_DEFAULT;
-import static
org.apache.hadoop.ozone.OzoneConsts.OZONE_DB_CHECKPOINT_HTTP_ENDPOINT;
-import static
org.apache.hadoop.ozone.OzoneConsts.OZONE_DB_CHECKPOINT_REQUEST_FLUSH;
+import static
org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_DB_CHECKPOINT_USE_INODE_BASED_DEFAULT;
+import static
org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_DB_CHECKPOINT_USE_INODE_BASED_KEY;
import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_HTTP_AUTH_TYPE;
import static
org.apache.hadoop.ozone.recon.ReconConstants.RECON_OM_SNAPSHOT_DB;
import static org.apache.hadoop.ozone.recon.ReconConstants.STAGING;
@@ -45,10 +45,8 @@
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import java.io.File;
import java.io.IOException;
-import java.io.InputStream;
+import java.io.UncheckedIOException;
import java.nio.file.Files;
-import java.nio.file.Path;
-import java.nio.file.Paths;
import java.nio.file.attribute.PosixFilePermission;
import java.nio.file.attribute.PosixFilePermissions;
import java.util.Arrays;
@@ -73,23 +71,20 @@
import org.apache.hadoop.hdds.utils.db.DBCheckpoint;
import org.apache.hadoop.hdds.utils.db.RDBBatchOperation;
import org.apache.hadoop.hdds.utils.db.RDBStore;
-import org.apache.hadoop.hdds.utils.db.RocksDBCheckpoint;
import org.apache.hadoop.hdds.utils.db.RocksDatabase;
import org.apache.hadoop.hdds.utils.db.Table;
import org.apache.hadoop.hdds.utils.db.TableIterator;
import org.apache.hadoop.hdds.utils.db.managed.ManagedWriteBatch;
import org.apache.hadoop.hdds.utils.db.managed.ManagedWriteOptions;
import org.apache.hadoop.hdfs.web.URLConnectionFactory;
-import org.apache.hadoop.ozone.om.OMConfigKeys;
import org.apache.hadoop.ozone.om.OMMetadataManager;
import org.apache.hadoop.ozone.om.helpers.DBUpdates;
+import org.apache.hadoop.ozone.om.helpers.ServiceInfo;
import org.apache.hadoop.ozone.om.protocol.OzoneManagerProtocol;
import
org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.DBUpdatesRequest;
-import
org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.ServicePort.Type;
import org.apache.hadoop.ozone.recon.ReconContext;
import org.apache.hadoop.ozone.recon.ReconServerConfigKeys;
import org.apache.hadoop.ozone.recon.ReconUtils;
-import org.apache.hadoop.ozone.recon.TarExtractor;
import org.apache.hadoop.ozone.recon.metrics.OzoneManagerSyncMetrics;
import org.apache.hadoop.ozone.recon.metrics.ReconSyncMetrics;
import org.apache.hadoop.ozone.recon.recovery.ReconOMMetadataManager;
@@ -101,7 +96,6 @@
import org.apache.hadoop.ozone.recon.tasks.ReconTaskReInitializationEvent;
import org.apache.hadoop.ozone.recon.tasks.updater.ReconTaskStatusUpdater;
import
org.apache.hadoop.ozone.recon.tasks.updater.ReconTaskStatusUpdaterManager;
-import org.apache.hadoop.security.SecurityUtil;
import org.apache.hadoop.util.Time;
import org.rocksdb.RocksDBException;
import org.slf4j.Logger;
@@ -118,9 +112,8 @@ public class OzoneManagerServiceProviderImpl
LoggerFactory.getLogger(OzoneManagerServiceProviderImpl.class);
private URLConnectionFactory connectionFactory;
- private File omSnapshotDBParentDir = null;
private File reconDbDir = null;
- private String omDBSnapshotUrl;
+ private File omSnapshotDBParentDir = null;
private OzoneManagerProtocol ozoneManagerClient;
private final OzoneConfiguration configuration;
@@ -140,7 +133,7 @@ public class OzoneManagerServiceProviderImpl
private ThreadFactory threadFactory;
private ReconContext reconContext;
private ReconTaskStatusUpdaterManager taskStatusUpdaterManager;
- private TarExtractor tarExtractor;
+ private ReconRDBSnapshotProvider reconSnapshotProvider;
/**
* OM Snapshot related task names.
@@ -179,12 +172,6 @@ public OzoneManagerServiceProviderImpl(
URLConnectionFactory.newDefaultURLConnectionFactory(connectionTimeout,
connectionRequestTimeout, configuration);
- String ozoneManagerHttpAddress = configuration.get(OMConfigKeys
- .OZONE_OM_HTTP_ADDRESS_KEY);
-
- String ozoneManagerHttpsAddress = configuration.get(OMConfigKeys
- .OZONE_OM_HTTPS_ADDRESS_KEY);
-
long deltaUpdateLimits = configuration.getLong(RECON_OM_DELTA_UPDATE_LIMIT,
RECON_OM_DELTA_UPDATE_LIMIT_DEFAULT);
@@ -195,24 +182,23 @@ public OzoneManagerServiceProviderImpl(
HttpConfig.Policy policy = HttpConfig.getHttpPolicy(configuration);
- omDBSnapshotUrl = "http://"; + ozoneManagerHttpAddress +
- OZONE_DB_CHECKPOINT_HTTP_ENDPOINT;
-
- if (policy.isHttpsEnabled()) {
- omDBSnapshotUrl = "https://"; + ozoneManagerHttpsAddress +
- OZONE_DB_CHECKPOINT_HTTP_ENDPOINT;
- }
-
boolean flushParam = configuration.getBoolean(
OZONE_RECON_OM_SNAPSHOT_TASK_FLUSH_PARAM,
configuration.getBoolean(
ReconServerConfigKeys.RECON_OM_SNAPSHOT_TASK_FLUSH_PARAM,
false)
);
-
- if (flushParam) {
- omDBSnapshotUrl += "?" + OZONE_DB_CHECKPOINT_REQUEST_FLUSH + "=true";
- }
+ // Same switch OM followers honor: use the inode-based v2 checkpoint
endpoint
+ // by default, or fall back to the v1 endpoint when disabled (e.g. mixed
+ // versions during an upgrade).
+ // NOTE: this flag is read from Recon's OzoneConfiguration, not OM's.
During
+ // a mixed-version rollout, operators must set
+ // ozone.om.db.checkpoint.use.inode.based.transfer on Recon to match the
+ // OM cluster's setting; otherwise Recon may hit an endpoint the OM side
+ // does not serve.
+ boolean useV2CheckpointApi = configuration.getBoolean(
+ OZONE_OM_DB_CHECKPOINT_USE_INODE_BASED_KEY,
+ OZONE_OM_DB_CHECKPOINT_USE_INODE_BASED_DEFAULT);
this.reconUtils = reconUtils;
this.omMetadataManager = omMetadataManager;
@@ -228,13 +214,16 @@ public OzoneManagerServiceProviderImpl(
this.threadFactory =
new ThreadFactoryBuilder().setNameFormat(threadNamePrefix +
"SyncOM-%d")
.build();
- // Number of parallel workers
- int omDBTarProcessorThreadCount = Math.min(64,
Runtime.getRuntime().availableProcessors());
this.reconContext = reconContext;
this.taskStatusUpdaterManager = taskStatusUpdaterManager;
this.omDBLagThreshold =
configuration.getLong(RECON_OM_DELTA_UPDATE_LAG_THRESHOLD,
RECON_OM_DELTA_UPDATE_LAG_THRESHOLD_DEFAULT);
- this.tarExtractor = new TarExtractor(omDBTarProcessorThreadCount,
threadNamePrefix);
+ // Download the full OM DB snapshot by reusing OM's checkpoint transfer
path
+ // (the same one an OM follower uses): POST /v2/dbCheckpoint, or the v1
+ // /dbCheckpoint endpoint when the inode-based transfer is disabled.
+ this.reconSnapshotProvider = new ReconRDBSnapshotProvider(
+ omSnapshotDBParentDir, connectionFactory, isOmSpnegoEnabled(), policy,
+ flushParam, useV2CheckpointApi, this::getLeaderServiceInfo);
}
@Override
@@ -247,7 +236,6 @@ public void start() {
LOG.info("Starting Ozone Manager Service Provider.");
scheduler = Executors.newScheduledThreadPool(1, threadFactory);
try {
- tarExtractor.start();
omMetadataManager.start(configuration);
} catch (IOException ioEx) {
LOG.error("Error starting Recon OM Metadata Manager.", ioEx);
@@ -375,7 +363,6 @@ private void stopSyncDataFromOMThread() {
scheduler.shutdownNow();
Thread.currentThread().interrupt();
}
- tarExtractor.stop();
LOG.debug("Shutdown the OM DB sync scheduler.");
}
@@ -387,7 +374,6 @@ public boolean triggerSyncDataFromOMImmediately() {
// immediately.
stopSyncDataFromOMThread();
scheduler = Executors.newScheduledThreadPool(1, threadFactory);
- tarExtractor.start();
startSyncDataFromOM(0L);
return true;
} else {
@@ -402,54 +388,59 @@ public void stop() throws Exception {
reconTaskController.stop();
omMetadataManager.stop();
scheduler.shutdownNow();
- tarExtractor.stop();
+ reconSnapshotProvider.close();
metrics.unRegister();
reconSyncMetrics.unRegister();
connectionFactory.destroy();
}
+ private boolean isOmSpnegoEnabled() {
+ return configuration.get(OZONE_OM_HTTP_AUTH_TYPE, "simple")
+ .equals("kerberos");
+ }
+
/**
- * Find the OM leader's address to get the snapshot from.
+ * Return the current OM leader's {@link ServiceInfo} from the OM service
+ * list. The retained transfer checkpoint is local to the leader, so Recon
+ * always downloads from (and resumes against) the leader.
*/
- @VisibleForTesting
- public String getOzoneManagerSnapshotUrl() throws IOException {
- String omLeaderUrl = omDBSnapshotUrl;
- List<org.apache.hadoop.ozone.om.helpers.ServiceInfo> serviceList =
- ozoneManagerClient.getServiceList();
- HttpConfig.Policy policy = HttpConfig.getHttpPolicy(configuration);
- if (!serviceList.isEmpty()) {
- for (org.apache.hadoop.ozone.om.helpers.ServiceInfo info : serviceList) {
- if (info.getNodeType().equals(HddsProtos.NodeType.OM) &&
- info.getOmRoleInfo().hasServerRole() &&
- info.getOmRoleInfo().getServerRole().equals(LEADER.name())) {
- omLeaderUrl = (policy.isHttpsEnabled() ?
- "https://"; + info.getServiceAddress(Type.HTTPS) :
- "http://"; + info.getServiceAddress(Type.HTTP)) +
- OZONE_DB_CHECKPOINT_HTTP_ENDPOINT;
+ private ServiceInfo getLeaderServiceInfo() {
+ try {
+ List<ServiceInfo> serviceList = ozoneManagerClient.getServiceList();
+ for (ServiceInfo info : serviceList) {
+ if (info.getNodeType().equals(HddsProtos.NodeType.OM)
+ && info.getOmRoleInfo().hasServerRole()
+ && info.getOmRoleInfo().getServerRole().equals(LEADER.name())) {
+ return info;
}
}
+ } catch (IOException e) {
+ throw new UncheckedIOException("Failed to fetch OM service list", e);
}
- return omLeaderUrl;
+ throw new IllegalStateException("No OM leader found in the OM service
list.");
}
- private boolean isOmSpnegoEnabled() {
- return configuration.get(OZONE_OM_HTTP_AUTH_TYPE, "simple")
- .equals("kerberos");
+ private String getLeaderNodeId() {
+ return getLeaderServiceInfo().getOmRoleInfo().getNodeId();
}
/**
- * Method to obtain current OM DB Snapshot.
- * @return DBCheckpoint instance.
+ * Obtain the current OM DB snapshot using the same OM-follower bootstrap
+ * mechanism (chunked, resumable {@code POST /v2/dbCheckpoint} with hard-link
+ * dedup on the leader). The returned {@link DBCheckpoint} points at
+ * a stable {@code om.snapshot.db_<ts>} directory ready to be promoted by
+ * {@link #updateReconOmDBWithNewSnapshot()}; returns {@code null} on
failure,
+ * leaving the current active DB untouched.
+ *
+ * @return DBCheckpoint instance, or {@code null} on failure.
*/
@VisibleForTesting
public DBCheckpoint getOzoneManagerDBSnapshot() {
- String snapshotFileName = RECON_OM_SNAPSHOT_DB + "_" +
System.currentTimeMillis();
- Path untarredDbDir = Paths.get(omSnapshotDBParentDir.getAbsolutePath(),
snapshotFileName);
-
- // Before fetching full snapshot again and create a new OM DB snapshot
directory, check and delete
- // any existing OM DB snapshot directories under recon om db dir location
and delete all such
- // om db snapshot dirs including the last known om db snapshot dir
returned by reconUtils.getLastKnownDB
- File lastKnownDB = reconUtils.getLastKnownDB(omSnapshotDBParentDir,
RECON_OM_SNAPSHOT_DB);
+ // Before fetching a new full snapshot, delete the last known OM DB
snapshot
+ // dir so we don't hold two full copies at once (keeps peak disk ~1x). This
+ // also clears any snapshot dir left over after switching v1/v2 endpoints.
+ File lastKnownDB = reconUtils.getLastKnownDB(omSnapshotDBParentDir,
+ RECON_OM_SNAPSHOT_DB);
if (lastKnownDB != null) {
boolean existingOmSnapshotDBDeleted =
FileUtils.deleteQuietly(lastKnownDB);
if (existingOmSnapshotDBDeleted) {
@@ -461,56 +452,58 @@ public DBCheckpoint getOzoneManagerDBSnapshot() {
}
}
- // Now below cleanup operation will even remove any left over staging dirs
in recon om db dir location which
- // may be left due to any previous partial extraction of tar entries and
during copy sst files process by
- // tarExtractor.extractTar
- File[] leftOverStagingDirs = omSnapshotDBParentDir.listFiles(f ->
f.getName().startsWith(STAGING));
+ // Remove any leftover staging dirs from a previous partial extraction
+ // (for example artifacts left by the v1 tar-extraction path).
+ File[] leftOverStagingDirs =
+ omSnapshotDBParentDir.listFiles(f -> f.getName().startsWith(STAGING));
if (leftOverStagingDirs != null) {
for (File stagingDir : leftOverStagingDirs) {
- LOG.warn("Cleaning up leftover staging folder from failed extraction:
{}", stagingDir.getAbsolutePath());
- boolean stagingDirDeleted = FileUtils.deleteQuietly(stagingDir);
- if (stagingDirDeleted) {
- LOG.info("Successfully deleted leftover staging folder: {}",
stagingDir.getAbsolutePath());
+ LOG.warn("Cleaning up leftover staging folder from failed extraction:
{}",
+ stagingDir.getAbsolutePath());
+ if (FileUtils.deleteQuietly(stagingDir)) {
+ LOG.info("Successfully deleted leftover staging folder: {}",
+ stagingDir.getAbsolutePath());
} else {
- LOG.warn("Failed to delete leftover staging folder: {}",
stagingDir.getAbsolutePath());
+ LOG.warn("Failed to delete leftover staging folder: {}",
+ stagingDir.getAbsolutePath());
}
}
}
try {
- SecurityUtil.doAsLoginUser(() -> {
- try (InputStream inputStream = reconUtils.makeHttpCall(
- connectionFactory, getOzoneManagerSnapshotUrl(),
isOmSpnegoEnabled()).getInputStream()) {
- tarExtractor.extractTar(inputStream, untarredDbDir);
- } catch (IOException | InterruptedException e) {
- reconContext.updateHealthStatus(new AtomicBoolean(false));
-
reconContext.updateErrors(ReconContext.ErrorCode.GET_OM_DB_SNAPSHOT_FAILED);
- throw new RuntimeException("Error while extracting OM DB Snapshot
TAR.", e);
- }
- return null;
- });
- // Validate extracted files
- File[] sstFiles = untarredDbDir.toFile().listFiles((dir, name) ->
name.endsWith(".sst"));
+ String leaderNodeId = getLeaderNodeId();
+ DBCheckpoint checkpoint =
+ reconSnapshotProvider.downloadDBSnapshotFromLeader(leaderNodeId);
+
+ // Validate the assembled snapshot: log how many SST files it contains.
+ File untarredDbDir = checkpoint.getCheckpointLocation().toFile();
+ File[] sstFiles =
+ untarredDbDir.listFiles((dir, name) -> name.endsWith(".sst"));
if (sstFiles != null && sstFiles.length > 0) {
- LOG.info("Number of SST files found in the OM snapshot directory: {} -
{}", untarredDbDir, sstFiles.length);
+ LOG.info("Number of SST files found in the OM snapshot directory: {} -
{}",
+ untarredDbDir, sstFiles.length);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Valid SST files found: {}", Arrays.stream(sstFiles)
+ .map(File::getName).collect(Collectors.toList()));
+ }
}
- List<String> sstFileNames = Arrays.stream(sstFiles)
- .map(File::getName)
- .collect(Collectors.toList());
- LOG.debug("Valid SST files found: {}", sstFileNames);
-
- // Currently, OM DB type is not configurable. Hence, defaulting to
- // RocksDB.
reconContext.updateHealthStatus(new AtomicBoolean(true));
-
reconContext.getErrors().remove(ReconContext.ErrorCode.GET_OM_DB_SNAPSHOT_FAILED);
- return new RocksDBCheckpoint(untarredDbDir);
- } catch (IOException e) {
+ reconContext.getErrors()
+ .remove(ReconContext.ErrorCode.GET_OM_DB_SNAPSHOT_FAILED);
+ return checkpoint;
+ } catch (IOException | RuntimeException e) {
LOG.error("Unable to obtain Ozone Manager DB Snapshot.", e);
reconContext.updateHealthStatus(new AtomicBoolean(false));
-
reconContext.updateErrors(ReconContext.ErrorCode.GET_OM_DB_SNAPSHOT_FAILED);
+ reconContext.updateErrors(
+ ReconContext.ErrorCode.GET_OM_DB_SNAPSHOT_FAILED);
+ // Do not wipe the candidate dir here. The shared RDBSnapshotProvider's
+ // checkLeaderConsistency() manages it on the next attempt: kept for the
+ // same leader (so a future batched/chunked transfer can resume the parts
+ // already received), reset on a leader change. This matches how the OM
+ // follower handles a failed download.
+ return null;
}
- return null;
}
/**
@@ -993,8 +986,9 @@ public OzoneManagerSyncMetrics getMetrics() {
}
@VisibleForTesting
- public TarExtractor getTarExtractor() {
- return tarExtractor;
+ public void setReconSnapshotProvider(
+ ReconRDBSnapshotProvider reconSnapshotProvider) {
+ this.reconSnapshotProvider = reconSnapshotProvider;
}
}
diff --git
a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/ReconRDBSnapshotProvider.java
b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/ReconRDBSnapshotProvider.java
new file mode 100644
index 00000000000..7241178e9b9
--- /dev/null
+++
b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/ReconRDBSnapshotProvider.java
@@ -0,0 +1,245 @@
+/*
+ * 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 java.net.HttpURLConnection.HTTP_CREATED;
+import static java.net.HttpURLConnection.HTTP_OK;
+import static org.apache.hadoop.ozone.OzoneConsts.MULTIPART_FORM_DATA_BOUNDARY;
+import static org.apache.hadoop.ozone.OzoneConsts.OM_DB_NAME;
+import static
org.apache.hadoop.ozone.OzoneConsts.OZONE_DB_CHECKPOINT_HTTP_ENDPOINT;
+import static
org.apache.hadoop.ozone.OzoneConsts.OZONE_DB_CHECKPOINT_HTTP_ENDPOINT_V2;
+import static
org.apache.hadoop.ozone.OzoneConsts.OZONE_DB_CHECKPOINT_INCLUDE_SNAPSHOT_DATA;
+import static
org.apache.hadoop.ozone.OzoneConsts.OZONE_DB_CHECKPOINT_REQUEST_FLUSH;
+import static
org.apache.hadoop.ozone.recon.ReconConstants.RECON_OM_SNAPSHOT_DB;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.HttpURLConnection;
+import java.net.MalformedURLException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Supplier;
+import org.apache.commons.io.FileUtils;
+import org.apache.hadoop.hdds.server.http.HttpConfig;
+import org.apache.hadoop.hdds.utils.HAUtils;
+import org.apache.hadoop.hdds.utils.RDBSnapshotProvider;
+import org.apache.hadoop.hdds.utils.db.DBCheckpoint;
+import org.apache.hadoop.hdds.utils.db.InodeMetadataRocksDBCheckpoint;
+import org.apache.hadoop.hdds.utils.db.RocksDBCheckpoint;
+import org.apache.hadoop.hdfs.web.URLConnectionFactory;
+import org.apache.hadoop.ozone.om.helpers.ServiceInfo;
+import org.apache.hadoop.ozone.om.ratis_snapshot.OmRatisSnapshotProvider;
+import
org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.ServicePort.Type;
+import org.apache.hadoop.security.SecurityUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Recon's {@link RDBSnapshotProvider} implementation that downloads the OM DB
+ * checkpoint using the same bootstrap mechanism an OM follower uses: a chunked
+ * {@code POST /v2/dbCheckpoint} request that carries a {@code toExcludeList[]}
+ * of the parts already received so an interrupted transfer can resume, with
+ * hard-link dedup on the leader and a completion sentinel to end the transfer.
+ *
+ * <p>This mirrors {@link OmRatisSnapshotProvider}: it overrides
+ * {@link #downloadSnapshot} to POST to the leader's {@code /v2/dbCheckpoint}
+ * endpoint and {@link #getCheckpointFromUntarredDb} to assemble and promote
the
+ * downloaded DB into a stable snapshot dir Recon can open. Like the OM
follower,
+ * it honors {@code ozone.om.db.checkpoint.use.inode.based.transfer}: when
that is
+ * {@code false} it falls back to the v1 {@code /dbCheckpoint} endpoint.
+ */
+public class ReconRDBSnapshotProvider extends RDBSnapshotProvider {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(ReconRDBSnapshotProvider.class);
+
+ private final URLConnectionFactory connectionFactory;
+ private final boolean spnegoEnabled;
+ private final boolean httpsEnabled;
+ private final boolean flushBeforeCheckpoint;
+ private final boolean useV2CheckpointApi;
+ private final Supplier<ServiceInfo> leaderInfoSupplier;
+ // Leader pinned for the duration of a single (possibly multi-part) transfer.
+ private final AtomicReference<ServiceInfo> pinnedLeader =
+ new AtomicReference<>();
+
+ public ReconRDBSnapshotProvider(File snapshotDir,
+ URLConnectionFactory connectionFactory, boolean spnegoEnabled,
+ HttpConfig.Policy httpPolicy, boolean flushBeforeCheckpoint,
+ boolean useV2CheckpointApi,
+ Supplier<ServiceInfo> leaderInfoSupplier) {
+ super(snapshotDir, RECON_OM_SNAPSHOT_DB);
+ this.connectionFactory = connectionFactory;
+ this.spnegoEnabled = spnegoEnabled;
+ this.httpsEnabled = httpPolicy.isHttpsEnabled();
+ this.flushBeforeCheckpoint = flushBeforeCheckpoint;
+ this.useV2CheckpointApi = useV2CheckpointApi;
+ this.leaderInfoSupplier = leaderInfoSupplier;
+ }
+
+ /**
+ * Pin the OM leader for the whole transfer before delegating to the shared
+ * driver loop. The base loop resolves the leader only once (via
+ * {#checkLeaderConsistency}); re-resolving per part would let a
+ * mid-transfer OM failover merge parts from two leaders into the same
+ * candidate dir. Recon's transfer is single-part today (it excludes snapshot
+ * data), but pinning keeps this correct if that ever changes, matching how
+ * {@link OmRatisSnapshotProvider} pins the leader for every part.
+ */
+ @Override
+ public DBCheckpoint downloadDBSnapshotFromLeader(String leaderNodeID)
+ throws IOException {
+ pinnedLeader.set(leaderInfoSupplier.get());
+ try {
+ return super.downloadDBSnapshotFromLeader(leaderNodeID);
+ } finally {
+ pinnedLeader.set(null);
+ }
+ }
+
+ @Override
+ public void downloadSnapshot(String leaderNodeID, File targetFile)
+ throws IOException {
+ // Use the leader pinned for this transfer. The fallback only applies to a
+ // direct downloadSnapshot call outside downloadDBSnapshotFromLeader.
+ ServiceInfo leader = pinnedLeader.get();
+ if (leader == null) {
+ leader = leaderInfoSupplier.get();
+ }
+ URL checkpointUrl = buildCheckpointUrl(leader);
+ LOG.info("Downloading OM DB checkpoint from leader {}. Checkpoint: {}, "
+ + "URL: {}", leaderNodeID, targetFile.getName(), checkpointUrl);
+ SecurityUtil.doAsLoginUser(() -> {
+ HttpURLConnection connection = (HttpURLConnection)
+ connectionFactory.openConnection(checkpointUrl, spnegoEnabled);
+ connection.setRequestMethod("POST");
+ connection.setRequestProperty("Content-Type",
+ "multipart/form-data; boundary=" + MULTIPART_FORM_DATA_BOUNDARY);
+ connection.setDoOutput(true);
+
+ List<String> existingFiles = useV2CheckpointApi
+ ? HAUtils.getExistingFiles(getCandidateDir())
+ : HAUtils.getExistingSstFilesRelativeToDbDir(getCandidateDir());
+ OmRatisSnapshotProvider.writeFormData(connection, existingFiles);
+
+ connection.connect();
+ int errorCode = connection.getResponseCode();
+ if (errorCode != HTTP_OK && errorCode != HTTP_CREATED) {
+ throw new IOException("Unexpected response code " + errorCode
+ + " when downloading OM DB checkpoint from " + checkpointUrl);
+ }
+ try (InputStream inputStream = connection.getInputStream()) {
+ OmRatisSnapshotProvider.downloadFileWithProgress(inputStream,
+ targetFile);
+ } catch (IOException ex) {
+ if (!FileUtils.deleteQuietly(targetFile)) {
+ LOG.error("Failed to delete partial checkpoint file {}", targetFile);
+ }
+ throw ex;
+ } finally {
+ connection.disconnect();
+ }
+ return null;
+ });
+ }
+
+ /**
+ * After the transfer completes, install the leader's hard-link inventory,
+ * normalize the layout to {@code <candidate>/om.db}, then move that DB out
of
+ * the reused candidate dir into a stable timestamped snapshot dir that Recon
+ * opens as its new live DB. The candidate dir is emptied so the next sync
+ * starts from a clean candidate dir.
+ */
+ @Override
+ public DBCheckpoint getCheckpointFromUntarredDb(Path untarredDbDir)
+ throws IOException {
+ // The base class only calls this once it has seen the leader's
+ // end-of-tarball marker. That marker is named "ratis snapshot complete"
+ // for historical reasons, but it is not Ratis-specific: OM's shared
+ // DBCheckpointServlet appends it to the end of every /v2/dbCheckpoint
+ // response (the same one an OM follower bootstraps from), so here it just
+ // means "the leader finished sending the checkpoint".
+
+ // Installs hard links from hardLinkFile (tolerates a missing/empty file)
+ // and moves root-level DB files into <untarredDbDir>/om.db.
deleteSourceFiles
+ // follows the endpoint: true for v2 (inode-based), false for the v1
layout.
+ new InodeMetadataRocksDBCheckpoint(untarredDbDir, useV2CheckpointApi);
+
+ Path omDbDir = untarredDbDir.resolve(OM_DB_NAME);
+ if (!Files.isDirectory(omDbDir)) {
+ throw new IOException("Expected RocksDB directory not found after "
+ + "assembling checkpoint: " + omDbDir);
+ }
+
+ String stableName = RECON_OM_SNAPSHOT_DB + "_" +
System.currentTimeMillis();
+ Path stablePath = getSnapshotDir().toPath().resolve(stableName);
+ Files.move(omDbDir, stablePath);
+ LOG.info("Assembled OM DB moved from {} to {}", omDbDir, stablePath);
+
+ // Clear residual entries (completion flag, orphan seeded files, empty
dirs)
+ // so the candidate dir is empty for the next sync cycle.
+ cleanupCandidateDir(untarredDbDir.toFile());
+
+ return new RocksDBCheckpoint(stablePath);
+ }
+
+ URL buildCheckpointUrl(ServiceInfo leader) throws IOException {
+ Type portType = httpsEnabled ? Type.HTTPS : Type.HTTP;
+ String scheme = httpsEnabled ? "https" : "http";
+ String path = useV2CheckpointApi ? OZONE_DB_CHECKPOINT_HTTP_ENDPOINT_V2
+ : OZONE_DB_CHECKPOINT_HTTP_ENDPOINT;
+ // Recon does not need OM's nested snapshot data.
+ String query = OZONE_DB_CHECKPOINT_INCLUDE_SNAPSHOT_DATA + "=false&"
+ + OZONE_DB_CHECKPOINT_REQUEST_FLUSH + "="
+ + (flushBeforeCheckpoint ? "true" : "false");
+ try {
+ return new URI(scheme, null, leader.getHostname(),
+ leader.getPort(portType), path, query, null).toURL();
+ } catch (URISyntaxException | MalformedURLException e) {
+ throw new IOException("Could not build OM DB checkpoint URL", e);
+ }
+ }
+
+ ServiceInfo getPinnedLeader() {
+ return pinnedLeader.get();
+ }
+
+ private void cleanupCandidateDir(File candidate) {
+ File[] entries = candidate.listFiles();
+ if (entries == null) {
+ return;
+ }
+ for (File entry : entries) {
+ if (!FileUtils.deleteQuietly(entry)) {
+ LOG.warn("Failed to clean up candidate dir entry {}", entry);
+ }
+ }
+ }
+
+ @Override
+ public void close() {
+ // The URLConnectionFactory is owned and destroyed by
+ // OzoneManagerServiceProviderImpl; nothing to release here.
+ }
+}
diff --git
a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/api/TestTriggerDBSyncEndpoint.java
b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/api/TestTriggerDBSyncEndpoint.java
index 6e2ad040ce1..5ec99ee9b44 100644
---
a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/api/TestTriggerDBSyncEndpoint.java
+++
b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/api/TestTriggerDBSyncEndpoint.java
@@ -120,6 +120,7 @@ public void setUp() throws IOException,
AuthenticationException {
}
when(reconUtilsMock.makeHttpCall(any(), anyString(), anyBoolean()))
.thenReturn(httpURLConnectionMock);
+ when(reconUtilsMock.getReconDbDir(any(),
anyString())).thenCallRealMethod();
when(reconUtilsMock.getReconNodeDetails(
any(OzoneConfiguration.class))).thenReturn(
ReconTestUtils.getReconNodeDetails());
diff --git
a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/spi/impl/TestOzoneManagerServiceProviderImpl.java
b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/spi/impl/TestOzoneManagerServiceProviderImpl.java
index e6690e37703..c5664934116 100644
---
a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/spi/impl/TestOzoneManagerServiceProviderImpl.java
+++
b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/spi/impl/TestOzoneManagerServiceProviderImpl.java
@@ -22,21 +22,19 @@
import static
org.apache.hadoop.ozone.recon.OMMetadataManagerTestUtils.initializeEmptyOmMetadataManager;
import static
org.apache.hadoop.ozone.recon.OMMetadataManagerTestUtils.initializeNewOmMetadataManager;
import static
org.apache.hadoop.ozone.recon.OMMetadataManagerTestUtils.writeDataToOm;
+import static
org.apache.hadoop.ozone.recon.ReconConstants.RECON_OM_SNAPSHOT_DB;
import static
org.apache.hadoop.ozone.recon.ReconServerConfigKeys.OZONE_RECON_DB_DIR;
import static
org.apache.hadoop.ozone.recon.ReconServerConfigKeys.OZONE_RECON_OM_SNAPSHOT_DB_DIR;
import static
org.apache.hadoop.ozone.recon.ReconServerConfigKeys.RECON_OM_DELTA_UPDATE_LAG_THRESHOLD;
import static
org.apache.hadoop.ozone.recon.ReconServerConfigKeys.RECON_OM_DELTA_UPDATE_LIMIT;
-import static org.apache.hadoop.ozone.recon.ReconUtils.createTarFile;
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.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
-import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.any;
-import static org.mockito.Mockito.anyBoolean;
import static org.mockito.Mockito.anyString;
import static org.mockito.Mockito.doCallRealMethod;
import static org.mockito.Mockito.doNothing;
@@ -47,13 +45,12 @@
import java.io.File;
import java.io.IOException;
-import java.io.InputStream;
-import java.net.HttpURLConnection;
-import java.nio.file.Files;
import java.nio.file.Paths;
+import java.util.Collections;
import java.util.List;
import org.apache.commons.io.FileUtils;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
import org.apache.hadoop.hdds.utils.db.DBCheckpoint;
import org.apache.hadoop.hdds.utils.db.RDBStore;
import org.apache.hadoop.hdds.utils.db.RocksDatabase;
@@ -62,6 +59,7 @@
import org.apache.hadoop.ozone.om.OMMetadataManager;
import org.apache.hadoop.ozone.om.helpers.BucketLayout;
import org.apache.hadoop.ozone.om.helpers.DBUpdates;
+import org.apache.hadoop.ozone.om.helpers.ServiceInfo;
import org.apache.hadoop.ozone.om.protocol.OzoneManagerProtocol;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
import org.apache.hadoop.ozone.recon.ReconContext;
@@ -92,6 +90,7 @@ public class TestOzoneManagerServiceProviderImpl {
private OzoneConfiguration configuration;
private OzoneManagerProtocol ozoneManagerProtocol;
private ReconContext reconContext;
+ private File omSnapshotDbParentDir;
@BeforeEach
public void setUp(@TempDir File dirReconSnapDB, @TempDir File dirReconDB)
@@ -104,6 +103,8 @@ public void setUp(@TempDir File dirReconSnapDB, @TempDir
File dirReconDB)
configuration.set("ozone.om.address", "localhost:9862");
ozoneManagerProtocol = getMockOzoneManagerClient(new DBUpdates());
reconContext = new ReconContext(configuration, new ReconUtils());
+ omSnapshotDbParentDir = new ReconUtils()
+ .getReconDbDir(configuration, OZONE_RECON_OM_SNAPSHOT_DB_DIR);
}
@AfterEach
@@ -130,42 +131,37 @@ public void testUpdateReconOmDBWithNewSnapshot(
DBCheckpoint checkpoint = omMetadataManager.getStore()
.getCheckpoint(true);
- File tarFile = createTarFile(checkpoint.getCheckpointLocation());
- try (InputStream inputStream = Files.newInputStream(tarFile.toPath())) {
- ReconUtils reconUtilsMock = getMockReconUtils();
- HttpURLConnection httpURLConnectionMock = mock(HttpURLConnection.class);
- when(httpURLConnectionMock.getInputStream()).thenReturn(inputStream);
- when(reconUtilsMock.makeHttpCall(any(), anyString(), anyBoolean()))
- .thenReturn(httpURLConnectionMock);
- when(reconUtilsMock.getReconNodeDetails(
- any(OzoneConfiguration.class))).thenReturn(
- ReconTestUtils.getReconNodeDetails());
- ReconTaskController reconTaskController = getMockTaskController();
-
- OzoneManagerServiceProviderImpl ozoneManagerServiceProvider =
- new OzoneManagerServiceProviderImpl(configuration,
- reconOMMetadataManager, reconTaskController, reconUtilsMock,
ozoneManagerProtocol,
- reconContext, getMockTaskStatusUpdaterManager());
-
- assertNull(reconOMMetadataManager.getKeyTable(getBucketLayout())
- .get("/sampleVol/bucketOne/key_one"));
- assertNull(reconOMMetadataManager.getKeyTable(getBucketLayout())
- .get("/sampleVol/bucketOne/key_two"));
-
- ozoneManagerServiceProvider.getTarExtractor().start();
- assertTrue(ozoneManagerServiceProvider.updateReconOmDBWithNewSnapshot());
-
- assertNotNull(reconOMMetadataManager.getKeyTable(getBucketLayout())
- .get("/sampleVol/bucketOne/key_one"));
- assertNotNull(reconOMMetadataManager.getKeyTable(getBucketLayout())
- .get("/sampleVol/bucketOne/key_two"));
-
- // Verifying if context error GET_OM_DB_SNAPSHOT_FAILED is removed
-
assertFalse(reconContext.getErrors().contains(ReconContext.ErrorCode.GET_OM_DB_SNAPSHOT_FAILED));
- omMetadataManager.stop();
- reconOMMetadataManager.stop();
- reconTaskController.stop();
- }
+ ReconUtils reconUtilsMock = getMockReconUtils();
+ when(reconUtilsMock.getReconNodeDetails(
+ any(OzoneConfiguration.class))).thenReturn(
+ ReconTestUtils.getReconNodeDetails());
+ ReconTaskController reconTaskController = getMockTaskController();
+
+ OzoneManagerServiceProviderImpl ozoneManagerServiceProvider =
+ new OzoneManagerServiceProviderImpl(configuration,
+ reconOMMetadataManager, reconTaskController, reconUtilsMock,
ozoneManagerProtocol,
+ reconContext, getMockTaskStatusUpdaterManager());
+ ozoneManagerServiceProvider.setReconSnapshotProvider(
+ newStubSnapshotProvider(omSnapshotDbParentDir,
+ checkpoint.getCheckpointLocation(), false));
+
+ assertNull(reconOMMetadataManager.getKeyTable(getBucketLayout())
+ .get("/sampleVol/bucketOne/key_one"));
+ assertNull(reconOMMetadataManager.getKeyTable(getBucketLayout())
+ .get("/sampleVol/bucketOne/key_two"));
+
+ assertTrue(ozoneManagerServiceProvider.updateReconOmDBWithNewSnapshot());
+
+ assertNotNull(reconOMMetadataManager.getKeyTable(getBucketLayout())
+ .get("/sampleVol/bucketOne/key_one"));
+ assertNotNull(reconOMMetadataManager.getKeyTable(getBucketLayout())
+ .get("/sampleVol/bucketOne/key_two"));
+
+ // Verifying if context error GET_OM_DB_SNAPSHOT_FAILED is removed
+
assertFalse(reconContext.getErrors().contains(ReconContext.ErrorCode.GET_OM_DB_SNAPSHOT_FAILED));
+ omMetadataManager.stop();
+ reconOMMetadataManager.stop();
+ reconTaskController.stop();
}
@Test
@@ -180,9 +176,6 @@ public void testUpdateReconOmDBWithNewSnapshotFailure(
dirReconMetadata);
ReconUtils reconUtilsMock = getMockReconUtils();
-
- when(reconUtilsMock.makeHttpCall(any(), anyString(), anyBoolean()))
- .thenThrow(new IOException("Mocked IOException"));
when(reconUtilsMock.getReconNodeDetails(
any(OzoneConfiguration.class))).thenReturn(
ReconTestUtils.getReconNodeDetails());
@@ -192,12 +185,11 @@ public void testUpdateReconOmDBWithNewSnapshotFailure(
new OzoneManagerServiceProviderImpl(configuration,
reconOMMetadataManager, reconTaskController, reconUtilsMock,
ozoneManagerProtocol,
reconContext, getMockTaskStatusUpdaterManager());
+ // Stub provider that fails the download; the active DB must be kept.
+ ozoneManagerServiceProvider.setReconSnapshotProvider(
+ newStubSnapshotProvider(omSnapshotDbParentDir, null, true));
- Exception exception = assertThrows(RuntimeException.class, () -> {
- ozoneManagerServiceProvider.updateReconOmDBWithNewSnapshot();
- });
-
- assertTrue(exception.getCause() instanceof IOException);
+ assertFalse(ozoneManagerServiceProvider.updateReconOmDBWithNewSnapshot());
// Verifying if context error GET_OM_DB_SNAPSHOT_FAILED is added
assertTrue(reconContext.getErrors().contains(ReconContext.ErrorCode.GET_OM_DB_SNAPSHOT_FAILED));
@@ -219,34 +211,29 @@ public void testUpdateReconOmDBWithNewSnapshotSuccess(
writeDataToOm(omMetadataManager, "key_two");
DBCheckpoint checkpoint = omMetadataManager.getStore().getCheckpoint(true);
- File tarFile = createTarFile(checkpoint.getCheckpointLocation());
- try (InputStream inputStream = Files.newInputStream(tarFile.toPath())) {
- ReconUtils reconUtilsMock = getMockReconUtils();
- HttpURLConnection httpURLConnectionMock = mock(HttpURLConnection.class);
- when(httpURLConnectionMock.getInputStream()).thenReturn(inputStream);
- when(reconUtilsMock.makeHttpCall(any(), anyString(), anyBoolean()))
- .thenReturn(httpURLConnectionMock);
- when(reconUtilsMock.getReconNodeDetails(any(OzoneConfiguration.class)))
- .thenReturn(ReconTestUtils.getReconNodeDetails());
- ReconTaskController reconTaskController = getMockTaskController();
-
-
reconContext.updateErrors(ReconContext.ErrorCode.GET_OM_DB_SNAPSHOT_FAILED);
-
- OzoneManagerServiceProviderImpl ozoneManagerServiceProvider =
- new OzoneManagerServiceProviderImpl(configuration,
- reconOMMetadataManager, reconTaskController, reconUtilsMock,
ozoneManagerProtocol,
- reconContext, getMockTaskStatusUpdaterManager());
-
-
assertTrue(reconContext.getErrors().contains(ReconContext.ErrorCode.GET_OM_DB_SNAPSHOT_FAILED));
- ozoneManagerServiceProvider.getTarExtractor().start();
- assertTrue(ozoneManagerServiceProvider.updateReconOmDBWithNewSnapshot());
-
assertFalse(reconContext.getErrors().contains(ReconContext.ErrorCode.GET_OM_DB_SNAPSHOT_FAILED));
-
- assertNotNull(reconOMMetadataManager.getKeyTable(getBucketLayout())
- .get("/sampleVol/bucketOne/key_one"));
- assertNotNull(reconOMMetadataManager.getKeyTable(getBucketLayout())
- .get("/sampleVol/bucketOne/key_two"));
- }
+ ReconUtils reconUtilsMock = getMockReconUtils();
+ when(reconUtilsMock.getReconNodeDetails(any(OzoneConfiguration.class)))
+ .thenReturn(ReconTestUtils.getReconNodeDetails());
+ ReconTaskController reconTaskController = getMockTaskController();
+
+
reconContext.updateErrors(ReconContext.ErrorCode.GET_OM_DB_SNAPSHOT_FAILED);
+
+ OzoneManagerServiceProviderImpl ozoneManagerServiceProvider =
+ new OzoneManagerServiceProviderImpl(configuration,
+ reconOMMetadataManager, reconTaskController, reconUtilsMock,
ozoneManagerProtocol,
+ reconContext, getMockTaskStatusUpdaterManager());
+ ozoneManagerServiceProvider.setReconSnapshotProvider(
+ newStubSnapshotProvider(omSnapshotDbParentDir,
+ checkpoint.getCheckpointLocation(), false));
+
+
assertTrue(reconContext.getErrors().contains(ReconContext.ErrorCode.GET_OM_DB_SNAPSHOT_FAILED));
+ assertTrue(ozoneManagerServiceProvider.updateReconOmDBWithNewSnapshot());
+
assertFalse(reconContext.getErrors().contains(ReconContext.ErrorCode.GET_OM_DB_SNAPSHOT_FAILED));
+
+ assertNotNull(reconOMMetadataManager.getKeyTable(getBucketLayout())
+ .get("/sampleVol/bucketOne/key_one"));
+ assertNotNull(reconOMMetadataManager.getKeyTable(getBucketLayout())
+ .get("/sampleVol/bucketOne/key_two"));
omMetadataManager.stop();
reconOMMetadataManager.stop();
}
@@ -265,39 +252,30 @@ public void testReconOmDBCloseAndOpenNewSnapshotDb(
DBCheckpoint checkpoint = omMetadataManager.getStore()
.getCheckpoint(true);
- File tarFile1 = createTarFile(checkpoint.getCheckpointLocation());
- File tarFile2 = createTarFile(checkpoint.getCheckpointLocation());
ReconUtils reconUtilsMock = getMockReconUtils();
+ when(reconUtilsMock.getReconNodeDetails(
+ any(OzoneConfiguration.class))).thenReturn(
+ ReconTestUtils.getReconNodeDetails());
ReconTaskController reconTaskController = getMockTaskController();
- try (InputStream inputStream1 = Files.newInputStream(tarFile1.toPath())) {
- HttpURLConnection httpURLConnectionMock1 = mock(HttpURLConnection.class);
- when(httpURLConnectionMock1.getInputStream()).thenReturn(inputStream1);
- when(reconUtilsMock.makeHttpCall(any(), anyString(), anyBoolean()))
- .thenReturn(httpURLConnectionMock1);
- when(reconUtilsMock.getReconNodeDetails(
- any(OzoneConfiguration.class))).thenReturn(
- ReconTestUtils.getReconNodeDetails());
-
- OzoneManagerServiceProviderImpl ozoneManagerServiceProvider1 =
- new OzoneManagerServiceProviderImpl(configuration,
- reconOMMetadataManager, reconTaskController, reconUtilsMock,
ozoneManagerProtocol,
- reconContext, getMockTaskStatusUpdaterManager());
- ozoneManagerServiceProvider1.getTarExtractor().start();
-
assertTrue(ozoneManagerServiceProvider1.updateReconOmDBWithNewSnapshot());
- }
- try (InputStream inputStream2 = Files.newInputStream(tarFile2.toPath())) {
- HttpURLConnection httpURLConnectionMock2 = mock(HttpURLConnection.class);
- when(httpURLConnectionMock2.getInputStream()).thenReturn(inputStream2);
- when(reconUtilsMock.makeHttpCall(any(), anyString(), anyBoolean()))
- .thenReturn(httpURLConnectionMock2);
- OzoneManagerServiceProviderImpl ozoneManagerServiceProvider2 =
- new OzoneManagerServiceProviderImpl(configuration,
- reconOMMetadataManager, reconTaskController, reconUtilsMock,
ozoneManagerProtocol,
- reconContext, getMockTaskStatusUpdaterManager());
- ozoneManagerServiceProvider2.getTarExtractor().start();
-
assertTrue(ozoneManagerServiceProvider2.updateReconOmDBWithNewSnapshot());
- }
+ OzoneManagerServiceProviderImpl ozoneManagerServiceProvider1 =
+ new OzoneManagerServiceProviderImpl(configuration,
+ reconOMMetadataManager, reconTaskController, reconUtilsMock,
ozoneManagerProtocol,
+ reconContext, getMockTaskStatusUpdaterManager());
+ ozoneManagerServiceProvider1.setReconSnapshotProvider(
+ newStubSnapshotProvider(omSnapshotDbParentDir,
+ checkpoint.getCheckpointLocation(), false));
+ assertTrue(ozoneManagerServiceProvider1.updateReconOmDBWithNewSnapshot());
+
+ OzoneManagerServiceProviderImpl ozoneManagerServiceProvider2 =
+ new OzoneManagerServiceProviderImpl(configuration,
+ reconOMMetadataManager, reconTaskController, reconUtilsMock,
ozoneManagerProtocol,
+ reconContext, getMockTaskStatusUpdaterManager());
+ ozoneManagerServiceProvider2.setReconSnapshotProvider(
+ newStubSnapshotProvider(omSnapshotDbParentDir,
+ checkpoint.getCheckpointLocation(), false));
+ assertTrue(ozoneManagerServiceProvider2.updateReconOmDBWithNewSnapshot());
+
omMetadataManager.stop();
reconOMMetadataManager.stop();
reconTaskController.stop();
@@ -319,35 +297,29 @@ public void testGetOzoneManagerDBSnapshot(@TempDir File
dirReconMetadata)
.toFile();
FileUtils.write(file2, "File2 Contents", UTF_8);
- //Create test tar file.
- File tarFile = createTarFile(checkpointDir.toPath());
- try (InputStream fileInputStream = Files.newInputStream(tarFile.toPath()))
{
- ReconUtils reconUtilsMock = getMockReconUtils();
- HttpURLConnection httpURLConnectionMock = mock(HttpURLConnection.class);
- when(httpURLConnectionMock.getInputStream()).thenReturn(fileInputStream);
- when(reconUtilsMock.makeHttpCall(any(), anyString(), anyBoolean()))
- .thenReturn(httpURLConnectionMock);
- when(reconUtilsMock.getReconNodeDetails(
- any(OzoneConfiguration.class))).thenReturn(
- ReconTestUtils.getReconNodeDetails());
- ReconOMMetadataManager reconOMMetadataManager =
- mock(ReconOMMetadataManager.class);
- ReconTaskController reconTaskController = getMockTaskController();
- OzoneManagerServiceProviderImpl ozoneManagerServiceProvider =
- new OzoneManagerServiceProviderImpl(configuration,
- reconOMMetadataManager, reconTaskController, reconUtilsMock,
ozoneManagerProtocol,
- reconContext, getMockTaskStatusUpdaterManager());
-
- ozoneManagerServiceProvider.getTarExtractor().start();
- DBCheckpoint checkpoint = ozoneManagerServiceProvider
- .getOzoneManagerDBSnapshot();
- assertNotNull(checkpoint);
- assertTrue(checkpoint.getCheckpointLocation().toFile().isDirectory());
-
- File[] files = checkpoint.getCheckpointLocation().toFile().listFiles();
- assertNotNull(files);
- assertEquals(2, files.length);
- }
+ ReconUtils reconUtilsMock = getMockReconUtils();
+ when(reconUtilsMock.getReconNodeDetails(
+ any(OzoneConfiguration.class))).thenReturn(
+ ReconTestUtils.getReconNodeDetails());
+ ReconOMMetadataManager reconOMMetadataManager =
+ mock(ReconOMMetadataManager.class);
+ ReconTaskController reconTaskController = getMockTaskController();
+ OzoneManagerServiceProviderImpl ozoneManagerServiceProvider =
+ new OzoneManagerServiceProviderImpl(configuration,
+ reconOMMetadataManager, reconTaskController, reconUtilsMock,
ozoneManagerProtocol,
+ reconContext, getMockTaskStatusUpdaterManager());
+ ozoneManagerServiceProvider.setReconSnapshotProvider(
+ newStubSnapshotProvider(omSnapshotDbParentDir, checkpointDir.toPath(),
+ false));
+
+ DBCheckpoint checkpoint = ozoneManagerServiceProvider
+ .getOzoneManagerDBSnapshot();
+ assertNotNull(checkpoint);
+ assertTrue(checkpoint.getCheckpointLocation().toFile().isDirectory());
+
+ File[] files = checkpoint.getCheckpointLocation().toFile().listFiles();
+ assertNotNull(files);
+ assertEquals(2, files.length);
}
static RocksDatabase getRocksDatabase(OMMetadataManager om) {
@@ -636,9 +608,44 @@ private OzoneManagerProtocol getMockOzoneManagerClient(
mock(OzoneManagerProtocol.class);
when(ozoneManagerProtocolMock.getDBUpdates(any(OzoneManagerProtocolProtos
.DBUpdatesRequest.class))).thenReturn(dbUpdatesWrapper);
+ // Provide an OM leader so getLeaderNodeId() can resolve the download
source.
+ ServiceInfo leaderInfo = mock(ServiceInfo.class);
+ when(leaderInfo.getNodeType()).thenReturn(HddsProtos.NodeType.OM);
+ OzoneManagerProtocolProtos.OMRoleInfo roleInfo =
+ OzoneManagerProtocolProtos.OMRoleInfo.newBuilder()
+ .setNodeId("om1").setServerRole("LEADER").build();
+ when(leaderInfo.getOmRoleInfo()).thenReturn(roleInfo);
+ when(ozoneManagerProtocolMock.getServiceList())
+ .thenReturn(Collections.singletonList(leaderInfo));
return ozoneManagerProtocolMock;
}
+ /**
+ * A stub {@link ReconRDBSnapshotProvider} that bypasses HTTP/tar and simply
+ * promotes a copy of {@code sourceCheckpoint} into a fresh timestamped
+ * snapshot dir (or fails). Used to drive the provider's install path
+ * deterministically.
+ */
+ private ReconRDBSnapshotProvider newStubSnapshotProvider(
+ File snapshotParent, java.nio.file.Path sourceCheckpoint,
+ boolean fail) {
+ return new ReconRDBSnapshotProvider(snapshotParent, null, false,
+ org.apache.hadoop.hdds.server.http.HttpConfig.Policy.HTTP_ONLY, false,
+ true, () -> null) {
+ @Override
+ public DBCheckpoint downloadDBSnapshotFromLeader(String leaderNodeID)
+ throws IOException {
+ if (fail) {
+ throw new IOException("Mocked download failure");
+ }
+ java.nio.file.Path dest = snapshotParent.toPath().resolve(
+ RECON_OM_SNAPSHOT_DB + "_" + System.nanoTime());
+ FileUtils.copyDirectory(sourceCheckpoint.toFile(), dest.toFile());
+ return new org.apache.hadoop.hdds.utils.db.RocksDBCheckpoint(dest);
+ }
+ };
+ }
+
// Mock the case of SNNFE
private OzoneManagerProtocol getMockOzoneManagerClientWithThrow()
throws IOException {
diff --git
a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/spi/impl/TestReconRDBSnapshotProvider.java
b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/spi/impl/TestReconRDBSnapshotProvider.java
new file mode 100644
index 00000000000..9c93adc4a33
--- /dev/null
+++
b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/spi/impl/TestReconRDBSnapshotProvider.java
@@ -0,0 +1,223 @@
+/*
+ * 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 java.nio.charset.StandardCharsets.UTF_8;
+import static org.apache.hadoop.ozone.OzoneConsts.HARDLINK_SEPARATOR;
+import static
org.apache.hadoop.ozone.OzoneConsts.OZONE_DB_CHECKPOINT_HTTP_ENDPOINT;
+import static
org.apache.hadoop.ozone.OzoneConsts.OZONE_DB_CHECKPOINT_HTTP_ENDPOINT_V2;
+import static
org.apache.hadoop.ozone.recon.ReconConstants.RECON_OM_SNAPSHOT_DB;
+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.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.URL;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Supplier;
+import org.apache.commons.compress.archivers.ArchiveOutputStream;
+import org.apache.commons.compress.archivers.tar.TarArchiveEntry;
+import org.apache.commons.io.FileUtils;
+import org.apache.hadoop.hdds.server.http.HttpConfig;
+import org.apache.hadoop.hdds.utils.Archiver;
+import org.apache.hadoop.hdds.utils.HddsServerUtil;
+import org.apache.hadoop.hdds.utils.db.DBCheckpoint;
+import org.apache.hadoop.hdds.utils.db.InodeMetadataRocksDBCheckpoint;
+import org.apache.hadoop.hdds.utils.db.RocksDBCheckpoint;
+import org.apache.hadoop.ozone.om.helpers.ServiceInfo;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+/**
+ * Tests for {@link ReconRDBSnapshotProvider}: normalization and promotion of
+ * the assembled checkpoint after the transfer completes.
+ */
+public class TestReconRDBSnapshotProvider {
+
+ private static final Supplier<ServiceInfo> NO_LEADER = () -> null;
+
+ private ReconRDBSnapshotProvider newProvider(File snapshotDir) {
+ return newProvider(snapshotDir, true);
+ }
+
+ private ReconRDBSnapshotProvider newProvider(File snapshotDir,
+ boolean useV2CheckpointApi) {
+ return new ReconRDBSnapshotProvider(snapshotDir, null, false,
+ HttpConfig.Policy.HTTP_ONLY, false, useV2CheckpointApi, NO_LEADER);
+ }
+
+ private void writeFile(File dir, String name, String content)
+ throws IOException {
+ FileUtils.write(new File(dir, name), content, UTF_8);
+ }
+
+ @Test
+ public void testGetCheckpointPromotesDbAndClearsCandidate(
+ @TempDir File snapshotDir) throws IOException {
+ ReconRDBSnapshotProvider provider = newProvider(snapshotDir);
+ File candidate = provider.getCandidateDir();
+
+ // Simulate a fully untarred v2 checkpoint: flat DB files at the root plus
+ // the completion sentinel (no hardLinkFile - it should be tolerated).
+ writeFile(candidate, "000010.sst", "data-a");
+ writeFile(candidate, "CURRENT", "current");
+ writeFile(candidate,
HddsServerUtil.OZONE_RATIS_SNAPSHOT_COMPLETE_FLAG_NAME,
+ "");
+
+ DBCheckpoint checkpoint =
+ provider.getCheckpointFromUntarredDb(candidate.toPath());
+
+ File promoted = checkpoint.getCheckpointLocation().toFile();
+ assertTrue(promoted.getName().startsWith(RECON_OM_SNAPSHOT_DB + "_"),
+ "Promoted DB should be a timestamped snapshot dir");
+ assertEquals(snapshotDir, promoted.getParentFile());
+ assertTrue(new File(promoted, "000010.sst").exists());
+ assertTrue(new File(promoted, "CURRENT").exists());
+ // The completion sentinel must not leak into the DB.
+ assertFalse(new File(promoted,
+ HddsServerUtil.OZONE_RATIS_SNAPSHOT_COMPLETE_FLAG_NAME).exists());
+ // Candidate dir must be emptied so the next sync starts clean.
+ assertEquals(0, candidate.list().length);
+ }
+
+ @Test
+ public void testGetCheckpointInstallsHardLinks(@TempDir File snapshotDir)
+ throws IOException {
+ ReconRDBSnapshotProvider provider = newProvider(snapshotDir);
+ File candidate = provider.getCandidateDir();
+
+ writeFile(candidate, "000001.sst", "shared-content");
+ // hardLinkFile: create 000002.sst as a hard link to 000001.sst.
+ writeFile(candidate, InodeMetadataRocksDBCheckpoint.OM_HARDLINK_FILE,
+ "000002.sst" + HARDLINK_SEPARATOR + "000001.sst" + "\n");
+
+ DBCheckpoint checkpoint =
+ provider.getCheckpointFromUntarredDb(candidate.toPath());
+
+ File promoted = checkpoint.getCheckpointLocation().toFile();
+ File linked = new File(promoted, "000002.sst");
+ assertTrue(linked.exists(), "Hard-linked SST should be materialized");
+ assertEquals("shared-content",
+ FileUtils.readFileToString(linked, UTF_8));
+ assertFalse(new File(promoted,
+ InodeMetadataRocksDBCheckpoint.OM_HARDLINK_FILE).exists(),
+ "hardLinkFile must be consumed, not left in the DB");
+ }
+
+ @Test
+ public void testCandidateDirLocation(@TempDir File snapshotDir) {
+ ReconRDBSnapshotProvider provider = newProvider(snapshotDir);
+ assertEquals(RECON_OM_SNAPSHOT_DB + ".candidate",
+ provider.getCandidateDir().getName());
+ assertEquals(snapshotDir, provider.getCandidateDir().getParentFile());
+ }
+
+ @Test
+ public void testLeaderPinnedForEntireTransfer(@TempDir File snapshotDir)
+ throws IOException {
+ ServiceInfo leaderA = mock(ServiceInfo.class);
+ when(leaderA.getHostname()).thenReturn("om-leader-a");
+ ServiceInfo leaderB = mock(ServiceInfo.class);
+ when(leaderB.getHostname()).thenReturn("om-leader-b");
+
+ // Supplier resolves leaderA first, then leaderB, mimicking an OM failover
+ // during the transfer.
+ AtomicInteger supplierCalls = new AtomicInteger();
+ Supplier<ServiceInfo> supplier = () ->
+ supplierCalls.getAndIncrement() == 0 ? leaderA : leaderB;
+
+ List<String> leadersUsedPerPart = new ArrayList<>();
+
+ ReconRDBSnapshotProvider provider =
+ new ReconRDBSnapshotProvider(snapshotDir, null, false,
+ HttpConfig.Policy.HTTP_ONLY, false, true, supplier) {
+ @Override
+ public void downloadSnapshot(String leaderNodeID, File targetFile)
+ throws IOException {
+ // Mirror the production leader resolution: use the pinned leader,
+ // falling back to the supplier only when nothing is pinned.
+ ServiceInfo leader = getPinnedLeader();
+ if (leader == null) {
+ leader = supplier.get();
+ }
+ leadersUsedPerPart.add(leader.getHostname());
+ // First part is partial; the second carries the completion flag.
+ writeTar(targetFile, leadersUsedPerPart.size() > 1);
+ }
+
+ @Override
+ public DBCheckpoint getCheckpointFromUntarredDb(Path untarredDbDir) {
+ return new RocksDBCheckpoint(untarredDbDir);
+ }
+ };
+
+ provider.downloadDBSnapshotFromLeader("om-leader-a-node-id");
+
+ assertEquals(2, leadersUsedPerPart.size(),
+ "Test should exercise a two-part transfer");
+ assertTrue(leadersUsedPerPart.stream().allMatch("om-leader-a"::equals),
+ "All parts must come from the leader pinned at the start of the "
+ + "transfer, even if the resolved leader changes mid-transfer; "
+ + "leadersUsedPerPart=" + leadersUsedPerPart);
+ assertEquals(1, supplierCalls.get(),
+ "Leader should be resolved exactly once per transfer");
+ assertNull(provider.getPinnedLeader(),
+ "Pinned leader must be cleared after the transfer completes");
+ }
+
+ private void writeTar(File targetFile, boolean complete) throws IOException {
+ try (ArchiveOutputStream<TarArchiveEntry> out =
+ Archiver.tar(Files.newOutputStream(targetFile.toPath()))) {
+ if (complete) {
+ HddsServerUtil.includeRatisSnapshotCompleteFlag(out);
+ } else {
+ File tmp = File.createTempFile("recon-part", ".sst");
+ FileUtils.write(tmp, "partial", UTF_8);
+ Archiver.includeFile(tmp, "000100.sst", out);
+ }
+ }
+ }
+
+ @Test
+ public void testBuildCheckpointUrlHonorsInodeBasedConfig(
+ @TempDir File snapshotDir) throws IOException {
+ ServiceInfo leader = mock(ServiceInfo.class);
+ when(leader.getHostname()).thenReturn("om-host");
+ when(leader.getPort(any())).thenReturn(9874);
+
+ // Inode-based transfer on (default) -> v2 endpoint.
+ URL v2Url = newProvider(snapshotDir, true).buildCheckpointUrl(leader);
+ assertEquals(OZONE_DB_CHECKPOINT_HTTP_ENDPOINT_V2, v2Url.getPath());
+ assertTrue(v2Url.getQuery().contains("includeSnapshotData=false"));
+ assertTrue(v2Url.getQuery().contains("flushBeforeCheckpoint=false"));
+ // Disabled -> fall back to the v1 endpoint.
+ URL v1Url = newProvider(snapshotDir, false).buildCheckpointUrl(leader);
+ assertEquals(OZONE_DB_CHECKPOINT_HTTP_ENDPOINT, v1Url.getPath());
+ assertTrue(v1Url.getQuery().contains("includeSnapshotData=false"));
+ assertTrue(v1Url.getQuery().contains("flushBeforeCheckpoint=false"));
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
