Backport to branch(3.13) : Fix the DynamoDB metadata table upsert operation (#2790)

Co-authored-by: Kodai Doki <52027276+KodaiD@users.noreply.github.com>

Author: feeblefakie

core/src/main/java/com/scalar/db/storage/dynamo/DynamoAdmin.java

@@ -95,6 +95,7 @@ public class DynamoAdmin implements DistributedStorageAdmin {
   public static final String DEFAULT_NO_BACKUP = "false";
   public static final String DEFAULT_REQUEST_UNIT = "10";
   private static final int DEFAULT_WAITING_DURATION_SECS = 3;
+  @VisibleForTesting static final int MAX_RETRY_COUNT = 10;
 
   @VisibleForTesting static final String PARTITION_KEY = "concatenatedPartitionKey";
   @VisibleForTesting static final String CLUSTERING_KEY = "concatenatedClusteringKey";
@@ -418,15 +419,28 @@ private void putTableMetadata(Namespace namespace, String table, TableMetadata m
           METADATA_ATTR_SECONDARY_INDEX,
           AttributeValue.builder().ss(metadata.getSecondaryIndexNames()).build());
     }
-    try {
-      client.putItem(
-          PutItemRequest.builder()
-              .tableName(ScalarDbUtils.getFullTableName(metadataNamespace, METADATA_TABLE))
-              .item(itemValues)
-              .build());
-    } catch (Exception e) {
-      throw new ExecutionException(
-          "Adding the meta data for table " + getFullTableName(namespace, table) + " failed", e);
+    int retryCount = 0;
+    while (true) {
+      try {
+        client.putItem(
+            PutItemRequest.builder()
+                .tableName(ScalarDbUtils.getFullTableName(metadataNamespace, METADATA_TABLE))
+                .item(itemValues)
+                .build());
+        return;
+      } catch (ResourceNotFoundException e) {
+        if (retryCount >= MAX_RETRY_COUNT) {
+          throw new ExecutionException(
+              "Adding the metadata for the " + getFullTableName(namespace, table) + " table failed",
+              e);
+        }
+        Uninterruptibles.sleepUninterruptibly(waitingDurationSecs, TimeUnit.SECONDS);
+        retryCount++;
+      } catch (Exception e) {
+        throw new ExecutionException(
+            "Adding the metadata for the " + getFullTableName(namespace, table) + " table failed",
+            e);
+      }
     }
   }
 

core/src/test/java/com/scalar/db/storage/dynamo/DynamoAdminTestBase.java

@@ -640,6 +640,47 @@ public void createTable_WhenMetadataTableExists_ShouldCreateOnlyTable()
         .isInstanceOf(IllegalArgumentException.class);
   }
 
+  @Test
+  public void createTable_WhenActualMetadataTableCreationIsDelayed_ShouldFailAfterRetries() {
+    // Arrange
+    // prepare tableIsActiveResponse
+    TableDescription tableDescription = mock(TableDescription.class);
+    when(tableIsActiveResponse.table()).thenReturn(tableDescription);
+    when(tableDescription.tableStatus()).thenReturn(TableStatus.ACTIVE);
+    when(client.describeTable(any(DescribeTableRequest.class))).thenReturn(tableIsActiveResponse);
+    // prepare backupIsEnabledResponse
+    DescribeContinuousBackupsResponse backupIsEnabledResponse =
+        mock(DescribeContinuousBackupsResponse.class);
+    when(client.describeContinuousBackups(any(DescribeContinuousBackupsRequest.class)))
+        .thenReturn(backupIsEnabledResponse);
+    ContinuousBackupsDescription continuousBackupsDescription =
+        mock(ContinuousBackupsDescription.class);
+    when(backupIsEnabledResponse.continuousBackupsDescription())
+        .thenReturn(continuousBackupsDescription);
+    when(continuousBackupsDescription.continuousBackupsStatus())
+        .thenReturn(ContinuousBackupsStatus.ENABLED);
+    when(client.putItem(any(PutItemRequest.class))).thenThrow(ResourceNotFoundException.class);
+    TableMetadata metadata =
+        TableMetadata.newBuilder()
+            .addPartitionKey("c1")
+            .addClusteringKey("c2", Order.DESC)
+            .addClusteringKey("c3", Order.ASC)
+            .addColumn("c1", DataType.TEXT)
+            .addColumn("c2", DataType.BIGINT)
+            .addColumn("c3", DataType.BOOLEAN)
+            .addColumn("c4", DataType.INT)
+            .addColumn("c5", DataType.BLOB)
+            .addColumn("c6", DataType.DOUBLE)
+            .addColumn("c7", DataType.FLOAT)
+            .addSecondaryIndex("c4")
+            .build();
+
+    // Act Assert
+    assertThatThrownBy(() -> admin.createTable(NAMESPACE, TABLE, metadata))
+        .isInstanceOf(ExecutionException.class);
+    verify(client, times(DynamoAdmin.MAX_RETRY_COUNT + 1)).putItem(any(PutItemRequest.class));
+  }
+
   @Test
   public void dropTable_WithNoMetadataLeft_ShouldDropTableAndDeleteMetadata()
       throws ExecutionException {