public void commitSegmentMetadata(String realtimeTableName, CommittingSegmentDescriptor committingSegmentDescriptor) {
    Preconditions.checkState(!_isStopping, "Segment manager is stopping");

    try {
      _numCompletingSegments.addAndGet(1);
      commitSegmentMetadataInternal(realtimeTableName, committingSegmentDescriptor);
    } finally {
      _numCompletingSegments.addAndGet(-1);
    }
  }
@Test
  public void testCommitSegmentMetadata() {
    // Set up a new table with 2 replicas, 5 instances, 4 partition
    PinotHelixResourceManager mockHelixResourceManager = mock(PinotHelixResourceManager.class);
    when(mockHelixResourceManager.getIdealStateUpdaterLock(anyString())).thenReturn(new Object());
    FakePinotLLCRealtimeSegmentManager segmentManager =
        new FakePinotLLCRealtimeSegmentManager(mockHelixResourceManager);
    setUpNewTable(segmentManager, 2, 5, 4);

    // Test case 1: segment location with vip format.
    // Commit a segment for partition group 0
    String committingSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 0, CURRENT_TIME_MS).getSegmentName();
    String segmentLocationVIP = "http://control_vip/segments/segment1";
    CommittingSegmentDescriptor committingSegmentDescriptor = new CommittingSegmentDescriptor(committingSegment,
        new LongMsgOffset(PARTITION_OFFSET.getOffset() + NUM_DOCS).toString(), 0L, segmentLocationVIP);
    committingSegmentDescriptor.setSegmentMetadata(mockSegmentMetadata());
    segmentManager.commitSegmentMetadata(REALTIME_TABLE_NAME, committingSegmentDescriptor);

    SegmentZKMetadata segmentZKMetadata =
        segmentManager.getSegmentZKMetadata(REALTIME_TABLE_NAME, committingSegment, null);
    Assert.assertEquals(segmentZKMetadata.getDownloadUrl(), segmentLocationVIP);

    // Test case 2: segment location with peer format: peer://segment1, verify that an empty string is stored in zk.
    committingSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 1, CURRENT_TIME_MS).getSegmentName();
    String peerSegmentLocation = CommonConstants.Segment.PEER_SEGMENT_DOWNLOAD_SCHEME + "/segment1";
    committingSegmentDescriptor = new CommittingSegmentDescriptor(committingSegment,
        new LongMsgOffset(PARTITION_OFFSET.getOffset() + NUM_DOCS).toString(), 0L, peerSegmentLocation);
    committingSegmentDescriptor.setSegmentMetadata(mockSegmentMetadata());
    segmentManager.commitSegmentMetadata(REALTIME_TABLE_NAME, committingSegmentDescriptor);

    segmentZKMetadata = segmentManager.getSegmentZKMetadata(REALTIME_TABLE_NAME, committingSegment, null);
    Assert.assertEquals(segmentZKMetadata.getDownloadUrl(), "");
  }