This is an automated email from the ASF dual-hosted git repository.
ArafatKhan2198 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/master by this push:
new 734435aa846 HDDS-15839. Implement SCM container ID export manager
(#11097).
734435aa846 is described below
commit 734435aa84632f12ff5cf6c2aeced1c6613fb46a
Author: Sarveksha Yeshavantha Raju
<[email protected]>
AuthorDate: Tue Oct 13 11:40:16 2026 +0530
HDDS-15839. Implement SCM container ID export manager (#11097).
---
.../container/export/ContainerExportManager.java | 299 ++++++++++++
.../scm/container/export/ExportFileManager.java | 124 +++--
.../hdds/scm/container/export/ExportJob.java | 145 +++++-
.../hdds/scm/container/export/ExportScope.java | 7 -
.../hdds/scm/server/StorageContainerManager.java | 20 +
.../export/TestContainerExportManager.java | 525 +++++++++++++++++++++
.../container/export/TestExportFileManager.java | 231 ---------
.../hdds/scm/container/export/TestExportJob.java | 67 ---
8 files changed, 1018 insertions(+), 400 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
new file mode 100644
index 00000000000..3cfd1d25ce6
--- /dev/null
+++
b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportManager.java
@@ -0,0 +1,299 @@
+/*
+ * 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.IOException;
+import java.io.OutputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.BooleanSupplier;
+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.scm.exceptions.SCMException;
+import org.apache.ratis.util.Preconditions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Manages asynchronous container ID export jobs on the SCM leader.
+ *
+ * <p>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.
+ *
+ * <p>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. {@link
ExportFileManager} owns on-disk layout,
+ * locking, part files, and completed archives; this class tracks {@link
ExportJob} state and
+ * schedules work.
+ */
+public class ContainerExportManager {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(ContainerExportManager.class);
+
+ private static final int DEFAULT_BATCH_SIZE = 100_000;
+ private static final int DEFAULT_PART_SIZE = 500_000;
+ private static final long SHUTDOWN_TIMEOUT_MS = 5_000;
+
+ private final Map<ExportJob.Id, ExportJob> jobMap = new
ConcurrentHashMap<>();
+ private final AtomicReference<ExportJob> runningJob = new
AtomicReference<>();
+ private final ExecutorService workerPool;
+ private final ContainerManager containerManager;
+ private final ExportFileManager fileManager;
+ private final BooleanSupplier isLeaderReady;
+ private final int partSize;
+ private final int batchSize;
+
+ public ContainerExportManager(String scmId, ContainerManager
containerManager, BooleanSupplier isLeaderReady,
+ OzoneConfiguration conf) {
+ this(scmId, containerManager, isLeaderReady,
+ ExportFileManager.resolveExportDirectory(conf), DEFAULT_PART_SIZE,
DEFAULT_BATCH_SIZE);
+ }
+
+ ContainerExportManager(String scmId, ContainerManager containerManager,
BooleanSupplier isLeaderReady,
+ String exportDirectory, int partSize, int batchSize) {
+ if (partSize <= 0 || batchSize <= 0) {
+ throw new IllegalArgumentException("partSize and batchSize must be
positive: partSize=" + partSize
+ + ", batchSize=" + batchSize);
+ }
+ this.containerManager = Objects.requireNonNull(containerManager,
"containerManager == null");
+ this.isLeaderReady = Objects.requireNonNull(isLeaderReady, "isLeaderReady
== null");
+ this.fileManager = new ExportFileManager(exportDirectory);
+ this.partSize = partSize;
+ this.batchSize = batchSize;
+ this.workerPool = newWorkerPool(scmId);
+ }
+
+ private ExecutorService newWorkerPool(String scmId) {
+ return new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new
LinkedBlockingQueue<>(), r -> {
+ Thread t = new Thread(r, scmId + "-ContainerExportWorker");
+ t.setDaemon(true);
+ return t;
+ }) {
+ @Override
+ protected void terminated() {
+ onWorkerTerminated();
+ }
+ };
+ }
+
+ /**
+ * Initializes the export directory. Must be called once before submitting
jobs.
+ */
+ public void start() throws IOException {
+ fileManager.start();
+ LOG.info("ContainerExportManager started (dir={}, partSize={},
batchSize={})",
+ fileManager.getExportDirectory(), partSize, batchSize);
+ }
+
+ /**
+ * Submit a container ID export job on the SCM leader.
+ *
+ * @return the submitted job
+ * @throws SCMException with {@link SCMException.ResultCodes#SCM_NOT_LEADER}
if this SCM is not the ready leader
+ * @throws IOException if another export is already running or the export
worker is shut down
+ */
+ public ExportJob submitJob(ContainerID start, LifeCycleState lifeCycleState,
ContainerHealthState healthState)
+ throws IOException {
+ if (!isLeaderReady.getAsBoolean()) {
+ throw new SCMException("Container ID export can only be submitted on the
SCM leader",
+ SCMException.ResultCodes.SCM_NOT_LEADER);
+ }
+
+ final ExportScope scope = ExportScope.of(lifeCycleState, healthState);
+ final ExportJob job = new ExportJob(scope, start, batchSize, partSize);
+ if (!runningJob.compareAndSet(null, job)) {
+ throw new IOException("Another container ID export is already running");
+ }
+
+ jobMap.put(job.getId(), job);
+ try {
+ workerPool.submit(() -> executeExport(job));
+ } catch (RejectedExecutionException e) {
+ jobMap.remove(job.getId());
+ runningJob.compareAndSet(job, null);
+ throw new IOException("Container ID export worker is shut down", e);
+ }
+ LOG.info("Submitted container ID export job {} (scope={}, start={},
batchSize={}, partSize={})",
+ job, scope, start, batchSize, partSize);
+ return job;
+ }
+
+ public ExportJob.Status getExportStatus(ExportJob.Id jobId) {
+ ExportJob job = jobMap.get(jobId);
+ return job != null ? job.getStatus() : null;
+ }
+
+ public void shutdown() {
+ LOG.info("Shutting down ContainerExportManager");
+ workerPool.shutdownNow();
+ try {
+ if (!workerPool.awaitTermination(SHUTDOWN_TIMEOUT_MS,
TimeUnit.MILLISECONDS)) {
+ LOG.warn("Export worker did not stop within {} ms; the container
export directory lock is "
+ + "released once it stops", SHUTDOWN_TIMEOUT_MS);
+ }
+ } catch (InterruptedException e) {
+ LOG.warn("Interrupted waiting for export worker shutdown; the container
export directory lock is "
+ + "released once it stops");
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ /**
+ * Runs once the worker has stopped, so the export directory is no longer in
use: the lock is kept
+ * until then so that a new manager cannot start and delete files the worker
is still writing.
+ */
+ private void onWorkerTerminated() {
+ // A job accepted but drained from the executor queue by shutdownNow()
never ran; fail it.
+ final ExportJob notStarted = runningJob.getAndSet(null);
+ if (notStarted != null) {
+ notStarted.getFuture().completeExceptionally(new
InterruptedException("Export cancelled: SCM shutting down"));
+ }
+ try {
+ fileManager.unlock();
+ } catch (IOException e) {
+ LOG.warn("Failed to unlock container export directory", e);
+ }
+ }
+
+ private void executeExport(ExportJob job) {
+ Long idsWritten = null;
+ Throwable failure = new IllegalStateException(job + " terminated
unexpectedly");
+ try {
+ fileManager.createJobDirectory(job.getId());
+
+ idsWritten = writeContainerIDs(job);
+ if (idsWritten == null) {
+ failure = new InterruptedException("Export cancelled");
+ LOG.info("Export {} was cancelled", job);
+ } else {
+ if (idsWritten > 0) {
+ fileManager.writeArchive(job, isLeaderReady);
+ LOG.info("{} completed: {} ids, archive={}", job, idsWritten,
job.getFileName());
+ } else {
+ LOG.info("{} completed with {} matching containers", job,
idsWritten);
+ }
+ fileManager.deleteJobDirectory(job.getId());
+ failure = null;
+ }
+ } catch (Exception e) {
+ failure = e;
+ LOG.error("Export {} failed", job, e);
+ } finally {
+ if (failure != null) {
+ fileManager.cleanupFailedJob(job);
+ }
+ final boolean cleared = runningJob.compareAndSet(job, null);
+ if (failure == null) {
+ job.getFuture().complete(idsWritten);
+ } else {
+ job.getFuture().completeExceptionally(failure);
+ }
+ Preconditions.assertTrue(cleared);
+ }
+ }
+
+ private Long writeContainerIDs(ExportJob job) throws IOException {
+ final byte[] buffer = new byte[job.getBatchSize() * 12];
+ int partNumber = 1;
+ long idsWritten = 0;
+ ContainerID cursor = job.getStartContainerId();
+ List<ContainerID> batch = Collections.emptyList();
+ int batchIndex = 0;
+
+ while (cursor != null || batchIndex < batch.size()) {
+ if (Thread.interrupted()) {
+ return null;
+ }
+ if (!isLeaderReady.getAsBoolean()) {
+ throw new IOException("SCM lost leadership during " + job);
+ }
+
+ if (batchIndex >= batch.size()) {
+ batch = fetchContainerIds(job, cursor);
+ batchIndex = 0;
+ if (batch.isEmpty()) {
+ break;
+ }
+ cursor = nextCursor(batch);
+ }
+
+ final ContainerID partStartContainerId = batch.get(batchIndex);
+ try (OutputStream out = fileManager.newPartOutputStream(job.getId(),
job.partFileName(partNumber))) {
+ job.writeMetadataHeader(out, partNumber, partStartContainerId);
+ LOG.info("{} created part{}", job, partNumber);
+
+ long idsInCurrentPart = 0;
+ int offset = 0;
+ while (idsInCurrentPart < job.getPartSize()
+ && (batchIndex < batch.size() || cursor != null)) {
+ if (batchIndex >= batch.size()) {
+ batch = fetchContainerIds(job, cursor);
+ batchIndex = 0;
+ if (batch.isEmpty()) {
+ cursor = null;
+ break;
+ }
+ cursor = nextCursor(batch);
+ }
+
+ final ContainerID containerId = batch.get(batchIndex++);
+ final byte[] idBytes = (containerId.getProtobuf().getId() +
"\n").getBytes(StandardCharsets.UTF_8);
+ if (offset + idBytes.length > buffer.length) {
+ out.write(buffer, 0, offset);
+ offset = 0;
+ }
+ System.arraycopy(idBytes, 0, buffer, offset, idBytes.length);
+ offset += idBytes.length;
+ idsWritten++;
+ idsInCurrentPart++;
+ }
+
+ if (offset > 0) {
+ out.write(buffer, 0, offset);
+ }
+ }
+ partNumber++;
+ }
+ return idsWritten;
+ }
+
+ private static ContainerID nextCursor(List<ContainerID> batch) {
+ return ContainerID.valueOf(batch.get(batch.size() -
1).getProtobuf().getId() + 1);
+ }
+
+ private List<ContainerID> fetchContainerIds(ExportJob job, ContainerID
cursor) throws IOException {
+ if (!isLeaderReady.getAsBoolean()) {
+ throw new IOException("SCM lost leadership during " + job);
+ }
+ return containerManager.getContainerIDs(
+ cursor, job.getBatchSize(), job.getLifeCycleState(),
job.getHealthState());
+ }
+}
diff --git
a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportFileManager.java
b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportFileManager.java
index 5ebbf012970..7dc37847cc8 100644
---
a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportFileManager.java
+++
b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportFileManager.java
@@ -17,26 +17,28 @@
package org.apache.hadoop.hdds.scm.container.export;
-import java.io.BufferedWriter;
import java.io.File;
import java.io.IOException;
+import java.io.OutputStream;
import java.io.RandomAccessFile;
import java.nio.channels.FileLock;
import java.nio.channels.OverlappingFileLockException;
-import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
-import java.util.ArrayList;
+import java.time.Instant;
+import java.time.ZoneOffset;
+import java.time.format.DateTimeFormatter;
import java.util.Arrays;
-import java.util.Collections;
import java.util.Comparator;
-import java.util.List;
import java.util.Objects;
+import java.util.function.BooleanSupplier;
import java.util.zip.GZIPOutputStream;
import org.apache.commons.compress.archivers.ArchiveOutputStream;
import org.apache.commons.compress.archivers.tar.TarArchiveEntry;
import org.apache.commons.io.FileUtils;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.server.ServerUtils;
import org.apache.hadoop.hdds.utils.Archiver;
import org.apache.hadoop.ozone.util.UUIDUtil;
import org.apache.ratis.util.AtomicFileOutputStream;
@@ -67,22 +69,23 @@
* </pre>
*
* <p><b>Incomplete work</b> ({@code export_{jobId}/} and {@code .tar.gz.tmp})
is removed by
- * {@link #cleanupFailedJob(ExportJob.Id, String)} on failure or cancel, and
by {@link #start()} for every
+ * {@link #cleanupFailedJob(ExportJob)} on failure or cancel, and by {@link
#start()} for every
* leftover directory and temp file after SCM restart. Completed {@code
.tar.gz} files are kept.
*
* <p><b>SCM restart:</b> {@link #start()} acquires the export directory lock
and clears incomplete work.
- * {@link #listCompletedArchivePaths()} returns existing archive paths (oldest
first).
*/
final class ExportFileManager {
private static final Logger LOG =
LoggerFactory.getLogger(ExportFileManager.class);
+ static final String EXPORT_SUBDIR = "exports";
static final String EXPORT_JOB_DIR_PREFIX = "export_";
static final String EXPORT_ARCHIVE_JOB_INFIX = "_job";
static final String EXPORT_ARCHIVE_SUFFIX = ".tar.gz";
static final String EXPORT_ARCHIVE_TMP_SUFFIX = EXPORT_ARCHIVE_SUFFIX +
AtomicFileOutputStream.TMP_EXTENSION;
static final String EXPORT_LOCK_NAME = "in_use.lock";
- private static final int EXPORT_JOB_START_TIME_LENGTH = 19;
+ private static final DateTimeFormatter EXPORT_JOB_START_TIME_FORMAT =
+
DateTimeFormatter.ofPattern("yyyy-MM-dd-HH-mm-ss").withZone(ZoneOffset.UTC);
private final String exportDirectory;
private FileLock exportDirectoryLock;
@@ -91,13 +94,22 @@ final class ExportFileManager {
this.exportDirectory = Objects.requireNonNull(exportDirectory,
"exportDirectory == null");
}
+ static String resolveExportDirectory(OzoneConfiguration conf) {
+ File scmDbDir = ServerUtils.getScmDbDir(conf);
+ return new File(scmDbDir, EXPORT_SUBDIR).getAbsolutePath();
+ }
+
+ String getExportDirectory() {
+ return exportDirectory;
+ }
+
void start() throws IOException {
Files.createDirectories(Paths.get(exportDirectory));
lock();
removeIncompleteWorkOnStartup();
}
- private void lock() throws IOException {
+ private synchronized void lock() throws IOException {
if (exportDirectoryLock != null) {
return;
}
@@ -117,96 +129,66 @@ private void lock() throws IOException {
}
}
- File resolveArchiveFile(ExportScope scope, String jobStartTime, ExportJob.Id
jobId) {
- return new File(exportDirectory, String.format("container-ids_%s_%s%s%s%s",
- scope.getValue(), jobStartTime, EXPORT_ARCHIVE_JOB_INFIX,
jobId.getValue(), EXPORT_ARCHIVE_SUFFIX));
+ synchronized void unlock() throws IOException {
+ if (exportDirectoryLock == null) {
+ return;
+ }
+ exportDirectoryLock.release();
+ exportDirectoryLock.channel().close();
+ exportDirectoryLock = null;
}
- File resolveArchiveTempFile(ExportScope scope, String jobStartTime,
ExportJob.Id jobId) {
- return AtomicFileOutputStream.getTemporaryFile(resolveArchiveFile(scope,
jobStartTime, jobId));
+ static String archiveFileName(ExportScope scope, String jobStartTime,
ExportJob.Id jobId) {
+ return String.format("container-ids_%s_%s%s%s%s",
+ scope, jobStartTime, EXPORT_ARCHIVE_JOB_INFIX, jobId,
EXPORT_ARCHIVE_SUFFIX);
}
void createJobDirectory(ExportJob.Id jobId) throws IOException {
Files.createDirectories(jobDirectory(jobId));
}
- BufferedWriter newPartWriter(ExportJob.Id jobId, String partFileName) throws
IOException {
- return Files.newBufferedWriter(jobDirectory(jobId).resolve(partFileName),
StandardCharsets.UTF_8);
+ OutputStream newPartOutputStream(ExportJob.Id jobId, String partFileName)
throws IOException {
+ return Files.newOutputStream(jobDirectory(jobId).resolve(partFileName));
}
- void writeArchive(ExportJob.Id jobId, String archivePath) throws IOException
{
- Path jobDir = jobDirectory(jobId);
+ void writeArchive(ExportJob job, BooleanSupplier isLeaderReady) throws
IOException {
+ Path jobDir = jobDirectory(job.getId());
File[] parts = jobDir.toFile().listFiles((dir, name) ->
name.endsWith(".txt"));
if (parts == null || parts.length == 0) {
- throw new IOException("No part files found for export job " + jobId);
+ throw new IOException("No part files found for " + job);
}
Arrays.sort(parts, Comparator.comparing(File::getName));
- File archiveFile = new File(archivePath);
+ File archiveFile = new File(exportDirectory, job.getFileName());
try (AtomicFileOutputStream atomicOut = new
AtomicFileOutputStream(archiveFile);
GZIPOutputStream gzipOut = new GZIPOutputStream(atomicOut);
ArchiveOutputStream<TarArchiveEntry> tarOut = Archiver.tar(gzipOut)) {
for (File part : parts) {
+ if (!isLeaderReady.getAsBoolean()) {
+ // close() would publish the partial archive; abort() discards the
temp file instead.
+ atomicOut.abort();
+ throw new IOException("SCM lost leadership while archiving " + job);
+ }
Archiver.includeFile(part, part.getName(), tarOut);
}
}
}
- void deleteJobDirectory(ExportJob.Id jobId) throws IOException {
+ void deleteJobDirectory(ExportJob.Id jobId) {
Path jobDir = jobDirectory(jobId);
- if (Files.exists(jobDir)) {
- FileUtils.deleteDirectory(jobDir.toFile());
+ if (Files.exists(jobDir) && !FileUtils.deleteQuietly(jobDir.toFile())) {
+ LOG.warn("Failed to delete container export job directory {}; it will be
removed on next SCM start", jobDir);
}
}
- /**
- * Returns completed archive paths, oldest first.
- */
- List<String> listCompletedArchivePaths() {
- File exportDir = new File(exportDirectory);
- File[] matches = exportDir.listFiles((dir, fileName) ->
fileName.endsWith(EXPORT_ARCHIVE_SUFFIX)
- && !fileName.endsWith(EXPORT_ARCHIVE_TMP_SUFFIX));
- if (matches == null || matches.length == 0) {
- return Collections.emptyList();
- }
- Arrays.sort(matches, Comparator.comparing(
- file -> jobStartTimeFromArchiveFileName(file.getName())));
- List<String> archivePaths = new ArrayList<>(matches.length);
- for (File archive : matches) {
- archivePaths.add(archive.getAbsolutePath());
- }
- return archivePaths;
+ static String formatJobStartTime(Instant jobStartTime) {
+ return EXPORT_JOB_START_TIME_FORMAT.format(jobStartTime);
}
- static String jobStartTimeFromArchiveFileName(String fileName) {
- int jobIndex = fileName.lastIndexOf(EXPORT_ARCHIVE_JOB_INFIX);
- if (jobIndex < EXPORT_JOB_START_TIME_LENGTH + 1
- || !fileName.endsWith(EXPORT_ARCHIVE_SUFFIX)
- || fileName.endsWith(EXPORT_ARCHIVE_TMP_SUFFIX)) {
- return null;
- }
- return fileName.substring(jobIndex - EXPORT_JOB_START_TIME_LENGTH,
jobIndex);
- }
-
- static ExportJob.Id jobIdFromArchiveFileName(String fileName) {
- if (!fileName.endsWith(EXPORT_ARCHIVE_SUFFIX)) {
- return null;
- }
- String nameWithoutSuffix = fileName.substring(0, fileName.length() -
EXPORT_ARCHIVE_SUFFIX.length());
- int jobIndex = nameWithoutSuffix.lastIndexOf(EXPORT_ARCHIVE_JOB_INFIX);
- if (jobIndex < 0) {
- return null;
- }
- String jobId = nameWithoutSuffix.substring(jobIndex +
EXPORT_ARCHIVE_JOB_INFIX.length());
- return UUIDUtil.isValidUuidString(jobId) ? ExportJob.Id.of(jobId) : null;
- }
-
- void cleanupFailedJob(ExportJob.Id jobId, String archivePath) {
- FileUtils.deleteQuietly(jobDirectory(jobId).toFile());
- if (archivePath != null) {
- File archive = new File(archivePath);
- FileUtils.deleteQuietly(archive);
-
FileUtils.deleteQuietly(AtomicFileOutputStream.getTemporaryFile(archive));
- }
+ void cleanupFailedJob(ExportJob job) {
+ FileUtils.deleteQuietly(jobDirectory(job.getId()).toFile());
+ File archive = new File(exportDirectory, job.getFileName());
+ FileUtils.deleteQuietly(archive);
+ FileUtils.deleteQuietly(AtomicFileOutputStream.getTemporaryFile(archive));
}
private Path jobDirectory(ExportJob.Id jobId) {
@@ -235,7 +217,7 @@ private void removeIncompleteWorkOnStartup() {
}
static String exportJobDirName(ExportJob.Id jobId) {
- return EXPORT_JOB_DIR_PREFIX + jobId.getValue();
+ return EXPORT_JOB_DIR_PREFIX + jobId;
}
private static ExportJob.Id jobIdFromExportDirName(String dirName) {
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 ba66a4907f1..1d7a1e79353 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
@@ -19,17 +19,33 @@
import java.io.BufferedWriter;
import java.io.IOException;
+import java.io.OutputStream;
+import java.io.OutputStreamWriter;
+import java.nio.charset.StandardCharsets;
+import java.time.Instant;
import java.util.Objects;
import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+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;
/**
- * Metadata for a container ID export job.
+ * In-memory state for a container ID export job.
*/
public final class ExportJob {
private final Id id;
private final ExportScope scope;
- private final String jobStartTime;
+ private final String jobStartTime =
ExportFileManager.formatJobStartTime(Instant.now());
+ private final ContainerID startContainerId;
+ private final int batchSize;
+ private final int partSize;
+ /** The .tar.gz output path, fixed at job creation; used by the worker to
write the archive. */
+ private final String fileName;
+
+ private final CompletableFuture<Long> idsWrittenFuture = new
CompletableFuture<>();
/**
* Unique job identifier.
@@ -49,10 +65,6 @@ public static Id of(String value) {
return new Id(value);
}
- public String getValue() {
- return value;
- }
-
@Override
public String toString() {
return value;
@@ -75,37 +87,122 @@ public int hashCode() {
}
}
- ExportJob(Id id, ExportScope scope, String jobStartTime) {
- this.id = id;
+ /**
+ * Immutable snapshot of export progress.
+ */
+ public final class Status {
+ private final Long idsWritten;
+ private final Throwable exception;
+
+ private Status() {
+ if (!idsWrittenFuture.isDone()) {
+ this.idsWritten = null;
+ this.exception = null;
+ } else {
+ long rows = -1;
+ Throwable cause = null;
+ try {
+ rows = idsWrittenFuture.join();
+ } catch (CompletionException e) {
+ cause = e.getCause();
+ }
+ this.idsWritten = rows;
+ this.exception = cause;
+ }
+ }
+
+ public ExportJob.Id getId() {
+ return id;
+ }
+
+ public boolean isDone() {
+ return idsWritten != null || exception != null;
+ }
+
+ public long getIdsWritten() {
+ if (!isDone()) {
+ throw new IllegalStateException("Export is not done");
+ }
+ if (exception != null) {
+ throw new IllegalStateException("Export failed", exception);
+ }
+ return idsWritten;
+ }
+
+ public Throwable getException() {
+ return exception;
+ }
+ }
+
+ ExportJob(ExportScope scope, ContainerID startContainerId, int batchSize,
int partSize) {
+ this.id = Id.newId();
this.scope = scope;
- this.jobStartTime = jobStartTime;
+ this.startContainerId = startContainerId != null ? startContainerId :
ContainerID.valueOf(0);
+ this.fileName = ExportFileManager.archiveFileName(scope, jobStartTime, id);
+ this.batchSize = batchSize;
+ this.partSize = partSize;
}
- String partFileName(int partIndex) {
- return String.format("container-ids_%s_%s_part%03d.txt",
- scope.getValue(), jobStartTime, partIndex);
+ Id getId() {
+ return id;
}
- void writeMetadataHeader(BufferedWriter writer, int partNumber, long
partStartContainerId)
- throws IOException {
- writer.write("# jobId=" + id.getValue());
+ LifeCycleState getLifeCycleState() {
+ return scope.getLifeCycleState();
+ }
+
+ ContainerHealthState getHealthState() {
+ return scope.getHealthState();
+ }
+
+ String getFileName() {
+ return fileName;
+ }
+
+ ContainerID getStartContainerId() {
+ return startContainerId;
+ }
+
+ int getBatchSize() {
+ return batchSize;
+ }
+
+ int getPartSize() {
+ return partSize;
+ }
+
+ CompletableFuture<Long> getFuture() {
+ return idsWrittenFuture;
+ }
+
+ Status getStatus() {
+ return new Status();
+ }
+
+ String partFileName(int partNumber) {
+ return String.format("container-ids_%s_%s_part%03d.txt", scope,
jobStartTime, partNumber);
+ }
+
+ void writeMetadataHeader(OutputStream out, int partNumber, ContainerID
partStartContainerId) throws IOException {
+ final BufferedWriter writer = new BufferedWriter(new
OutputStreamWriter(out, StandardCharsets.UTF_8));
+ writer.write("# jobId=" + id);
writer.newLine();
writer.write("# jobStartTime=" + jobStartTime);
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=" + partStartContainerId);
+ writer.write("# scope=" + scope);
+ writer.newLine();
+ writer.write("# partStartContainerId=" + partStartContainerId);
writer.newLine();
writer.write("# part=" + partNumber);
writer.newLine();
writer.write("# format=container-id-per-line");
writer.newLine();
writer.newLine();
+ writer.flush();
+ }
+
+ @Override
+ public String toString() {
+ return "ExportJob-" + id;
}
}
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
index c6f174a3299..7d8a3c4b42f 100644
---
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
@@ -58,13 +58,6 @@ public ContainerHealthState getHealthState() {
return healthState;
}
- /**
- * Stable filter name segment used in export TAR and part 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/server/StorageContainerManager.java
b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/StorageContainerManager.java
index fa32b0b9705..8004619d772 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;
@@ -307,6 +308,7 @@ public final class StorageContainerManager extends
ServiceRuntimeInfoImpl
private final ContainerBalancer containerBalancer;
// MoveManager is used by ContainerBalancer to schedule container moves
private final MoveManager moveManager;
+ private final ContainerExportManager containerExportManager;
private StatefulServiceStateManager statefulServiceStateManager;
// Used to keep track of pending replication and pending deletes for
// container replicas.
@@ -431,6 +433,9 @@ private StorageContainerManager(OzoneConfiguration conf,
initializeSystemManagers(conf, configurator);
+ containerExportManager = new ContainerExportManager(
+ getScmId(), containerManager, this::checkLeader, conf);
+
if (isSecretKeyEnable(securityConfig)) {
secretKeyManagerService = new SecretKeyManagerService(scmContext, conf,
scmHAManager.getRatisServer());
@@ -1570,6 +1575,7 @@ public void start() throws IOException {
scmBlockManager.start();
leaseManager.start();
+ containerExportManager.start();
try {
httpServer = new StorageContainerManagerHttpServer(configuration, this);
@@ -1647,6 +1653,7 @@ public void stop() {
}
stopReplicationManager();
+ stopContainerExportManager();
try {
LOG.info("Stopping the Datanode Admin Monitor.");
@@ -1780,6 +1787,15 @@ private void stopReplicationManager() {
}
}
+ private void stopContainerExportManager() {
+ try {
+ LOG.info("Shutting down Container Export Manager.");
+ containerExportManager.shutdown();
+ } catch (Exception ex) {
+ LOG.error("Container Export Manager shutdown failed.", ex);
+ }
+ }
+
@Override
public void shutDown(String message) {
stop();
@@ -1917,6 +1933,10 @@ public MoveManager getMoveManager() {
return moveManager;
}
+ public ContainerExportManager getContainerExportManager() {
+ return containerExportManager;
+ }
+
/**
* Returns SCM root CA rotation manager.
*/
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 00000000000..9c60af26672
--- /dev/null
+++
b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/export/TestContainerExportManager.java
@@ -0,0 +1,525 @@
+/*
+ * 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.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.verify;
+import static org.mockito.Mockito.when;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.BooleanSupplier;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import java.util.zip.GZIPInputStream;
+import org.apache.commons.compress.archivers.ArchiveInputStream;
+import org.apache.commons.compress.archivers.tar.TarArchiveEntry;
+import org.apache.commons.io.FileUtils;
+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.scm.exceptions.SCMException;
+import org.apache.hadoop.hdds.utils.Archiver;
+import org.apache.ozone.test.GenericTestUtils;
+import org.apache.ratis.util.AtomicFileOutputStream;
+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_BATCH_SIZE = 2;
+ private static final int TEST_PART_SIZE = 3;
+ private static final String TEST_SCM_ID = "test-scm";
+
+ @TempDir
+ private File tempDir;
+
+ private ContainerManager containerManager;
+ private ContainerExportManager exportManager;
+
+ @BeforeEach
+ public void setup() throws Exception {
+ containerManager = mock(ContainerManager.class);
+ exportManager = newExportManager(TEST_PART_SIZE, TEST_BATCH_SIZE, () ->
true);
+ }
+
+ @AfterEach
+ public void teardown() {
+ if (exportManager != null) {
+ exportManager.shutdown();
+ }
+ }
+
+ @Test
+ public void testRejectMissingFilters() {
+ assertThrows(IllegalArgumentException.class, () ->
+ exportManager.submitJob(ContainerID.valueOf(0), null, null));
+ }
+
+ @Test
+ public void testRejectSubmitWhenNotLeader() throws Exception {
+ exportManager.shutdown();
+ exportManager = newExportManager(TEST_PART_SIZE, TEST_BATCH_SIZE, () ->
false);
+ SCMException e = assertThrows(SCMException.class, () ->
+ exportManager.submitJob(ContainerID.valueOf(0), null,
ContainerHealthState.MISSING));
+ assertEquals(SCMException.ResultCodes.SCM_NOT_LEADER, e.getResult());
+ }
+
+ @Test
+ public void testRejectSubmitAfterShutdown() {
+ exportManager.shutdown();
+ IOException e = assertThrows(IOException.class, () ->
+ exportManager.submitJob(ContainerID.valueOf(0), null,
ContainerHealthState.MISSING));
+ assertTrue(e.getMessage().contains("shut down"));
+ }
+
+ @Test
+ public void testRejectNonPositiveSizes() {
+ assertThrows(IllegalArgumentException.class, () ->
+ new ContainerExportManager(TEST_SCM_ID, containerManager, () -> true,
tempDir.getAbsolutePath(), 0, 1));
+ assertThrows(IllegalArgumentException.class, () ->
+ new ContainerExportManager(TEST_SCM_ID, containerManager, () -> true,
tempDir.getAbsolutePath(), 1, 0));
+ }
+
+ @Test
+ public void testGetExportStatusUnknownJobReturnsNull() {
+ assertNull(exportManager.getExportStatus(ExportJob.Id.newId()));
+ }
+
+ @Test
+ public void testRejectConcurrentExport() 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)))
+ .thenAnswer(invocation -> {
+ Thread.sleep(60_000);
+ return Collections.emptyList();
+ });
+
+ ExportJob first = exportManager.submitJob(ContainerID.valueOf(0), null,
+ ContainerHealthState.MISSING);
+ assertNotNull(first);
+ IOException e = assertThrows(IOException.class, () ->
+ exportManager.submitJob(ContainerID.valueOf(0), null,
ContainerHealthState.EMPTY));
+ assertTrue(e.getMessage().contains("already running"));
+ }
+
+ @Test
+ public void testRejectSubmitWhileArchivingThenAcceptAfterDone() throws
Exception {
+ final AtomicBoolean allIdsWritten = new AtomicBoolean();
+ final CountDownLatch archivingStarted = new CountDownLatch(1);
+ final CountDownLatch releaseArchiving = new CountDownLatch(1);
+ exportManager.shutdown();
+ exportManager = newExportManager(TEST_PART_SIZE, TEST_BATCH_SIZE, () -> {
+ // submitJob also calls this on the caller thread; only hold the worker.
+ if (allIdsWritten.get() &&
Thread.currentThread().getName().endsWith("-ContainerExportWorker")) {
+ archivingStarted.countDown();
+ try {
+ releaseArchiving.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ return true;
+ });
+
+ 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 -> {
+ allIdsWritten.set(true);
+ return Collections.emptyList();
+ });
+
+ ExportJob first = exportManager.submitJob(ContainerID.valueOf(0), null,
ContainerHealthState.MISSING);
+ assertTrue(archivingStarted.await(30, TimeUnit.SECONDS));
+ assertFalse(exportManager.getExportStatus(first.getId()).isDone());
+
+ IOException e = assertThrows(IOException.class, () ->
+ exportManager.submitJob(ContainerID.valueOf(0), null,
ContainerHealthState.EMPTY));
+ assertTrue(e.getMessage().contains("already running"));
+
+ releaseArchiving.countDown();
+ ExportJob.Status status = waitForDone(first.getId());
+ assertNull(status.getException());
+ assertTrue(archiveFile(first).exists());
+ assertNotNull(exportManager.submitJob(ContainerID.valueOf(0), null,
ContainerHealthState.EMPTY));
+ }
+
+ @Test
+ public void testNullStartDefaultsToZeroCursor() throws Exception {
+ when(containerManager.getContainerIDs(
+ eq(ContainerID.valueOf(0)), anyInt(), isNull(),
eq(ContainerHealthState.MISSING)))
+ .thenReturn(Collections.emptyList());
+
+ ExportJob job = exportManager.submitJob(null, null,
ContainerHealthState.MISSING);
+ assertNotNull(job);
+ waitForDone(job.getId());
+
+ verify(containerManager).getContainerIDs(
+ eq(ContainerID.valueOf(0)), anyInt(), isNull(),
eq(ContainerHealthState.MISSING));
+ }
+
+ @Test
+ public void testExportStartsFromRequestedCursor() throws Exception {
+ when(containerManager.getContainerIDs(
+ eq(ContainerID.valueOf(5)), anyInt(), isNull(),
eq(ContainerHealthState.MISSING)))
+ .thenReturn(ids(5, 6));
+ when(containerManager.getContainerIDs(
+ eq(ContainerID.valueOf(7)), anyInt(), isNull(),
eq(ContainerHealthState.MISSING)))
+ .thenReturn(Collections.emptyList());
+
+ ExportJob job = exportManager.submitJob(ContainerID.valueOf(5), null,
+ ContainerHealthState.MISSING);
+ assertNotNull(job);
+
+ ExportJob.Status status = waitForDone(job.getId());
+ assertNull(status.getException());
+ assertEquals(2, status.getIdsWritten());
+
+ verify(containerManager).getContainerIDs(
+ eq(ContainerID.valueOf(5)), anyInt(), isNull(),
eq(ContainerHealthState.MISSING));
+ }
+
+ @Test
+ public void testExportWithLifeCycleFilter() throws Exception {
+ when(containerManager.getContainerIDs(
+ eq(ContainerID.valueOf(0)), anyInt(), eq(LifeCycleState.OPEN),
isNull()))
+ .thenReturn(ids(1));
+ when(containerManager.getContainerIDs(
+ eq(ContainerID.valueOf(2)), anyInt(), eq(LifeCycleState.OPEN),
isNull()))
+ .thenReturn(Collections.emptyList());
+
+ ExportJob job = exportManager.submitJob(ContainerID.valueOf(0),
LifeCycleState.OPEN, null);
+ assertNotNull(job);
+ waitForDone(job.getId());
+
+ verify(containerManager).getContainerIDs(
+ eq(ContainerID.valueOf(0)), anyInt(), eq(LifeCycleState.OPEN),
isNull());
+ }
+
+ @Test
+ public void testZeroMatchesSucceedsWithoutArchive() throws Exception {
+ when(containerManager.getContainerIDs(
+ eq(ContainerID.valueOf(0)), anyInt(), isNull(),
eq(ContainerHealthState.MISSING)))
+ .thenReturn(Collections.emptyList());
+
+ ExportJob job = exportManager.submitJob(ContainerID.valueOf(0), null,
+ ContainerHealthState.MISSING);
+ assertNotNull(job);
+
+ ExportJob.Status status = waitForDone(job.getId());
+ assertNull(status.getException());
+ assertEquals(0, status.getIdsWritten());
+ assertFalse(archiveFile(job).exists());
+ }
+
+ @Test
+ public void testSinglePartExportCreatesTar() throws Exception {
+ exportManager.shutdown();
+ exportManager = newExportManager(100, TEST_BATCH_SIZE, () -> true);
+
+ 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());
+
+ ExportJob job = exportManager.submitJob(ContainerID.valueOf(0), null,
+ ContainerHealthState.MISSING);
+ assertNotNull(job);
+
+ ExportJob.Status status = waitForDone(job.getId());
+ assertNull(status.getException());
+ assertEquals(4, status.getIdsWritten());
+
assertTrue(job.getFileName().endsWith(ExportFileManager.EXPORT_ARCHIVE_SUFFIX));
+ assertTrue(archiveFile(job).exists());
+ }
+
+ @Test
+ public void testMultiPartExportCreatesTar() throws Exception {
+ // partSize=3, batchSize=2: part 1 ends mid-batch (after 3), part 2 ends
exactly where batch (5,6) runs out.
+ 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(ids(5, 6));
+ when(containerManager.getContainerIDs(
+ eq(ContainerID.valueOf(7)), anyInt(), isNull(),
eq(ContainerHealthState.MISSING)))
+ .thenReturn(ids(7));
+ when(containerManager.getContainerIDs(
+ eq(ContainerID.valueOf(8)), anyInt(), isNull(),
eq(ContainerHealthState.MISSING)))
+ .thenReturn(Collections.emptyList());
+
+ ExportJob job = exportManager.submitJob(ContainerID.valueOf(0), null,
+ ContainerHealthState.MISSING);
+ assertNotNull(job);
+
+ ExportJob.Status status = waitForDone(job.getId());
+ assertNull(status.getException());
+ assertEquals(7, status.getIdsWritten());
+
assertTrue(job.getFileName().endsWith(ExportFileManager.EXPORT_ARCHIVE_SUFFIX));
+ File archive = archiveFile(job);
+ assertTrue(archive.exists());
+
assertFalse(Files.exists(tempDir.toPath().resolve(ExportFileManager.exportJobDirName(job.getId()))));
+
+ Path extractDir = Files.createTempDirectory("export-archive");
+ try {
+ extractGzTar(archive, extractDir);
+ List<Path> parts;
+ try (Stream<Path> stream = Files.list(extractDir)) {
+ parts = stream.sorted().collect(Collectors.toList());
+ }
+ assertEquals(3, parts.size());
+
assertTrue(parts.get(0).getFileName().toString().endsWith("part001.txt"));
+
assertTrue(parts.get(2).getFileName().toString().endsWith("part003.txt"));
+ assertTrue(Files.readAllLines(parts.get(1)).contains("#
partStartContainerId=#4"));
+ assertTrue(Files.readAllLines(parts.get(2)).contains("#
partStartContainerId=#7"));
+
+ List<String> exportedIds = new ArrayList<>();
+ for (Path part : parts) {
+ for (String line : Files.readAllLines(part)) {
+ if (!line.isEmpty() && !line.startsWith("#")) {
+ exportedIds.add(line);
+ }
+ }
+ }
+ assertEquals(Arrays.asList("1", "2", "3", "4", "5", "6", "7"),
exportedIds);
+ } finally {
+ FileUtils.deleteQuietly(extractDir.toFile());
+ }
+ }
+
+ @Test
+ public void testLeadershipLostWhileArchivingLeavesNoArchive() throws
Exception {
+ ExportFileManager fileManager = new
ExportFileManager(tempDir.getAbsolutePath());
+ ExportJob job = new ExportJob(ExportScope.of(null,
ContainerHealthState.MISSING),
+ ContainerID.valueOf(0), TEST_BATCH_SIZE, TEST_PART_SIZE);
+ fileManager.createJobDirectory(job.getId());
+ try (OutputStream out = fileManager.newPartOutputStream(job.getId(),
job.partFileName(1))) {
+ job.writeMetadataHeader(out, 1, ContainerID.valueOf(1));
+ }
+
+ // Lost before the first part is copied.
+ IOException e = assertThrows(IOException.class, () ->
fileManager.writeArchive(job, () -> false));
+ assertTrue(e.getMessage().contains("lost leadership"));
+ assertFalse(archiveFile(job).exists());
+
assertFalse(AtomicFileOutputStream.getTemporaryFile(archiveFile(job)).exists());
+
+ // Lost only after the last (here: the only) part is copied; the finished
archive is kept.
+ final AtomicInteger checks = new AtomicInteger();
+ fileManager.writeArchive(job, () -> checks.incrementAndGet() <= 1);
+ assertEquals(1, checks.get());
+ assertTrue(archiveFile(job).exists());
+
assertFalse(AtomicFileOutputStream.getTemporaryFile(archiveFile(job)).exists());
+ }
+
+ @Test
+ public void testLockReleasedOnlyAfterWorkerStops() throws Exception {
+ final CountDownLatch workerBusy = new CountDownLatch(1);
+ final CountDownLatch releaseWorker = new CountDownLatch(1);
+ when(containerManager.getContainerIDs(
+ eq(ContainerID.valueOf(0)), anyInt(), isNull(),
eq(ContainerHealthState.MISSING)))
+ .thenAnswer(invocation -> {
+ workerBusy.countDown();
+ try {
+ releaseWorker.await();
+ } catch (InterruptedException ignored) {
+ // shutdownNow() interrupts; keep running to simulate a worker
that is slow to stop.
+ }
+ releaseWorker.await();
+ return Collections.emptyList();
+ });
+
+ ExportJob job = exportManager.submitJob(ContainerID.valueOf(0), null,
ContainerHealthState.MISSING);
+ assertTrue(workerBusy.await(30, TimeUnit.SECONDS));
+
+ // Returns after the 5s timeout while the worker is still running.
+ exportManager.shutdown();
+ ExportFileManager other = new ExportFileManager(tempDir.getAbsolutePath());
+ assertThrows(IOException.class, other::start);
+
+ releaseWorker.countDown();
+ GenericTestUtils.waitFor(() -> {
+ try {
+ other.start();
+ return true;
+ } catch (IOException ex) {
+ return false;
+ }
+ }, 100, 30_000);
+ other.unlock();
+ assertTrue(waitForDone(job.getId()).isDone());
+ }
+
+ @Test
+ public void testSubmitAfterPreviousJobCompletes() 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());
+ 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());
+
+ ExportJob first = exportManager.submitJob(ContainerID.valueOf(0), null,
+ ContainerHealthState.MISSING);
+ waitForDone(first.getId());
+
+ ExportJob second = exportManager.submitJob(ContainerID.valueOf(0), null,
+ ContainerHealthState.EMPTY);
+ assertNotNull(second);
+ ExportJob.Status status = waitForDone(second.getId());
+ assertNull(status.getException());
+ assertEquals(1, status.getIdsWritten());
+ }
+
+ @Test
+ public void testFailsWhenLeadershipLostDuringExport() throws Exception {
+ AtomicBoolean leader = new AtomicBoolean(true);
+ exportManager.shutdown();
+ exportManager = newExportManager(TEST_PART_SIZE, TEST_BATCH_SIZE,
leader::get);
+
+ when(containerManager.getContainerIDs(
+ eq(ContainerID.valueOf(0)), anyInt(), isNull(),
eq(ContainerHealthState.MISSING)))
+ .thenAnswer(invocation -> {
+ leader.set(false);
+ return ids(1);
+ });
+
+ ExportJob job = exportManager.submitJob(ContainerID.valueOf(0), null,
+ ContainerHealthState.MISSING);
+ assertNotNull(job);
+
+ ExportJob.Status status = waitForDone(job.getId());
+ assertNotNull(status.getException());
+ assertTrue(status.getException().getMessage().contains("lost leadership"));
+ }
+
+ @Test
+ public void testFailsWhenContainerManagerThrows() throws Exception {
+ when(containerManager.getContainerIDs(
+ eq(ContainerID.valueOf(0)), anyInt(), isNull(),
eq(ContainerHealthState.MISSING)))
+ .thenThrow(new RuntimeException("container listing failed"));
+
+ ExportJob job = exportManager.submitJob(ContainerID.valueOf(0), null,
+ ContainerHealthState.MISSING);
+ assertNotNull(job);
+
+ ExportJob.Status status = waitForDone(job.getId());
+ assertNotNull(status.getException());
+ assertTrue(status.getException().getMessage().contains("container listing
failed"));
+
assertFalse(Files.exists(tempDir.toPath().resolve(ExportFileManager.exportJobDirName(job.getId()))));
+ }
+
+ @Test
+ public void testShutdownCancelsRunningExport() throws Exception {
+ when(containerManager.getContainerIDs(
+ eq(ContainerID.valueOf(0)), anyInt(), isNull(),
eq(ContainerHealthState.MISSING)))
+ .thenAnswer(invocation -> {
+ Thread.sleep(60_000);
+ return Collections.emptyList();
+ });
+
+ ExportJob job = exportManager.submitJob(ContainerID.valueOf(0), null,
+ ContainerHealthState.MISSING);
+ assertNotNull(job);
+
+ exportManager.shutdown();
+ ExportJob.Status status = waitForDone(job.getId());
+ assertNotNull(status.getException());
+ exportManager = null;
+ }
+
+ private File archiveFile(ExportJob job) {
+ return new File(tempDir, job.getFileName());
+ }
+
+ private ContainerExportManager newExportManager(int partSize, int batchSize,
+ BooleanSupplier isLeaderReady) throws Exception {
+ ContainerExportManager manager = new ContainerExportManager(TEST_SCM_ID,
containerManager, isLeaderReady,
+ tempDir.getAbsolutePath(), partSize, batchSize);
+ manager.start();
+ return manager;
+ }
+
+ private static List<ContainerID> ids(long... values) {
+ return Arrays.stream(values).mapToObj(ContainerID::valueOf)
+ .collect(Collectors.toList());
+ }
+
+ private ExportJob.Status waitForDone(ExportJob.Id jobId) throws Exception {
+ GenericTestUtils.waitFor(() -> {
+ ExportJob.Status status = exportManager.getExportStatus(jobId);
+ return status != null && status.isDone();
+ }, 100, 30_000);
+ return exportManager.getExportStatus(jobId);
+ }
+
+ private static void extractGzTar(File archive, Path extractDir) throws
Exception {
+ Files.createDirectories(extractDir);
+ try (InputStream in = new
GZIPInputStream(Files.newInputStream(archive.toPath()));
+ ArchiveInputStream<TarArchiveEntry> tarIn = Archiver.untar(in)) {
+ TarArchiveEntry entry;
+ while ((entry = tarIn.getNextEntry()) != null) {
+ Archiver.extractEntry(entry, tarIn, entry.getSize(), extractDir,
extractDir.resolve(entry.getName()));
+ }
+ }
+ }
+}
diff --git
a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/export/TestExportFileManager.java
b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/export/TestExportFileManager.java
deleted file mode 100644
index a97255ba198..00000000000
---
a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/export/TestExportFileManager.java
+++ /dev/null
@@ -1,231 +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 static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertFalse;
-import static org.junit.jupiter.api.Assertions.assertNull;
-import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-
-import java.io.BufferedWriter;
-import java.io.File;
-import java.io.InputStream;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.util.List;
-import java.util.UUID;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
-import java.util.zip.GZIPInputStream;
-import org.apache.commons.compress.archivers.ArchiveInputStream;
-import org.apache.commons.compress.archivers.tar.TarArchiveEntry;
-import org.apache.commons.io.FileUtils;
-import org.apache.hadoop.hdds.protocol.proto.HddsProtos.LifeCycleState;
-import org.apache.hadoop.hdds.scm.container.ContainerHealthState;
-import org.apache.hadoop.hdds.utils.Archiver;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.io.TempDir;
-
-/**
- * Tests for {@link ExportFileManager}.
- */
-public class TestExportFileManager {
-
- private static final String TEST_JOB_START_TIME = "2026-01-01-12-00-00";
-
- @TempDir
- private File tempDir;
-
- private ExportFileManager fileManager;
-
- @BeforeEach
- public void setup() throws Exception {
- fileManager = new ExportFileManager(tempDir.getAbsolutePath());
- fileManager.start();
- }
-
- @Test
- public void testExportScopeUsesAnyForNullFilters() {
- assertEquals("health-MISSING_lifecycle-ANY",
- ExportScope.of(null, ContainerHealthState.MISSING).getValue());
- assertEquals("health-ANY_lifecycle-OPEN",
- ExportScope.of(LifeCycleState.OPEN, null).getValue());
- }
-
- @Test
- public void testRejectMissingFilters() {
- assertThrows(IllegalArgumentException.class, () -> ExportScope.of(null,
null));
- }
-
- @Test
- public void testResolveArchiveFile() {
- ExportJob.Id jobId = ExportJob.Id.newId();
- ExportScope scope = ExportScope.of(null, ContainerHealthState.MISSING);
- File archive = fileManager.resolveArchiveFile(scope, TEST_JOB_START_TIME,
jobId);
- assertTrue(archive.getName().contains("health-MISSING_lifecycle-ANY_" +
TEST_JOB_START_TIME));
-
assertTrue(archive.getName().endsWith(ExportFileManager.EXPORT_ARCHIVE_JOB_INFIX
+ jobId.getValue()
- + ExportFileManager.EXPORT_ARCHIVE_SUFFIX));
- }
-
- @Test
- public void testResolveArchiveTempFile() {
- ExportJob.Id jobId = ExportJob.Id.newId();
- ExportScope scope = ExportScope.of(null, ContainerHealthState.MISSING);
- File tempFile = fileManager.resolveArchiveTempFile(scope,
TEST_JOB_START_TIME, jobId);
-
assertTrue(tempFile.getName().endsWith(ExportFileManager.EXPORT_ARCHIVE_TMP_SUFFIX));
- }
-
- @Test
- public void testJobIdFromArchiveFileName() {
- String jobId = UUID.randomUUID().toString();
- String fileName = "container-ids_health-MISSING_lifecycle-ANY_" +
TEST_JOB_START_TIME
- + ExportFileManager.EXPORT_ARCHIVE_JOB_INFIX + jobId +
ExportFileManager.EXPORT_ARCHIVE_SUFFIX;
- assertEquals(ExportJob.Id.of(jobId),
ExportFileManager.jobIdFromArchiveFileName(fileName));
-
assertNull(ExportFileManager.jobIdFromArchiveFileName("container-ids_health-MISSING_lifecycle-ANY_"
- + TEST_JOB_START_TIME + ExportFileManager.EXPORT_ARCHIVE_SUFFIX));
- }
-
- @Test
- public void testJobStartTimeFromArchiveFileName() {
- String fileName = "container-ids_health-MISSING_lifecycle-ANY_" +
TEST_JOB_START_TIME
- + ExportFileManager.EXPORT_ARCHIVE_JOB_INFIX + UUID.randomUUID() +
ExportFileManager.EXPORT_ARCHIVE_SUFFIX;
- assertEquals(TEST_JOB_START_TIME,
ExportFileManager.jobStartTimeFromArchiveFileName(fileName));
- }
-
- @Test
- public void testListCompletedArchivePaths() throws Exception {
- ExportScope scope = ExportScope.of(null, ContainerHealthState.MISSING);
- ExportJob.Id olderJobId = ExportJob.Id.newId();
- File olderArchive = fileManager.resolveArchiveFile(scope,
"2026-01-01-12-00-00", olderJobId);
- assertTrue(olderArchive.createNewFile());
- assertTrue(olderArchive.setLastModified(2_000L));
- ExportJob.Id newerJobId = ExportJob.Id.newId();
- File newerArchive = fileManager.resolveArchiveFile(scope,
"2026-01-01-12-00-01", newerJobId);
- assertTrue(newerArchive.createNewFile());
- assertTrue(newerArchive.setLastModified(1_000L));
- ExportJob.Id tempJobId = ExportJob.Id.newId();
- File tempArchive = fileManager.resolveArchiveTempFile(scope,
"2026-01-01-12-00-02", tempJobId);
- assertTrue(tempArchive.createNewFile());
-
- List<String> completedPaths = fileManager.listCompletedArchivePaths();
- assertEquals(2, completedPaths.size());
- assertEquals(olderArchive.getAbsolutePath(), completedPaths.get(0));
- assertEquals(newerArchive.getAbsolutePath(), completedPaths.get(1));
- }
-
- @Test
- public void testWriteArchiveFromPartFiles() throws Exception {
- ExportJob.Id jobId = ExportJob.Id.newId();
- ExportScope scope = ExportScope.of(null, ContainerHealthState.MISSING);
- ExportJob job = new ExportJob(jobId, scope, TEST_JOB_START_TIME);
- File archive = fileManager.resolveArchiveFile(scope, TEST_JOB_START_TIME,
jobId);
-
- fileManager.createJobDirectory(jobId);
- try (BufferedWriter writer = fileManager.newPartWriter(jobId,
job.partFileName(1))) {
- job.writeMetadataHeader(writer, 1, 1L);
- writer.write("1\n2\n");
- }
-
- fileManager.writeArchive(jobId, archive.getAbsolutePath());
- fileManager.deleteJobDirectory(jobId);
-
- assertTrue(archive.exists());
- Path extractDir = Files.createTempDirectory("export-archive");
- try {
- extractGzTar(archive, extractDir);
- List<String> partNames;
- try (Stream<Path> stream = Files.list(extractDir)) {
- partNames = stream.map(path ->
path.getFileName().toString()).collect(Collectors.toList());
- }
- assertEquals(1, partNames.size());
- assertTrue(partNames.get(0).endsWith("part001.txt"));
- } finally {
- FileUtils.deleteQuietly(extractDir.toFile());
- }
- }
-
- @Test
- public void testCleanupFailedJobRemovesJobDirAndTempArchive() throws
Exception {
- ExportJob.Id jobId = ExportJob.Id.newId();
- ExportScope scope = ExportScope.of(null, ContainerHealthState.MISSING);
- File archive = fileManager.resolveArchiveFile(scope, TEST_JOB_START_TIME,
jobId);
- File tempArchive = fileManager.resolveArchiveTempFile(scope,
TEST_JOB_START_TIME, jobId);
- assertTrue(tempArchive.createNewFile());
-
- fileManager.createJobDirectory(jobId);
- fileManager.cleanupFailedJob(jobId, archive.getAbsolutePath());
-
-
assertFalse(Files.exists(tempDir.toPath().resolve(ExportFileManager.exportJobDirName(jobId))));
- assertFalse(tempArchive.exists());
- assertFalse(archive.exists());
- }
-
- @Test
- public void testOrphanJobDirRemovedOnStartup() throws Exception {
- ExportJob.Id jobId = ExportJob.Id.newId();
- Path orphanJobDir =
tempDir.toPath().resolve(ExportFileManager.exportJobDirName(jobId));
- Files.createDirectories(orphanJobDir);
-
- fileManager.start();
-
- assertFalse(Files.exists(orphanJobDir));
- }
-
- @Test
- public void testIncompleteExportArtifactsRemovedOnStartup() throws Exception
{
- ExportJob.Id jobId = ExportJob.Id.newId();
- Path jobDir =
tempDir.toPath().resolve(ExportFileManager.exportJobDirName(jobId));
- Files.createDirectories(jobDir);
- ExportScope scope = ExportScope.of(null, ContainerHealthState.MISSING);
- File partialArchiveTemp = fileManager.resolveArchiveTempFile(scope,
"2026-01-01-00-00-00", jobId);
- assertTrue(partialArchiveTemp.createNewFile());
-
- fileManager.start();
-
- assertFalse(Files.exists(jobDir));
- assertFalse(partialArchiveTemp.exists());
- }
-
- @Test
- public void testOrphanJobDirDoesNotDeleteCompletedTar() throws Exception {
- ExportJob.Id jobId = ExportJob.Id.newId();
- ExportScope scope = ExportScope.of(null, ContainerHealthState.MISSING);
- File completedArchive = fileManager.resolveArchiveFile(scope,
"2026-01-01-00-00-00", jobId);
- assertTrue(completedArchive.createNewFile());
- Path orphanJobDir =
tempDir.toPath().resolve(ExportFileManager.exportJobDirName(jobId));
- Files.createDirectories(orphanJobDir);
-
- fileManager.start();
-
- assertTrue(completedArchive.exists());
- assertFalse(Files.exists(orphanJobDir));
- }
-
- private static void extractGzTar(File archive, Path extractDir) throws
Exception {
- Files.createDirectories(extractDir);
- try (InputStream in = new
GZIPInputStream(Files.newInputStream(archive.toPath()));
- ArchiveInputStream<TarArchiveEntry> tarIn = Archiver.untar(in)) {
- TarArchiveEntry entry;
- while ((entry = tarIn.getNextEntry()) != null) {
- Archiver.extractEntry(entry, tarIn, entry.getSize(), extractDir,
extractDir.resolve(entry.getName()));
- }
- }
- }
-}
diff --git
a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/export/TestExportJob.java
b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/export/TestExportJob.java
deleted file mode 100644
index fdad57b4dc3..00000000000
---
a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/export/TestExportJob.java
+++ /dev/null
@@ -1,67 +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 static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-
-import java.io.BufferedWriter;
-import java.io.File;
-import java.nio.file.Files;
-import java.util.List;
-import org.apache.hadoop.hdds.protocol.proto.HddsProtos.LifeCycleState;
-import org.apache.hadoop.hdds.scm.container.ContainerHealthState;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.io.TempDir;
-
-/**
- * Tests for {@link ExportJob}.
- */
-public class TestExportJob {
-
- private static final String TEST_JOB_START_TIME = "2026-01-01-12-00-00";
-
- @TempDir
- private File tempDir;
-
- @Test
- public void testPartFileName() {
- ExportJob job = newJob(ContainerHealthState.MISSING, null);
-
assertEquals("container-ids_health-MISSING_lifecycle-ANY_2026-01-01-12-00-00_part001.txt",
job.partFileName(1));
- }
-
- @Test
- public void testWriteMetadataHeader() throws Exception {
- ExportJob job = newJob(ContainerHealthState.MISSING, LifeCycleState.OPEN);
- File headerFile = new File(tempDir, "part-header.txt");
- try (BufferedWriter writer = Files.newBufferedWriter(headerFile.toPath()))
{
- job.writeMetadataHeader(writer, 2, 42L);
- }
- List<String> lines = Files.readAllLines(headerFile.toPath());
- assertTrue(lines.contains("# jobStartTime=" + TEST_JOB_START_TIME));
- assertTrue(lines.contains("# healthState=MISSING"));
- assertTrue(lines.contains("# lifecycleState=OPEN"));
- assertTrue(lines.contains("# startContainerId=42"));
- assertTrue(lines.contains("# part=2"));
- }
-
- private static ExportJob newJob(ContainerHealthState healthState,
LifeCycleState lifeCycleState) {
- ExportScope scope = ExportScope.of(lifeCycleState, healthState);
- return new ExportJob(ExportJob.Id.newId(), scope, TEST_JOB_START_TIME);
- }
-}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
