public SegmentCommitter createSegmentCommitter(SegmentCompletionProtocol.Request.Params params,
      String controllerVipUrl)
      throws URISyntaxException {
    boolean uploadToFs = _streamConfig.isServerUploadToDeepStore();
    String peerSegmentDownloadScheme = _tableConfig.getValidationConfig().getPeerSegmentDownloadScheme();
    String segmentStoreUri = _indexLoadingConfig.getSegmentStoreURI();

    SegmentUploader segmentUploader;
    if (uploadToFs || peerSegmentDownloadScheme != null) {
      // TODO: peer scheme non-null check exists for backwards compatibility. remove check once users have migrated
      segmentUploader = new PinotFSSegmentUploader(segmentStoreUri,
          ServerSegmentCompletionProtocolHandler.getSegmentUploadRequestTimeoutMs(), _serverMetrics);
    } else {
      segmentUploader = new Server2ControllerSegmentUploader(_logger,
          _protocolHandler.getFileUploadDownloadClient(),
          _protocolHandler.getSegmentCommitUploadURL(params, controllerVipUrl), params.getSegmentName(),
          ServerSegmentCompletionProtocolHandler.getSegmentUploadRequestTimeoutMs(), _serverMetrics,
          _protocolHandler.getAuthProvider(), _tableConfig.getTableName());
    }

    return new SplitSegmentCommitter(_logger, _protocolHandler, params, segmentUploader, peerSegmentDownloadScheme);
  }
@Test(description = "when controller supports split commit, server should always use split segment commit")
  public void testSplitSegmentCommitterIsDefault()
      throws URISyntaxException {
    TableConfig config = createRealtimeTableConfig("test").build();
    ServerSegmentCompletionProtocolHandler protocolHandler =
        new ServerSegmentCompletionProtocolHandler(Mockito.mock(ServerMetrics.class), "test_REALTIME");
    String controllerVipUrl = "http://localhost:1234";
    SegmentCompletionProtocol.Request.Params requestParams = new SegmentCompletionProtocol.Request.Params();
    SegmentCommitterFactory factory = new SegmentCommitterFactory(Mockito.mock(Logger.class), protocolHandler, config,
        Mockito.mock(IndexLoadingConfig.class), Mockito.mock(ServerMetrics.class));
    SegmentCommitter committer = factory.createSegmentCommitter(requestParams, controllerVipUrl);
    Assert.assertNotNull(committer);
    Assert.assertTrue(committer instanceof SplitSegmentCommitter);
  }