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 = "use upload to deepstore when either serverUploadToDeepStore is set or peer segment download "
      + "scheme is non-null")
  public void testUploadToDeepStoreConfig()
      throws URISyntaxException {
    ServerSegmentCompletionProtocolHandler protocolHandler =
        new ServerSegmentCompletionProtocolHandler(Mockito.mock(ServerMetrics.class), "test_REALTIME");
    String controllerVipUrl = "http://localhost:1234";
    SegmentCompletionProtocol.Request.Params requestParams = new SegmentCompletionProtocol.Request.Params();

    // No peer segment download scheme, serverUploadToDeepStore = true
    Map<String, String> streamConfigMap = new HashMap<>(getMinimumStreamConfigMap());
    streamConfigMap.put(StreamConfigProperties.SERVER_UPLOAD_TO_DEEPSTORE, "true");
    TableConfig config = createRealtimeTableConfig("testDeepStoreConfig", streamConfigMap).build();
    IndexLoadingConfig indexLoadingConfig = Mockito.mock(IndexLoadingConfig.class);
    Mockito.when(indexLoadingConfig.getSegmentStoreURI()).thenReturn("file:///path/to/segment/store.txt");

    SegmentCommitterFactory factory = new SegmentCommitterFactory(Mockito.mock(Logger.class), protocolHandler, config,
        indexLoadingConfig, Mockito.mock(ServerMetrics.class));
    SegmentCommitter committer = factory.createSegmentCommitter(requestParams, controllerVipUrl);
    Assert.assertNotNull(committer);
    Assert.assertTrue(committer instanceof SplitSegmentCommitter);
    Assert.assertTrue(((SplitSegmentCommitter) committer).getSegmentUploader() instanceof PinotFSSegmentUploader);

    // Peer segment download scheme is set, serverUploadToDeepStore = false (for backwards compatibility)
    Map<String, String> streamConfigMap1 = new HashMap<>(getMinimumStreamConfigMap());
    streamConfigMap1.put(StreamConfigProperties.SERVER_UPLOAD_TO_DEEPSTORE, "false");
    TableConfig config1 = createRealtimeTableConfig("testDeepStoreConfig", streamConfigMap1)
        .setPeerSegmentDownloadScheme("http")
        .build();

    factory = new SegmentCommitterFactory(Mockito.mock(Logger.class), protocolHandler, config1,
        indexLoadingConfig, Mockito.mock(ServerMetrics.class));
    committer = factory.createSegmentCommitter(requestParams, controllerVipUrl);
    Assert.assertNotNull(committer);
    Assert.assertTrue(committer instanceof SplitSegmentCommitter);
    Assert.assertTrue(((SplitSegmentCommitter) committer).getSegmentUploader() instanceof PinotFSSegmentUploader);
  }