@VisibleForTesting
  Pair<String, File> encryptSegmentIfNeeded(File tempDecryptedFile, File tempEncryptedFile,
      boolean isUploadedSegmentEncrypted, String crypterUsedInUploadedSegment, String crypterClassNameInTableConfig,
      String segmentName, String tableNameWithType) {

    boolean segmentNeedsEncryption = StringUtils.isNotEmpty(crypterClassNameInTableConfig);

    // form the output
    File finalSegmentFile =
        (isUploadedSegmentEncrypted || segmentNeedsEncryption) ? tempEncryptedFile : tempDecryptedFile;
    String crypterClassName = StringUtils.isEmpty(crypterClassNameInTableConfig) ? crypterUsedInUploadedSegment
        : crypterClassNameInTableConfig;
    ImmutablePair<String, File> out = ImmutablePair.of(crypterClassName, finalSegmentFile);

    if (!segmentNeedsEncryption) {
      return out;
    }

    if (isUploadedSegmentEncrypted && !crypterClassNameInTableConfig.equals(crypterUsedInUploadedSegment)) {
      throw new ControllerApplicationException(LOGGER, String.format(
          "Uploaded segment is encrypted with '%s' while table config requires '%s' as crypter "
              + "(segment name = '%s', table name = '%s').", crypterUsedInUploadedSegment,
          crypterClassNameInTableConfig, segmentName, tableNameWithType), Response.Status.INTERNAL_SERVER_ERROR);
    }

    // encrypt segment
    PinotCrypter pinotCrypter = PinotCrypterFactory.create(crypterClassNameInTableConfig);
    LOGGER.info("Using crypter class '{}' for encrypting '{}' to '{}' (segment name = '{}', table name = '{}').",
        crypterClassNameInTableConfig, tempDecryptedFile, tempEncryptedFile, segmentName, tableNameWithType);
    pinotCrypter.encrypt(tempDecryptedFile, tempEncryptedFile);

    return out;
  }
@Test
  public void testEncryptSegmentIfNeededCrypterInTableConfig() {

    // arrange
    boolean uploadedSegmentIsEncrypted = false;
    String crypterClassNameInTableConfig = "NoOpPinotCrypter";
    String crypterClassNameUsedInUploadedSegment = null;

    // act
    Pair<String, File> encryptionInfo = _resource
        .encryptSegmentIfNeeded(_decryptedFile, _encryptedFile, uploadedSegmentIsEncrypted,
            crypterClassNameUsedInUploadedSegment, crypterClassNameInTableConfig, SEGMENT_NAME, TABLE_NAME);

    // assert
    assertEquals("NoOpPinotCrypter", encryptionInfo.getLeft());
    assertEquals(_encryptedFile, encryptionInfo.getRight());
  }