HDDS-12233. Atomically import a container

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueContainer.java

@@ -17,6 +17,7 @@
 
 package org.apache.hadoop.ozone.container.keyvalue;
 
+import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerDataProto.State.RECOVERING;
 import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result.CONTAINER_ALREADY_EXISTS;
 import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result.CONTAINER_FILES_CREATE_ERROR;
 import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result.CONTAINER_INTERNAL_ERROR;
@@ -666,7 +667,8 @@ public void importContainerData(InputStream input,
 
   public void importContainerData(KeyValueContainerData originalContainerData)
       throws IOException {
-    containerData.setState(originalContainerData.getState());
+    // place the container in the Recovering state while it is being imported
+    containerData.setState(RECOVERING);
     containerData
         .setContainerDBType(originalContainerData.getContainerDBType());
     containerData.setSchemaVersion(originalContainerData.getSchemaVersion());
@@ -681,6 +683,10 @@ public void importContainerData(KeyValueContainerData originalContainerData)
 
     //fill in memory stat counter (keycount, byte usage)
     KeyValueContainerUtil.parseKVContainerData(containerData, config);
+
+    // restore imported container's state to the original state and flush the yaml file
+    containerData.setState(originalContainerData.getState());
+    update(originalContainerData.getMetadata(), true);
   }
 
   @Override

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/helpers/BlockUtils.java

@@ -328,7 +328,7 @@ public static void loadKVContainerDataFromFiles(
       File metaDir = new File(containerData.getMetadataPath());
       File dumpDir = DatanodeStoreSchemaThreeImpl.getDumpDir(metaDir);
       try {
-        store.loadKVContainerData(dumpDir);
+        store.loadKVContainerData(dumpDir, store);
       } catch (IOException e) {
         // Don't delete unloaded or partially loaded files on failure,
         // but delete all partially loaded metadata.

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/metadata/DatanodeStoreSchemaThreeImpl.java

@@ -31,17 +31,25 @@
 import org.apache.hadoop.hdds.upgrade.HDDSLayoutFeature;
 import org.apache.hadoop.hdds.utils.MetadataKeyFilters;
 import org.apache.hadoop.hdds.utils.db.BatchOperation;
+import org.apache.hadoop.hdds.utils.db.Codec;
 import org.apache.hadoop.hdds.utils.db.FixedLengthStringCodec;
+import org.apache.hadoop.hdds.utils.db.LongCodec;
+import org.apache.hadoop.hdds.utils.db.Proto2Codec;
 import org.apache.hadoop.hdds.utils.db.RDBStore;
 import org.apache.hadoop.hdds.utils.db.RocksDatabase;
 import org.apache.hadoop.hdds.utils.db.RocksDatabase.ColumnFamily;
 import org.apache.hadoop.hdds.utils.db.Table;
 import org.apache.hadoop.hdds.utils.db.managed.ManagedCompactRangeOptions;
+import org.apache.hadoop.hdds.utils.db.managed.ManagedOptions;
+import org.apache.hadoop.hdds.utils.db.managed.ManagedReadOptions;
+import org.apache.hadoop.hdds.utils.db.managed.ManagedSstFileReader;
+import org.apache.hadoop.hdds.utils.db.managed.ManagedSstFileReaderIterator;
 import org.apache.hadoop.ozone.container.common.helpers.BlockData;
 import org.apache.hadoop.ozone.container.common.interfaces.BlockIterator;
 import org.apache.hadoop.ozone.container.common.statemachine.DatanodeConfiguration;
 import org.apache.hadoop.ozone.container.upgrade.VersionedDatanodeFeatures;
 import org.rocksdb.LiveFileMetaData;
+import org.rocksdb.RocksDBException;
 
 /**
  * Constructs a datanode store in accordance with schema version 3, which uses
@@ -131,18 +139,39 @@ public void dumpKVContainerData(long containerID, File dumpDir)
         prefix);
   }
 
-  public void loadKVContainerData(File dumpDir)
-      throws IOException {
-    getMetadataTable().loadFromFile(
-        getTableDumpFile(getMetadataTable(), dumpDir));
-    getBlockDataTable().loadFromFile(
-        getTableDumpFile(getBlockDataTable(), dumpDir));
-    if (VersionedDatanodeFeatures.isFinalized(HDDSLayoutFeature.HBASE_SUPPORT)) {
-      getLastChunkInfoTable().loadFromFile(
-          getTableDumpFile(getLastChunkInfoTable(), dumpDir));
+  public void loadKVContainerData(File dumpDir, DatanodeStoreSchemaThreeImpl store) throws IOException {
+    try (BatchOperation batch = store.getBatchHandler().initBatchOperation()) {
+      processTable(batch, getTableDumpFile(getMetadataTable(), dumpDir),
+          FixedLengthStringCodec.get(), LongCodec.get(), getMetadataTable());
+      processTable(batch, getTableDumpFile(getBlockDataTable(), dumpDir),
+          FixedLengthStringCodec.get(), BlockData.getCodec(), getBlockDataTable());
+      if (VersionedDatanodeFeatures.isFinalized(HDDSLayoutFeature.HBASE_SUPPORT)) {
+        processTable(batch, getTableDumpFile(getLastChunkInfoTable(), dumpDir),
+            FixedLengthStringCodec.get(), BlockData.getCodec(), getLastChunkInfoTable());
+      }
+      processTable(batch, getTableDumpFile(getDeleteTransactionTable(), dumpDir), FixedLengthStringCodec.get(),
+          Proto2Codec.get(DeletedBlocksTransaction.getDefaultInstance()), getDeleteTransactionTable());
+
+      store.getStore().commitBatchOperation(batch);
+    }
+  }
+
+  private <K, V> void processTable(BatchOperation batch, File tableDumpFile,
+      Codec<K> keyCodec, Codec<V> valueCodec, Table<K, V> table) throws IOException {
+    try (ManagedSstFileReader sstFileReader = new ManagedSstFileReader(new ManagedOptions());
+         ManagedSstFileReaderIterator iterator =
+             ManagedSstFileReaderIterator.managed(sstFileReader.newIterator(new ManagedReadOptions()))) {
+      sstFileReader.open(tableDumpFile.getAbsolutePath());
+      for (iterator.get().seekToFirst(); iterator.get().isValid(); iterator.get().next()) {
+        byte[] key = iterator.get().key();
+        byte[] value = iterator.get().value();
+        K decodedKey = keyCodec.fromPersistedFormat(key);
+        V decodedValue = valueCodec.fromPersistedFormat(value);
+        table.putWithBatch(batch, decodedKey, decodedValue);
+      }
+    } catch (RocksDBException e) {
+      LOG.error("Failed to import SST file", e);
     }
-    getDeleteTransactionTable().loadFromFile(
-        getTableDumpFile(getDeleteTransactionTable(), dumpDir));
   }
 
   public static File getTableDumpFile(Table<String, ?> table,