Copilot commented on code in PR #10673:
URL: https://github.com/apache/ozone/pull/10673#discussion_r3530441126
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/protocol/StorageContainerLocationProtocolServerSideTranslatorPB.java:
##########
@@ -1455,4 +1481,69 @@ public SuppressContainerResponseProto
suppressContainer(SuppressContainerRequest
.addAllFailedContainerIDs(failedContainerIDs)
.build();
}
+
+ public SubmitContainerIdExportResponseProto submitContainerIdExport(
+ SubmitContainerIdExportRequestProto request) throws IOException {
+ ContainerID start = ContainerID.valueOf(0);
+ if (request.hasStartContainerID()) {
+ start = ContainerID.valueOf(request.getStartContainerID().getId());
+ }
+ HddsProtos.LifeCycleState lifecycle = null;
+ if (request.hasLifecycleState()) {
+ lifecycle = request.getLifecycleState();
+ }
+ ContainerHealthState health = null;
+ if (request.hasHealthState() && !request.getHealthState().isEmpty()) {
+ health = ContainerHealthState.valueOf(request.getHealthState());
+ }
Review Comment:
`submitContainerIdExport` parses `healthState` via
`ContainerHealthState.valueOf(...)` without handling invalid values, so a bad
string will throw `IllegalArgumentException` rather than an `IOException` with
a clear message. Since this is a public RPC boundary, convert invalid values
into an `IOException` (similar to other request validation).
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/protocol/StorageContainerLocationProtocolServerSideTranslatorPB.java:
##########
@@ -1436,11 +1457,16 @@ public SCMListContainerIDsResponseProto
listContainerIDs(
state = request.getState();
}
+ ContainerHealthState healthState = null;
+ if (request.hasHealthState()) {
+ healthState = ContainerHealthState.valueOf(request.getHealthState());
+ }
Review Comment:
SCMListContainerIDsRequestProto.healthState is an optional string;
`hasHealthState()` can be true even when the value is empty, and
`ContainerHealthState.valueOf(request.getHealthState())` will throw
`IllegalArgumentException` (bypassing the method's declared `IOException`).
This can break the ListContainerIDs RPC for malformed/empty inputs. Guard
against empty strings and wrap invalid values as `IOException`.
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportManager.java:
##########
@@ -0,0 +1,500 @@
+/*
+ * 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.hdds.scm.container.export;
+
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.time.Instant;
+import java.time.ZoneOffset;
+import java.time.format.DateTimeFormatter;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.apache.commons.io.FileUtils;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos.LifeCycleState;
+import org.apache.hadoop.hdds.scm.container.ContainerHealthState;
+import org.apache.hadoop.hdds.scm.container.ContainerID;
+import org.apache.hadoop.hdds.scm.container.ContainerManager;
+import org.apache.hadoop.hdds.server.ServerUtils;
+import org.apache.hadoop.hdds.utils.Archiver;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Manages asynchronous container ID export jobs on SCM leader.
+ */
+public class ContainerExportManager {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(ContainerExportManager.class);
+
+ private static final DateTimeFormatter METADATA_TIMESTAMP_FORMAT =
+
DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss'Z'").withZone(ZoneOffset.UTC);
+ private static final DateTimeFormatter FILENAME_TIMESTAMP_FORMAT =
+
DateTimeFormatter.ofPattern("yyyyMMdd'T'HHmmss'Z'").withZone(ZoneOffset.UTC);
+
+ private static final String EXPORT_SUBDIR = "exports";
+ private static final int DEFAULT_SHARD_SIZE = 500_000;
+ private static final int DEFAULT_PAGE_SIZE = 100_000;
+ private static final int MAX_SHARD_SIZE = 5_000_000;
+ private static final int MAX_PAGE_SIZE = 1_000_000;
+
+ private final Map<String, ExportJob> jobTracker = new ConcurrentHashMap<>();
+ private final ExecutorService workerPool;
+ private final Map<String, Future<?>> runningTasks = new
ConcurrentHashMap<>();
+ private final ContainerManager containerManager;
+ private final String exportDirectory;
+ private final int defaultShardSize;
+ private final int defaultPageSize;
+ private final Object submitLock = new Object();
+
+ public ContainerExportManager(ContainerManager containerManager,
+ OzoneConfiguration conf) {
+ this(containerManager, resolveExportDirectory(conf),
+ DEFAULT_SHARD_SIZE, DEFAULT_PAGE_SIZE);
+ }
+
+ ContainerExportManager(ContainerManager containerManager,
+ String exportDirectory, int defaultShardSize, int defaultPageSize) {
+ this.containerManager = containerManager;
+ this.exportDirectory = exportDirectory;
+ this.defaultShardSize = defaultShardSize;
+ this.defaultPageSize = defaultPageSize;
+ this.workerPool = Executors.newSingleThreadExecutor(r -> {
+ Thread t = new Thread(r, "ContainerExportWorker");
+ t.setDaemon(true);
+ return t;
+ });
+
+ try {
+ Files.createDirectories(Paths.get(exportDirectory));
+ } catch (IOException e) {
+ LOG.error("Failed to create export directory: {}", exportDirectory, e);
+ }
+ LOG.info("ContainerExportManager initialized (dir={}, defaultShardSize={},
defaultPageSize={})",
+ exportDirectory, defaultShardSize, defaultPageSize);
+ }
+
+ private static String resolveExportDirectory(OzoneConfiguration conf) {
+ File scmDbDir = ServerUtils.getScmDbDir(conf);
+ return new File(scmDbDir, EXPORT_SUBDIR).getAbsolutePath();
+ }
+
+ /**
+ * Submit a container ID export job.
+ *
+ * @param start optional inclusive start container ID (0 for beginning)
+ * @param lifeCycleState optional lifecycle filter
+ * @param healthState optional health filter
+ * @param maxRows optional row limit (0 = unlimited)
+ * @param pageSize IDs fetched per SCM read (0 = manager default)
+ * @param shardSize IDs per TAR entry (0 = manager default)
+ * @return job id
+ */
+ public String submitJob(ContainerID start, LifeCycleState lifeCycleState,
+ ContainerHealthState healthState, long maxRows, int pageSize, int
shardSize) {
+ if (lifeCycleState == null && healthState == null) {
+ throw new IllegalArgumentException("At least one of healthState or
lifecycleState filter is required.");
+ }
+ validateRequest(start, maxRows, pageSize, shardSize);
+ int resolvedPageSize = pageSize > 0 ? pageSize : defaultPageSize;
+ int resolvedShardSize = shardSize > 0 ? shardSize : defaultShardSize;
+
+ String jobId = UUID.randomUUID().toString();
+ String scope = buildScope(lifeCycleState, healthState);
+ Instant now = Instant.now();
+ String metadataTimestamp = METADATA_TIMESTAMP_FORMAT.format(now);
+ String fileTimestamp = FILENAME_TIMESTAMP_FORMAT.format(now);
+ String tarFileName = String.format("container-ids-%s-%s-%s.tar", scope,
fileTimestamp, jobId);
+ String tarPath = exportDirectory + File.separator + tarFileName;
+
+ ExportJob job = new ExportJob(jobId, scope, metadataTimestamp, tarPath,
+ start, lifeCycleState, healthState, maxRows, resolvedPageSize,
resolvedShardSize);
+
+ synchronized (submitLock) {
+ boolean exportInProgress = jobTracker.values().stream()
+ .anyMatch(j -> j.getState() == ContainerExportStatus.State.RUNNING);
+ if (exportInProgress) {
+ throw new IllegalStateException("Another container ID export is
already running.");
+ }
+ jobTracker.put(jobId, job);
+ }
+
+ Future<?> future = workerPool.submit(() -> executeExport(job));
+ runningTasks.put(jobId, future);
+ LOG.info("Submitted container ID export job {} (scope={}, start={},
maxRows={}, pageSize={}, shardSize={})",
+ jobId, scope, start, maxRows, resolvedPageSize, resolvedShardSize);
+ return jobId;
+ }
+
+ private static void validateRequest(ContainerID start, long maxRows, int
pageSize, int shardSize) {
+ if (start != null && start.getProtobuf().getId() < 0) {
+ throw new IllegalArgumentException("start container ID must be
non-negative.");
+ }
+ if (maxRows < 0) {
+ throw new IllegalArgumentException("count must be non-negative.");
+ }
Review Comment:
The validation error message for `maxRows < 0` says "count must be
non-negative", but the parameter name and API surface use `maxRows`. Aligning
the message makes failures easier to understand when surfaced through the
RPC/CLI.
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportManager.java:
##########
@@ -0,0 +1,500 @@
+/*
+ * 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.hdds.scm.container.export;
+
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.time.Instant;
+import java.time.ZoneOffset;
+import java.time.format.DateTimeFormatter;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.apache.commons.io.FileUtils;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos.LifeCycleState;
+import org.apache.hadoop.hdds.scm.container.ContainerHealthState;
+import org.apache.hadoop.hdds.scm.container.ContainerID;
+import org.apache.hadoop.hdds.scm.container.ContainerManager;
+import org.apache.hadoop.hdds.server.ServerUtils;
+import org.apache.hadoop.hdds.utils.Archiver;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Manages asynchronous container ID export jobs on SCM leader.
+ */
+public class ContainerExportManager {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(ContainerExportManager.class);
+
+ private static final DateTimeFormatter METADATA_TIMESTAMP_FORMAT =
+
DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss'Z'").withZone(ZoneOffset.UTC);
+ private static final DateTimeFormatter FILENAME_TIMESTAMP_FORMAT =
+
DateTimeFormatter.ofPattern("yyyyMMdd'T'HHmmss'Z'").withZone(ZoneOffset.UTC);
+
+ private static final String EXPORT_SUBDIR = "exports";
+ private static final int DEFAULT_SHARD_SIZE = 500_000;
+ private static final int DEFAULT_PAGE_SIZE = 100_000;
+ private static final int MAX_SHARD_SIZE = 5_000_000;
+ private static final int MAX_PAGE_SIZE = 1_000_000;
+
+ private final Map<String, ExportJob> jobTracker = new ConcurrentHashMap<>();
+ private final ExecutorService workerPool;
+ private final Map<String, Future<?>> runningTasks = new
ConcurrentHashMap<>();
+ private final ContainerManager containerManager;
+ private final String exportDirectory;
+ private final int defaultShardSize;
+ private final int defaultPageSize;
+ private final Object submitLock = new Object();
+
+ public ContainerExportManager(ContainerManager containerManager,
+ OzoneConfiguration conf) {
+ this(containerManager, resolveExportDirectory(conf),
+ DEFAULT_SHARD_SIZE, DEFAULT_PAGE_SIZE);
+ }
+
+ ContainerExportManager(ContainerManager containerManager,
+ String exportDirectory, int defaultShardSize, int defaultPageSize) {
+ this.containerManager = containerManager;
+ this.exportDirectory = exportDirectory;
+ this.defaultShardSize = defaultShardSize;
+ this.defaultPageSize = defaultPageSize;
+ this.workerPool = Executors.newSingleThreadExecutor(r -> {
+ Thread t = new Thread(r, "ContainerExportWorker");
+ t.setDaemon(true);
+ return t;
+ });
+
+ try {
+ Files.createDirectories(Paths.get(exportDirectory));
+ } catch (IOException e) {
+ LOG.error("Failed to create export directory: {}", exportDirectory, e);
+ }
+ LOG.info("ContainerExportManager initialized (dir={}, defaultShardSize={},
defaultPageSize={})",
+ exportDirectory, defaultShardSize, defaultPageSize);
+ }
+
+ private static String resolveExportDirectory(OzoneConfiguration conf) {
+ File scmDbDir = ServerUtils.getScmDbDir(conf);
+ return new File(scmDbDir, EXPORT_SUBDIR).getAbsolutePath();
+ }
Review Comment:
`resolveExportDirectory` always uses `{scm.db.dirs}/exports` and ignores the
`ozone.scm.container.export.dir` config that is being set in the compose
`docker-config` in this PR. This makes the new config knob ineffective and can
surprise operators. Consider honoring the explicit config value with a fallback
to the current default.
##########
hadoop-ozone/cli-debug/src/main/java/org/apache/hadoop/ozone/debug/scm/container/ContainerSubCommand.java:
##########
@@ -0,0 +1,36 @@
+/*
+ * 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.debug.scm.container;
+
+import org.apache.hadoop.hdds.cli.AbstractSubcommand;
+import org.apache.hadoop.hdds.cli.HddsVersionProvider;
+import picocli.CommandLine;
+
+/**
+ * Container debug related commands.
+ */
[email protected](
+ name = "container",
+ description = "Container debug commands.",
+ mixinStandardHelpOptions = true,
+ versionProvider = HddsVersionProvider.class,
Review Comment:
There is trailing whitespace after `HddsVersionProvider.class,` which can
trip formatting/checkstyle checks. Remove the trailing spaces.
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportManager.java:
##########
@@ -0,0 +1,500 @@
+/*
+ * 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.hdds.scm.container.export;
+
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.time.Instant;
+import java.time.ZoneOffset;
+import java.time.format.DateTimeFormatter;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.apache.commons.io.FileUtils;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos.LifeCycleState;
+import org.apache.hadoop.hdds.scm.container.ContainerHealthState;
+import org.apache.hadoop.hdds.scm.container.ContainerID;
+import org.apache.hadoop.hdds.scm.container.ContainerManager;
+import org.apache.hadoop.hdds.server.ServerUtils;
+import org.apache.hadoop.hdds.utils.Archiver;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Manages asynchronous container ID export jobs on SCM leader.
+ */
+public class ContainerExportManager {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(ContainerExportManager.class);
+
+ private static final DateTimeFormatter METADATA_TIMESTAMP_FORMAT =
+
DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss'Z'").withZone(ZoneOffset.UTC);
+ private static final DateTimeFormatter FILENAME_TIMESTAMP_FORMAT =
+
DateTimeFormatter.ofPattern("yyyyMMdd'T'HHmmss'Z'").withZone(ZoneOffset.UTC);
+
+ private static final String EXPORT_SUBDIR = "exports";
+ private static final int DEFAULT_SHARD_SIZE = 500_000;
+ private static final int DEFAULT_PAGE_SIZE = 100_000;
+ private static final int MAX_SHARD_SIZE = 5_000_000;
+ private static final int MAX_PAGE_SIZE = 1_000_000;
+
+ private final Map<String, ExportJob> jobTracker = new ConcurrentHashMap<>();
+ private final ExecutorService workerPool;
+ private final Map<String, Future<?>> runningTasks = new
ConcurrentHashMap<>();
+ private final ContainerManager containerManager;
+ private final String exportDirectory;
+ private final int defaultShardSize;
+ private final int defaultPageSize;
+ private final Object submitLock = new Object();
+
+ public ContainerExportManager(ContainerManager containerManager,
+ OzoneConfiguration conf) {
+ this(containerManager, resolveExportDirectory(conf),
+ DEFAULT_SHARD_SIZE, DEFAULT_PAGE_SIZE);
+ }
+
+ ContainerExportManager(ContainerManager containerManager,
+ String exportDirectory, int defaultShardSize, int defaultPageSize) {
+ this.containerManager = containerManager;
+ this.exportDirectory = exportDirectory;
+ this.defaultShardSize = defaultShardSize;
+ this.defaultPageSize = defaultPageSize;
+ this.workerPool = Executors.newSingleThreadExecutor(r -> {
+ Thread t = new Thread(r, "ContainerExportWorker");
+ t.setDaemon(true);
+ return t;
+ });
+
+ try {
+ Files.createDirectories(Paths.get(exportDirectory));
+ } catch (IOException e) {
+ LOG.error("Failed to create export directory: {}", exportDirectory, e);
+ }
+ LOG.info("ContainerExportManager initialized (dir={}, defaultShardSize={},
defaultPageSize={})",
+ exportDirectory, defaultShardSize, defaultPageSize);
+ }
+
+ private static String resolveExportDirectory(OzoneConfiguration conf) {
+ File scmDbDir = ServerUtils.getScmDbDir(conf);
+ return new File(scmDbDir, EXPORT_SUBDIR).getAbsolutePath();
+ }
+
+ /**
+ * Submit a container ID export job.
+ *
+ * @param start optional inclusive start container ID (0 for beginning)
+ * @param lifeCycleState optional lifecycle filter
+ * @param healthState optional health filter
+ * @param maxRows optional row limit (0 = unlimited)
+ * @param pageSize IDs fetched per SCM read (0 = manager default)
+ * @param shardSize IDs per TAR entry (0 = manager default)
+ * @return job id
+ */
+ public String submitJob(ContainerID start, LifeCycleState lifeCycleState,
+ ContainerHealthState healthState, long maxRows, int pageSize, int
shardSize) {
+ if (lifeCycleState == null && healthState == null) {
+ throw new IllegalArgumentException("At least one of healthState or
lifecycleState filter is required.");
+ }
+ validateRequest(start, maxRows, pageSize, shardSize);
+ int resolvedPageSize = pageSize > 0 ? pageSize : defaultPageSize;
+ int resolvedShardSize = shardSize > 0 ? shardSize : defaultShardSize;
+
+ String jobId = UUID.randomUUID().toString();
+ String scope = buildScope(lifeCycleState, healthState);
+ Instant now = Instant.now();
+ String metadataTimestamp = METADATA_TIMESTAMP_FORMAT.format(now);
+ String fileTimestamp = FILENAME_TIMESTAMP_FORMAT.format(now);
+ String tarFileName = String.format("container-ids-%s-%s-%s.tar", scope,
fileTimestamp, jobId);
+ String tarPath = exportDirectory + File.separator + tarFileName;
+
+ ExportJob job = new ExportJob(jobId, scope, metadataTimestamp, tarPath,
+ start, lifeCycleState, healthState, maxRows, resolvedPageSize,
resolvedShardSize);
+
+ synchronized (submitLock) {
+ boolean exportInProgress = jobTracker.values().stream()
+ .anyMatch(j -> j.getState() == ContainerExportStatus.State.RUNNING);
+ if (exportInProgress) {
+ throw new IllegalStateException("Another container ID export is
already running.");
+ }
+ jobTracker.put(jobId, job);
+ }
+
+ Future<?> future = workerPool.submit(() -> executeExport(job));
+ runningTasks.put(jobId, future);
+ LOG.info("Submitted container ID export job {} (scope={}, start={},
maxRows={}, pageSize={}, shardSize={})",
+ jobId, scope, start, maxRows, resolvedPageSize, resolvedShardSize);
+ return jobId;
+ }
+
+ private static void validateRequest(ContainerID start, long maxRows, int
pageSize, int shardSize) {
+ if (start != null && start.getProtobuf().getId() < 0) {
+ throw new IllegalArgumentException("start container ID must be
non-negative.");
+ }
+ if (maxRows < 0) {
+ throw new IllegalArgumentException("count must be non-negative.");
+ }
+ if (pageSize < 0 || pageSize > MAX_PAGE_SIZE) {
+ throw new IllegalArgumentException("pageSize must be between 0 and " +
MAX_PAGE_SIZE + ".");
+ }
+ if (shardSize < 0 || shardSize > MAX_SHARD_SIZE) {
+ throw new IllegalArgumentException("shardSize must be between 0 and " +
MAX_SHARD_SIZE + ".");
+ }
+ }
+
+ public ContainerExportStatus getJobStatus(String jobId) {
+ ExportJob job = jobTracker.get(jobId);
+ if (job == null) {
+ return null;
+ }
+ return job.toStatus();
+ }
+
+ public void shutdown() {
+ LOG.info("Shutting down ContainerExportManager");
+ workerPool.shutdownNow();
+ try {
+ workerPool.awaitTermination(30, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ LOG.warn("Timeout waiting for export worker shutdown", e);
+ Thread.currentThread().interrupt();
+ }
+ runningTasks.clear();
+ }
+
+ Map<String, ExportJob> getJobTracker() {
+ return jobTracker;
+ }
+
+ private void executeExport(ExportJob job) {
+ Path jobDir = Paths.get(exportDirectory, job.getJobId());
+ Path workDir = jobDir.resolve("work");
+ File tarFile = new File(job.getTarPath());
+ long startTimeMs = System.currentTimeMillis();
+ job.setStartTimeMs(startTimeMs);
+
+ try {
+ Files.createDirectories(workDir);
+ job.setState(ContainerExportStatus.State.RUNNING);
+
+ ContainerID cursor = job.getStartContainerId();
+ int pageSize = job.getPageSize();
+ int shardSize = job.getShardSize();
+ int fileIndex = 1;
+ long totalRows = 0;
+ long recordsInCurrentFile = 0;
+ BufferedWriter writer = null;
+ Path currentShardPath = null;
+ // Pre-allocated buffer: ~12 chars per ID (up to 20 digits + newline)
per page.
+ StringBuilder buf = new StringBuilder(pageSize * 12);
+
+ try {
+ while (true) {
+ if (Thread.currentThread().isInterrupted()) {
+ throw new InterruptedException("Job cancelled");
+ }
+
+ int fetchCount = pageSize;
+ if (job.getMaxRows() > 0) {
+ long remaining = job.getMaxRows() - totalRows;
+ if (remaining <= 0) {
+ break;
+ }
+ fetchCount = (int) Math.min(fetchCount, remaining);
+ }
+
+ List<ContainerID> page = containerManager.getContainerIDs(
+ cursor, fetchCount, job.getLifeCycleState(),
job.getHealthState());
+ if (page.isEmpty()) {
+ break;
+ }
+
+ for (ContainerID containerId : page) {
+ if (recordsInCurrentFile == 0) {
+ writer = closeWriter(writer);
+ currentShardPath = workDir.resolve(shardFileName(job,
fileIndex));
+ writer = Files.newBufferedWriter(currentShardPath,
StandardCharsets.UTF_8);
+ writeMetadataHeader(writer, job, fileIndex, containerId);
+ LOG.info("Export job {} created shard part{}", job.getJobId(),
fileIndex);
+ }
+
+ buf.append(containerId.getProtobuf().getId()).append('\n');
+ totalRows++;
+ recordsInCurrentFile++;
+ job.setTotalRows(totalRows);
+
+ if (recordsInCurrentFile >= shardSize) {
+ writer.write(buf.toString());
+ buf.setLength(0);
+ writer = closeWriter(writer);
+ appendShardToTar(tarFile, currentShardPath, shardEntryName(job,
fileIndex));
+ currentShardPath = null;
+ recordsInCurrentFile = 0;
+ fileIndex++;
+ }
+ }
+
+ // Flush the batch buffer at the end of each page.
+ if (buf.length() > 0 && writer != null) {
+ writer.write(buf.toString());
+ buf.setLength(0);
+ }
+
+ cursor = ContainerID.valueOf(
+ page.get(page.size() - 1).getProtobuf().getId() + 1);
+ }
+
+ writer = closeWriter(writer);
+ if (totalRows == 0) {
+ FileUtils.deleteQuietly(workDir.toFile());
+ FileUtils.deleteQuietly(jobDir.toFile());
+ job.setState(ContainerExportStatus.State.SUCCEEDED);
+ job.setTarPath(null);
+ LOG.info("Export job {} completed with zero matching containers",
job.getJobId());
+ return;
+ }
+
+ if (currentShardPath != null) {
+ appendShardToTar(tarFile, currentShardPath, shardEntryName(job,
fileIndex));
+ }
+
+ FileUtils.deleteQuietly(workDir.toFile());
+ FileUtils.deleteQuietly(jobDir.toFile());
+ job.setState(ContainerExportStatus.State.SUCCEEDED);
+ LOG.info("Export job {} completed ({} rows, tar={}). "
+ + "Delete the TAR file manually on the SCM leader when no
longer needed.",
+ job.getJobId(), totalRows, tarFile.getAbsolutePath());
+ } finally {
+ closeWriter(writer);
+ }
+ } catch (InterruptedException e) {
+ job.setState(ContainerExportStatus.State.FAILED);
+ job.setErrorMessage("Job was cancelled");
+ cleanupFailedArtifacts(jobDir, tarFile);
+ LOG.info("Export job {} was cancelled", job.getJobId());
+ Thread.currentThread().interrupt();
+ } catch (IOException | RuntimeException e) {
+ job.setState(ContainerExportStatus.State.FAILED);
+ job.setErrorMessage(e.getMessage() != null ? e.getMessage() :
e.toString());
+ cleanupFailedArtifacts(jobDir, tarFile);
+ LOG.error("Export job {} failed", job.getJobId(), e);
+ } finally {
+ runningTasks.remove(job.getJobId());
+ }
+ }
+
+ private static String shardFileName(ExportJob job, int partIndex) {
+ return String.format("container-ids-%s-%s-part%03d.txt",
+ job.getScope(), job.getTimestamp(), partIndex);
+ }
+
+ private static String shardEntryName(ExportJob job, int partIndex) {
+ return shardFileName(job, partIndex);
+ }
+
+ private static void appendShardToTar(File tarFile, Path shardPath, String
entryName)
+ throws IOException {
+ Archiver.appendFile(tarFile, shardPath.toFile(), entryName);
+ FileUtils.deleteQuietly(shardPath.toFile());
+ }
+
+ /** Remove partial work artifacts after a failed or cancelled export. */
+ private void cleanupFailedArtifacts(Path jobDir, File tarFile) {
+ if (jobDir != null) {
+ FileUtils.deleteQuietly(jobDir.toFile());
+ }
+ if (tarFile != null) {
+ FileUtils.deleteQuietly(tarFile);
+ }
+ }
+
+ private static BufferedWriter closeWriter(BufferedWriter writer) throws
IOException {
+ if (writer != null) {
+ writer.flush();
+ writer.close();
+ }
+ return null;
+ }
+
+ private static void writeMetadataHeader(BufferedWriter writer, ExportJob job,
+ int partNumber, ContainerID shardStartContainerId) throws IOException {
+ writer.newLine();
+ writer.write("# jobId=" + job.getJobId());
+ writer.newLine();
+ writer.write("# timestamp=" + job.getTimestamp());
Review Comment:
`writeMetadataHeader` writes an initial blank line before `# jobId=...`. The
PR description examples show the shard content starting directly with the `#
jobId` header. Dropping the leading newline keeps shard files consistent with
the documented format and avoids an empty first line in each part file.
##########
hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/Archiver.java:
##########
@@ -61,6 +64,51 @@ public static void create(File tarFile, Path from) throws
IOException {
}
}
+ /**
+ * Append a single file as a new entry to an existing tarball, or create the
+ * tarball if it does not exist yet.
+ */
+ public static void appendFile(File tarFile, File file, String entryName)
+ throws IOException {
+ if (tarFile.exists() && tarFile.length() > 0) {
+ stripTarEofBlocks(tarFile);
+ }
+ OpenOption[] options = tarFile.exists() && tarFile.length() > 0
+ ? new OpenOption[] {StandardOpenOption.WRITE,
StandardOpenOption.APPEND}
+ : new OpenOption[] {StandardOpenOption.WRITE,
StandardOpenOption.CREATE};
+ try (OutputStream fos = Files.newOutputStream(tarFile.toPath(), options);
+ ArchiveOutputStream<TarArchiveEntry> out = simpleTar(fos)) {
+ includeSimpleFile(file, entryName, out);
+ out.finish();
+ }
+ }
+
+ /**
+ * Remove trailing zero blocks so new entries can be appended to a tarball.
+ */
+ private static void stripTarEofBlocks(File tarFile) throws IOException {
+ try (RandomAccessFile raf = new RandomAccessFile(tarFile, "rw")) {
+ long size = raf.length();
+ while (size >= TarConstants.DEFAULT_RCDSIZE) {
+ raf.seek(size - TarConstants.DEFAULT_RCDSIZE);
+ byte[] block = new byte[TarConstants.DEFAULT_RCDSIZE];
+ raf.readFully(block);
+ boolean allZero = true;
Review Comment:
`stripTarEofBlocks` truncates *all* trailing 512-byte blocks that are
entirely zero. In tar archives it's possible (especially for binary files) for
the last entry's data to legitimately end with one or more all-zero 512-byte
blocks, which would be truncated here and corrupt the archive when appending.
Appending safely generally requires locating the actual end-of-archive marker
by parsing the tar (or rewriting to a new tar) rather than relying on
trailing-zero heuristics.
##########
hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportStatus.java:
##########
@@ -0,0 +1,91 @@
+/*
+ * 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.hdds.scm.container.export;
+
+/**
+ * Client-side view of a container ID export job on SCM.
+ */
+public final class ContainerExportStatus {
+
+ private final String jobId;
+ private final State state;
+ private final String lifecycleState;
+ private final String healthState;
+ private final long totalRows;
+ private final long elapsedMs;
+ private final String tarPath;
+ private final String errorMessage;
+
+ /**
+ * States of export job.
+ */
+ public enum State {
+ RUNNING,
+ SUCCEEDED,
+ FAILED
+ }
+
+ @SuppressWarnings("checkstyle:ParameterNumber")
+ public ContainerExportStatus(String jobId, State state, String
lifecycleState, String healthState,
+ long totalRows, long elapsedMs, String tarPath, String errorMessage) {
Review Comment:
There is trailing whitespace at the end of the constructor signature line,
which can trigger checkstyle/formatting checks. Remove the trailing space and
keep the wrap consistent.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
