@Test public void testCommitSegmentFile() 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 segmentFileName = SegmentCompletionUtils.generateSegmentFileName(segmentName); File segmentFile = new File(tableDir, segmentFileName); FileUtils.write(segmentFile, "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()); }
public void commitSegmentFile(String realtimeTableName, CommittingSegmentDescriptor committingSegmentDescriptor) throws Exception { Preconditions.checkState(!_isStopping, "Segment manager is stopping"); if (isPeerSegmentDownloadScheme(committingSegmentDescriptor)) { LOGGER.info("No moving needed for segment on peer servers: {}", committingSegmentDescriptor.getSegmentLocation()); return; } String rawTableName = TableNameBuilder.extractRawTableName(realtimeTableName); String segmentName = committingSegmentDescriptor.getSegmentName(); LOGGER.info("Committing segment file for segment: {}", segmentName); String segmentLocation = committingSegmentDescriptor.getSegmentLocation(); URI segmentFileURI = URIUtils.getUri(segmentLocation); URI tableDirURI = URIUtils.getUri(_controllerConf.getDataDir(), rawTableName); URI uriToMoveTo = URIUtils.getUri(_controllerConf.getDataDir(), rawTableName, URIUtils.encode(segmentName)); PinotFS pinotFS = PinotFSFactory.create(tableDirURI.getScheme()); Preconditions.checkState(pinotFS.move(segmentFileURI, uriToMoveTo, true), "Failed to move segment file for segment: %s from: %s to: %s", segmentName, segmentLocation, uriToMoveTo); try { for (String uri : pinotFS.listFiles(tableDirURI, false)) { if (uri.contains(SegmentCompletionUtils.getSegmentNamePrefix(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.toString()); }
PinotLLCRealtimeSegmentManager { public void commitSegmentFile(String realtimeTableName, CommittingSegmentDescriptor committingSegmentDescriptor) throws Exception { Preconditions.checkState(!_isStopping, "Segment manager is stopping"); if (isPeerSegmentDownloadScheme(committingSegmentDescriptor)) { LOGGER.info("No moving needed for segment on peer servers: {}", committingSegmentDescriptor.getSegmentLocation()); return; } String rawTableName = TableNameBuilder.extractRawTableName(realtimeTableName); String segmentName = committingSegmentDescriptor.getSegmentName(); LOGGER.info("Committing segment file for segment: {}", segmentName); String segmentLocation = committingSegmentDescriptor.getSegmentLocation(); URI segmentFileURI = URIUtils.getUri(segmentLocation); URI tableDirURI = URIUtils.getUri(_controllerConf.getDataDir(), rawTableName); URI uriToMoveTo = URIUtils.getUri(_controllerConf.getDataDir(), rawTableName, URIUtils.encode(segmentName)); PinotFS pinotFS = PinotFSFactory.create(tableDirURI.getScheme()); Preconditions.checkState(pinotFS.move(segmentFileURI, uriToMoveTo, true), "Failed to move segment file for segment: %s from: %s to: %s", segmentName, segmentLocation, uriToMoveTo); try { for (String uri : pinotFS.listFiles(tableDirURI, false)) { if (uri.contains(SegmentCompletionUtils.getSegmentNamePrefix(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.toString()); } }
PinotLLCRealtimeSegmentManager { public void commitSegmentFile(String realtimeTableName, CommittingSegmentDescriptor committingSegmentDescriptor) throws Exception { Preconditions.checkState(!_isStopping, "Segment manager is stopping"); if (isPeerSegmentDownloadScheme(committingSegmentDescriptor)) { LOGGER.info("No moving needed for segment on peer servers: {}", committingSegmentDescriptor.getSegmentLocation()); return; } String rawTableName = TableNameBuilder.extractRawTableName(realtimeTableName); String segmentName = committingSegmentDescriptor.getSegmentName(); LOGGER.info("Committing segment file for segment: {}", segmentName); String segmentLocation = committingSegmentDescriptor.getSegmentLocation(); URI segmentFileURI = URIUtils.getUri(segmentLocation); URI tableDirURI = URIUtils.getUri(_controllerConf.getDataDir(), rawTableName); URI uriToMoveTo = URIUtils.getUri(_controllerConf.getDataDir(), rawTableName, URIUtils.encode(segmentName)); PinotFS pinotFS = PinotFSFactory.create(tableDirURI.getScheme()); Preconditions.checkState(pinotFS.move(segmentFileURI, uriToMoveTo, true), "Failed to move segment file for segment: %s from: %s to: %s", segmentName, segmentLocation, uriToMoveTo); try { for (String uri : pinotFS.listFiles(tableDirURI, false)) { if (uri.contains(SegmentCompletionUtils.getSegmentNamePrefix(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.toString()); } PinotLLCRealtimeSegmentManager(PinotHelixResourceManager helixResourceManager, ControllerConf controllerConf,
      ControllerMetrics controllerMetrics); }
PinotLLCRealtimeSegmentManager { public void commitSegmentFile(String realtimeTableName, CommittingSegmentDescriptor committingSegmentDescriptor) throws Exception { Preconditions.checkState(!_isStopping, "Segment manager is stopping"); if (isPeerSegmentDownloadScheme(committingSegmentDescriptor)) { LOGGER.info("No moving needed for segment on peer servers: {}", committingSegmentDescriptor.getSegmentLocation()); return; } String rawTableName = TableNameBuilder.extractRawTableName(realtimeTableName); String segmentName = committingSegmentDescriptor.getSegmentName(); LOGGER.info("Committing segment file for segment: {}", segmentName); String segmentLocation = committingSegmentDescriptor.getSegmentLocation(); URI segmentFileURI = URIUtils.getUri(segmentLocation); URI tableDirURI = URIUtils.getUri(_controllerConf.getDataDir(), rawTableName); URI uriToMoveTo = URIUtils.getUri(_controllerConf.getDataDir(), rawTableName, URIUtils.encode(segmentName)); PinotFS pinotFS = PinotFSFactory.create(tableDirURI.getScheme()); Preconditions.checkState(pinotFS.move(segmentFileURI, uriToMoveTo, true), "Failed to move segment file for segment: %s from: %s to: %s", segmentName, segmentLocation, uriToMoveTo); try { for (String uri : pinotFS.listFiles(tableDirURI, false)) { if (uri.contains(SegmentCompletionUtils.getSegmentNamePrefix(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.toString()); } PinotLLCRealtimeSegmentManager(PinotHelixResourceManager helixResourceManager, ControllerConf controllerConf,
      ControllerMetrics controllerMetrics); boolean getIsSplitCommitEnabled(); String getControllerVipUrl(); void stop(); void setUpNewTable(TableConfig tableConfig, IdealState idealState); void removeLLCSegments(IdealState idealState); @VisibleForTesting TableConfig getTableConfig(String realtimeTableName); void commitSegmentFile(String realtimeTableName, CommittingSegmentDescriptor committingSegmentDescriptor); void commitSegmentMetadata(String realtimeTableName, CommittingSegmentDescriptor committingSegmentDescriptor); long getCommitTimeoutMS(String realtimeTableName); void segmentStoppedConsuming(LLCSegmentName llcSegmentName, String instanceName); void ensureAllPartitionsConsuming(TableConfig tableConfig, PartitionLevelStreamConfig streamConfig); }
PinotLLCRealtimeSegmentManager { public void commitSegmentFile(String realtimeTableName, CommittingSegmentDescriptor committingSegmentDescriptor) throws Exception { Preconditions.checkState(!_isStopping, "Segment manager is stopping"); if (isPeerSegmentDownloadScheme(committingSegmentDescriptor)) { LOGGER.info("No moving needed for segment on peer servers: {}", committingSegmentDescriptor.getSegmentLocation()); return; } String rawTableName = TableNameBuilder.extractRawTableName(realtimeTableName); String segmentName = committingSegmentDescriptor.getSegmentName(); LOGGER.info("Committing segment file for segment: {}", segmentName); String segmentLocation = committingSegmentDescriptor.getSegmentLocation(); URI segmentFileURI = URIUtils.getUri(segmentLocation); URI tableDirURI = URIUtils.getUri(_controllerConf.getDataDir(), rawTableName); URI uriToMoveTo = URIUtils.getUri(_controllerConf.getDataDir(), rawTableName, URIUtils.encode(segmentName)); PinotFS pinotFS = PinotFSFactory.create(tableDirURI.getScheme()); Preconditions.checkState(pinotFS.move(segmentFileURI, uriToMoveTo, true), "Failed to move segment file for segment: %s from: %s to: %s", segmentName, segmentLocation, uriToMoveTo); try { for (String uri : pinotFS.listFiles(tableDirURI, false)) { if (uri.contains(SegmentCompletionUtils.getSegmentNamePrefix(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.toString()); } PinotLLCRealtimeSegmentManager(PinotHelixResourceManager helixResourceManager, ControllerConf controllerConf,
      ControllerMetrics controllerMetrics); boolean getIsSplitCommitEnabled(); String getControllerVipUrl(); void stop(); void setUpNewTable(TableConfig tableConfig, IdealState idealState); void removeLLCSegments(IdealState idealState); @VisibleForTesting TableConfig getTableConfig(String realtimeTableName); void commitSegmentFile(String realtimeTableName, CommittingSegmentDescriptor committingSegmentDescriptor); void commitSegmentMetadata(String realtimeTableName, CommittingSegmentDescriptor committingSegmentDescriptor); long getCommitTimeoutMS(String realtimeTableName); void segmentStoppedConsuming(LLCSegmentName llcSegmentName, String instanceName); void ensureAllPartitionsConsuming(TableConfig tableConfig, PartitionLevelStreamConfig streamConfig);  }