@ProcessElement
  public void processElement(OutputReceiver<PartitionMetadata> receiver) {
    PartitionMetadataDao partitionMetadataDao = daoFactory.getPartitionMetadataDao();
    if (!partitionMetadataDao.tableExists()) {
      daoFactory.getPartitionMetadataAdminDao().createPartitionMetadataTable();
      createFakeParentPartition();
    }
    final PartitionMetadata initialPartition =
        Optional.ofNullable(partitionMetadataDao.getPartition(InitialPartition.PARTITION_TOKEN))
            .map(mapperFactory.partitionMetadataMapper()::from)
            .orElseThrow(
                () -> new IllegalStateException("Initial partition not found in metadata table."));
    receiver.output(initialPartition);
  }
@Test
  public void testInitializeWithNoPartition() {
    when(daoFactory.getPartitionMetadataDao()).thenReturn(partitionMetadataDao);
    when(partitionMetadataDao.tableExists()).thenReturn(false);
    when(daoFactory.getPartitionMetadataAdminDao()).thenReturn(partitionMetadataAdminDao);
    doNothing().when(partitionMetadataAdminDao).createPartitionMetadataTable();
    when(partitionMetadataDao.insert(any())).thenReturn(Timestamp.ofTimeMicroseconds(1L));
    when(mapperFactory.partitionMetadataMapper()).thenReturn(partitionMetadataMapper);
    when(partitionMetadataMapper.from(any())).thenReturn(mock(PartitionMetadata.class));
    try {
      initializeDoFn.processElement(receiver);
      fail();
    } catch (IllegalStateException e) {
      assertEquals("Initial partition not found in metadata table.", e.getMessage());
    }
  }