### Question:
PinotFSSegmentUploader implements SegmentUploader { public URI uploadSegment(File segmentFile, LLCSegmentName segmentName) { if (_segmentStoreUriStr == null || _segmentStoreUriStr.isEmpty()) { return null; } Callable<URI> uploadTask = () -> { URI destUri = new URI(StringUtil .join(File.separator, _segmentStoreUriStr, segmentName.getTableName(), segmentName.getSegmentName() + UUID.randomUUID().toString())); try { PinotFS pinotFS = PinotFSFactory.create(new URI(_segmentStoreUriStr).getScheme()); if (pinotFS.exists(destUri)) { pinotFS.delete(destUri, true); } pinotFS.copyFromLocalFile(segmentFile, destUri); return destUri; } catch (Exception e) { LOGGER.warn("Failed copy segment tar file {} to segment store {}: {}", segmentFile.getName(), destUri, e); } return null; }; Future<URI> future = _executorService.submit(uploadTask); try { URI segmentLocation = future.get(_timeoutInMs, TimeUnit.MILLISECONDS); LOGGER.info("Successfully upload segment {} to {}.", segmentName, segmentLocation); return segmentLocation; } catch (InterruptedException e) { LOGGER.info("Interrupted while waiting for segment upload of {} to {}.", segmentName, _segmentStoreUriStr); Thread.currentThread().interrupt(); } catch (Exception e) { LOGGER .warn("Failed to upload file {} of segment {} for table {} ", segmentFile.getAbsolutePath(), segmentName, e); } return null; } PinotFSSegmentUploader(String segmentStoreDirUri, int timeoutMillis); URI uploadSegment(File segmentFile, LLCSegmentName segmentName); static final int DEFAULT_SEGMENT_UPLOAD_TIMEOUT_MILLIS; }### Answer:
@Test public void testSuccessfulUpload() { SegmentUploader segmentUploader = new PinotFSSegmentUploader("hdfs: URI segmentURI = segmentUploader.uploadSegment(_file, _llcSegmentName); Assert.assertTrue(segmentURI.toString().startsWith(StringUtil.join(File.separator,"hdfs: }
@Test public void testSegmentAlreadyExist() { SegmentUploader segmentUploader = new PinotFSSegmentUploader("existing: URI segmentURI = segmentUploader.uploadSegment(_file, _llcSegmentName); Assert.assertTrue(segmentURI.toString().startsWith(StringUtil.join(File.separator,"existing: }
@Test public void testUploadTimeOut() { SegmentUploader segmentUploader = new PinotFSSegmentUploader("timeout: URI segmentURI = segmentUploader.uploadSegment(_file, _llcSegmentName); Assert.assertNull(segmentURI); }
@Test public void testNoSegmentStoreConfigured() { SegmentUploader segmentUploader = new PinotFSSegmentUploader("", TIMEOUT_IN_MS); URI segmentURI = segmentUploader.uploadSegment(_file, _llcSegmentName); Assert.assertNull(segmentURI); }