Fix based on review

Author: KodaiD

core/src/main/java/com/scalar/db/common/CoreError.java

@@ -931,6 +931,18 @@ public enum CoreError implements ScalarDbError {
       "Conditions on indexed columns in cross-partition scan operations are not allowed in the SERIALIZABLE isolation level",
       "",
       ""),
+  OBJECT_STORAGE_CLOUD_STORAGE_SERVICE_ACCOUNT_KEY_NOT_FOUND(
+      Category.USER_ERROR,
+      "0263",
+      "The service account key for Cloud Storage is not found.",
+      "",
+      ""),
+  OBJECT_STORAGE_CLOUD_STORAGE_SERVICE_ACCOUNT_KEY_LOAD_FAILED(
+      Category.USER_ERROR,
+      "0264",
+      "Failed to load the service account key for Cloud Storage.",
+      "",
+      ""),
 
   //
   // Errors for the concurrency error category

core/src/main/java/com/scalar/db/storage/objectstorage/ObjectStorageWrapper.java

@@ -68,7 +68,7 @@ public interface ObjectStorageWrapper {
   /**
    * Delete objects with the specified prefix from the storage. <br>
    * <br>
-   * <strong>Attention:</strong> This method does not guarantee atomicity and assume to be used
+   * <strong>Attention:</strong> This method does not guarantee atomicity and is assumed to be used
    * where concurrent operations do not occur.
    *
    * @param prefix the prefix of the objects to delete

core/src/main/java/com/scalar/db/storage/objectstorage/cloudstorage/CloudStorageConfig.java

@@ -2,8 +2,14 @@
 
 import static com.scalar.db.config.ConfigUtils.getInt;
 
+import com.google.auth.Credentials;
+import com.google.auth.oauth2.ServiceAccountCredentials;
+import com.scalar.db.common.CoreError;
 import com.scalar.db.config.DatabaseConfig;
 import com.scalar.db.storage.objectstorage.ObjectStorageConfig;
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
 import java.util.Optional;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -71,6 +77,21 @@ public String getProjectId() {
     return projectId;
   }
 
+  public Credentials getCredentials() {
+    String serviceAccountJson = getPassword();
+    if (serviceAccountJson == null) {
+      throw new IllegalArgumentException(
+          CoreError.OBJECT_STORAGE_CLOUD_STORAGE_SERVICE_ACCOUNT_KEY_NOT_FOUND.buildMessage());
+    }
+    try (ByteArrayInputStream keyStream =
+        new ByteArrayInputStream(serviceAccountJson.getBytes(StandardCharsets.UTF_8))) {
+      return ServiceAccountCredentials.fromStream(keyStream);
+    } catch (IOException e) {
+      throw new IllegalArgumentException(
+          CoreError.OBJECT_STORAGE_CLOUD_STORAGE_SERVICE_ACCOUNT_KEY_LOAD_FAILED.buildMessage());
+    }
+  }
+
   public Optional<Integer> getParallelUploadBlockSizeInBytes() {
     return Optional.ofNullable(parallelUploadBlockSizeInBytes);
   }

core/src/main/java/com/scalar/db/storage/objectstorage/cloudstorage/CloudStorageWrapper.java

@@ -1,7 +1,6 @@
 package com.scalar.db.storage.objectstorage.cloudstorage;
 
 import com.google.api.gax.paging.Page;
-import com.google.auth.oauth2.ServiceAccountCredentials;
 import com.google.cloud.WriteChannel;
 import com.google.cloud.storage.Blob;
 import com.google.cloud.storage.BlobId;
@@ -16,7 +15,6 @@
 import com.scalar.db.storage.objectstorage.ObjectStorageWrapperResponse;
 import com.scalar.db.storage.objectstorage.PreconditionFailedException;
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
-import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
@@ -36,22 +34,10 @@ public class CloudStorageWrapper implements ObjectStorageWrapper {
   private final Integer parallelUploadBlockSizeInBytes;
 
   public CloudStorageWrapper(CloudStorageConfig config) {
-    ServiceAccountCredentials credentials;
-    if (config.getPassword() == null) {
-      throw new RuntimeException(
-          "Service account credentials are not provided in the password field");
-    }
-    try (ByteArrayInputStream keyStream =
-        new ByteArrayInputStream(config.getPassword().getBytes(StandardCharsets.UTF_8))) {
-      credentials = ServiceAccountCredentials.fromStream(keyStream);
-    } catch (IOException e) {
-      throw new RuntimeException("Failed to load the service account credentials", e);
-    }
-
     storage =
         StorageOptions.newBuilder()
             .setProjectId(config.getProjectId())
-            .setCredentials(credentials)
+            .setCredentials(config.getCredentials())
             .build()
             .getService();
     bucket = config.getBucket();
@@ -108,18 +94,8 @@ public Set<String> getKeys(String prefix) throws ObjectStorageWrapperException {
   @Override
   public void insert(String key, String object) throws ObjectStorageWrapperException {
     try {
-      byte[] data = object.getBytes(StandardCharsets.UTF_8);
-      BlobInfo blobInfo = BlobInfo.newBuilder(BlobId.of(bucket, key)).build();
       Storage.BlobWriteOption precondition = Storage.BlobWriteOption.doesNotExist();
-      try (WriteChannel writer = storage.writer(blobInfo, precondition)) {
-        if (parallelUploadBlockSizeInBytes != null) {
-          writer.setChunkSize(parallelUploadBlockSizeInBytes);
-        }
-        ByteBuffer buffer = ByteBuffer.wrap(data);
-        while (buffer.hasRemaining()) {
-          writer.write(buffer);
-        }
-      }
+      writeData(key, object, precondition);
     } catch (StorageException e) {
       if (e.getCode() == CloudStorageErrorCode.PRECONDITION_FAILED.get()) {
         throw new PreconditionFailedException(
@@ -139,19 +115,9 @@ public void insert(String key, String object) throws ObjectStorageWrapperExcepti
   public void update(String key, String object, String version)
       throws ObjectStorageWrapperException {
     try {
-      byte[] data = object.getBytes(StandardCharsets.UTF_8);
-      BlobInfo blobInfo = BlobInfo.newBuilder(BlobId.of(bucket, key)).build();
       Storage.BlobWriteOption precondition =
           Storage.BlobWriteOption.generationMatch(Long.parseLong(version));
-      try (WriteChannel writer = storage.writer(blobInfo, precondition)) {
-        if (parallelUploadBlockSizeInBytes != null) {
-          writer.setChunkSize(parallelUploadBlockSizeInBytes);
-        }
-        ByteBuffer buffer = ByteBuffer.wrap(data);
-        while (buffer.hasRemaining()) {
-          writer.write(buffer);
-        }
-      }
+      writeData(key, object, precondition);
     } catch (StorageException e) {
       if (e.getCode() == CloudStorageErrorCode.PRECONDITION_FAILED.get()) {
         throw new PreconditionFailedException(
@@ -246,4 +212,20 @@ public void close() throws ObjectStorageWrapperException {
       throw new ObjectStorageWrapperException("Failed to close the storage wrapper", e);
     }
   }
+
+  private void writeData(String key, String object, Storage.BlobWriteOption precondition)
+      throws IOException {
+    byte[] data = object.getBytes(StandardCharsets.UTF_8);
+    BlobInfo blobInfo = BlobInfo.newBuilder(BlobId.of(bucket, key)).build();
+
+    try (WriteChannel writer = storage.writer(blobInfo, precondition)) {
+      if (parallelUploadBlockSizeInBytes != null) {
+        writer.setChunkSize(parallelUploadBlockSizeInBytes);
+      }
+      ByteBuffer buffer = ByteBuffer.wrap(data);
+      while (buffer.hasRemaining()) {
+        writer.write(buffer);
+      }
+    }
+  }
 }