public void markNewPartitionForDeletion(NewPartition newPartition) {
    ByteString rowKey = convertPartitionToNewPartitionRowKey(newPartition.getPartition());
    RowMutation rowMutation = RowMutation.create(tableId, rowKey);
    for (ChangeStreamContinuationToken token : newPartition.getChangeStreamContinuationTokens()) {
      rowMutation.setCell(
          MetadataTableAdminDao.CF_SHOULD_DELETE,
          ByteStringRange.serializeToByteString(token.getPartition()),
          1);
    }
    mutateRowWithHardTimeout(rowMutation);
  }
@Test
  public void testMarkNewPartitionForDeletion() throws InvalidProtocolBufferException {
    ByteStringRange childPartition = ByteStringRange.create("A", "C");
    ChangeStreamContinuationToken token =
        ChangeStreamContinuationToken.create(ByteStringRange.create("B", "C"), "BC");
    NewPartition newPartition =
        new NewPartition(childPartition, Collections.singletonList(token), Instant.now());
    metadataTableDao.writeNewPartition(newPartition);
    List<NewPartition> newPartitions = metadataTableDao.readNewPartitions();
    assertEquals(1, newPartitions.size());
    assertEquals(childPartition, newPartitions.get(0).getPartition());

    // Once it's marked for deletion, it is not visible to readNewPartitions.
    metadataTableDao.markNewPartitionForDeletion(newPartitions.get(0));
    assertEquals(0, metadataTableDao.readNewPartitions().size());
  }