@Test
  public void testCreateTableEntity() throws ReflectiveOperationException {
    v2TestOnly();
    setupCloudStore(false, false, 0, true);
    MockMessageWriteSet messageWriteSet = new MockMessageWriteSet();

    /*
      Add some permanent blobs.
      set isVcr = true so that the lifeVersion is 0, else it is -1
     */
    final int NUM_BLOBS = 10;
    IntStream.range(0,NUM_BLOBS).forEach(i -> CloudTestUtil.addBlobToMessageSet(messageWriteSet, 1024,
        Utils.Infinite_Time, refAccountId, refContainerId, false, false, partitionId,
        System.currentTimeMillis(), isVcr));

    // Test constants
    final String TABLE_NAME = "corruptBlobIds";
    final String HOST_1 = "localhost";
    final String HOST_2 = "localhost2";
    final String REPLICA_COLUMN = "replicaPath";
    final String REPLICA_1 = "/mnt/disk/1234";
    final String REPLICA_2 = "/mnt/disk/5678";

    // Clear table
    TableClient tableClient = dest.getTableClient(TABLE_NAME);
    // Don't delete the table, because getTableClient populates its cache with tableClient ref.
    // If we delete the table, then the cached ref is dangling.
    tableClient.listEntities().forEach(tableEntity -> tableClient.deleteEntity(tableEntity));
    assertEquals(0, tableClient.listEntities().stream().count());

    // Add some rows to the table
    // TableEntity = (partition-key, row-key) + more columns
    // Table entity = Table row
    // =========================================
    // | partition-key | row-key | replicaPath |
    // =========================================
    // | blob-id-1     | host1   | replica1    |
    // | blob-id-1     | host2   | replica2    |
    // =========================================
    // Add several times to assert row is added only once
    messageWriteSet.getMessageSetInfo().forEach(messageInfo ->
        IntStream.range(0,3).forEach(i ->
            dest.createTableEntity(TABLE_NAME,
                new TableEntity(messageInfo.getStoreKey().getID(), HOST_1).addProperty(REPLICA_COLUMN, REPLICA_1))));
    assertEquals(NUM_BLOBS, tableClient.listEntities().stream().count());
    // Simulate same corrupted blob from a different host
    messageWriteSet.getMessageSetInfo().forEach(messageInfo ->
        IntStream.range(0,3).forEach(i ->
            dest.createTableEntity(TABLE_NAME,
                new TableEntity(messageInfo.getStoreKey().getID(), HOST_2).addProperty(REPLICA_COLUMN, REPLICA_2))));
    assertEquals(NUM_BLOBS*2, tableClient.listEntities().stream().count());

    // Check rows are as expected
    messageWriteSet.getMessageSetInfo().forEach(messageInfo ->
        assertEquals(REPLICA_1,
            tableClient.getEntity(messageInfo.getStoreKey().getID(), HOST_1).getProperties().get(REPLICA_COLUMN)));
    messageWriteSet.getMessageSetInfo().forEach(messageInfo ->
        assertEquals(REPLICA_2,
            tableClient.getEntity(messageInfo.getStoreKey().getID(), HOST_2).getProperties().get(REPLICA_COLUMN)));
  }
Test PUT(with TTL) and TtlUpdate record replication.
Replication may happen after PUT and after TtlUpdate, or after TtlUpdate only.
PUT may already expired, expiration time < upload threshold or expiration time >= upload threshold.
@throws Exception
ambry-vcr/src/test/java/com/github/ambry/vcr/CloudBlobStoreTest.java