Backport to branch(3) : Add support for S3

.github/workflows/object-storage-adapter-check.yaml

@@ -0,0 +1,102 @@
+name: Test Object Storage Adapter
+
+on:
+  workflow_dispatch:
+    inputs:
+      INT_TEST_JAVA_RUNTIME_VERSION:
+        description: JDK version used to run the integration test
+        type: choice
+        required: false
+        default: '8'
+        options:
+          - '8'
+          - '11'
+          - '17'
+          - '21'
+      INT_TEST_JAVA_RUNTIME_VENDOR:
+        description: Vendor of the JDK used to run the integration test
+        type: choice
+        required: false
+        default: 'temurin'
+        options:
+          - 'corretto'
+          - 'microsoft'
+          - 'oracle'
+          - 'temurin'
+
+env:
+  TERM: dumb
+  JAVA_VERSION: '8'
+  JAVA_VENDOR: 'temurin'
+  INT_TEST_JAVA_RUNTIME_VERSION: "${{ github.event_name != 'workflow_dispatch' && '8' || inputs.INT_TEST_JAVA_RUNTIME_VERSION }}"
+  INT_TEST_JAVA_RUNTIME_VENDOR: "${{ github.event_name != 'workflow_dispatch' && 'temurin' || inputs.INT_TEST_JAVA_RUNTIME_VENDOR }}"
+  # Gradle will parse 'ORG_GRADLE_PROJECT_<project_property_name>' environment variables as project properties.
+  # The following variables configure the 'com.scalar.db.jdk-configuration' Gradle plugin.
+  ORG_GRADLE_PROJECT_javaVersion: '8'
+  ORG_GRADLE_PROJECT_javaVendor: 'temurin'
+  ORG_GRADLE_PROJECT_integrationTestJavaRuntimeVersion: "${{ github.event_name != 'workflow_dispatch' && '8' || inputs.INT_TEST_JAVA_RUNTIME_VERSION }}"
+  ORG_GRADLE_PROJECT_integrationTestJavaRuntimeVendor: "${{ github.event_name != 'workflow_dispatch' && 'temurin' || inputs.INT_TEST_JAVA_RUNTIME_VENDOR }}"
+  # This variable evaluates to: if {!(Temurin JDK 8) && !(Oracle JDK)} then {true} else {false}
+  # Oracle JDK that are linux compatible and publicly available through direct download exist for all LTS versions
+  SET_UP_INT_TEST_RUNTIME_NON_ORACLE_JDK: "${{ (github.event_name == 'workflow_dispatch' && !(inputs.INT_TEST_JAVA_RUNTIME_VERSION == '8' && inputs.INT_TEST_JAVA_RUNTIME_VENDOR == 'temurin') && !(inputs.INT_TEST_JAVA_RUNTIME_VENDOR == 'oracle')) && 'true' || 'false' }}"
+  INT_TEST_GRADLE_OPTIONS_FOR_GROUP_COMMIT: '"-Dscalardb.consensus_commit.coordinator.group_commit.enabled=true" "-Dscalardb.consensus_commit.coordinator.group_commit.old_group_abort_timeout_millis=15000" --tests "**.ConsensusCommit**"'
+  AWS_ACCESS_KEY_ID: ${{ secrets.S3_ACCESS_KEY }}
+  AWS_SECRET_ACCESS_KEY: ${{ secrets.S3_SECRET_ACCESS_KEY }}
+  S3_REGION: ap-northeast-1
+  S3_BUCKET_NAME: scalardb-test-bucket
+
+jobs:
+  integration-test-s3:
+    name: S3 integration test (${{ matrix.mode.label }})
+    runs-on: ubuntu-latest
+
+    strategy:
+      fail-fast: false
+      matrix:
+        mode:
+          - label: default
+            group_commit_enabled: false
+          - label: with_group_commit
+            group_commit_enabled: true
+
+    steps:
+      - uses: actions/checkout@v5
+
+      - name: Set up JDK ${{ env.JAVA_VERSION }} (${{ env.JAVA_VENDOR }})
+        uses: actions/setup-java@v5
+        with:
+          java-version: ${{ env.JAVA_VERSION }}
+          distribution: ${{ env.JAVA_VENDOR }}
+
+      - name: Set up JDK ${{ env.INT_TEST_JAVA_RUNTIME_VERSION }} (${{ env.INT_TEST_JAVA_RUNTIME_VENDOR }}) to run integration test
+        uses: actions/setup-java@v5
+        if: ${{ env.SET_UP_INT_TEST_RUNTIME_NON_ORACLE_JDK == 'true'}}
+        with:
+          java-version: ${{ env.INT_TEST_JAVA_RUNTIME_VERSION }}
+          distribution: ${{ env.INT_TEST_JAVA_RUNTIME_VENDOR }}
+
+      - name: Login to Oracle container registry
+        uses: docker/login-action@v3
+        if: ${{ env.INT_TEST_JAVA_RUNTIME_VENDOR == 'oracle' }}
+        with:
+          registry: container-registry.oracle.com
+          username: ${{ secrets.OCR_USERNAME }}
+          password: ${{ secrets.OCR_TOKEN }}
+
+      - name: Set up JDK ${{ env.INT_TEST_JAVA_RUNTIME_VERSION }} (oracle) to run the integration test
+        if: ${{ env.INT_TEST_JAVA_RUNTIME_VENDOR == 'oracle' }}
+        run: |
+          container_id=$(docker create "container-registry.oracle.com/java/jdk:${{ env.INT_TEST_JAVA_RUNTIME_VERSION }}")
+          docker cp -L "$container_id:/usr/java/default" /usr/lib/jvm/oracle-jdk && docker rm "$container_id"
+      - name: Setup Gradle
+        uses: gradle/actions/setup-gradle@v5
+
+      - name: Execute Gradle 'integrationTestObjectStorage' task
+        run: ./gradlew integrationTestObjectStorage -Dscalardb.object_storage.storage=s3 -Dscalardb.object_storage.endpoint=${{ env.S3_REGION }}/${{ env.S3_BUCKET_NAME }} ${{ matrix.mode.group_commit_enabled && env.INT_TEST_GRADLE_OPTIONS_FOR_GROUP_COMMIT || '' }}
+
+      - name: Upload Gradle test reports
+        if: always()
+        uses: actions/upload-artifact@v5
+        with:
+          name: cassandra_3.0_integration_test_reports_${{ matrix.mode.label }}
+          path: core/build/reports/tests/integrationTestObjectStorage

core/build.gradle

@@ -174,6 +174,8 @@ dependencies {
     implementation platform("software.amazon.awssdk:bom:${awssdkVersion}")
     implementation 'software.amazon.awssdk:applicationautoscaling'
     implementation 'software.amazon.awssdk:dynamodb'
+    implementation 'software.amazon.awssdk:s3'
+    implementation 'software.amazon.awssdk:aws-crt-client'
     testImplementation 'software.amazon.awssdk:iam'
     testImplementation 'software.amazon.awssdk:iam-policy-builder'
     implementation "org.apache.commons:commons-dbcp2:${commonsDbcp2Version}"

core/src/integration-test/java/com/scalar/db/storage/objectstorage/ObjectStorageEnv.java

@@ -2,15 +2,18 @@
 
 import com.scalar.db.config.DatabaseConfig;
 import com.scalar.db.storage.objectstorage.blobstorage.BlobStorageConfig;
+import com.scalar.db.storage.objectstorage.s3.S3Config;
 import java.util.Collections;
 import java.util.Map;
 import java.util.Properties;
 
 public class ObjectStorageEnv {
+  private static final String PROP_OBJECT_STORAGE_STORAGE = "scalardb.object_storage.storage";
   private static final String PROP_OBJECT_STORAGE_ENDPOINT = "scalardb.object_storage.endpoint";
   private static final String PROP_OBJECT_STORAGE_USERNAME = "scalardb.object_storage.username";
   private static final String PROP_OBJECT_STORAGE_PASSWORD = "scalardb.object_storage.password";
 
+  private static final String DEFAULT_OBJECT_STORAGE_STORAGE = BlobStorageConfig.STORAGE_NAME;
   private static final String DEFAULT_OBJECT_STORAGE_ENDPOINT =
       "http://localhost:10000/test/test-container";
   private static final String DEFAULT_OBJECT_STORAGE_USERNAME = "test";
@@ -19,6 +22,8 @@ public class ObjectStorageEnv {
   private ObjectStorageEnv() {}
 
   public static Properties getProperties(String testName) {
+    String storage =
+        System.getProperty(PROP_OBJECT_STORAGE_STORAGE, DEFAULT_OBJECT_STORAGE_STORAGE);
     String accountName =
         System.getProperty(PROP_OBJECT_STORAGE_USERNAME, DEFAULT_OBJECT_STORAGE_USERNAME);
     String accountKey =
@@ -30,7 +35,7 @@ public static Properties getProperties(String testName) {
     properties.setProperty(DatabaseConfig.CONTACT_POINTS, endpoint);
     properties.setProperty(DatabaseConfig.USERNAME, accountName);
     properties.setProperty(DatabaseConfig.PASSWORD, accountKey);
-    properties.setProperty(DatabaseConfig.STORAGE, BlobStorageConfig.STORAGE_NAME);
+    properties.setProperty(DatabaseConfig.STORAGE, storage);
     properties.setProperty(DatabaseConfig.CROSS_PARTITION_SCAN, "true");
     properties.setProperty(DatabaseConfig.CROSS_PARTITION_SCAN_FILTERING, "true");
     properties.setProperty(DatabaseConfig.CROSS_PARTITION_SCAN_ORDERING, "false");
@@ -43,7 +48,36 @@ public static Properties getProperties(String testName) {
     return properties;
   }
 
+  public static Properties getPropertiesWithPerformanceOptions(String testName) {
+    Properties properties = getProperties(testName);
+
+    // For Blob Storage
+    properties.setProperty(BlobStorageConfig.PARALLEL_UPLOAD_BLOCK_SIZE_IN_BYTES, "5242880"); // 5MB
+    properties.setProperty(BlobStorageConfig.PARALLEL_UPLOAD_MAX_PARALLELISM, "4");
+    properties.setProperty(
+        BlobStorageConfig.PARALLEL_UPLOAD_THRESHOLD_IN_BYTES, "10485760"); // 10MB
+    properties.setProperty(BlobStorageConfig.REQUEST_TIMEOUT_IN_SECONDS, "30");
+
+    // For S3
+    properties.setProperty(S3Config.PARALLEL_UPLOAD_BLOCK_SIZE_IN_BYTES, "5242880"); // 5MB
+    properties.setProperty(S3Config.PARALLEL_UPLOAD_MAX_PARALLELISM, "4");
+    properties.setProperty(S3Config.PARALLEL_UPLOAD_THRESHOLD_IN_BYTES, "10485760"); // 10MB
+    properties.setProperty(S3Config.REQUEST_TIMEOUT_IN_SECONDS, "30");
+
+    return properties;
+  }
+
   public static Map<String, String> getCreationOptions() {
     return Collections.emptyMap();
   }
+
+  public static boolean isBlobStorage() {
+    return System.getProperty(PROP_OBJECT_STORAGE_STORAGE, DEFAULT_OBJECT_STORAGE_STORAGE)
+        .equals(BlobStorageConfig.STORAGE_NAME);
+  }
+
+  public static boolean isS3() {
+    return System.getProperty(PROP_OBJECT_STORAGE_STORAGE, DEFAULT_OBJECT_STORAGE_STORAGE)
+        .equals(S3Config.STORAGE_NAME);
+  }
 }

core/src/integration-test/java/com/scalar/db/storage/objectstorage/ObjectStorageWrapperIntegrationTest.java

@@ -270,7 +270,11 @@ public void getKeys_WithPrefixForTheNumberOfObjectsExceedingTheListLimit_ShouldR
       }
     } finally {
       for (int i = 0; i < numberOfObjects; i++) {
-        wrapper.delete(prefix + i);
+        try {
+          wrapper.delete(prefix + i);
+        } catch (PreconditionFailedException e) {
+          // The object may not exist if the setup failed partially, so do nothing
+        }
       }
     }
   }
@@ -336,8 +340,12 @@ public void deleteByPrefix_WithNonExistingPrefix_ShouldDoNothing() throws Except
   @Test
   public void close_ShouldNotThrowException() {
     // Arrange
+    Properties properties = getProperties(TEST_NAME);
+    ObjectStorageConfig objectStorageConfig =
+        ObjectStorageUtils.getObjectStorageConfig(new DatabaseConfig(properties));
+    ObjectStorageWrapper wrapper = ObjectStorageWrapperFactory.create(objectStorageConfig);
 
     // Act Assert
-    assertThatCode(() -> wrapper.close()).doesNotThrowAnyException();
+    assertThatCode(wrapper::close).doesNotThrowAnyException();
   }
 }

core/src/integration-test/java/com/scalar/db/storage/objectstorage/ObjectStorageWrapperLargeObjectWriteIntegrationTest.java

@@ -5,10 +5,10 @@
 
 import com.scalar.db.config.DatabaseConfig;
 import com.scalar.db.storage.objectstorage.blobstorage.BlobStorageConfig;
+import com.scalar.db.storage.objectstorage.s3.S3Config;
 import java.util.Arrays;
 import java.util.Optional;
 import java.util.Properties;
-import java.util.stream.LongStream;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
@@ -35,23 +35,42 @@ public class ObjectStorageWrapperLargeObjectWriteIntegrationTest {
   @BeforeAll
   public void beforeAll() throws ObjectStorageWrapperException {
     Properties properties = getProperties(TEST_NAME);
-    ObjectStorageConfig objectStorageConfig =
-        ObjectStorageUtils.getObjectStorageConfig(new DatabaseConfig(properties));
-    wrapper = ObjectStorageWrapperFactory.create(objectStorageConfig);
-    long objectSizeInBytes =
-        LongStream.of(BlobStorageConfig.DEFAULT_PARALLEL_UPLOAD_THRESHOLD_IN_BYTES)
-                .max()
-                .getAsLong()
-            + 1;
+    long parallelUploadThresholdInBytes;
+
+    if (ObjectStorageEnv.isBlobStorage()) {
+      // Minimum block size must be greater than or equal to 256KB for Blob Storage
+      Long parallelUploadUnit = 256 * 1024L; // 256KB
+      properties.setProperty(
+          BlobStorageConfig.PARALLEL_UPLOAD_BLOCK_SIZE_IN_BYTES,
+          String.valueOf(parallelUploadUnit));
+      properties.setProperty(
+          BlobStorageConfig.PARALLEL_UPLOAD_THRESHOLD_IN_BYTES,
+          String.valueOf(parallelUploadUnit * 2));
+      parallelUploadThresholdInBytes = parallelUploadUnit * 2;
+    } else if (ObjectStorageEnv.isS3()) {
+      // Minimum part size must be greater than or equal to 5MB for S3
+      Long parallelUploadUnit = 5 * 1024 * 1024L; // 5MB
+      properties.setProperty(
+          S3Config.PARALLEL_UPLOAD_BLOCK_SIZE_IN_BYTES, String.valueOf(parallelUploadUnit));
+      properties.setProperty(
+          S3Config.PARALLEL_UPLOAD_THRESHOLD_IN_BYTES, String.valueOf(parallelUploadUnit * 2));
+      parallelUploadThresholdInBytes = parallelUploadUnit * 2;
+    } else {
+      throw new AssertionError();
+    }
 
-    char[] charArray = new char[(int) objectSizeInBytes];
+    char[] charArray = new char[(int) parallelUploadThresholdInBytes];
     Arrays.fill(charArray, 'a');
     testObject1 = new String(charArray);
     Arrays.fill(charArray, 'b');
     testObject2 = new String(charArray);
     Arrays.fill(charArray, 'c');
     testObject3 = new String(charArray);
 
+    ObjectStorageConfig objectStorageConfig =
+        ObjectStorageUtils.getObjectStorageConfig(new DatabaseConfig(properties));
+    wrapper = ObjectStorageWrapperFactory.create(objectStorageConfig);
+
     createObjects();
   }
 
@@ -73,7 +92,7 @@ public void afterAll() {
   }
 
   protected Properties getProperties(String testName) {
-    return ObjectStorageEnv.getProperties(testName);
+    return ObjectStorageEnv.getPropertiesWithPerformanceOptions(testName);
   }
 
   private void createObjects() throws ObjectStorageWrapperException {

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

@@ -0,0 +1,8 @@
+package com.scalar.db.storage.objectstorage;
+
+public class ConflictOccurredException extends ObjectStorageWrapperException {
+
+  public ConflictOccurredException(String message, Throwable cause) {
+    super(message, cause);
+  }
+}

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

@@ -10,7 +10,9 @@
 import com.scalar.db.exception.storage.RetriableExecutionException;
 import java.util.Collections;
 import java.util.List;
+import javax.annotation.concurrent.ThreadSafe;
 
+@ThreadSafe
 public class MutateStatementHandler extends StatementHandler {
   public MutateStatementHandler(
       ObjectStorageWrapper wrapper, TableMetadataManager metadataManager) {
@@ -78,7 +80,7 @@ private void writePartition(ObjectStoragePartitionSnapshot snapshot) throws Exec
           wrapper.insert(snapshot.getObjectKey(), snapshot.getPartition().serialize());
         }
       }
-    } catch (PreconditionFailedException e) {
+    } catch (PreconditionFailedException | ConflictOccurredException e) {
       throw new RetriableExecutionException(
           CoreError.OBJECT_STORAGE_CONFLICT_OCCURRED_IN_MUTATION.buildMessage(e.getMessage()), e);
     } catch (ObjectStorageWrapperException e) {

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

@@ -21,7 +21,9 @@
 import java.util.Set;
 import java.util.stream.Collectors;
 import javax.annotation.Nullable;
+import javax.annotation.concurrent.ThreadSafe;
 
+@ThreadSafe
 public class ObjectStorageAdmin implements DistributedStorageAdmin {
   public static final String NAMESPACE_METADATA_TABLE = "namespaces";
   public static final String TABLE_METADATA_TABLE = "metadata";
@@ -49,6 +51,29 @@ public ObjectStorageAdmin(DatabaseConfig databaseConfig) {
     metadataNamespace = objectStorageConfig.getMetadataNamespace();
   }
 
+  private static String getTableMetadataKey(String namespace, String table) {
+    return String.join(
+        String.valueOf(ObjectStorageUtils.CONCATENATED_KEY_DELIMITER), namespace, table);
+  }
+
+  private static String getNamespaceNameFromTableMetadataKey(String tableMetadataKey) {
+    List<String> parts =
+        Splitter.on(ObjectStorageUtils.CONCATENATED_KEY_DELIMITER).splitToList(tableMetadataKey);
+    if (parts.size() != 2 || parts.get(0).isEmpty()) {
+      throw new IllegalArgumentException("Invalid table metadata key: " + tableMetadataKey);
+    }
+    return parts.get(0);
+  }
+
+  private static String getTableNameFromTableMetadataKey(String tableMetadataKey) {
+    List<String> parts =
+        Splitter.on(ObjectStorageUtils.CONCATENATED_KEY_DELIMITER).splitToList(tableMetadataKey);
+    if (parts.size() != 2 || parts.get(1).isEmpty()) {
+      throw new IllegalArgumentException("Invalid table metadata key: " + tableMetadataKey);
+    }
+    return parts.get(1);
+  }
+
   @Override
   public StorageInfo getStorageInfo(String namespace) throws ExecutionException {
     return STORAGE_INFO;
@@ -425,29 +450,6 @@ private void deleteTableData(String namespace, String table) throws ExecutionExc
     }
   }
 
-  private static String getTableMetadataKey(String namespace, String table) {
-    return String.join(
-        String.valueOf(ObjectStorageUtils.CONCATENATED_KEY_DELIMITER), namespace, table);
-  }
-
-  private static String getNamespaceNameFromTableMetadataKey(String tableMetadataKey) {
-    List<String> parts =
-        Splitter.on(ObjectStorageUtils.CONCATENATED_KEY_DELIMITER).splitToList(tableMetadataKey);
-    if (parts.size() != 2 || parts.get(0).isEmpty()) {
-      throw new IllegalArgumentException("Invalid table metadata key: " + tableMetadataKey);
-    }
-    return parts.get(0);
-  }
-
-  private static String getTableNameFromTableMetadataKey(String tableMetadataKey) {
-    List<String> parts =
-        Splitter.on(ObjectStorageUtils.CONCATENATED_KEY_DELIMITER).splitToList(tableMetadataKey);
-    if (parts.size() != 2 || parts.get(1).isEmpty()) {
-      throw new IllegalArgumentException("Invalid table metadata key: " + tableMetadataKey);
-    }
-    return parts.get(1);
-  }
-
   private void checkTableMetadata(TableMetadata metadata) {
     Set<String> secondaryIndexNames = metadata.getSecondaryIndexNames();
     if (!secondaryIndexNames.isEmpty()) {

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

@@ -9,13 +9,6 @@ public interface ObjectStorageConfig {
    */
   String getStorageName();
 
-  /**
-   * Returns the endpoint for the object storage service.
-   *
-   * @return the endpoint
-   */
-  String getEndpoint();
-
   /**
    * Returns the username for authentication.
    *