@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 testInitialize() {
    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(partitionMetadataDao.getPartition(InitialPartition.PARTITION_TOKEN))
        .thenReturn(Struct.newBuilder().build());
    when(mapperFactory.partitionMetadataMapper()).thenReturn(partitionMetadataMapper);
    when(partitionMetadataMapper.from(any())).thenReturn(mock(PartitionMetadata.class));
    initializeDoFn.processElement(receiver);
    verify(daoFactory, times(2)).getPartitionMetadataDao();
    verify(daoFactory, times(1)).getPartitionMetadataAdminDao();
    verify(partitionMetadataDao, times(1)).insert(any());
    verify(partitionMetadataDao, times(1)).getPartition(InitialPartition.PARTITION_TOKEN);
    verify(partitionMetadataDao, times(1)).tableExists();
    verify(mapperFactory, times(1)).partitionMetadataMapper();
    verify(partitionMetadataMapper, times(1)).from(any());
  }