@Override
  public SegmentPublishResult perform(Task task, TaskActionToolbox toolbox)
  {
    TaskActionPreconditions.checkLockCoversSegments(task, toolbox.getTaskLockbox(), segments);

    final SegmentPublishResult retVal;
    try {
      retVal = toolbox.getTaskLockbox().doInCriticalSection(
          task,
          segments.stream().map(DataSegment::getInterval).collect(Collectors.toList()),
          CriticalAction.<SegmentPublishResult>builder()
              .onValidLocks(
                  () -> toolbox.getIndexerMetadataStorageCoordinator().announceHistoricalSegments(
                      segments,
                      startMetadata,
                      endMetadata
                  )
              )
              .onInvalidLocks(
                  () -> SegmentPublishResult.fail(
                      "Invalid task locks. Maybe they are revoked by a higher priority task."
                      + " Please check the overlord log for details."
                  )
              )
              .build()
      );
    }
    catch (Exception e) {
      throw new RuntimeException(e);
    }

    // Emit metrics
    final ServiceMetricEvent.Builder metricBuilder = new ServiceMetricEvent.Builder();
    IndexTaskUtils.setTaskDimensions(metricBuilder, task);

    if (retVal.isSuccess()) {
      toolbox.getEmitter().emit(metricBuilder.build("segment/txn/success", 1));
    } else {
      toolbox.getEmitter().emit(metricBuilder.build("segment/txn/failure", 1));
    }

    // getSegments() should return an empty set if announceHistoricalSegments() failed
    for (DataSegment segment : retVal.getSegments()) {
      metricBuilder.setDimension(DruidMetrics.INTERVAL, segment.getInterval().toString());
      toolbox.getEmitter().emit(metricBuilder.build("segment/added/bytes", segment.getSize()));
    }

    return retVal;
  }
@ Override public SegmentPublishResult perform ( Task task , TaskActionToolbox toolbox ) { TaskActionPreconditions . checkLockCoversSegments ( task , toolbox . getTaskLockbox ( ) , segments ) ; final SegmentPublishResult retVal ; try { retVal = toolbox . getTaskLockbox ( ) . doInCriticalSection ( task , segments . stream ( ) . map ( DataSegment :: getInterval ) . collect ( Collectors . toList ( ) ) , CriticalAction . < SegmentPublishResult > builder ( ) . onValidLocks ( ( ) -> toolbox . getIndexerMetadataStorageCoordinator ( ) . announceHistoricalSegments ( segments , startMetadata , endMetadata ) ) . onInvalidLocks ( ( ) -> SegmentPublishResult . fail ( "Invalid task locks. Maybe they are revoked by a higher priority task." + " Please check the overlord log for details." ) ) . build ( ) ) ; } catch ( Exception e ) { throw new RuntimeException ( e ) ; } final ServiceMetricEvent . Builder metricBuilder = new ServiceMetricEvent . Builder ( ) ; IndexTaskUtils . setTaskDimensions ( metricBuilder , task ) ; if ( retVal . isSuccess ( ) ) { toolbox . getEmitter ( ) . emit ( metricBuilder . build ( "segment/txn/success" , 1 ) ) ; } else { toolbox . getEmitter ( ) . emit ( metricBuilder . build ( "segment/txn/failure" , 1 ) ) ; } for ( DataSegment segment : retVal . getSegments ( ) ) { metricBuilder . setDimension ( DruidMetrics . INTERVAL , segment . getInterval ( ) . toString ( ) ) ; toolbox . getEmitter ( ) . emit ( metricBuilder . build ( "segment/added/bytes" , segment . getSize ( ) ) ) ; } return retVal ; }
Behaves similarly to
{@link org.apache.druid.indexing.overlord.IndexerMetadataStorageCoordinator#announceHistoricalSegments(Set, DataSourceMetadata, DataSourceMetadata)}.