public List<StreamPartitionWithWatermark> readStreamPartitionsWithWatermark()
      throws InvalidProtocolBufferException {
    LOG.debug(
        "Reading stream partitions from metadata table: "
            + getFullStreamPartitionPrefix().toStringUtf8());
    Filter filterForWatermark =
        FILTERS
            .chain()
            .filter(Filters.FILTERS.limit().cellsPerColumn(1))
            .filter(FILTERS.family().exactMatch(MetadataTableAdminDao.CF_WATERMARK))
            .filter(FILTERS.qualifier().exactMatch(MetadataTableAdminDao.QUALIFIER_DEFAULT));
    Filter filterForLock =
        FILTERS
            .chain()
            .filter(Filters.FILTERS.limit().cellsPerColumn(1))
            .filter(FILTERS.family().exactMatch(MetadataTableAdminDao.CF_LOCK))
            .filter(FILTERS.qualifier().exactMatch(MetadataTableAdminDao.QUALIFIER_DEFAULT));
    Query query =
        Query.create(tableId)
            .prefix(getFullStreamPartitionPrefix())
            .filter(FILTERS.interleave().filter(filterForWatermark).filter(filterForLock));
    ServerStream<Row> rows = dataClient.readRows(query);
    List<StreamPartitionWithWatermark> partitions = new ArrayList<>();
    for (Row row : rows) {
      if (!isRowLocked(row)) {
        continue;
      }
      Instant watermark = MetadataTableEncoder.parseWatermarkFromRow(row);
      if (watermark == null) {
        continue;
      }
      ByteStringRange partition = convertStreamPartitionRowKeyToPartition(row.getKey());
      partitions.add(new StreamPartitionWithWatermark(partition, watermark));
    }
    return partitions;
  }
@Test
  public void testReadStreamPartitionsWithWatermark() throws InvalidProtocolBufferException {
    ByteStringRange lockedPartition = ByteStringRange.create("", "a");
    PartitionRecord partitionRecord =
        new PartitionRecord(
            lockedPartition,
            Instant.now(),
            UniqueIdGenerator.getNextId(),
            Instant.now(),
            Collections.emptyList(),
            null);
    metadataTableDao.lockAndRecordPartition(partitionRecord);

    // Only one row has both watermark and is locked.
    List<StreamPartitionWithWatermark> streamPartitionsWithWatermark =
        metadataTableDao.readStreamPartitionsWithWatermark();
    assertEquals(1, streamPartitionsWithWatermark.size());
    assertEquals(
        partitionRecord.getParentLowWatermark(),
        streamPartitionsWithWatermark.get(0).getWatermark());
    assertEquals(
        partitionRecord.getPartition(), streamPartitionsWithWatermark.get(0).getPartition());

    Instant watermark = Instant.now();
    // Update the watermark on the locked partition.
    metadataTableDao.updateWatermark(lockedPartition, watermark, null);

    // Watermark is updated.
    streamPartitionsWithWatermark = metadataTableDao.readStreamPartitionsWithWatermark();
    assertEquals(1, streamPartitionsWithWatermark.size());
    assertEquals(watermark, streamPartitionsWithWatermark.get(0).getWatermark());
  }