public SegmentLineage(String tableNameWithType) {
    _tableNameWithType = tableNameWithType;
    _lineageEntries = new HashMap<>();
  }
@Test
  public void testSegmentLineage() {
    SegmentLineage segmentLineage = new SegmentLineage("test_OFFLINE");
    String id = SegmentLineageUtils.generateLineageEntryId();
    segmentLineage.addLineageEntry(id,
        new LineageEntry(Arrays.asList("s1", "s2", "s3"), Arrays.asList("s4", "s5"), LineageEntryState.COMPLETED,
            11111L));
    LineageEntry lineageEntry = segmentLineage.getLineageEntry(id);
    Assert.assertEquals(lineageEntry.getSegmentsFrom(), Arrays.asList("s1", "s2", "s3"));
    Assert.assertEquals(lineageEntry.getSegmentsTo(), Arrays.asList("s4", "s5"));
    Assert.assertEquals(lineageEntry.getState(), LineageEntryState.COMPLETED);
    Assert.assertEquals(lineageEntry.getTimestamp(), 11111L);

    String id2 = SegmentLineageUtils.generateLineageEntryId();
    segmentLineage.addLineageEntry(id2,
        new LineageEntry(Arrays.asList("s6", "s6", "s8"), Arrays.asList("s9", "s10"), LineageEntryState.COMPLETED,
            22222L));
    LineageEntry lineageEntry2 = segmentLineage.getLineageEntry(id2);
    Assert.assertEquals(lineageEntry2.getSegmentsFrom(), Arrays.asList("s6", "s6", "s8"));
    Assert.assertEquals(lineageEntry2.getSegmentsTo(), Arrays.asList("s9", "s10"));
    Assert.assertEquals(lineageEntry2.getState(), LineageEntryState.COMPLETED);
    Assert.assertEquals(lineageEntry2.getTimestamp(), 22222L);

    String id3 = SegmentLineageUtils.generateLineageEntryId();
    segmentLineage.addLineageEntry(id3,
        new LineageEntry(Arrays.asList("s5", "s9"), Arrays.asList("s11"), LineageEntryState.IN_PROGRESS, 33333L));
    LineageEntry lineageEntry3 = segmentLineage.getLineageEntry(id3);
    Assert.assertEquals(lineageEntry3.getSegmentsFrom(), Arrays.asList("s5", "s9"));
    Assert.assertEquals(lineageEntry3.getSegmentsTo(), Arrays.asList("s11"));
    Assert.assertEquals(lineageEntry3.getState(), LineageEntryState.IN_PROGRESS);
    Assert.assertEquals(lineageEntry3.getTimestamp(), 33333L);

    String id4 = SegmentLineageUtils.generateLineageEntryId();
    segmentLineage.addLineageEntry(id4,
        new LineageEntry(new ArrayList<>(), Arrays.asList("s12"), LineageEntryState.IN_PROGRESS, 44444L));
    LineageEntry lineageEntry4 = segmentLineage.getLineageEntry(id4);
    Assert.assertEquals(lineageEntry4.getSegmentsFrom(), new ArrayList<>());
    Assert.assertEquals(lineageEntry4.getSegmentsTo(), Arrays.asList("s12"));
    Assert.assertEquals(lineageEntry4.getState(), LineageEntryState.IN_PROGRESS);
    Assert.assertEquals(lineageEntry4.getTimestamp(), 44444L);

    // Test the convesion from the segment lineage to the znRecord
    ZNRecord znRecord = segmentLineage.toZNRecord();
    Assert.assertEquals(znRecord.getId(), "test_OFFLINE");

    Map<String, List<String>> listFields = znRecord.getListFields();
    List<String> entry = listFields.get(id);
    Assert.assertEquals(entry.get(0), String.join(",", Arrays.asList("s1", "s2", "s3")));
    Assert.assertEquals(entry.get(1), String.join(",", Arrays.asList("s4", "s5")));
    Assert.assertEquals(entry.get(2), LineageEntryState.COMPLETED.toString());
    Assert.assertEquals(entry.get(3), Long.toString(11111L));

    List<String> entry2 = listFields.get(id2);
    Assert.assertEquals(entry2.get(0), String.join(",", Arrays.asList("s6", "s6", "s8")));
    Assert.assertEquals(entry2.get(1), String.join(",", Arrays.asList("s9", "s10")));
    Assert.assertEquals(entry2.get(2), LineageEntryState.COMPLETED.toString());
    Assert.assertEquals(entry2.get(3), Long.toString(22222L));

    List<String> entry3 = listFields.get(id3);
    Assert.assertEquals(entry3.get(0), String.join(",", Arrays.asList("s5", "s9")));
    Assert.assertEquals(entry3.get(1), String.join(",", Arrays.asList("s11")));
    Assert.assertEquals(entry3.get(2), LineageEntryState.IN_PROGRESS.toString());
    Assert.assertEquals(entry3.get(3), Long.toString(33333L));

    List<String> entry4 = listFields.get(id4);
    Assert.assertEquals(entry4.get(0), "");
    Assert.assertEquals(entry4.get(1), String.join(",", Arrays.asList("s12")));
    Assert.assertEquals(entry4.get(2), LineageEntryState.IN_PROGRESS.toString());
    Assert.assertEquals(entry4.get(3), Long.toString(44444L));

    // Test the conversion from the znRecord to the segment lineage
    SegmentLineage segmentLineageFromZNRecord = SegmentLineage.fromZNRecord(segmentLineage.toZNRecord());
    Assert.assertEquals(segmentLineageFromZNRecord.getLineageEntry(id), lineageEntry);
    Assert.assertEquals(segmentLineageFromZNRecord.getLineageEntry(id2), lineageEntry2);
    Assert.assertEquals(segmentLineageFromZNRecord.getLineageEntry(id3), lineageEntry3);
    Assert.assertEquals(segmentLineageFromZNRecord.getLineageEntry(id4), lineageEntry4);

    // Try to delete by iterating through the lineage entry ids
    for (String lineageId : segmentLineage.getLineageEntryIds()) {
      segmentLineage.deleteLineageEntry(lineageId);
    }
    Assert.assertEquals(segmentLineage.getLineageEntryIds().size(), 0);
  }