@Override
  public void reloadSegment(String segmentName, IndexLoadingConfig indexLoadingConfig, SegmentZKMetadata zkMetadata,
      SegmentMetadata localMetadata, @Nullable Schema schema, boolean forceDownload)
      throws Exception {
    Preconditions.checkState(!_shutDown,
        "Table data manager is already shut down, cannot reload segment: %s of table: %s", segmentName,
        _tableNameWithType);
    _logger.info("Reloading segment: {}", segmentName);
    String segmentTier = getSegmentCurrentTier(segmentName);
    indexLoadingConfig.setSegmentTier(segmentTier);
    indexLoadingConfig.setTableDataDir(_tableDataDir);
    indexLoadingConfig.setInstanceTierConfigs(_instanceDataManagerConfig.getTierConfigs());
    File indexDir = getSegmentDataDir(segmentName, segmentTier, indexLoadingConfig.getTableConfig());
    Lock segmentLock = getSegmentLock(segmentName);
    segmentLock.lock();
    try {
      // Download segment from deep store if CRC changes or forced to download;
      // otherwise, copy backup directory back to the original index directory.
      // And then continue to load the segment from the index directory.
      boolean shouldDownload = forceDownload || !hasSameCRC(zkMetadata, localMetadata);
      if (shouldDownload) {
        // Create backup directory to handle failure of segment reloading.
        createBackup(indexDir);
        if (forceDownload) {
          _logger.info("Force downloading segment: {}", segmentName);
        } else {
          _logger.info("Downloading segment: {} because its CRC has changed from: {} to: {}", segmentName,
              localMetadata.getCrc(), zkMetadata.getCrc());
        }
        indexDir = downloadSegment(zkMetadata);
      } else {
        _logger.info("Reloading existing segment: {} on tier: {}", segmentName,
            TierConfigUtils.normalizeTierName(segmentTier));
        SegmentDirectory segmentDirectory =
            initSegmentDirectory(segmentName, String.valueOf(zkMetadata.getCrc()), indexLoadingConfig);
        // We should first try to reuse existing segment directory
        if (canReuseExistingDirectoryForReload(zkMetadata, segmentTier, segmentDirectory, indexLoadingConfig, schema)) {
          _logger.info("Reloading segment: {} using existing segment directory as no reprocessing needed", segmentName);
          // No reprocessing needed, reuse the same segment
          ImmutableSegment segment = ImmutableSegmentLoader.load(segmentDirectory, indexLoadingConfig, schema);
          addSegment(segment);
          return;
        }
        // Create backup directory to handle failure of segment reloading.
        createBackup(indexDir);
        // The indexDir is empty after calling createBackup, as it's renamed to a backup directory.
        // The SegmentDirectory should initialize accordingly. Like for SegmentLocalFSDirectory, it
        // doesn't load anything from an empty indexDir, but gets the info to complete the copyTo.
        try {
          segmentDirectory.copyTo(indexDir);
        } finally {
          segmentDirectory.close();
        }
      }

      // Load from indexDir and replace the old segment in memory. What's inside indexDir
      // may come from SegmentDirectory.copyTo() or the segment downloaded from deep store.
      indexLoadingConfig.setSegmentTier(zkMetadata.getTier());
      _logger.info("Loading segment: {} from indexDir: {} to tier: {}", segmentName, indexDir,
          TierConfigUtils.normalizeTierName(zkMetadata.getTier()));
      ImmutableSegment segment = ImmutableSegmentLoader.load(indexDir, indexLoadingConfig, schema);
      addSegment(segment);

      // Remove backup directory to mark the completion of segment reloading.
      removeBackup(indexDir);
    } catch (Exception reloadFailureException) {
      try {
        LoaderUtils.reloadFailureRecovery(indexDir);
      } catch (Exception recoveryFailureException) {
        _logger.error("Failed to recover segment: {} after reload failure", segmentName, recoveryFailureException);
        reloadFailureException.addSuppressed(recoveryFailureException);
      }
      addSegmentError(segmentName,
          new SegmentErrorInfo(System.currentTimeMillis(), "Caught exception while reloading segment",
              reloadFailureException));
      throw reloadFailureException;
    } finally {
      segmentLock.unlock();
    }
    _logger.info("Reloaded segment: {}", segmentName);
  }
@Test
  public void testReloadSegmentForceDownload()
      throws Exception {
    File indexDir = createSegment(SegmentVersion.v3, 5);
    SegmentZKMetadata zkMetadata =
        makeRawSegment(indexDir, new File(TEMP_DIR, SEGMENT_NAME + TarCompressionUtils.TAR_GZ_FILE_EXTENSION), false);

    // Same CRC but force to download.
    BaseTableDataManager tableDataManager = createTableManager();
    SegmentMetadataImpl segmentMetadata = new SegmentMetadataImpl(indexDir);
    assertEquals(Long.parseLong(segmentMetadata.getCrc()), zkMetadata.getCrc());

    // Remove the local segment dir. Segment reloading fails unless force to download.
    FileUtils.deleteQuietly(indexDir);
    try {
      tableDataManager.reloadSegment(SEGMENT_NAME, new IndexLoadingConfig(), zkMetadata, segmentMetadata, null, false);
      fail();
    } catch (Exception e) {
      // As expected, segment reloading fails due to missing the local segment dir.
    }

    tableDataManager.reloadSegment(SEGMENT_NAME, new IndexLoadingConfig(), zkMetadata, segmentMetadata, null, true);
    assertTrue(indexDir.exists());
    segmentMetadata = new SegmentMetadataImpl(indexDir);
    assertEquals(Long.parseLong(segmentMetadata.getCrc()), zkMetadata.getCrc());
    assertEquals(segmentMetadata.getTotalDocs(), 5);
  }