@Test public void testCompleteSegmentOperations() throws Exception { ZKOperator zkOperator = new ZKOperator(_helixResourceManager, mock(ControllerConf.class), mock(ControllerMetrics.class)); SegmentMetadata segmentMetadata = mock(SegmentMetadata.class); when(segmentMetadata.getName()).thenReturn(SEGMENT_NAME); when(segmentMetadata.getCrc()).thenReturn("12345"); when(segmentMetadata.getIndexCreationTime()).thenReturn(123L); HttpHeaders httpHeaders = mock(HttpHeaders.class); zkOperator.completeSegmentOperations(RAW_TABLE_NAME, segmentMetadata, null, null, false, httpHeaders, "downloadUrl", false, "crypter"); OfflineSegmentZKMetadata segmentZKMetadata = _helixResourceManager.getOfflineSegmentZKMetadata(RAW_TABLE_NAME, SEGMENT_NAME); assertEquals(segmentZKMetadata.getCrc(), 12345L); assertEquals(segmentZKMetadata.getCreationTime(), 123L); long pushTime = segmentZKMetadata.getPushTime(); assertTrue(pushTime > 0); assertEquals(segmentZKMetadata.getRefreshTime(), Long.MIN_VALUE); assertEquals(segmentZKMetadata.getDownloadUrl(), "downloadUrl"); assertEquals(segmentZKMetadata.getCrypterName(), "crypter"); when(httpHeaders.getHeaderString(HttpHeaders.IF_MATCH)).thenReturn("123"); try { zkOperator.completeSegmentOperations(RAW_TABLE_NAME, segmentMetadata, null, null, false, httpHeaders, "otherDownloadUrl", false, null); fail(); } catch (Exception e) { } when(httpHeaders.getHeaderString(HttpHeaders.IF_MATCH)).thenReturn("12345"); when(segmentMetadata.getIndexCreationTime()).thenReturn(456L); zkOperator .completeSegmentOperations(RAW_TABLE_NAME, segmentMetadata, null, null, false, httpHeaders, "otherDownloadUrl", false, "otherCrypter"); segmentZKMetadata = _helixResourceManager.getOfflineSegmentZKMetadata(RAW_TABLE_NAME, SEGMENT_NAME); assertEquals(segmentZKMetadata.getCrc(), 12345L); assertEquals(segmentZKMetadata.getPushTime(), pushTime); assertEquals(segmentZKMetadata.getCreationTime(), 456L); long refreshTime = segmentZKMetadata.getRefreshTime(); assertTrue(refreshTime > 0); assertEquals(segmentZKMetadata.getDownloadUrl(), "downloadUrl"); assertEquals(segmentZKMetadata.getCrypterName(), "crypter"); when(segmentMetadata.getCrc()).thenReturn("23456"); when(segmentMetadata.getIndexCreationTime()).thenReturn(789L); Thread.sleep(10L); zkOperator .completeSegmentOperations(RAW_TABLE_NAME, segmentMetadata, null, null, false, httpHeaders, "otherDownloadUrl", false, "otherCrypter"); segmentZKMetadata = _helixResourceManager.getOfflineSegmentZKMetadata(RAW_TABLE_NAME, SEGMENT_NAME); assertEquals(segmentZKMetadata.getCrc(), 23456L); assertEquals(segmentZKMetadata.getPushTime(), pushTime); assertEquals(segmentZKMetadata.getCreationTime(), 789L); assertTrue(segmentZKMetadata.getRefreshTime() > refreshTime); assertEquals(segmentZKMetadata.getDownloadUrl(), "otherDownloadUrl"); assertEquals(segmentZKMetadata.getCrypterName(), "otherCrypter"); }
public void completeSegmentOperations(String rawTableName, SegmentMetadata segmentMetadata, URI finalSegmentLocationURI, File currentSegmentLocation, boolean enableParallelPushProtection, HttpHeaders headers, String zkDownloadURI, boolean moveSegmentToFinalLocation, String crypter) throws Exception { String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName); String segmentName = segmentMetadata.getName(); ZNRecord segmentMetadataZnRecord = _pinotHelixResourceManager.getSegmentMetadataZnRecord(offlineTableName, segmentName); if (segmentMetadataZnRecord == null) { LOGGER.info("Adding new segment {} from table {}", segmentName, rawTableName); processNewSegment(segmentMetadata, finalSegmentLocationURI, currentSegmentLocation, zkDownloadURI, crypter, rawTableName, segmentName, moveSegmentToFinalLocation); return; } LOGGER.info("Segment {} from table {} already exists, refreshing if necessary", segmentName, rawTableName); processExistingSegment(segmentMetadata, finalSegmentLocationURI, currentSegmentLocation, enableParallelPushProtection, headers, zkDownloadURI, crypter, offlineTableName, segmentName, segmentMetadataZnRecord, moveSegmentToFinalLocation); }
ZKOperator { public void completeSegmentOperations(String rawTableName, SegmentMetadata segmentMetadata, URI finalSegmentLocationURI, File currentSegmentLocation, boolean enableParallelPushProtection, HttpHeaders headers, String zkDownloadURI, boolean moveSegmentToFinalLocation, String crypter) throws Exception { String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName); String segmentName = segmentMetadata.getName(); ZNRecord segmentMetadataZnRecord = _pinotHelixResourceManager.getSegmentMetadataZnRecord(offlineTableName, segmentName); if (segmentMetadataZnRecord == null) { LOGGER.info("Adding new segment {} from table {}", segmentName, rawTableName); processNewSegment(segmentMetadata, finalSegmentLocationURI, currentSegmentLocation, zkDownloadURI, crypter, rawTableName, segmentName, moveSegmentToFinalLocation); return; } LOGGER.info("Segment {} from table {} already exists, refreshing if necessary", segmentName, rawTableName); processExistingSegment(segmentMetadata, finalSegmentLocationURI, currentSegmentLocation, enableParallelPushProtection, headers, zkDownloadURI, crypter, offlineTableName, segmentName, segmentMetadataZnRecord, moveSegmentToFinalLocation); } }
ZKOperator { public void completeSegmentOperations(String rawTableName, SegmentMetadata segmentMetadata, URI finalSegmentLocationURI, File currentSegmentLocation, boolean enableParallelPushProtection, HttpHeaders headers, String zkDownloadURI, boolean moveSegmentToFinalLocation, String crypter) throws Exception { String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName); String segmentName = segmentMetadata.getName(); ZNRecord segmentMetadataZnRecord = _pinotHelixResourceManager.getSegmentMetadataZnRecord(offlineTableName, segmentName); if (segmentMetadataZnRecord == null) { LOGGER.info("Adding new segment {} from table {}", segmentName, rawTableName); processNewSegment(segmentMetadata, finalSegmentLocationURI, currentSegmentLocation, zkDownloadURI, crypter, rawTableName, segmentName, moveSegmentToFinalLocation); return; } LOGGER.info("Segment {} from table {} already exists, refreshing if necessary", segmentName, rawTableName); processExistingSegment(segmentMetadata, finalSegmentLocationURI, currentSegmentLocation, enableParallelPushProtection, headers, zkDownloadURI, crypter, offlineTableName, segmentName, segmentMetadataZnRecord, moveSegmentToFinalLocation); } ZKOperator(PinotHelixResourceManager pinotHelixResourceManager, ControllerConf controllerConf,
      ControllerMetrics controllerMetrics); }
ZKOperator { public void completeSegmentOperations(String rawTableName, SegmentMetadata segmentMetadata, URI finalSegmentLocationURI, File currentSegmentLocation, boolean enableParallelPushProtection, HttpHeaders headers, String zkDownloadURI, boolean moveSegmentToFinalLocation, String crypter) throws Exception { String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName); String segmentName = segmentMetadata.getName(); ZNRecord segmentMetadataZnRecord = _pinotHelixResourceManager.getSegmentMetadataZnRecord(offlineTableName, segmentName); if (segmentMetadataZnRecord == null) { LOGGER.info("Adding new segment {} from table {}", segmentName, rawTableName); processNewSegment(segmentMetadata, finalSegmentLocationURI, currentSegmentLocation, zkDownloadURI, crypter, rawTableName, segmentName, moveSegmentToFinalLocation); return; } LOGGER.info("Segment {} from table {} already exists, refreshing if necessary", segmentName, rawTableName); processExistingSegment(segmentMetadata, finalSegmentLocationURI, currentSegmentLocation, enableParallelPushProtection, headers, zkDownloadURI, crypter, offlineTableName, segmentName, segmentMetadataZnRecord, moveSegmentToFinalLocation); } ZKOperator(PinotHelixResourceManager pinotHelixResourceManager, ControllerConf controllerConf,
      ControllerMetrics controllerMetrics); void completeSegmentOperations(String rawTableName, SegmentMetadata segmentMetadata,
      URI finalSegmentLocationURI, File currentSegmentLocation, boolean enableParallelPushProtection,
      HttpHeaders headers, String zkDownloadURI, boolean moveSegmentToFinalLocation, String crypter); }
ZKOperator { public void completeSegmentOperations(String rawTableName, SegmentMetadata segmentMetadata, URI finalSegmentLocationURI, File currentSegmentLocation, boolean enableParallelPushProtection, HttpHeaders headers, String zkDownloadURI, boolean moveSegmentToFinalLocation, String crypter) throws Exception { String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(rawTableName); String segmentName = segmentMetadata.getName(); ZNRecord segmentMetadataZnRecord = _pinotHelixResourceManager.getSegmentMetadataZnRecord(offlineTableName, segmentName); if (segmentMetadataZnRecord == null) { LOGGER.info("Adding new segment {} from table {}", segmentName, rawTableName); processNewSegment(segmentMetadata, finalSegmentLocationURI, currentSegmentLocation, zkDownloadURI, crypter, rawTableName, segmentName, moveSegmentToFinalLocation); return; } LOGGER.info("Segment {} from table {} already exists, refreshing if necessary", segmentName, rawTableName); processExistingSegment(segmentMetadata, finalSegmentLocationURI, currentSegmentLocation, enableParallelPushProtection, headers, zkDownloadURI, crypter, offlineTableName, segmentName, segmentMetadataZnRecord, moveSegmentToFinalLocation); } ZKOperator(PinotHelixResourceManager pinotHelixResourceManager, ControllerConf controllerConf,
      ControllerMetrics controllerMetrics); void completeSegmentOperations(String rawTableName, SegmentMetadata segmentMetadata,
      URI finalSegmentLocationURI, File currentSegmentLocation, boolean enableParallelPushProtection,
      HttpHeaders headers, String zkDownloadURI, boolean moveSegmentToFinalLocation, String crypter);  }