szetszwo commented on code in PR #10813:
URL: https://github.com/apache/ozone/pull/10813#discussion_r3797611625
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/StorageContainerManager.java:
##########
@@ -436,6 +439,10 @@ private StorageContainerManager(OzoneConfiguration conf,
initializeSystemManagers(conf, configurator);
+ containerExportManager = new ContainerExportManager(
+ getScmId(), containerManager, this::checkLeader, conf);
+ containerExportManager.start();
Review Comment:
containerExportManager.start() should be called in
StorageContainerManager.start() but not the constructor.
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportJob.java:
##########
@@ -69,6 +85,165 @@ public int hashCode() {
}
}
- private ExportJob() {
+ /**
+ * 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 ExportJob.this.getExecutionState();
+ }
+
+ public long getTotalRows() {
+ return ExportJob.this.getTotalRows();
+ }
+
+ public String getTarPath() {
+ return ExportJob.this.getTarPath();
+ }
+
+ public String getErrorMessage() {
+ return ExportJob.this.getErrorMessage();
+ }
+ }
Review Comment:
Remove State for now. It is currently not used and the synchronization is
incorrect. Let's add it later.
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportManager.java:
##########
@@ -0,0 +1,256 @@
+/*
+ * 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.time.Instant;
+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.Executors;
+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.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, shard 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_PAGE_SIZE = 100_000;
+ private static final int DEFAULT_SHARD_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.Id> runningJobId = new
AtomicReference<>();
+ private final ExecutorService workerPool;
+ private final ContainerManager containerManager;
+ private final ExportFileManager fileManager;
+ private final BooleanSupplier isLeaderReady;
+ private final int shardSize;
+ private final int pageSize;
Review Comment:
If I understand correctly, "shard" and "page" may not be the best names here
since
- shard: a specific name for horizontal partition in a database but here we
don't have a database.
- pageSize: it is actually a batchSize since the size information is
forgotten after the file is written.
Google suggested to call the sub-files "parts".
<img width="605" height="428" alt="Image"
src="https://github.com/user-attachments/assets/5fc66003-7ab1-4e3b-8ede-0f6497775e22";
/>
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportJob.java:
##########
@@ -17,14 +17,30 @@
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 org.apache.hadoop.hdds.protocol.proto.HddsProtos.LifeCycleState;
+import org.apache.hadoop.hdds.scm.container.ContainerHealthState;
+import org.apache.hadoop.hdds.scm.container.ContainerID;
/**
- * Container ID export job identifier.
+ * 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 ContainerID startContainerId;
+ private final int pageSize;
+ private final int shardSize;
+ private String tarPath;
Review Comment:
Why tarPath is not final? Explain how could it be changed?
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportJob.java:
##########
@@ -69,6 +85,165 @@ public int hashCode() {
}
}
- private ExportJob() {
+ /**
+ * 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 ExportJob.this.getExecutionState();
+ }
+
+ public long getTotalRows() {
+ return ExportJob.this.getTotalRows();
+ }
+
+ public String getTarPath() {
+ return ExportJob.this.getTarPath();
+ }
+
+ public String getErrorMessage() {
+ return ExportJob.this.getErrorMessage();
+ }
+ }
+
+ /**
+ * Job execution state.
+ */
+ public enum ExecutionState {
+ RUNNING(false),
+ SUCCEEDED(true),
+ FAILED(true);
+
+ private final boolean terminal;
+
+ ExecutionState(boolean terminal) {
+ this.terminal = terminal;
+ }
+
+ public boolean isTerminal() {
+ return terminal;
+ }
+ }
+
+ ExportJob(Id id, ExportScope scope, String jobStartTime, String tarPath,
ContainerID startContainerId,
+ int pageSize, int shardSize) {
+ this.id = id;
+ this.scope = scope;
+ this.jobStartTime = jobStartTime;
+ this.tarPath = tarPath;
+ this.startContainerId = startContainerId != null ? startContainerId :
ContainerID.valueOf(0);
+ this.pageSize = pageSize;
+ this.shardSize = shardSize;
+ }
+
+ Id getId() {
+ return id;
+ }
+
+ ContainerID getStartContainerId() {
+ return startContainerId;
+ }
+
+ LifeCycleState getLifeCycleState() {
+ return scope.getLifeCycleState();
+ }
+
+ ContainerHealthState getHealthState() {
+ return scope.getHealthState();
+ }
+
+ int getPageSize() {
+ return pageSize;
+ }
+
+ int getShardSize() {
+ return shardSize;
+ }
+
+ synchronized String getTarPath() {
+ return tarPath;
+ }
+
+ synchronized ExecutionState getExecutionState() {
+ return executionState;
+ }
+
+ synchronized long getTotalRows() {
+ return totalRows;
+ }
+
+ synchronized String getErrorMessage() {
+ return errorMessage;
+ }
+
+ synchronized void startExecution() {
+ if (executionState.isTerminal()) {
+ throw new IllegalStateException("Export job " + id + " is already
terminal: " + executionState);
+ }
+ }
+
+ synchronized void updateTotalRows(long rows) {
+ totalRows = rows;
+ }
+
+ synchronized void completeWithNoMatches() {
+ tarPath = null;
+ transitionToTerminal(ExecutionState.SUCCEEDED);
+ }
+
+ synchronized void completeWithArchive(String archivePath) {
+ tarPath = archivePath;
+ transitionToTerminal(ExecutionState.SUCCEEDED);
+ }
+
+ synchronized void fail(String message) {
+ errorMessage = message;
+ transitionToTerminal(ExecutionState.FAILED);
+ }
+
+ private synchronized void transitionToTerminal(ExecutionState terminalState)
{
+ if (executionState.isTerminal()) {
+ throw new IllegalStateException("Export job " + id + " is already
terminal: " + executionState);
+ }
+ executionState = terminalState;
+ }
+
+ Status toStatus() {
+ return new Status();
+ }
+
+ String shardFileName(int partIndex) {
+ return String.format("container-ids_%s_%s_part%03d.txt",
Review Comment:
Indeed, you are already using "part" -- don't use different words for the
same thing.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
