public boolean deleteNewPartition(NewPartition newPartition) {
    ByteString rowKey = convertPartitionToNewPartitionRowKey(newPartition.getPartition());

    boolean success = true;

    for (ChangeStreamContinuationToken token : newPartition.getChangeStreamContinuationTokens()) {
      Filter shouldDelete =
          FILTERS
              .chain()
              .filter(FILTERS.family().exactMatch(MetadataTableAdminDao.CF_SHOULD_DELETE))
              .filter(
                  FILTERS
                      .qualifier()
                      .exactMatch(ByteStringRange.serializeToByteString(token.getPartition())))
              .filter(FILTERS.value().exactMatch(ByteString.copyFrom(Longs.toByteArray(1))));
      Mutation mutation =
          Mutation.create()
              .deleteCells(
                  MetadataTableAdminDao.CF_INITIAL_TOKEN,
                  ByteStringRange.serializeToByteString(token.getPartition()))
              .deleteCells(
                  MetadataTableAdminDao.CF_PARENT_LOW_WATERMARKS,
                  ByteStringRange.serializeToByteString(token.getPartition()))
              .deleteCells(
                  MetadataTableAdminDao.CF_SHOULD_DELETE,
                  ByteStringRange.serializeToByteString(token.getPartition()));

      ConditionalRowMutation conditionalRowMutation =
          ConditionalRowMutation.create(tableId, rowKey).condition(shouldDelete).then(mutation);
      success = dataClient.checkAndMutateRow(conditionalRowMutation) && success;
    }

    return success;
  }
@Test
  public void testDeleteNewPartition() 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);
    ChangeStreamContinuationToken token2 =
        ChangeStreamContinuationToken.create(ByteStringRange.create("A", "B"), "AB");
    NewPartition newPartition2 =
        new NewPartition(childPartition, Collections.singletonList(token2), Instant.now());
    metadataTableDao.writeNewPartition(newPartition2);

    List<NewPartition> newPartitions = metadataTableDao.readNewPartitions();
    assertEquals(1, newPartitions.size());
    assertEquals(childPartition, newPartitions.get(0).getPartition());

    // Did not mark new partition for deletion. Cannot delete.
    assertFalse(metadataTableDao.deleteNewPartition(newPartitions.get(0)));
    assertEquals(1, metadataTableDao.readNewPartitions().size());

    metadataTableDao.markNewPartitionForDeletion(newPartitions.get(0));
    assertTrue(metadataTableDao.deleteNewPartition(newPartitions.get(0)));
    assertEquals(0, metadataTableDao.readNewPartitions().size());
  }