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

    String rawTableName = TableNameBuilder.extractRawTableName(realtimeTableName);
    String segmentName = committingSegmentDescriptor.getSegmentName();
    LOGGER.info("Committing segment file for segment: {}", segmentName);

    // Copy the segment file to the controller
    String segmentLocation = committingSegmentDescriptor.getSegmentLocation();
    Preconditions.checkArgument(segmentLocation != null, "Segment location must be provided");
    if (segmentLocation.regionMatches(true, 0, CommonConstants.Segment.PEER_SEGMENT_DOWNLOAD_SCHEME, 0,
        CommonConstants.Segment.PEER_SEGMENT_DOWNLOAD_SCHEME.length())) {
      LOGGER.info("No moving needed for segment on peer servers: {}", segmentLocation);
      return;
    }

    URI tableDirURI = URIUtils.getUri(_controllerConf.getDataDir(), rawTableName);
    PinotFS pinotFS = PinotFSFactory.create(tableDirURI.getScheme());
    String uriToMoveTo = moveSegmentFile(rawTableName, segmentName, segmentLocation, pinotFS);

    if (!isTmpSegmentAsyncDeletionEnabled()) {
      try {
        for (String uri : pinotFS.listFiles(tableDirURI, false)) {
          if (uri.contains(SegmentCompletionUtils.getTmpSegmentNamePrefix(segmentName))) {
            LOGGER.warn("Deleting temporary segment file: {}", uri);
            Preconditions.checkState(pinotFS.delete(new URI(uri), true), "Failed to delete file: %s", uri);
          }
        }
      } catch (Exception e) {
        LOGGER.warn("Caught exception while deleting temporary segment files for segment: {}", segmentName, e);
      }
    }
    committingSegmentDescriptor.setSegmentLocation(uriToMoveTo);
  }
@Test
  public void testSegmentAlreadyThereAndExtraneousFilesDeleted()
      throws Exception {
    PinotFSFactory.init(new PinotConfiguration());
    File tableDir = new File(TEMP_DIR, RAW_TABLE_NAME);
    String segmentName = new LLCSegmentName(RAW_TABLE_NAME, 0, 0, CURRENT_TIME_MS).getSegmentName();
    String otherSegmentName = new LLCSegmentName(RAW_TABLE_NAME, 1, 0, CURRENT_TIME_MS).getSegmentName();
    String segmentFileName = SegmentCompletionUtils.generateTmpSegmentFileName(segmentName);
    String extraSegmentFileName = SegmentCompletionUtils.generateTmpSegmentFileName(segmentName);
    String otherSegmentFileName = SegmentCompletionUtils.generateTmpSegmentFileName(otherSegmentName);
    File segmentFile = new File(tableDir, segmentFileName);
    File extraSegmentFile = new File(tableDir, extraSegmentFileName);
    File otherSegmentFile = new File(tableDir, otherSegmentFileName);
    FileUtils.write(segmentFile, "temporary file contents");
    FileUtils.write(extraSegmentFile, "temporary file contents");
    FileUtils.write(otherSegmentFile, "temporary file contents");

    FakePinotLLCRealtimeSegmentManager segmentManager = new FakePinotLLCRealtimeSegmentManager();
    String segmentLocation = SCHEME + tableDir + "/" + segmentFileName;
    CommittingSegmentDescriptor committingSegmentDescriptor =
        new CommittingSegmentDescriptor(segmentName, PARTITION_OFFSET.toString(), 0, segmentLocation);
    segmentManager.commitSegmentFile(REALTIME_TABLE_NAME, committingSegmentDescriptor);
    Assert.assertEquals(committingSegmentDescriptor.getSegmentLocation(),
        URIUtils.getUri(tableDir.toString(), URIUtils.encode(segmentName)).toString());
    assertFalse(segmentFile.exists());
    assertFalse(extraSegmentFile.exists());
    assertTrue(otherSegmentFile.exists());
  }