From 2a9ece729c62cb937e985efa72febfec2d3011b2 Mon Sep 17 00:00:00 2001 From: sarvekshayr Date: Mon, 20 Jul 2026 10:43:35 +0530 Subject: [PATCH 1/5] HDDS-15839. Implement SCM container ID export manager --- .../export/ContainerExportLimits.java | 34 + .../export/ContainerExportStatus.java | 95 +++ .../scm/container/export/package-info.java | 21 + .../export/ContainerExportManager.java | 647 ++++++++++++++++++ .../export/ContainerExportMetrics.java | 80 +++ .../scm/container/export/package-info.java | 21 + .../scm/server/StorageContainerManager.java | 12 + .../export/TestContainerExportManager.java | 326 +++++++++ 8 files changed, 1236 insertions(+) create mode 100644 hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportLimits.java create mode 100644 hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportStatus.java create mode 100644 hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/export/package-info.java create mode 100644 hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportManager.java create mode 100644 hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportMetrics.java create mode 100644 hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/package-info.java create mode 100644 hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/export/TestContainerExportManager.java diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportLimits.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportLimits.java new file mode 100644 index 000000000000..d78f86873786 --- /dev/null +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportLimits.java @@ -0,0 +1,34 @@ +/* + * 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; + +/** + * Shared limits and defaults for container ID export. + */ +public final class ContainerExportLimits { + + public static final String EXPORT_SUBDIR = "exports"; + + public static final int DEFAULT_PAGE_SIZE = 100_000; + public static final int DEFAULT_SHARD_SIZE = 500_000; + public static final int MAX_PAGE_SIZE = 1_000_000; + public static final int MAX_SHARD_SIZE = 5_000_000; + + private ContainerExportLimits() { + } +} diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportStatus.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportStatus.java new file mode 100644 index 000000000000..6ed1d48072d1 --- /dev/null +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportStatus.java @@ -0,0 +1,95 @@ +/* + * 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 org.apache.hadoop.hdds.protocol.proto.HddsProtos.LifeCycleState; +import org.apache.hadoop.hdds.scm.container.ContainerHealthState; + +/** + * Status of a container ID export job on the SCM leader. + */ +public final class ContainerExportStatus { + + private final String jobId; + private final State state; + private final LifeCycleState lifecycleState; + private final ContainerHealthState 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, LifeCycleState lifecycleState, + ContainerHealthState healthState, long totalRows, long elapsedMs, String tarPath, + String errorMessage) { + this.jobId = jobId; + this.state = state; + this.lifecycleState = lifecycleState; + this.healthState = healthState; + this.totalRows = totalRows; + this.elapsedMs = elapsedMs; + this.tarPath = tarPath; + this.errorMessage = errorMessage; + } + + public String getJobId() { + return jobId; + } + + public State getState() { + return state; + } + + public LifeCycleState getLifecycleState() { + return lifecycleState; + } + + public ContainerHealthState getHealthState() { + return healthState; + } + + public long getTotalRows() { + return totalRows; + } + + public long getElapsedMs() { + return elapsedMs; + } + + public String getTarPath() { + return tarPath; + } + + public String getErrorMessage() { + return errorMessage; + } + + public boolean isTerminal() { + return state == State.SUCCEEDED || state == State.FAILED; + } +} diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/export/package-info.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/export/package-info.java new file mode 100644 index 000000000000..103c9519fcab --- /dev/null +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/export/package-info.java @@ -0,0 +1,21 @@ +/* + * 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. + */ + +/** + * This package contains classes related to container export. + */ +package org.apache.hadoop.hdds.scm.container.export; diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportManager.java new file mode 100644 index 000000000000..ce8f604dade0 --- /dev/null +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportManager.java @@ -0,0 +1,647 @@ +/* + * 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 static org.apache.hadoop.hdds.scm.container.export.ContainerExportLimits.DEFAULT_PAGE_SIZE; +import static org.apache.hadoop.hdds.scm.container.export.ContainerExportLimits.DEFAULT_SHARD_SIZE; +import static org.apache.hadoop.hdds.scm.container.export.ContainerExportLimits.EXPORT_SUBDIR; +import static org.apache.hadoop.hdds.scm.container.export.ContainerExportLimits.MAX_PAGE_SIZE; +import static org.apache.hadoop.hdds.scm.container.export.ContainerExportLimits.MAX_SHARD_SIZE; + +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.Comparator; +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 java.util.function.BooleanSupplier; +import java.util.stream.Collectors; +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 the SCM leader. + * + *

Health filters read {@link org.apache.hadoop.hdds.scm.container.ContainerInfo#getHealthState()} + * as last written by Replication Manager; they are not recomputed during export and may be stale + * if RM has not yet evaluated a container. + * + *

Job status is kept in memory only. On SCM restart or leader failover, in-flight jobs are lost + * and the operator must re-submit on the new leader. Completed TAR files remain on disk until + * manually deleted. On startup, incomplete work ({@code {jobId}.inprogress} markers, UUID work + * directories, and matching partial TAR files) is removed. + */ +public class ContainerExportManager { + + private static final Logger LOG = LoggerFactory.getLogger(ContainerExportManager.class); + + private static final String INPROGRESS_MARKER_SUFFIX = ".inprogress"; + + //TODO: make ozone.scm.container.export.max.terminal.jobs configurable. + private static final int DEFAULT_MAX_TERMINAL_JOBS = 10; + + 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 final Map jobTracker = new ConcurrentHashMap<>(); + private final ExecutorService workerPool; + private final Map> runningTasks = new ConcurrentHashMap<>(); + private final ContainerManager containerManager; + private final ContainerExportMetrics metrics; + private final BooleanSupplier isLeaderReady; + private final String exportDirectory; + private final int defaultShardSize; + private final int defaultPageSize; + private final int maxTerminalJobs; + private final Object submitLock = new Object(); + + public ContainerExportManager(ContainerManager containerManager, + OzoneConfiguration conf, BooleanSupplier isLeaderReady) { + this(containerManager, resolveExportDirectory(conf), DEFAULT_SHARD_SIZE, DEFAULT_PAGE_SIZE, + DEFAULT_MAX_TERMINAL_JOBS, isLeaderReady, ContainerExportMetrics.create()); + } + + ContainerExportManager(ContainerManager containerManager, String exportDirectory, + int defaultShardSize, int defaultPageSize, int maxTerminalJobs, + BooleanSupplier isLeaderReady, ContainerExportMetrics metrics) { + this.containerManager = containerManager; + this.exportDirectory = exportDirectory; + this.defaultShardSize = defaultShardSize; + this.defaultPageSize = defaultPageSize; + this.maxTerminalJobs = maxTerminalJobs; + this.isLeaderReady = isLeaderReady; + this.metrics = metrics; + this.workerPool = Executors.newSingleThreadExecutor(r -> { + Thread t = new Thread(r, "ContainerExportWorker"); + t.setDaemon(true); + return t; + }); + + try { + Files.createDirectories(Paths.get(exportDirectory)); + cleanupOrphanedExportArtifacts(); + } catch (IOException e) { + LOG.error("Failed to initialize export directory: {}", exportDirectory, e); + } + LOG.info("ContainerExportManager initialized (dir={}, defaultShardSize={}, defaultPageSize={}, maxTerminalJobs={})", + exportDirectory, defaultShardSize, defaultPageSize, maxTerminalJobs); + } + + // TODO: make ozone.scm.container.export.dir configurable. + private static String resolveExportDirectory(OzoneConfiguration conf) { + File scmDbDir = ServerUtils.getScmDbDir(conf); + return new File(scmDbDir, EXPORT_SUBDIR).getAbsolutePath(); + } + + /** + * Submit a container ID export job on the SCM leader. + */ + public String submitJob(ContainerID start, LifeCycleState lifeCycleState, + ContainerHealthState healthState, long maxRows, int pageSize, int shardSize) { + if (!isLeaderReady.getAsBoolean()) { + throw new IllegalStateException( + "Container ID export must be submitted on the SCM leader."); + } + 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."); + } + evictOldTerminalJobs(); + jobTracker.put(jobId, job); + } + + if (metrics != null) { + metrics.incrExportJobsSubmitted(); + } + + 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("maxRows 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); + return job != null ? job.toStatus() : null; + } + + 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(); + if (metrics != null) { + metrics.unRegister(); + } + } + + Map getJobTracker() { + return jobTracker; + } + + private void evictOldTerminalJobs() { + List> terminalJobs = jobTracker.entrySet().stream() + .filter(e -> e.getValue().getState() != ContainerExportStatus.State.RUNNING) + .sorted(Comparator.comparingLong(e -> e.getValue().getEndTimeMs())) + .collect(Collectors.toList()); + int excess = terminalJobs.size() - maxTerminalJobs; + for (int i = 0; i < excess; i++) { + jobTracker.remove(terminalJobs.get(i).getKey()); + } + } + + private void cleanupOrphanedExportArtifacts() throws IOException { + File exportDir = new File(exportDirectory); + File[] children = exportDir.listFiles(); + if (children == null) { + return; + } + for (File child : children) { + if (child.isFile() && child.getName().endsWith(INPROGRESS_MARKER_SUFFIX)) { + String jobId = child.getName().substring( + 0, child.getName().length() - INPROGRESS_MARKER_SUFFIX.length()); + if (isUuidDirectoryName(jobId)) { + removeIncompleteExportArtifacts(jobId); + } + } + } + for (File child : children) { + if (child.isDirectory() && isUuidDirectoryName(child.getName())) { + removeIncompleteExportArtifacts(child.getName()); + } + } + } + + private void removeIncompleteExportArtifacts(String jobId) { + LOG.info("Removing incomplete container export artifacts for job {}", jobId); + FileUtils.deleteQuietly(inprogressMarkerFile(jobId)); + File tar = findTarForJobId(jobId); + if (tar != null) { + FileUtils.deleteQuietly(tar); + LOG.info("Removed incomplete container export TAR for job {}: {}", jobId, tar.getName()); + } + File jobWorkDir = new File(exportDirectory, jobId); + if (jobWorkDir.isDirectory()) { + FileUtils.deleteQuietly(jobWorkDir); + LOG.info("Removed orphaned container export work directory: {}", + jobWorkDir.getAbsolutePath()); + } + } + + private File findTarForJobId(String jobId) { + File exportDir = new File(exportDirectory); + File[] matches = exportDir.listFiles( + (dir, name) -> name.endsWith("-" + jobId + ".tar")); + if (matches == null || matches.length == 0) { + return null; + } + return matches[0]; + } + + private File inprogressMarkerFile(String jobId) { + return new File(exportDirectory, jobId + INPROGRESS_MARKER_SUFFIX); + } + + private void markExportInProgress(String jobId) throws IOException { + Files.createFile(inprogressMarkerFile(jobId).toPath()); + } + + private void clearExportInProgress(String jobId) { + FileUtils.deleteQuietly(inprogressMarkerFile(jobId)); + } + + private static boolean isUuidDirectoryName(String name) { + try { + UUID.fromString(name); + return true; + } catch (IllegalArgumentException e) { + return false; + } + } + + 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); + boolean succeeded = false; + Archiver.AppendableTar appendableTar = null; + + try { + Files.createDirectories(workDir); + job.setState(ContainerExportStatus.State.RUNNING); + markExportInProgress(job.getJobId()); + + 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"); + } + if (!isLeaderReady.getAsBoolean()) { + throw new IOException("SCM lost leadership during export"); + } + + int fetchCount = pageSize; + if (job.getMaxRows() > 0) { + long remaining = job.getMaxRows() - totalRows; + if (remaining <= 0) { + break; + } + fetchCount = (int) Math.min(fetchCount, remaining); + } + + List 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); + if (appendableTar == null) { + appendableTar = Archiver.openForAppend(tarFile); + } + appendShardToTar(appendableTar, 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) { + clearExportInProgress(job.getJobId()); + FileUtils.deleteQuietly(workDir.toFile()); + FileUtils.deleteQuietly(jobDir.toFile()); + job.setState(ContainerExportStatus.State.SUCCEEDED); + job.setTarPath(null); + succeeded = true; + LOG.info("Export job {} completed with zero matching containers", job.getJobId()); + return; + } + + if (currentShardPath != null) { + if (appendableTar == null) { + appendableTar = Archiver.openForAppend(tarFile); + } + appendShardToTar(appendableTar, currentShardPath, shardEntryName(job, fileIndex)); + } + + clearExportInProgress(job.getJobId()); + FileUtils.deleteQuietly(workDir.toFile()); + FileUtils.deleteQuietly(jobDir.toFile()); + job.setState(ContainerExportStatus.State.SUCCEEDED); + succeeded = true; + 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); + if (appendableTar != null) { + appendableTar.close(); + } + } + } catch (InterruptedException e) { + job.setState(ContainerExportStatus.State.FAILED); + job.setErrorMessage("Job was cancelled"); + cleanupFailedArtifacts(jobDir, tarFile, job.getJobId()); + 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, job.getJobId()); + LOG.error("Export job {} failed", job.getJobId(), e); + } finally { + runningTasks.remove(job.getJobId()); + if (metrics != null) { + if (succeeded) { + metrics.incrExportJobsSucceeded(); + long bytesWritten = tarFile.isFile() ? tarFile.length() : 0L; + metrics.recordLastSuccessfulExport(job.getTotalRows(), bytesWritten); + } else if (job.getState() == ContainerExportStatus.State.FAILED) { + metrics.incrExportJobsFailed(); + } + } + if (job.getState() == ContainerExportStatus.State.SUCCEEDED + || job.getState() == ContainerExportStatus.State.FAILED) { + synchronized (submitLock) { + evictOldTerminalJobs(); + } + } + } + } + + 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(Archiver.AppendableTar tar, Path shardPath, String entryName) + throws IOException { + tar.appendFile(shardPath.toFile(), entryName); + FileUtils.deleteQuietly(shardPath.toFile()); + } + + /** Remove partial work artifacts after a failed or cancelled export. */ + private void cleanupFailedArtifacts(Path jobDir, File tarFile, String jobId) { + if (jobDir != null) { + FileUtils.deleteQuietly(jobDir.toFile()); + } + if (tarFile != null) { + FileUtils.deleteQuietly(tarFile); + } + clearExportInProgress(jobId); + } + + 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.write("# jobId=" + job.getJobId()); + writer.newLine(); + writer.write("# timestamp=" + job.getTimestamp()); + writer.newLine(); + if (job.getHealthState() != null) { + writer.write("# healthState=" + job.getHealthState().name()); + writer.newLine(); + } + if (job.getLifeCycleState() != null) { + writer.write("# lifecycleState=" + job.getLifeCycleState().name()); + writer.newLine(); + } + writer.write("# startContainerId=" + shardStartContainerId.getProtobuf().getId()); + writer.newLine(); + writer.write("# part=" + partNumber); + writer.newLine(); + writer.write("# format=container-id-per-line"); + writer.newLine(); + writer.newLine(); + } + + static String buildScope(LifeCycleState lifeCycleState, + ContainerHealthState healthState) { + List parts = new ArrayList<>(2); + if (healthState != null) { + parts.add("health-" + healthState.name()); + } + if (lifeCycleState != null) { + parts.add("lifecycle-" + lifeCycleState.name()); + } + return String.join("_", parts); + } + + static final class ExportJob { + private final String jobId; + private final String scope; + private final String timestamp; + private final ContainerID startContainerId; + private final LifeCycleState lifeCycleState; + private final ContainerHealthState healthState; + private final long maxRows; + private final int pageSize; + private final int shardSize; + private volatile String tarPath; + private volatile ContainerExportStatus.State state = + ContainerExportStatus.State.RUNNING; + private volatile long totalRows; + private volatile long startTimeMs; + private volatile long endTimeMs; + private volatile String errorMessage; + + @SuppressWarnings("checkstyle:ParameterNumber") + ExportJob(String jobId, String scope, String timestamp, String tarPath, + ContainerID startContainerId, LifeCycleState lifeCycleState, + ContainerHealthState healthState, long maxRows, int pageSize, int shardSize) { + this.jobId = jobId; + this.scope = scope; + this.timestamp = timestamp; + this.tarPath = tarPath; + this.startContainerId = startContainerId != null + ? startContainerId : ContainerID.valueOf(0); + this.lifeCycleState = lifeCycleState; + this.healthState = healthState; + this.maxRows = maxRows; + this.pageSize = pageSize; + this.shardSize = shardSize; + } + + int getPageSize() { + return pageSize; + } + + int getShardSize() { + return shardSize; + } + + String getJobId() { + return jobId; + } + + String getScope() { + return scope; + } + + String getTimestamp() { + return timestamp; + } + + ContainerID getStartContainerId() { + return startContainerId; + } + + LifeCycleState getLifeCycleState() { + return lifeCycleState; + } + + ContainerHealthState getHealthState() { + return healthState; + } + + long getMaxRows() { + return maxRows; + } + + ContainerExportStatus.State getState() { + return state; + } + + void setState(ContainerExportStatus.State newState) { + this.state = newState; + if ((newState == ContainerExportStatus.State.SUCCEEDED + || newState == ContainerExportStatus.State.FAILED) + && endTimeMs == 0) { + this.endTimeMs = System.currentTimeMillis(); + } + } + + long getEndTimeMs() { + return endTimeMs; + } + + long getTotalRows() { + return totalRows; + } + + void setTotalRows(long rows) { + this.totalRows = rows; + } + + void setStartTimeMs(long startTimeMs) { + this.startTimeMs = startTimeMs; + } + + void setErrorMessage(String message) { + this.errorMessage = message; + } + + String getTarPath() { + return tarPath; + } + + void setTarPath(String path) { + this.tarPath = path; + } + + ContainerExportStatus toStatus() { + long elapsed; + if (startTimeMs <= 0) { + elapsed = 0; + } else if (endTimeMs > 0) { + elapsed = endTimeMs - startTimeMs; + } else { + elapsed = System.currentTimeMillis() - startTimeMs; + } + return new ContainerExportStatus(jobId, state, lifeCycleState, healthState, totalRows, + elapsed, tarPath, errorMessage); + } + } +} diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportMetrics.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportMetrics.java new file mode 100644 index 000000000000..1fdeac22445a --- /dev/null +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportMetrics.java @@ -0,0 +1,80 @@ +/* + * 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 org.apache.hadoop.metrics2.MetricsSystem; +import org.apache.hadoop.metrics2.annotation.Metric; +import org.apache.hadoop.metrics2.annotation.Metrics; +import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem; +import org.apache.hadoop.metrics2.lib.MutableCounterLong; +import org.apache.hadoop.metrics2.lib.MutableGaugeLong; +import org.apache.hadoop.ozone.OzoneConsts; + +/** + * Metrics for async container ID export jobs on SCM. + */ +@Metrics(about = "SCM Container Export Metrics", context = OzoneConsts.OZONE) +public final class ContainerExportMetrics { + + private static final String SOURCE_NAME = ContainerExportMetrics.class.getSimpleName(); + + @Metric(about = "Number of export jobs submitted") + private MutableCounterLong numExportJobsSubmitted; + + @Metric(about = "Number of export jobs succeeded") + private MutableCounterLong numExportJobsSucceeded; + + @Metric(about = "Number of export jobs failed") + private MutableCounterLong numExportJobsFailed; + + @Metric(about = "Rows exported by the most recent successful export job.") + private MutableGaugeLong lastExportRows; + + @Metric(about = "TAR bytes written by the most recent successful export job.") + private MutableGaugeLong lastExportBytesWritten; + + private ContainerExportMetrics() { + } + + public static ContainerExportMetrics create() { + MetricsSystem ms = DefaultMetricsSystem.instance(); + return ms.register(SOURCE_NAME, "SCM Container Export IDs Metrics", new ContainerExportMetrics()); + } + + public void unRegister() { + MetricsSystem ms = DefaultMetricsSystem.instance(); + ms.unregisterSource(SOURCE_NAME); + } + + public void incrExportJobsSubmitted() { + numExportJobsSubmitted.incr(); + } + + public void incrExportJobsSucceeded() { + numExportJobsSucceeded.incr(); + } + + public void incrExportJobsFailed() { + numExportJobsFailed.incr(); + } + + public void recordLastSuccessfulExport(long rows, long bytesWritten) { + lastExportRows.set(rows); + lastExportBytesWritten.set(bytesWritten); + } +} diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/package-info.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/package-info.java new file mode 100644 index 000000000000..103c9519fcab --- /dev/null +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/package-info.java @@ -0,0 +1,21 @@ +/* + * 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. + */ + +/** + * This package contains classes related to container export. + */ +package org.apache.hadoop.hdds.scm.container.export; diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/StorageContainerManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/StorageContainerManager.java index 100c9feb9b67..13a1ac9bc04c 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/StorageContainerManager.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/StorageContainerManager.java @@ -90,6 +90,7 @@ import org.apache.hadoop.hdds.scm.container.IncrementalContainerReportHandler; import org.apache.hadoop.hdds.scm.container.balancer.ContainerBalancer; import org.apache.hadoop.hdds.scm.container.balancer.MoveManager; +import org.apache.hadoop.hdds.scm.container.export.ContainerExportManager; import org.apache.hadoop.hdds.scm.container.placement.algorithms.ContainerPlacementPolicyFactory; import org.apache.hadoop.hdds.scm.container.placement.algorithms.SCMContainerPlacementMetrics; import org.apache.hadoop.hdds.scm.container.placement.metrics.SCMMetrics; @@ -282,6 +283,8 @@ public final class StorageContainerManager extends ServiceRuntimeInfoImpl private ReplicationManager replicationManager; + private ContainerExportManager containerExportManager; + private SCMSafeModeManager scmSafeModeManager; private SCMCertificateClient scmCertificateClient; private RootCARotationManager rootCARotationManager; @@ -875,6 +878,7 @@ private void initializeSystemManagers(OzoneConfiguration conf, // RM gets notified of expired pending delete from containerReplicaPendingOps by subscribing to it // so it can resend them. containerReplicaPendingOps.registerSubscriber(replicationManager); + containerExportManager = new ContainerExportManager(containerManager, conf, this::checkLeader); if (configurator.getScmSafeModeManager() != null) { scmSafeModeManager = configurator.getScmSafeModeManager(); } else { @@ -1679,6 +1683,10 @@ public void stop() { stopReplicationManager(); + if (containerExportManager != null) { + containerExportManager.shutdown(); + } + try { LOG.info("Stopping the Datanode Admin Monitor."); scmDecommissionManager.stop(); @@ -1930,6 +1938,10 @@ public ReplicationManager getReplicationManager() { return replicationManager; } + public ContainerExportManager getContainerExportManager() { + return containerExportManager; + } + public PlacementPolicy getContainerPlacementPolicy() { return containerPlacementPolicy; } diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/export/TestContainerExportManager.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/export/TestContainerExportManager.java new file mode 100644 index 000000000000..52e5a68fb326 --- /dev/null +++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/export/TestContainerExportManager.java @@ -0,0 +1,326 @@ +/* + * 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 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.junit.jupiter.api.Assertions.fail; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.ArgumentMatchers.isNull; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.io.File; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.UUID; +import java.util.function.BooleanSupplier; +import java.util.stream.Collectors; +import org.apache.commons.io.FileUtils; +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.utils.Archiver; +import org.apache.ozone.test.GenericTestUtils; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +/** + * Tests for {@link ContainerExportManager}. + */ +public class TestContainerExportManager { + + private static final int TEST_DEFAULT_PAGE_SIZE = 2; + private static final int TEST_DEFAULT_SHARD_SIZE = 3; + private static final int TEST_MAX_TERMINAL_JOBS = 10; + + @TempDir + private File tempDir; + + private ContainerManager containerManager; + private ContainerExportManager exportManager; + + @BeforeEach + public void setup() { + containerManager = mock(ContainerManager.class); + exportManager = newExportManager(TEST_DEFAULT_SHARD_SIZE, TEST_DEFAULT_PAGE_SIZE, + TEST_MAX_TERMINAL_JOBS, () -> true, null); + } + + @AfterEach + public void teardown() { + exportManager.shutdown(); + } + + @Test + public void testMultiShardExportCreatesTar() throws Exception { + when(containerManager.getContainerIDs( + eq(ContainerID.valueOf(0)), anyInt(), isNull(), eq(ContainerHealthState.MISSING))) + .thenReturn(ids(1, 2)); + when(containerManager.getContainerIDs( + eq(ContainerID.valueOf(3)), anyInt(), isNull(), eq(ContainerHealthState.MISSING))) + .thenReturn(ids(3, 4)); + when(containerManager.getContainerIDs( + eq(ContainerID.valueOf(5)), anyInt(), isNull(), eq(ContainerHealthState.MISSING))) + .thenReturn(Collections.emptyList()); + + String jobId = exportManager.submitJob(ContainerID.valueOf(0), null, + ContainerHealthState.MISSING, 0, 0, 0); + + ContainerExportStatus status = waitForTerminal(jobId); + if (status.getState() == ContainerExportStatus.State.FAILED) { + fail(status.getErrorMessage()); + } + assertEquals(ContainerExportStatus.State.SUCCEEDED, status.getState()); + assertEquals(4, status.getTotalRows()); + assertNotNull(status.getTarPath()); + assertTrue(status.getTarPath().endsWith(".tar")); + File tarPath = new File(status.getTarPath()); + assertTrue(tarPath.exists()); + + Path extractDir = Files.createTempDirectory("export-tar"); + try { + Archiver.extract(tarPath, extractDir); + String part2Name; + try (java.util.stream.Stream stream = Files.list(extractDir)) { + part2Name = stream.map(path -> path.getFileName().toString()) + .filter(name -> name.endsWith("part002.txt")) + .findFirst() + .orElseThrow(() -> new AssertionError("part002.txt not found in TAR")); + } + assertTrue(Files.readAllLines(extractDir.resolve(part2Name)).contains( + "# startContainerId=4")); + } finally { + FileUtils.deleteQuietly(extractDir.toFile()); + } + } + + @Test + public void testSingleShardExportCreatesTar() throws Exception { + exportManager.shutdown(); + exportManager = newExportManager(100, TEST_DEFAULT_PAGE_SIZE, + TEST_MAX_TERMINAL_JOBS, () -> true, null); + + when(containerManager.getContainerIDs( + eq(ContainerID.valueOf(0)), anyInt(), isNull(), eq(ContainerHealthState.MISSING))) + .thenReturn(ids(1, 2, 3, 4)); + when(containerManager.getContainerIDs( + eq(ContainerID.valueOf(5)), anyInt(), isNull(), eq(ContainerHealthState.MISSING))) + .thenReturn(Collections.emptyList()); + + String jobId = exportManager.submitJob(ContainerID.valueOf(0), null, + ContainerHealthState.MISSING, 0, 0, 0); + + ContainerExportStatus status = waitForTerminal(jobId); + if (status.getState() == ContainerExportStatus.State.FAILED) { + fail(status.getErrorMessage()); + } + assertEquals(ContainerExportStatus.State.SUCCEEDED, status.getState()); + assertTrue(status.getTarPath().endsWith(".tar")); + assertTrue(new File(status.getTarPath()).exists()); + assertFalse(new File(status.getTarPath().replace(".tar", ".txt")).exists()); + } + + @Test + public void testRejectConcurrentExport() { + exportManager.getJobTracker().put("existing", + newRunningJob("existing")); + assertThrows(IllegalStateException.class, () -> + exportManager.submitJob(ContainerID.valueOf(0), null, + ContainerHealthState.EMPTY, 0, 0, 0)); + } + + @Test + public void testRejectMissingFilters() { + assertThrows(IllegalArgumentException.class, () -> + exportManager.submitJob(ContainerID.valueOf(0), null, null, 0, 0, 0)); + } + + @Test + public void testBuildScope() { + assertEquals("health-MISSING", + ContainerExportManager.buildScope(null, ContainerHealthState.MISSING)); + } + + @Test + public void testMetadataTimestampIsHumanReadable() throws Exception { + when(containerManager.getContainerIDs( + eq(ContainerID.valueOf(0)), anyInt(), isNull(), eq(ContainerHealthState.MISSING))) + .thenReturn(Collections.emptyList()); + + String jobId = exportManager.submitJob(ContainerID.valueOf(0), null, + ContainerHealthState.MISSING, 0, 0, 0); + waitForTerminal(jobId); + + ContainerExportManager.ExportJob job = exportManager.getJobTracker().get(jobId); + assertTrue(job.getTimestamp().matches("\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}Z")); + } + + @Test + public void testElapsedMsStableAfterCompletion() throws Exception { + when(containerManager.getContainerIDs( + eq(ContainerID.valueOf(0)), anyInt(), isNull(), eq(ContainerHealthState.MISSING))) + .thenReturn(ids(1)); + when(containerManager.getContainerIDs( + eq(ContainerID.valueOf(2)), anyInt(), isNull(), eq(ContainerHealthState.MISSING))) + .thenReturn(Collections.emptyList()); + + String jobId = exportManager.submitJob(ContainerID.valueOf(0), null, + ContainerHealthState.MISSING, 0, 0, 0); + ContainerExportStatus first = waitForTerminal(jobId); + Thread.sleep(200); + ContainerExportStatus second = exportManager.getJobStatus(jobId); + assertEquals(first.getElapsedMs(), second.getElapsedMs()); + } + + @Test + public void testTerminalJobEvictionWhenOverCap() throws Exception { + exportManager.shutdown(); + exportManager = newExportManager(TEST_DEFAULT_SHARD_SIZE, TEST_DEFAULT_PAGE_SIZE, 2, () -> true, null); + + when(containerManager.getContainerIDs( + eq(ContainerID.valueOf(0)), anyInt(), isNull(), eq(ContainerHealthState.MISSING))) + .thenReturn(ids(1)); + when(containerManager.getContainerIDs( + eq(ContainerID.valueOf(2)), anyInt(), isNull(), eq(ContainerHealthState.MISSING))) + .thenReturn(Collections.emptyList()); + String job1 = exportManager.submitJob(ContainerID.valueOf(0), null, ContainerHealthState.MISSING, 0, 0, 0); + waitForTerminal(job1); + + when(containerManager.getContainerIDs( + eq(ContainerID.valueOf(0)), anyInt(), isNull(), eq(ContainerHealthState.EMPTY))) + .thenReturn(ids(10)); + when(containerManager.getContainerIDs( + eq(ContainerID.valueOf(11)), anyInt(), isNull(), eq(ContainerHealthState.EMPTY))) + .thenReturn(Collections.emptyList()); + String job2 = exportManager.submitJob(ContainerID.valueOf(0), null, ContainerHealthState.EMPTY, 0, 0, 0); + waitForTerminal(job2); + + when(containerManager.getContainerIDs( + eq(ContainerID.valueOf(0)), anyInt(), isNull(), eq(ContainerHealthState.UNDER_REPLICATED))) + .thenReturn(ids(20)); + when(containerManager.getContainerIDs( + eq(ContainerID.valueOf(21)), anyInt(), isNull(), eq(ContainerHealthState.UNDER_REPLICATED))) + .thenReturn(Collections.emptyList()); + String job3 = exportManager.submitJob(ContainerID.valueOf(0), null, + ContainerHealthState.UNDER_REPLICATED, 0, 0, 0); + waitForTerminal(job3); + + assertNotNull(exportManager.getJobStatus(job2)); + assertNotNull(exportManager.getJobStatus(job3)); + assertNull(exportManager.getJobStatus(job1)); + } + + @Test + public void testOrphanWorkDirRemovedOnStartup() throws Exception { + String jobId = UUID.randomUUID().toString(); + Path orphan = tempDir.toPath().resolve(jobId).resolve("work"); + Files.createDirectories(orphan); + exportManager.shutdown(); + exportManager = newExportManager(TEST_DEFAULT_SHARD_SIZE, TEST_DEFAULT_PAGE_SIZE, + TEST_MAX_TERMINAL_JOBS, () -> true, null); + assertFalse(Files.exists(orphan)); + } + + @Test + public void testIncompleteExportArtifactsRemovedOnStartup() throws Exception { + String jobId = UUID.randomUUID().toString(); + Path jobDir = tempDir.toPath().resolve(jobId).resolve("work"); + Files.createDirectories(jobDir); + File partialTar = new File(tempDir, "container-ids-health-MISSING-20260101T000000Z-" + jobId + ".tar"); + assertTrue(partialTar.createNewFile()); + File inprogress = new File(tempDir, jobId + ".inprogress"); + assertTrue(inprogress.createNewFile()); + + exportManager.shutdown(); + exportManager = newExportManager(TEST_DEFAULT_SHARD_SIZE, TEST_DEFAULT_PAGE_SIZE, + TEST_MAX_TERMINAL_JOBS, () -> true, null); + + assertFalse(Files.exists(jobDir)); + assertFalse(partialTar.exists()); + assertFalse(inprogress.exists()); + } + + @Test + public void testCompletedExportTarRetainedOnRestart() throws Exception { + when(containerManager.getContainerIDs( + eq(ContainerID.valueOf(0)), anyInt(), isNull(), eq(ContainerHealthState.MISSING))) + .thenReturn(ids(1)); + when(containerManager.getContainerIDs( + eq(ContainerID.valueOf(2)), anyInt(), isNull(), eq(ContainerHealthState.MISSING))) + .thenReturn(Collections.emptyList()); + + String jobId = exportManager.submitJob(ContainerID.valueOf(0), null, ContainerHealthState.MISSING, 0, 0, 0); + ContainerExportStatus status = waitForTerminal(jobId); + assertEquals(ContainerExportStatus.State.SUCCEEDED, status.getState()); + File tar = new File(status.getTarPath()); + assertTrue(tar.exists()); + + exportManager.shutdown(); + exportManager = newExportManager(TEST_DEFAULT_SHARD_SIZE, TEST_DEFAULT_PAGE_SIZE, + TEST_MAX_TERMINAL_JOBS, () -> true, null); + assertTrue(tar.exists()); + assertNull(exportManager.getJobStatus(jobId)); + } + + @Test + public void testRejectSubmitWhenNotLeader() { + exportManager.shutdown(); + exportManager = newExportManager(TEST_DEFAULT_SHARD_SIZE, TEST_DEFAULT_PAGE_SIZE, 2, () -> false, null); + assertThrows(IllegalStateException.class, () -> + exportManager.submitJob(ContainerID.valueOf(0), null, ContainerHealthState.MISSING, 0, 0, 0)); + } + + private ContainerExportManager newExportManager(int defaultShardSize, int defaultPageSize, int maxTerminalJobs, + BooleanSupplier isLeaderReady, ContainerExportMetrics metrics) { + return new ContainerExportManager(containerManager, tempDir.getAbsolutePath(), + defaultShardSize, defaultPageSize, maxTerminalJobs, isLeaderReady, metrics); + } + + private static List ids(long... values) { + return Arrays.stream(values).mapToObj(ContainerID::valueOf) + .collect(Collectors.toList()); + } + + private ContainerExportStatus waitForTerminal(String jobId) throws Exception { + GenericTestUtils.waitFor(() -> { + ContainerExportStatus status = exportManager.getJobStatus(jobId); + return status != null && status.isTerminal(); + }, 100, 30_000); + return exportManager.getJobStatus(jobId); + } + + private static ContainerExportManager.ExportJob newRunningJob(String jobId) { + ContainerExportManager.ExportJob job = new ContainerExportManager.ExportJob( + jobId, "health-MISSING", "2026-01-01T00:00:00Z", + "/tmp/test.tar", ContainerID.valueOf(0), null, + ContainerHealthState.MISSING, 0, TEST_DEFAULT_PAGE_SIZE, TEST_DEFAULT_SHARD_SIZE); + job.setState(ContainerExportStatus.State.RUNNING); + return job; + } +} From 6d36753088e7c47a2b244cc88f00d88cacf83e88 Mon Sep 17 00:00:00 2001 From: sarvekshayr Date: Tue, 21 Jul 2026 14:08:27 +0530 Subject: [PATCH 2/5] Fixed comments in Container Export Manager --- .../export/ContainerExportManager.java | 38 ++++++++++++------- .../export/TestContainerExportManager.java | 5 ++- 2 files changed, 29 insertions(+), 14 deletions(-) diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportManager.java index ce8f604dade0..533ef2b7d39a 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportManager.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportManager.java @@ -41,7 +41,6 @@ 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 java.util.function.BooleanSupplier; import java.util.stream.Collectors; @@ -64,8 +63,9 @@ * if RM has not yet evaluated a container. * *

Job status is kept in memory only. On SCM restart or leader failover, in-flight jobs are lost - * and the operator must re-submit on the new leader. Completed TAR files remain on disk until - * manually deleted. On startup, incomplete work ({@code {jobId}.inprogress} markers, UUID work + * and the operator must re-submit on the new leader. Completed TAR files are kept on disk while their + * job status remains in {@code jobTracker}; when a terminal job is evicted past {@code maxTerminalJobs}, + * its TAR is deleted as well. On startup, incomplete work ({@code {jobId}.inprogress} markers, UUID work * directories, and matching partial TAR files) is removed. */ public class ContainerExportManager { @@ -84,7 +84,6 @@ public class ContainerExportManager { private final Map jobTracker = new ConcurrentHashMap<>(); private final ExecutorService workerPool; - private final Map> runningTasks = new ConcurrentHashMap<>(); private final ContainerManager containerManager; private final ContainerExportMetrics metrics; private final BooleanSupplier isLeaderReady; @@ -126,7 +125,9 @@ public ContainerExportManager(ContainerManager containerManager, exportDirectory, defaultShardSize, defaultPageSize, maxTerminalJobs); } - // TODO: make ozone.scm.container.export.dir configurable. + // TODO: make ozone.scm.container.export.dir configurable + // Set path separate from scm.db.dirs to avoid large export TAR I/O + // contending with SCM metadata DB access as the disk fills. private static String resolveExportDirectory(OzoneConfiguration conf) { File scmDbDir = ServerUtils.getScmDbDir(conf); return new File(scmDbDir, EXPORT_SUBDIR).getAbsolutePath(); @@ -173,8 +174,7 @@ public String submitJob(ContainerID start, LifeCycleState lifeCycleState, metrics.incrExportJobsSubmitted(); } - Future future = workerPool.submit(() -> executeExport(job)); - runningTasks.put(jobId, future); + workerPool.submit(() -> executeExport(job)); LOG.info("Submitted container ID export job {} (scope={}, start={}, maxRows={}, pageSize={}, shardSize={})", jobId, scope, start, maxRows, resolvedPageSize, resolvedShardSize); return jobId; @@ -209,7 +209,6 @@ public void shutdown() { LOG.warn("Timeout waiting for export worker shutdown", e); Thread.currentThread().interrupt(); } - runningTasks.clear(); if (metrics != null) { metrics.unRegister(); } @@ -226,7 +225,20 @@ private void evictOldTerminalJobs() { .collect(Collectors.toList()); int excess = terminalJobs.size() - maxTerminalJobs; for (int i = 0; i < excess; i++) { - jobTracker.remove(terminalJobs.get(i).getKey()); + Map.Entry evicted = terminalJobs.get(i); + deleteExportTar(evicted.getValue()); + jobTracker.remove(evicted.getKey()); + } + } + + private void deleteExportTar(ExportJob job) { + String tarPath = job.getTarPath(); + if (tarPath == null) { + return; + } + File tar = new File(tarPath); + if (tar.isFile() && FileUtils.deleteQuietly(tar)) { + LOG.debug("Removed container export TAR for evicted job {}: {}", job.getJobId(), tar.getName()); } } @@ -405,18 +417,19 @@ private void executeExport(ExportJob job) { appendShardToTar(appendableTar, currentShardPath, shardEntryName(job, fileIndex)); } - clearExportInProgress(job.getJobId()); FileUtils.deleteQuietly(workDir.toFile()); FileUtils.deleteQuietly(jobDir.toFile()); job.setState(ContainerExportStatus.State.SUCCEEDED); succeeded = true; - LOG.info("Export job {} completed ({} rows, tar={}). " - + "Delete the TAR file manually on the SCM leader when no longer needed.", + LOG.info("Export job {} completed ({} rows, tar={}).", job.getJobId(), totalRows, tarFile.getAbsolutePath()); } finally { closeWriter(writer); if (appendableTar != null) { appendableTar.close(); + if (succeeded) { + clearExportInProgress(job.getJobId()); + } } } } catch (InterruptedException e) { @@ -431,7 +444,6 @@ private void executeExport(ExportJob job) { cleanupFailedArtifacts(jobDir, tarFile, job.getJobId()); LOG.error("Export job {} failed", job.getJobId(), e); } finally { - runningTasks.remove(job.getJobId()); if (metrics != null) { if (succeeded) { metrics.incrExportJobsSucceeded(); diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/export/TestContainerExportManager.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/export/TestContainerExportManager.java index 52e5a68fb326..7789e544e06e 100644 --- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/export/TestContainerExportManager.java +++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/export/TestContainerExportManager.java @@ -210,7 +210,9 @@ public void testTerminalJobEvictionWhenOverCap() throws Exception { eq(ContainerID.valueOf(2)), anyInt(), isNull(), eq(ContainerHealthState.MISSING))) .thenReturn(Collections.emptyList()); String job1 = exportManager.submitJob(ContainerID.valueOf(0), null, ContainerHealthState.MISSING, 0, 0, 0); - waitForTerminal(job1); + ContainerExportStatus status1 = waitForTerminal(job1); + File tar1 = new File(status1.getTarPath()); + assertTrue(tar1.exists()); when(containerManager.getContainerIDs( eq(ContainerID.valueOf(0)), anyInt(), isNull(), eq(ContainerHealthState.EMPTY))) @@ -234,6 +236,7 @@ public void testTerminalJobEvictionWhenOverCap() throws Exception { assertNotNull(exportManager.getJobStatus(job2)); assertNotNull(exportManager.getJobStatus(job3)); assertNull(exportManager.getJobStatus(job1)); + assertFalse(tar1.exists()); //evicting a terminal job deletes its TAR too } @Test From c17dcb1bd14b1b353de8fa20b12cc95ab22350b8 Mon Sep 17 00:00:00 2001 From: sarvekshayr Date: Wed, 22 Jul 2026 10:55:34 +0530 Subject: [PATCH 3/5] Refactored ContainerExportManager --- .../export/ContainerExportStatus.java | 95 ---- ...nerExportLimits.java => ExportLimits.java} | 4 +- .../export/ContainerExportManager.java | 454 ++++++------------ .../hdds/scm/container/export/ExportJob.java | 269 +++++++++++ ...rExportMetrics.java => ExportMetrics.java} | 10 +- .../scm/container/export/ExportScope.java | 74 +++ .../scm/container/export/ExportSizing.java | 75 +++ .../scm/server/StorageContainerManager.java | 11 +- .../export/TestContainerExportManager.java | 131 ++--- 9 files changed, 651 insertions(+), 472 deletions(-) delete mode 100644 hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportStatus.java rename hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/export/{ContainerExportLimits.java => ExportLimits.java} (93%) create mode 100644 hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportJob.java rename hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/{ContainerExportMetrics.java => ExportMetrics.java} (90%) create mode 100644 hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportScope.java create mode 100644 hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportSizing.java diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportStatus.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportStatus.java deleted file mode 100644 index 6ed1d48072d1..000000000000 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportStatus.java +++ /dev/null @@ -1,95 +0,0 @@ -/* - * 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 org.apache.hadoop.hdds.protocol.proto.HddsProtos.LifeCycleState; -import org.apache.hadoop.hdds.scm.container.ContainerHealthState; - -/** - * Status of a container ID export job on the SCM leader. - */ -public final class ContainerExportStatus { - - private final String jobId; - private final State state; - private final LifeCycleState lifecycleState; - private final ContainerHealthState 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, LifeCycleState lifecycleState, - ContainerHealthState healthState, long totalRows, long elapsedMs, String tarPath, - String errorMessage) { - this.jobId = jobId; - this.state = state; - this.lifecycleState = lifecycleState; - this.healthState = healthState; - this.totalRows = totalRows; - this.elapsedMs = elapsedMs; - this.tarPath = tarPath; - this.errorMessage = errorMessage; - } - - public String getJobId() { - return jobId; - } - - public State getState() { - return state; - } - - public LifeCycleState getLifecycleState() { - return lifecycleState; - } - - public ContainerHealthState getHealthState() { - return healthState; - } - - public long getTotalRows() { - return totalRows; - } - - public long getElapsedMs() { - return elapsedMs; - } - - public String getTarPath() { - return tarPath; - } - - public String getErrorMessage() { - return errorMessage; - } - - public boolean isTerminal() { - return state == State.SUCCEEDED || state == State.FAILED; - } -} diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportLimits.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportLimits.java similarity index 93% rename from hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportLimits.java rename to hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportLimits.java index d78f86873786..49e7b914b99b 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportLimits.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportLimits.java @@ -20,7 +20,7 @@ /** * Shared limits and defaults for container ID export. */ -public final class ContainerExportLimits { +public final class ExportLimits { public static final String EXPORT_SUBDIR = "exports"; @@ -29,6 +29,6 @@ public final class ContainerExportLimits { public static final int MAX_PAGE_SIZE = 1_000_000; public static final int MAX_SHARD_SIZE = 5_000_000; - private ContainerExportLimits() { + private ExportLimits() { } } diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportManager.java index 533ef2b7d39a..6ab5c51aea64 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportManager.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportManager.java @@ -17,11 +17,9 @@ package org.apache.hadoop.hdds.scm.container.export; -import static org.apache.hadoop.hdds.scm.container.export.ContainerExportLimits.DEFAULT_PAGE_SIZE; -import static org.apache.hadoop.hdds.scm.container.export.ContainerExportLimits.DEFAULT_SHARD_SIZE; -import static org.apache.hadoop.hdds.scm.container.export.ContainerExportLimits.EXPORT_SUBDIR; -import static org.apache.hadoop.hdds.scm.container.export.ContainerExportLimits.MAX_PAGE_SIZE; -import static org.apache.hadoop.hdds.scm.container.export.ContainerExportLimits.MAX_SHARD_SIZE; +import static org.apache.hadoop.hdds.scm.container.export.ExportLimits.DEFAULT_PAGE_SIZE; +import static org.apache.hadoop.hdds.scm.container.export.ExportLimits.DEFAULT_SHARD_SIZE; +import static org.apache.hadoop.hdds.scm.container.export.ExportLimits.EXPORT_SUBDIR; import java.io.BufferedWriter; import java.io.File; @@ -33,15 +31,16 @@ import java.time.Instant; import java.time.ZoneOffset; import java.time.format.DateTimeFormatter; -import java.util.ArrayList; import java.util.Comparator; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; import java.util.function.BooleanSupplier; import java.util.stream.Collectors; import org.apache.commons.io.FileUtils; @@ -65,64 +64,81 @@ *

Job status is kept in memory only. On SCM restart or leader failover, in-flight jobs are lost * and the operator must re-submit on the new leader. Completed TAR files are kept on disk while their * job status remains in {@code jobTracker}; when a terminal job is evicted past {@code maxTerminalJobs}, - * its TAR is deleted as well. On startup, incomplete work ({@code {jobId}.inprogress} markers, UUID work - * directories, and matching partial TAR files) is removed. + * its TAR is deleted as well. On startup, incomplete work ({@code {jobId}.in-progress} markers, + * {@code export-{jobId}} work directories, and matching partial TAR files) is removed. */ public class ContainerExportManager { private static final Logger LOG = LoggerFactory.getLogger(ContainerExportManager.class); - private static final String INPROGRESS_MARKER_SUFFIX = ".inprogress"; + static final String IN_PROGRESS_MARKER_SUFFIX = ".in-progress"; + static final String EXPORT_JOB_DIR_PREFIX = "export-"; //TODO: make ozone.scm.container.export.max.terminal.jobs configurable. private static final int DEFAULT_MAX_TERMINAL_JOBS = 10; + private static final long SHUTDOWN_TIMEOUT_MS = 5_000; 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 final Map jobTracker = new ConcurrentHashMap<>(); + private final Map jobTracker = new ConcurrentHashMap<>(); + private final AtomicReference runningJobId = new AtomicReference<>(); private final ExecutorService workerPool; private final ContainerManager containerManager; - private final ContainerExportMetrics metrics; + private final ExportMetrics metrics; private final BooleanSupplier isLeaderReady; private final String exportDirectory; private final int defaultShardSize; private final int defaultPageSize; private final int maxTerminalJobs; - private final Object submitLock = new Object(); - - public ContainerExportManager(ContainerManager containerManager, - OzoneConfiguration conf, BooleanSupplier isLeaderReady) { - this(containerManager, resolveExportDirectory(conf), DEFAULT_SHARD_SIZE, DEFAULT_PAGE_SIZE, - DEFAULT_MAX_TERMINAL_JOBS, isLeaderReady, ContainerExportMetrics.create()); + /** SCM node id, used in log messages and the export worker thread name. */ + private final String scmId; + + public ContainerExportManager(ContainerManager containerManager, BooleanSupplier isLeaderReady, + OzoneConfiguration conf, String scmId) { + this.containerManager = Objects.requireNonNull(containerManager, "containerManager == null"); + this.isLeaderReady = Objects.requireNonNull(isLeaderReady, "isLeaderReady == null"); + this.exportDirectory = Objects.requireNonNull(resolveExportDirectory(conf), "exportDirectory == null"); + this.scmId = Objects.requireNonNull(scmId, "scmId == null"); + this.defaultShardSize = DEFAULT_SHARD_SIZE; + this.defaultPageSize = DEFAULT_PAGE_SIZE; + this.maxTerminalJobs = DEFAULT_MAX_TERMINAL_JOBS; + this.metrics = ExportMetrics.create(); + this.workerPool = newWorkerPool(this.scmId); } - ContainerExportManager(ContainerManager containerManager, String exportDirectory, - int defaultShardSize, int defaultPageSize, int maxTerminalJobs, - BooleanSupplier isLeaderReady, ContainerExportMetrics metrics) { - this.containerManager = containerManager; - this.exportDirectory = exportDirectory; + ContainerExportManager(ContainerManager containerManager, BooleanSupplier isLeaderReady, + String exportDirectory, int defaultShardSize, int defaultPageSize, int maxTerminalJobs, + String scmId) { + this.containerManager = Objects.requireNonNull(containerManager, "containerManager == null"); + this.isLeaderReady = Objects.requireNonNull(isLeaderReady, "isLeaderReady == null"); + this.exportDirectory = Objects.requireNonNull(exportDirectory, "exportDirectory == null"); + this.scmId = Objects.requireNonNull(scmId, "scmId == null"); this.defaultShardSize = defaultShardSize; this.defaultPageSize = defaultPageSize; this.maxTerminalJobs = maxTerminalJobs; - this.isLeaderReady = isLeaderReady; - this.metrics = metrics; - this.workerPool = Executors.newSingleThreadExecutor(r -> { - Thread t = new Thread(r, "ContainerExportWorker"); + this.metrics = null; + this.workerPool = newWorkerPool(this.scmId); + } + + private static ExecutorService newWorkerPool(String scmId) { + return Executors.newSingleThreadExecutor(r -> { + Thread t = new Thread(r, scmId + "-ContainerExportWorker"); t.setDaemon(true); return t; }); + } - try { - Files.createDirectories(Paths.get(exportDirectory)); - cleanupOrphanedExportArtifacts(); - } catch (IOException e) { - LOG.error("Failed to initialize export directory: {}", exportDirectory, e); - } - LOG.info("ContainerExportManager initialized (dir={}, defaultShardSize={}, defaultPageSize={}, maxTerminalJobs={})", - exportDirectory, defaultShardSize, defaultPageSize, maxTerminalJobs); + /** + * Initializes the export directory. Must be called once before submitting jobs. + */ + public void start() throws IOException { + Files.createDirectories(Paths.get(exportDirectory)); + cleanupOrphanedExportArtifacts(); + LOG.info("{}: ContainerExportManager started (dir={}, defaultShardSize={}, defaultPageSize={}, maxTerminalJobs={})", + scmId, exportDirectory, defaultShardSize, defaultPageSize, maxTerminalJobs); } // TODO: make ozone.scm.container.export.dir configurable @@ -135,78 +151,69 @@ private static String resolveExportDirectory(OzoneConfiguration conf) { /** * Submit a container ID export job on the SCM leader. + * Batch sizing is described by {@link ExportSizing}. + * + * @return job id, or {@code null} if not leader or another export is already running */ - public String submitJob(ContainerID start, LifeCycleState lifeCycleState, + public ExportJob.Id submitJob(ContainerID start, LifeCycleState lifeCycleState, ContainerHealthState healthState, long maxRows, int pageSize, int shardSize) { if (!isLeaderReady.getAsBoolean()) { - throw new IllegalStateException( - "Container ID export must be submitted on the SCM leader."); + return null; } 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; + validateRequest(start); + ExportSizing.validate(maxRows, pageSize, shardSize); + ExportSizing sizing = ExportSizing.resolve(maxRows, pageSize, shardSize, defaultPageSize, defaultShardSize); - String jobId = UUID.randomUUID().toString(); - String scope = buildScope(lifeCycleState, healthState); + ExportJob.Id jobId = ExportJob.Id.newId(); + if (!runningJobId.compareAndSet(null, jobId)) { + return null; + } + + ExportScope scope = ExportScope.of(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 tarFileName = String.format("container-ids-%s-%s-%s.tar", scope.getValue(), fileTimestamp, jobId.getValue()); String tarPath = exportDirectory + File.separator + tarFileName; - ExportJob job = new ExportJob(jobId, scope, metadataTimestamp, tarPath, - start, lifeCycleState, healthState, maxRows, resolvedPageSize, resolvedShardSize); + ExportJob job = new ExportJob(jobId, scope, metadataTimestamp, tarPath, start, sizing); - 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."); - } - evictOldTerminalJobs(); - jobTracker.put(jobId, job); - } + evictOldTerminalJobs(); + jobTracker.put(jobId, job); if (metrics != null) { metrics.incrExportJobsSubmitted(); } workerPool.submit(() -> executeExport(job)); - LOG.info("Submitted container ID export job {} (scope={}, start={}, maxRows={}, pageSize={}, shardSize={})", - jobId, scope, start, maxRows, resolvedPageSize, resolvedShardSize); + LOG.info("{}: submitted container ID export job {} (scope={}, start={}, maxRows={}, pageSize={}, shardSize={})", + scmId, jobId, scope, start, sizing.getMaxRows(), sizing.getPageSize(), sizing.getShardSize()); return jobId; } - private static void validateRequest(ContainerID start, long maxRows, int pageSize, int shardSize) { + private static void validateRequest(ContainerID start) { if (start != null && start.getProtobuf().getId() < 0) { - throw new IllegalArgumentException("start container ID must be non-negative."); - } - if (maxRows < 0) { - throw new IllegalArgumentException("maxRows 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 + "."); + throw new IllegalArgumentException("start container ID must be non-negative: " + start.getProtobuf().getId()); } } - public ContainerExportStatus getJobStatus(String jobId) { + public ExportJob.Status getExportStatus(ExportJob.Id jobId) { ExportJob job = jobTracker.get(jobId); return job != null ? job.toStatus() : null; } public void shutdown() { - LOG.info("Shutting down ContainerExportManager"); + LOG.info("{}: shutting down ContainerExportManager", scmId); workerPool.shutdownNow(); try { - workerPool.awaitTermination(30, TimeUnit.SECONDS); + if (!workerPool.awaitTermination(SHUTDOWN_TIMEOUT_MS, TimeUnit.MILLISECONDS)) { + LOG.warn("{}: timed out waiting for export worker shutdown", scmId); + } } catch (InterruptedException e) { - LOG.warn("Timeout waiting for export worker shutdown", e); + LOG.warn("{}: interrupted waiting for export worker shutdown", scmId); Thread.currentThread().interrupt(); } if (metrics != null) { @@ -214,20 +221,20 @@ public void shutdown() { } } - Map getJobTracker() { + Map getJobTracker() { return jobTracker; } private void evictOldTerminalJobs() { - List> terminalJobs = jobTracker.entrySet().stream() - .filter(e -> e.getValue().getState() != ContainerExportStatus.State.RUNNING) - .sorted(Comparator.comparingLong(e -> e.getValue().getEndTimeMs())) + List> terminalJobs = jobTracker.entrySet().stream() + .filter(e -> e.getValue().getExecutionState() != ExportJob.ExecutionState.RUNNING) + .sorted(Comparator.comparingLong(e -> e.getValue().getEndTimeNs())) .collect(Collectors.toList()); int excess = terminalJobs.size() - maxTerminalJobs; for (int i = 0; i < excess; i++) { - Map.Entry evicted = terminalJobs.get(i); - deleteExportTar(evicted.getValue()); - jobTracker.remove(evicted.getKey()); + ExportJob evicted = terminalJobs.get(i).getValue(); + deleteExportTar(evicted); + jobTracker.remove(evicted.getId()); } } @@ -238,41 +245,56 @@ private void deleteExportTar(ExportJob job) { } File tar = new File(tarPath); if (tar.isFile() && FileUtils.deleteQuietly(tar)) { - LOG.debug("Removed container export TAR for evicted job {}: {}", job.getJobId(), tar.getName()); + LOG.debug("Removed container export TAR for evicted job {}: {}", job.getId(), tar.getName()); } } - private void cleanupOrphanedExportArtifacts() throws IOException { + private void cleanupOrphanedExportArtifacts() { File exportDir = new File(exportDirectory); File[] children = exportDir.listFiles(); if (children == null) { return; } for (File child : children) { - if (child.isFile() && child.getName().endsWith(INPROGRESS_MARKER_SUFFIX)) { + if (child.isFile() && child.getName().endsWith(IN_PROGRESS_MARKER_SUFFIX)) { String jobId = child.getName().substring( - 0, child.getName().length() - INPROGRESS_MARKER_SUFFIX.length()); + 0, child.getName().length() - IN_PROGRESS_MARKER_SUFFIX.length()); if (isUuidDirectoryName(jobId)) { removeIncompleteExportArtifacts(jobId); } } } for (File child : children) { - if (child.isDirectory() && isUuidDirectoryName(child.getName())) { - removeIncompleteExportArtifacts(child.getName()); + if (child.isDirectory()) { + String jobId = jobIdFromExportDirName(child.getName()); + if (jobId != null) { + removeIncompleteExportArtifacts(jobId); + } } } } + private static String exportJobDirName(String jobId) { + return EXPORT_JOB_DIR_PREFIX + jobId; + } + + private static String jobIdFromExportDirName(String dirName) { + if (!dirName.startsWith(EXPORT_JOB_DIR_PREFIX)) { + return null; + } + String jobId = dirName.substring(EXPORT_JOB_DIR_PREFIX.length()); + return isUuidDirectoryName(jobId) ? jobId : null; + } + private void removeIncompleteExportArtifacts(String jobId) { LOG.info("Removing incomplete container export artifacts for job {}", jobId); - FileUtils.deleteQuietly(inprogressMarkerFile(jobId)); + FileUtils.deleteQuietly(inProgressMarkerFile(jobId)); File tar = findTarForJobId(jobId); if (tar != null) { FileUtils.deleteQuietly(tar); LOG.info("Removed incomplete container export TAR for job {}: {}", jobId, tar.getName()); } - File jobWorkDir = new File(exportDirectory, jobId); + File jobWorkDir = new File(exportDirectory, exportJobDirName(jobId)); if (jobWorkDir.isDirectory()) { FileUtils.deleteQuietly(jobWorkDir); LOG.info("Removed orphaned container export work directory: {}", @@ -283,69 +305,66 @@ private void removeIncompleteExportArtifacts(String jobId) { private File findTarForJobId(String jobId) { File exportDir = new File(exportDirectory); File[] matches = exportDir.listFiles( - (dir, name) -> name.endsWith("-" + jobId + ".tar")); + (dir, fileName) -> fileName.endsWith("-" + jobId + ".tar")); if (matches == null || matches.length == 0) { return null; } return matches[0]; } - private File inprogressMarkerFile(String jobId) { - return new File(exportDirectory, jobId + INPROGRESS_MARKER_SUFFIX); + private File inProgressMarkerFile(String jobId) { + return new File(exportDirectory, jobId + IN_PROGRESS_MARKER_SUFFIX); } private void markExportInProgress(String jobId) throws IOException { - Files.createFile(inprogressMarkerFile(jobId).toPath()); + Files.createFile(inProgressMarkerFile(jobId).toPath()); } private void clearExportInProgress(String jobId) { - FileUtils.deleteQuietly(inprogressMarkerFile(jobId)); + FileUtils.deleteQuietly(inProgressMarkerFile(jobId)); } - private static boolean isUuidDirectoryName(String name) { + private static boolean isUuidDirectoryName(String directoryName) { try { - UUID.fromString(name); - return true; + return directoryName.equals(UUID.fromString(directoryName).toString()); } catch (IllegalArgumentException e) { return false; } } private void executeExport(ExportJob job) { - Path jobDir = Paths.get(exportDirectory, job.getJobId()); + String jobIdValue = job.getId().getValue(); + Path jobDir = Paths.get(exportDirectory, exportJobDirName(jobIdValue)); Path workDir = jobDir.resolve("work"); File tarFile = new File(job.getTarPath()); - long startTimeMs = System.currentTimeMillis(); - job.setStartTimeMs(startTimeMs); + job.setStartTimeNs(System.nanoTime()); boolean succeeded = false; Archiver.AppendableTar appendableTar = null; try { Files.createDirectories(workDir); - job.setState(ContainerExportStatus.State.RUNNING); - markExportInProgress(job.getJobId()); + job.setExecutionState(ExportJob.ExecutionState.RUNNING); + markExportInProgress(jobIdValue); 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); + StringBuilder buf = new StringBuilder(job.getPageSize() * 12); try { while (true) { if (Thread.currentThread().isInterrupted()) { - throw new InterruptedException("Job cancelled"); + throw new InterruptedException("Export job " + jobIdValue + " cancelled"); } if (!isLeaderReady.getAsBoolean()) { - throw new IOException("SCM lost leadership during export"); + throw new IOException(scmId + ": SCM lost leadership during export job " + jobIdValue); } - int fetchCount = pageSize; + int fetchCount = job.getPageSize(); if (job.getMaxRows() > 0) { long remaining = job.getMaxRows() - totalRows; if (remaining <= 0) { @@ -363,10 +382,10 @@ private void executeExport(ExportJob job) { for (ContainerID containerId : page) { if (recordsInCurrentFile == 0) { writer = closeWriter(writer); - currentShardPath = workDir.resolve(shardFileName(job, fileIndex)); + currentShardPath = workDir.resolve(job.shardFileName(fileIndex)); writer = Files.newBufferedWriter(currentShardPath, StandardCharsets.UTF_8); - writeMetadataHeader(writer, job, fileIndex, containerId); - LOG.info("Export job {} created shard part{}", job.getJobId(), fileIndex); + job.writeMetadataHeader(writer, fileIndex, containerId); + LOG.info("{}: export job {} created shard part{}", scmId, jobIdValue, fileIndex); } buf.append(containerId.getProtobuf().getId()).append('\n'); @@ -374,14 +393,14 @@ private void executeExport(ExportJob job) { recordsInCurrentFile++; job.setTotalRows(totalRows); - if (recordsInCurrentFile >= shardSize) { + if (recordsInCurrentFile >= job.getShardSize()) { writer.write(buf.toString()); buf.setLength(0); writer = closeWriter(writer); if (appendableTar == null) { appendableTar = Archiver.openForAppend(tarFile); } - appendShardToTar(appendableTar, currentShardPath, shardEntryName(job, fileIndex)); + appendShardToTar(appendableTar, currentShardPath, job, fileIndex); currentShardPath = null; recordsInCurrentFile = 0; fileIndex++; @@ -400,13 +419,13 @@ private void executeExport(ExportJob job) { writer = closeWriter(writer); if (totalRows == 0) { - clearExportInProgress(job.getJobId()); + clearExportInProgress(jobIdValue); FileUtils.deleteQuietly(workDir.toFile()); FileUtils.deleteQuietly(jobDir.toFile()); - job.setState(ContainerExportStatus.State.SUCCEEDED); + job.setExecutionState(ExportJob.ExecutionState.SUCCEEDED); job.setTarPath(null); succeeded = true; - LOG.info("Export job {} completed with zero matching containers", job.getJobId()); + LOG.info("{}: export job {} completed with zero matching containers", scmId, jobIdValue); return; } @@ -414,66 +433,56 @@ private void executeExport(ExportJob job) { if (appendableTar == null) { appendableTar = Archiver.openForAppend(tarFile); } - appendShardToTar(appendableTar, currentShardPath, shardEntryName(job, fileIndex)); + appendShardToTar(appendableTar, currentShardPath, job, fileIndex); } FileUtils.deleteQuietly(workDir.toFile()); FileUtils.deleteQuietly(jobDir.toFile()); - job.setState(ContainerExportStatus.State.SUCCEEDED); + job.setExecutionState(ExportJob.ExecutionState.SUCCEEDED); succeeded = true; - LOG.info("Export job {} completed ({} rows, tar={}).", - job.getJobId(), totalRows, tarFile.getAbsolutePath()); + LOG.info("{}: export job {} completed ({} rows, tar={}).", + scmId, jobIdValue, totalRows, tarFile.getAbsolutePath()); } finally { closeWriter(writer); if (appendableTar != null) { appendableTar.close(); if (succeeded) { - clearExportInProgress(job.getJobId()); + clearExportInProgress(jobIdValue); } } } } catch (InterruptedException e) { - job.setState(ContainerExportStatus.State.FAILED); - job.setErrorMessage("Job was cancelled"); - cleanupFailedArtifacts(jobDir, tarFile, job.getJobId()); - LOG.info("Export job {} was cancelled", job.getJobId()); + job.setExecutionState(ExportJob.ExecutionState.FAILED); + job.setErrorMessage(e.getMessage()); + cleanupFailedArtifacts(jobDir, tarFile, jobIdValue); + LOG.info("{}: export job {} was cancelled", scmId, jobIdValue); Thread.currentThread().interrupt(); } catch (IOException | RuntimeException e) { - job.setState(ContainerExportStatus.State.FAILED); + job.setExecutionState(ExportJob.ExecutionState.FAILED); job.setErrorMessage(e.getMessage() != null ? e.getMessage() : e.toString()); - cleanupFailedArtifacts(jobDir, tarFile, job.getJobId()); - LOG.error("Export job {} failed", job.getJobId(), e); + cleanupFailedArtifacts(jobDir, tarFile, jobIdValue); + LOG.error("{}: export job {} failed", scmId, jobIdValue, e); } finally { + runningJobId.compareAndSet(job.getId(), null); if (metrics != null) { if (succeeded) { metrics.incrExportJobsSucceeded(); long bytesWritten = tarFile.isFile() ? tarFile.length() : 0L; metrics.recordLastSuccessfulExport(job.getTotalRows(), bytesWritten); - } else if (job.getState() == ContainerExportStatus.State.FAILED) { + } else if (job.getExecutionState() == ExportJob.ExecutionState.FAILED) { metrics.incrExportJobsFailed(); } } - if (job.getState() == ContainerExportStatus.State.SUCCEEDED - || job.getState() == ContainerExportStatus.State.FAILED) { - synchronized (submitLock) { - evictOldTerminalJobs(); - } + if (job.getExecutionState() == ExportJob.ExecutionState.SUCCEEDED + || job.getExecutionState() == ExportJob.ExecutionState.FAILED) { + evictOldTerminalJobs(); } } } - 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(Archiver.AppendableTar tar, Path shardPath, String entryName) + private void appendShardToTar(Archiver.AppendableTar tar, Path shardPath, ExportJob job, int partIndex) throws IOException { - tar.appendFile(shardPath.toFile(), entryName); + tar.appendFile(shardPath.toFile(), job.shardEntryName(partIndex)); FileUtils.deleteQuietly(shardPath.toFile()); } @@ -495,165 +504,4 @@ private static BufferedWriter closeWriter(BufferedWriter writer) throws IOExcept } return null; } - - private static void writeMetadataHeader(BufferedWriter writer, ExportJob job, - int partNumber, ContainerID shardStartContainerId) throws IOException { - writer.write("# jobId=" + job.getJobId()); - writer.newLine(); - writer.write("# timestamp=" + job.getTimestamp()); - writer.newLine(); - if (job.getHealthState() != null) { - writer.write("# healthState=" + job.getHealthState().name()); - writer.newLine(); - } - if (job.getLifeCycleState() != null) { - writer.write("# lifecycleState=" + job.getLifeCycleState().name()); - writer.newLine(); - } - writer.write("# startContainerId=" + shardStartContainerId.getProtobuf().getId()); - writer.newLine(); - writer.write("# part=" + partNumber); - writer.newLine(); - writer.write("# format=container-id-per-line"); - writer.newLine(); - writer.newLine(); - } - - static String buildScope(LifeCycleState lifeCycleState, - ContainerHealthState healthState) { - List parts = new ArrayList<>(2); - if (healthState != null) { - parts.add("health-" + healthState.name()); - } - if (lifeCycleState != null) { - parts.add("lifecycle-" + lifeCycleState.name()); - } - return String.join("_", parts); - } - - static final class ExportJob { - private final String jobId; - private final String scope; - private final String timestamp; - private final ContainerID startContainerId; - private final LifeCycleState lifeCycleState; - private final ContainerHealthState healthState; - private final long maxRows; - private final int pageSize; - private final int shardSize; - private volatile String tarPath; - private volatile ContainerExportStatus.State state = - ContainerExportStatus.State.RUNNING; - private volatile long totalRows; - private volatile long startTimeMs; - private volatile long endTimeMs; - private volatile String errorMessage; - - @SuppressWarnings("checkstyle:ParameterNumber") - ExportJob(String jobId, String scope, String timestamp, String tarPath, - ContainerID startContainerId, LifeCycleState lifeCycleState, - ContainerHealthState healthState, long maxRows, int pageSize, int shardSize) { - this.jobId = jobId; - this.scope = scope; - this.timestamp = timestamp; - this.tarPath = tarPath; - this.startContainerId = startContainerId != null - ? startContainerId : ContainerID.valueOf(0); - this.lifeCycleState = lifeCycleState; - this.healthState = healthState; - this.maxRows = maxRows; - this.pageSize = pageSize; - this.shardSize = shardSize; - } - - int getPageSize() { - return pageSize; - } - - int getShardSize() { - return shardSize; - } - - String getJobId() { - return jobId; - } - - String getScope() { - return scope; - } - - String getTimestamp() { - return timestamp; - } - - ContainerID getStartContainerId() { - return startContainerId; - } - - LifeCycleState getLifeCycleState() { - return lifeCycleState; - } - - ContainerHealthState getHealthState() { - return healthState; - } - - long getMaxRows() { - return maxRows; - } - - ContainerExportStatus.State getState() { - return state; - } - - void setState(ContainerExportStatus.State newState) { - this.state = newState; - if ((newState == ContainerExportStatus.State.SUCCEEDED - || newState == ContainerExportStatus.State.FAILED) - && endTimeMs == 0) { - this.endTimeMs = System.currentTimeMillis(); - } - } - - long getEndTimeMs() { - return endTimeMs; - } - - long getTotalRows() { - return totalRows; - } - - void setTotalRows(long rows) { - this.totalRows = rows; - } - - void setStartTimeMs(long startTimeMs) { - this.startTimeMs = startTimeMs; - } - - void setErrorMessage(String message) { - this.errorMessage = message; - } - - String getTarPath() { - return tarPath; - } - - void setTarPath(String path) { - this.tarPath = path; - } - - ContainerExportStatus toStatus() { - long elapsed; - if (startTimeMs <= 0) { - elapsed = 0; - } else if (endTimeMs > 0) { - elapsed = endTimeMs - startTimeMs; - } else { - elapsed = System.currentTimeMillis() - startTimeMs; - } - return new ContainerExportStatus(jobId, state, lifeCycleState, healthState, totalRows, - elapsed, tarPath, errorMessage); - } - } } diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportJob.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportJob.java new file mode 100644 index 000000000000..429c769f9c7d --- /dev/null +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportJob.java @@ -0,0 +1,269 @@ +/* + * 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.IOException; +import java.util.Objects; +import java.util.UUID; +import java.util.concurrent.TimeUnit; +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; + +/** + * In-memory state for a container ID export job. + */ +public final class ExportJob { + + /** + * Unique job identifier. + */ + public static final class Id { + private final String value; + + private Id(String value) { + this.value = Objects.requireNonNull(value, "value == null"); + } + + public static Id newId() { + return new Id(UUID.randomUUID().toString()); + } + + public static Id of(String value) { + return new Id(value); + } + + public String getValue() { + return value; + } + + @Override + public String toString() { + return value; + } + + @Override + public boolean equals(Object obj) { + if (this == obj) { + return true; + } + if (!(obj instanceof Id)) { + return false; + } + return value.equals(((Id) obj).value); + } + + @Override + public int hashCode() { + return value.hashCode(); + } + } + + /** + * Snapshot of export progress returned to callers. Reads live fields from the enclosing job. + */ + public final class Status { + private Status() { + } + + public Id getId() { + return id; + } + + public ExecutionState getExecutionState() { + return executionState; + } + + public LifeCycleState getLifeCycleState() { + return scope.getLifeCycleState(); + } + + public ContainerHealthState getHealthState() { + return scope.getHealthState(); + } + + public long getTotalRows() { + return totalRows; + } + + public long getElapsedMs() { + if (startTimeNs <= 0) { + return 0; + } + long endNs = endTimeNs > 0 ? endTimeNs : System.nanoTime(); + return TimeUnit.NANOSECONDS.toMillis(endNs - startTimeNs); + } + + public String getTarPath() { + return tarPath; + } + + public String getErrorMessage() { + return errorMessage; + } + + public boolean isTerminal() { + return executionState == ExecutionState.SUCCEEDED + || executionState == ExecutionState.FAILED; + } + } + + /** + * Job execution state. + */ + public enum ExecutionState { + RUNNING, + SUCCEEDED, + FAILED + } + + private final Id id; + private final ExportScope scope; + private final String timestamp; + private final ContainerID startContainerId; + private final ExportSizing sizing; + private volatile String tarPath; + private volatile ExecutionState executionState = ExecutionState.RUNNING; + private volatile long totalRows; + private volatile long startTimeNs; + private volatile long endTimeNs; + private volatile String errorMessage; + + ExportJob(Id id, ExportScope scope, String timestamp, String tarPath, ContainerID startContainerId, + ExportSizing sizing) { + this.id = id; + this.scope = scope; + this.timestamp = timestamp; + this.tarPath = tarPath; + this.startContainerId = startContainerId != null ? startContainerId : ContainerID.valueOf(0); + this.sizing = sizing; + } + + Id getId() { + return id; + } + + ExportScope getScope() { + return scope; + } + + String getTimestamp() { + return timestamp; + } + + ContainerID getStartContainerId() { + return startContainerId; + } + + LifeCycleState getLifeCycleState() { + return scope.getLifeCycleState(); + } + + ContainerHealthState getHealthState() { + return scope.getHealthState(); + } + + long getMaxRows() { + return sizing.getMaxRows(); + } + + int getPageSize() { + return sizing.getPageSize(); + } + + int getShardSize() { + return sizing.getShardSize(); + } + + ExecutionState getExecutionState() { + return executionState; + } + + void setExecutionState(ExecutionState newState) { + this.executionState = newState; + if ((newState == ExecutionState.SUCCEEDED || newState == ExecutionState.FAILED) + && endTimeNs == 0) { + this.endTimeNs = System.nanoTime(); + } + } + + long getEndTimeNs() { + return endTimeNs; + } + + long getTotalRows() { + return totalRows; + } + + void setTotalRows(long rows) { + this.totalRows = rows; + } + + void setStartTimeNs(long startTimeNs) { + this.startTimeNs = startTimeNs; + } + + void setErrorMessage(String message) { + this.errorMessage = message; + } + + String getTarPath() { + return tarPath; + } + + void setTarPath(String path) { + this.tarPath = path; + } + + Status toStatus() { + return new Status(); + } + + String shardFileName(int partIndex) { + return String.format("container-ids-%s-%s-part%03d.txt", + scope.getValue(), timestamp, partIndex); + } + + String shardEntryName(int partIndex) { + return shardFileName(partIndex); + } + + void writeMetadataHeader(BufferedWriter writer, int partNumber, ContainerID shardStartContainerId) + throws IOException { + writer.write("# jobId=" + id.getValue()); + writer.newLine(); + writer.write("# timestamp=" + timestamp); + writer.newLine(); + if (scope.getHealthState() != null) { + writer.write("# healthState=" + scope.getHealthState().name()); + writer.newLine(); + } + if (scope.getLifeCycleState() != null) { + writer.write("# lifecycleState=" + scope.getLifeCycleState().name()); + writer.newLine(); + } + writer.write("# startContainerId=" + shardStartContainerId.getProtobuf().getId()); + writer.newLine(); + writer.write("# part=" + partNumber); + writer.newLine(); + writer.write("# format=container-id-per-line"); + writer.newLine(); + writer.newLine(); + } +} diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportMetrics.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportMetrics.java similarity index 90% rename from hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportMetrics.java rename to hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportMetrics.java index 1fdeac22445a..7e0df1b6f54c 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportMetrics.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportMetrics.java @@ -29,9 +29,9 @@ * Metrics for async container ID export jobs on SCM. */ @Metrics(about = "SCM Container Export Metrics", context = OzoneConsts.OZONE) -public final class ContainerExportMetrics { +public final class ExportMetrics { - private static final String SOURCE_NAME = ContainerExportMetrics.class.getSimpleName(); + private static final String SOURCE_NAME = ExportMetrics.class.getSimpleName(); @Metric(about = "Number of export jobs submitted") private MutableCounterLong numExportJobsSubmitted; @@ -48,12 +48,12 @@ public final class ContainerExportMetrics { @Metric(about = "TAR bytes written by the most recent successful export job.") private MutableGaugeLong lastExportBytesWritten; - private ContainerExportMetrics() { + private ExportMetrics() { } - public static ContainerExportMetrics create() { + public static ExportMetrics create() { MetricsSystem ms = DefaultMetricsSystem.instance(); - return ms.register(SOURCE_NAME, "SCM Container Export IDs Metrics", new ContainerExportMetrics()); + return ms.register(SOURCE_NAME, "SCM Container Export IDs Metrics", new ExportMetrics()); } public void unRegister() { diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportScope.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportScope.java new file mode 100644 index 000000000000..c88cf886cc3f --- /dev/null +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportScope.java @@ -0,0 +1,74 @@ +/* + * 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 org.apache.hadoop.hdds.protocol.proto.HddsProtos.LifeCycleState; +import org.apache.hadoop.hdds.scm.container.ContainerHealthState; + +/** + * Container listing filters for an export job. + * An export job filters containers by {@link ContainerHealthState}, {@link LifeCycleState} or both. + * Example TAR name: + * {@code container-ids-health-MISSING_lifecycle-OPEN-20260101T120000Z-{jobId}.tar} + */ +public final class ExportScope { + + private final LifeCycleState lifeCycleState; + private final ContainerHealthState healthState; + private final String value; + + private ExportScope(LifeCycleState lifeCycleState, ContainerHealthState healthState, String value) { + this.lifeCycleState = lifeCycleState; + this.healthState = healthState; + this.value = value; + } + + public static ExportScope of(LifeCycleState lifeCycleState, ContainerHealthState healthState) { + StringBuilder sb = new StringBuilder(); + if (healthState != null) { + sb.append("health-").append(healthState.name()); + } + if (lifeCycleState != null) { + if (sb.length() > 0) { + sb.append('_'); + } + sb.append("lifecycle-").append(lifeCycleState.name()); + } + return new ExportScope(lifeCycleState, healthState, sb.toString()); + } + + public LifeCycleState getLifeCycleState() { + return lifeCycleState; + } + + public ContainerHealthState getHealthState() { + return healthState; + } + + /** + * Stable filter name segment used in export TAR and shard file names. + */ + public String getValue() { + return value; + } + + @Override + public String toString() { + return value; + } +} diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportSizing.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportSizing.java new file mode 100644 index 000000000000..21536020b883 --- /dev/null +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportSizing.java @@ -0,0 +1,75 @@ +/* + * 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 static org.apache.hadoop.hdds.scm.container.export.ExportLimits.MAX_PAGE_SIZE; +import static org.apache.hadoop.hdds.scm.container.export.ExportLimits.MAX_SHARD_SIZE; + +/** + * Per-job batch sizing for container ID export. + * These values are always used together when reading container IDs from SCM and writing them into the export TAR: + * {@code pageSize} container IDs returned per {@code ContainerManager} listing query. + * {@code shardSize} - container IDs written into each shard text file before that file is appended to the TAR. + * {@code maxRows} - upper bound on total exported container IDs for the job; {@code 0} means no limit. + *

+ * A {@code pageSize} or {@code shardSize} of {@code 0} in a submit request means use the + * manager defaults from {@link ExportLimits}. + */ +final class ExportSizing { + + private final long maxRows; + private final int pageSize; + private final int shardSize; + + ExportSizing(long maxRows, int pageSize, int shardSize) { + this.maxRows = maxRows; + this.pageSize = pageSize; + this.shardSize = shardSize; + } + + static void validate(long maxRows, int pageSize, int shardSize) { + if (maxRows < 0) { + throw new IllegalArgumentException("maxRows must be non-negative: " + maxRows); + } + if (pageSize < 0 || pageSize > MAX_PAGE_SIZE) { + throw new IllegalArgumentException("pageSize must be between 0 and " + MAX_PAGE_SIZE + ": " + pageSize); + } + if (shardSize < 0 || shardSize > MAX_SHARD_SIZE) { + throw new IllegalArgumentException("shardSize must be between 0 and " + MAX_SHARD_SIZE + ": " + shardSize); + } + } + + static ExportSizing resolve(long maxRows, int pageSize, int shardSize, int defaultPageSize, + int defaultShardSize) { + return new ExportSizing(maxRows, + pageSize > 0 ? pageSize : defaultPageSize, + shardSize > 0 ? shardSize : defaultShardSize); + } + + long getMaxRows() { + return maxRows; + } + + int getPageSize() { + return pageSize; + } + + int getShardSize() { + return shardSize; + } +} diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/StorageContainerManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/StorageContainerManager.java index 13a1ac9bc04c..e1082e7aee82 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/StorageContainerManager.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/StorageContainerManager.java @@ -283,7 +283,7 @@ public final class StorageContainerManager extends ServiceRuntimeInfoImpl private ReplicationManager replicationManager; - private ContainerExportManager containerExportManager; + private final ContainerExportManager containerExportManager; private SCMSafeModeManager scmSafeModeManager; private SCMCertificateClient scmCertificateClient; @@ -439,6 +439,10 @@ private StorageContainerManager(OzoneConfiguration conf, initializeSystemManagers(conf, configurator); + containerExportManager = new ContainerExportManager( + containerManager, this::checkLeader, conf, getScmId()); + containerExportManager.start(); + if (isSecretKeyEnable(securityConfig)) { secretKeyManagerService = new SecretKeyManagerService(scmContext, conf, scmHAManager.getRatisServer()); @@ -878,7 +882,6 @@ private void initializeSystemManagers(OzoneConfiguration conf, // RM gets notified of expired pending delete from containerReplicaPendingOps by subscribing to it // so it can resend them. containerReplicaPendingOps.registerSubscriber(replicationManager); - containerExportManager = new ContainerExportManager(containerManager, conf, this::checkLeader); if (configurator.getScmSafeModeManager() != null) { scmSafeModeManager = configurator.getScmSafeModeManager(); } else { @@ -1683,9 +1686,7 @@ public void stop() { stopReplicationManager(); - if (containerExportManager != null) { - containerExportManager.shutdown(); - } + containerExportManager.shutdown(); try { LOG.info("Stopping the Datanode Admin Monitor."); diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/export/TestContainerExportManager.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/export/TestContainerExportManager.java index 7789e544e06e..afc8662e6621 100644 --- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/export/TestContainerExportManager.java +++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/export/TestContainerExportManager.java @@ -58,6 +58,7 @@ public class TestContainerExportManager { private static final int TEST_DEFAULT_PAGE_SIZE = 2; private static final int TEST_DEFAULT_SHARD_SIZE = 3; private static final int TEST_MAX_TERMINAL_JOBS = 10; + private static final String TEST_SCM_ID = "test-scm"; @TempDir private File tempDir; @@ -66,10 +67,10 @@ public class TestContainerExportManager { private ContainerExportManager exportManager; @BeforeEach - public void setup() { + public void setup() throws Exception { containerManager = mock(ContainerManager.class); exportManager = newExportManager(TEST_DEFAULT_SHARD_SIZE, TEST_DEFAULT_PAGE_SIZE, - TEST_MAX_TERMINAL_JOBS, () -> true, null); + TEST_MAX_TERMINAL_JOBS, () -> true); } @AfterEach @@ -89,14 +90,15 @@ public void testMultiShardExportCreatesTar() throws Exception { eq(ContainerID.valueOf(5)), anyInt(), isNull(), eq(ContainerHealthState.MISSING))) .thenReturn(Collections.emptyList()); - String jobId = exportManager.submitJob(ContainerID.valueOf(0), null, + ExportJob.Id jobId = exportManager.submitJob(ContainerID.valueOf(0), null, ContainerHealthState.MISSING, 0, 0, 0); + assertNotNull(jobId); - ContainerExportStatus status = waitForTerminal(jobId); - if (status.getState() == ContainerExportStatus.State.FAILED) { + ExportJob.Status status = waitForTerminal(jobId); + if (status.getExecutionState() == ExportJob.ExecutionState.FAILED) { fail(status.getErrorMessage()); } - assertEquals(ContainerExportStatus.State.SUCCEEDED, status.getState()); + assertEquals(ExportJob.ExecutionState.SUCCEEDED, status.getExecutionState()); assertEquals(4, status.getTotalRows()); assertNotNull(status.getTarPath()); assertTrue(status.getTarPath().endsWith(".tar")); @@ -124,7 +126,7 @@ public void testMultiShardExportCreatesTar() throws Exception { public void testSingleShardExportCreatesTar() throws Exception { exportManager.shutdown(); exportManager = newExportManager(100, TEST_DEFAULT_PAGE_SIZE, - TEST_MAX_TERMINAL_JOBS, () -> true, null); + TEST_MAX_TERMINAL_JOBS, () -> true); when(containerManager.getContainerIDs( eq(ContainerID.valueOf(0)), anyInt(), isNull(), eq(ContainerHealthState.MISSING))) @@ -133,14 +135,15 @@ public void testSingleShardExportCreatesTar() throws Exception { eq(ContainerID.valueOf(5)), anyInt(), isNull(), eq(ContainerHealthState.MISSING))) .thenReturn(Collections.emptyList()); - String jobId = exportManager.submitJob(ContainerID.valueOf(0), null, + ExportJob.Id jobId = exportManager.submitJob(ContainerID.valueOf(0), null, ContainerHealthState.MISSING, 0, 0, 0); + assertNotNull(jobId); - ContainerExportStatus status = waitForTerminal(jobId); - if (status.getState() == ContainerExportStatus.State.FAILED) { + ExportJob.Status status = waitForTerminal(jobId); + if (status.getExecutionState() == ExportJob.ExecutionState.FAILED) { fail(status.getErrorMessage()); } - assertEquals(ContainerExportStatus.State.SUCCEEDED, status.getState()); + assertEquals(ExportJob.ExecutionState.SUCCEEDED, status.getExecutionState()); assertTrue(status.getTarPath().endsWith(".tar")); assertTrue(new File(status.getTarPath()).exists()); assertFalse(new File(status.getTarPath().replace(".tar", ".txt")).exists()); @@ -148,11 +151,21 @@ public void testSingleShardExportCreatesTar() throws Exception { @Test public void testRejectConcurrentExport() { - exportManager.getJobTracker().put("existing", - newRunningJob("existing")); - assertThrows(IllegalStateException.class, () -> - exportManager.submitJob(ContainerID.valueOf(0), null, - ContainerHealthState.EMPTY, 0, 0, 0)); + when(containerManager.getContainerIDs( + eq(ContainerID.valueOf(0)), anyInt(), isNull(), eq(ContainerHealthState.MISSING))) + .thenReturn(ids(1)); + when(containerManager.getContainerIDs( + eq(ContainerID.valueOf(2)), anyInt(), isNull(), eq(ContainerHealthState.MISSING))) + .thenAnswer(invocation -> { + Thread.sleep(60_000); + return Collections.emptyList(); + }); + + ExportJob.Id first = exportManager.submitJob(ContainerID.valueOf(0), null, + ContainerHealthState.MISSING, 0, 0, 0); + assertNotNull(first); + assertNull(exportManager.submitJob(ContainerID.valueOf(0), null, + ContainerHealthState.EMPTY, 0, 0, 0)); } @Test @@ -162,9 +175,9 @@ public void testRejectMissingFilters() { } @Test - public void testBuildScope() { + public void testExportScope() { assertEquals("health-MISSING", - ContainerExportManager.buildScope(null, ContainerHealthState.MISSING)); + ExportScope.of(null, ContainerHealthState.MISSING).getValue()); } @Test @@ -173,11 +186,12 @@ public void testMetadataTimestampIsHumanReadable() throws Exception { eq(ContainerID.valueOf(0)), anyInt(), isNull(), eq(ContainerHealthState.MISSING))) .thenReturn(Collections.emptyList()); - String jobId = exportManager.submitJob(ContainerID.valueOf(0), null, + ExportJob.Id jobId = exportManager.submitJob(ContainerID.valueOf(0), null, ContainerHealthState.MISSING, 0, 0, 0); + assertNotNull(jobId); waitForTerminal(jobId); - ContainerExportManager.ExportJob job = exportManager.getJobTracker().get(jobId); + ExportJob job = exportManager.getJobTracker().get(jobId); assertTrue(job.getTimestamp().matches("\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}Z")); } @@ -190,18 +204,19 @@ public void testElapsedMsStableAfterCompletion() throws Exception { eq(ContainerID.valueOf(2)), anyInt(), isNull(), eq(ContainerHealthState.MISSING))) .thenReturn(Collections.emptyList()); - String jobId = exportManager.submitJob(ContainerID.valueOf(0), null, + ExportJob.Id jobId = exportManager.submitJob(ContainerID.valueOf(0), null, ContainerHealthState.MISSING, 0, 0, 0); - ContainerExportStatus first = waitForTerminal(jobId); + assertNotNull(jobId); + ExportJob.Status first = waitForTerminal(jobId); Thread.sleep(200); - ContainerExportStatus second = exportManager.getJobStatus(jobId); + ExportJob.Status second = exportManager.getExportStatus(jobId); assertEquals(first.getElapsedMs(), second.getElapsedMs()); } @Test public void testTerminalJobEvictionWhenOverCap() throws Exception { exportManager.shutdown(); - exportManager = newExportManager(TEST_DEFAULT_SHARD_SIZE, TEST_DEFAULT_PAGE_SIZE, 2, () -> true, null); + exportManager = newExportManager(TEST_DEFAULT_SHARD_SIZE, TEST_DEFAULT_PAGE_SIZE, 2, () -> true); when(containerManager.getContainerIDs( eq(ContainerID.valueOf(0)), anyInt(), isNull(), eq(ContainerHealthState.MISSING))) @@ -209,8 +224,8 @@ public void testTerminalJobEvictionWhenOverCap() throws Exception { when(containerManager.getContainerIDs( eq(ContainerID.valueOf(2)), anyInt(), isNull(), eq(ContainerHealthState.MISSING))) .thenReturn(Collections.emptyList()); - String job1 = exportManager.submitJob(ContainerID.valueOf(0), null, ContainerHealthState.MISSING, 0, 0, 0); - ContainerExportStatus status1 = waitForTerminal(job1); + ExportJob.Id job1 = exportManager.submitJob(ContainerID.valueOf(0), null, ContainerHealthState.MISSING, 0, 0, 0); + ExportJob.Status status1 = waitForTerminal(job1); File tar1 = new File(status1.getTarPath()); assertTrue(tar1.exists()); @@ -220,7 +235,7 @@ public void testTerminalJobEvictionWhenOverCap() throws Exception { when(containerManager.getContainerIDs( eq(ContainerID.valueOf(11)), anyInt(), isNull(), eq(ContainerHealthState.EMPTY))) .thenReturn(Collections.emptyList()); - String job2 = exportManager.submitJob(ContainerID.valueOf(0), null, ContainerHealthState.EMPTY, 0, 0, 0); + ExportJob.Id job2 = exportManager.submitJob(ContainerID.valueOf(0), null, ContainerHealthState.EMPTY, 0, 0, 0); waitForTerminal(job2); when(containerManager.getContainerIDs( @@ -229,44 +244,44 @@ public void testTerminalJobEvictionWhenOverCap() throws Exception { when(containerManager.getContainerIDs( eq(ContainerID.valueOf(21)), anyInt(), isNull(), eq(ContainerHealthState.UNDER_REPLICATED))) .thenReturn(Collections.emptyList()); - String job3 = exportManager.submitJob(ContainerID.valueOf(0), null, + ExportJob.Id job3 = exportManager.submitJob(ContainerID.valueOf(0), null, ContainerHealthState.UNDER_REPLICATED, 0, 0, 0); waitForTerminal(job3); - assertNotNull(exportManager.getJobStatus(job2)); - assertNotNull(exportManager.getJobStatus(job3)); - assertNull(exportManager.getJobStatus(job1)); + assertNotNull(exportManager.getExportStatus(job2)); + assertNotNull(exportManager.getExportStatus(job3)); + assertNull(exportManager.getExportStatus(job1)); assertFalse(tar1.exists()); //evicting a terminal job deletes its TAR too } @Test public void testOrphanWorkDirRemovedOnStartup() throws Exception { String jobId = UUID.randomUUID().toString(); - Path orphan = tempDir.toPath().resolve(jobId).resolve("work"); + Path orphan = tempDir.toPath().resolve(ContainerExportManager.EXPORT_JOB_DIR_PREFIX + jobId).resolve("work"); Files.createDirectories(orphan); exportManager.shutdown(); exportManager = newExportManager(TEST_DEFAULT_SHARD_SIZE, TEST_DEFAULT_PAGE_SIZE, - TEST_MAX_TERMINAL_JOBS, () -> true, null); + TEST_MAX_TERMINAL_JOBS, () -> true); assertFalse(Files.exists(orphan)); } @Test public void testIncompleteExportArtifactsRemovedOnStartup() throws Exception { String jobId = UUID.randomUUID().toString(); - Path jobDir = tempDir.toPath().resolve(jobId).resolve("work"); + Path jobDir = tempDir.toPath().resolve(ContainerExportManager.EXPORT_JOB_DIR_PREFIX + jobId).resolve("work"); Files.createDirectories(jobDir); File partialTar = new File(tempDir, "container-ids-health-MISSING-20260101T000000Z-" + jobId + ".tar"); assertTrue(partialTar.createNewFile()); - File inprogress = new File(tempDir, jobId + ".inprogress"); - assertTrue(inprogress.createNewFile()); + File inProgress = new File(tempDir, jobId + ContainerExportManager.IN_PROGRESS_MARKER_SUFFIX); + assertTrue(inProgress.createNewFile()); exportManager.shutdown(); exportManager = newExportManager(TEST_DEFAULT_SHARD_SIZE, TEST_DEFAULT_PAGE_SIZE, - TEST_MAX_TERMINAL_JOBS, () -> true, null); + TEST_MAX_TERMINAL_JOBS, () -> true); assertFalse(Files.exists(jobDir)); assertFalse(partialTar.exists()); - assertFalse(inprogress.exists()); + assertFalse(inProgress.exists()); } @Test @@ -278,31 +293,32 @@ public void testCompletedExportTarRetainedOnRestart() throws Exception { eq(ContainerID.valueOf(2)), anyInt(), isNull(), eq(ContainerHealthState.MISSING))) .thenReturn(Collections.emptyList()); - String jobId = exportManager.submitJob(ContainerID.valueOf(0), null, ContainerHealthState.MISSING, 0, 0, 0); - ContainerExportStatus status = waitForTerminal(jobId); - assertEquals(ContainerExportStatus.State.SUCCEEDED, status.getState()); + ExportJob.Id jobId = exportManager.submitJob(ContainerID.valueOf(0), null, ContainerHealthState.MISSING, 0, 0, 0); + ExportJob.Status status = waitForTerminal(jobId); + assertEquals(ExportJob.ExecutionState.SUCCEEDED, status.getExecutionState()); File tar = new File(status.getTarPath()); assertTrue(tar.exists()); exportManager.shutdown(); exportManager = newExportManager(TEST_DEFAULT_SHARD_SIZE, TEST_DEFAULT_PAGE_SIZE, - TEST_MAX_TERMINAL_JOBS, () -> true, null); + TEST_MAX_TERMINAL_JOBS, () -> true); assertTrue(tar.exists()); - assertNull(exportManager.getJobStatus(jobId)); + assertNull(exportManager.getExportStatus(jobId)); } @Test - public void testRejectSubmitWhenNotLeader() { + public void testRejectSubmitWhenNotLeader() throws Exception { exportManager.shutdown(); - exportManager = newExportManager(TEST_DEFAULT_SHARD_SIZE, TEST_DEFAULT_PAGE_SIZE, 2, () -> false, null); - assertThrows(IllegalStateException.class, () -> - exportManager.submitJob(ContainerID.valueOf(0), null, ContainerHealthState.MISSING, 0, 0, 0)); + exportManager = newExportManager(TEST_DEFAULT_SHARD_SIZE, TEST_DEFAULT_PAGE_SIZE, 2, () -> false); + assertNull(exportManager.submitJob(ContainerID.valueOf(0), null, ContainerHealthState.MISSING, 0, 0, 0)); } private ContainerExportManager newExportManager(int defaultShardSize, int defaultPageSize, int maxTerminalJobs, - BooleanSupplier isLeaderReady, ContainerExportMetrics metrics) { - return new ContainerExportManager(containerManager, tempDir.getAbsolutePath(), - defaultShardSize, defaultPageSize, maxTerminalJobs, isLeaderReady, metrics); + BooleanSupplier isLeaderReady) throws Exception { + ContainerExportManager manager = new ContainerExportManager(containerManager, isLeaderReady, + tempDir.getAbsolutePath(), defaultShardSize, defaultPageSize, maxTerminalJobs, TEST_SCM_ID); + manager.start(); + return manager; } private static List ids(long... values) { @@ -310,20 +326,11 @@ private static List ids(long... values) { .collect(Collectors.toList()); } - private ContainerExportStatus waitForTerminal(String jobId) throws Exception { + private ExportJob.Status waitForTerminal(ExportJob.Id jobId) throws Exception { GenericTestUtils.waitFor(() -> { - ContainerExportStatus status = exportManager.getJobStatus(jobId); + ExportJob.Status status = exportManager.getExportStatus(jobId); return status != null && status.isTerminal(); }, 100, 30_000); - return exportManager.getJobStatus(jobId); - } - - private static ContainerExportManager.ExportJob newRunningJob(String jobId) { - ContainerExportManager.ExportJob job = new ContainerExportManager.ExportJob( - jobId, "health-MISSING", "2026-01-01T00:00:00Z", - "/tmp/test.tar", ContainerID.valueOf(0), null, - ContainerHealthState.MISSING, 0, TEST_DEFAULT_PAGE_SIZE, TEST_DEFAULT_SHARD_SIZE); - job.setState(ContainerExportStatus.State.RUNNING); - return job; + return exportManager.getExportStatus(jobId); } } From a597a4166545079c7588fbb0a7006924c2e23966 Mon Sep 17 00:00:00 2001 From: sarvekshayr Date: Wed, 22 Jul 2026 11:11:45 +0530 Subject: [PATCH 4/5] Fixed pmd error --- .../hdds/scm/container/export/ExportJob.java | 24 +++++++++---------- 1 file changed, 12 insertions(+), 12 deletions(-) diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportJob.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportJob.java index 429c769f9c7d..4d2ba160a893 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportJob.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportJob.java @@ -31,6 +31,18 @@ */ public final class ExportJob { + private final Id id; + private final ExportScope scope; + private final String timestamp; + private final ContainerID startContainerId; + private final ExportSizing sizing; + private volatile String tarPath; + private volatile ExecutionState executionState = ExecutionState.RUNNING; + private volatile long totalRows; + private volatile long startTimeNs; + private volatile long endTimeNs; + private volatile String errorMessage; + /** * Unique job identifier. */ @@ -133,18 +145,6 @@ public enum ExecutionState { FAILED } - private final Id id; - private final ExportScope scope; - private final String timestamp; - private final ContainerID startContainerId; - private final ExportSizing sizing; - private volatile String tarPath; - private volatile ExecutionState executionState = ExecutionState.RUNNING; - private volatile long totalRows; - private volatile long startTimeNs; - private volatile long endTimeNs; - private volatile String errorMessage; - ExportJob(Id id, ExportScope scope, String timestamp, String tarPath, ContainerID startContainerId, ExportSizing sizing) { this.id = id; From a5d5e6b7e53864d9dc75c7777ae972b67d89e2c3 Mon Sep 17 00:00:00 2001 From: sarvekshayr Date: Wed, 22 Jul 2026 22:35:37 +0530 Subject: [PATCH 5/5] Set succeeded = false in catch block --- .../hadoop/hdds/scm/container/export/ContainerExportManager.java | 1 + 1 file changed, 1 insertion(+) diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportManager.java index 6ab5c51aea64..e3bc0f33924c 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportManager.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportManager.java @@ -458,6 +458,7 @@ private void executeExport(ExportJob job) { LOG.info("{}: export job {} was cancelled", scmId, jobIdValue); Thread.currentThread().interrupt(); } catch (IOException | RuntimeException e) { + succeeded = false; job.setExecutionState(ExportJob.ExecutionState.FAILED); job.setErrorMessage(e.getMessage() != null ? e.getMessage() : e.toString()); cleanupFailedArtifacts(jobDir, tarFile, jobIdValue);