portability-transfer/src/main/java/org/datatransferproject/transfer/JobProcessor.java
/** Process our job, whose metadata is available via {@link JobMetadata}. */
Process our job, whose metadata is available via  JobMetadata.
Process our job whose metadata is available via JobMetadata .
void processJob() {
    boolean success = false;
    UUID jobId = JobMetadata.getJobId();
    monitor.debug(() -> format("Begin processing jobId: %s", jobId), EventCode.WORKER_JOB_STARTED);
    Collection<ErrorDetail> errors = null;
    try {
      markJobStarted(jobId);
      hooks.jobStarted(jobId);
      PortabilityJob job = store.findJob(jobId);
      JobAuthorization jobAuthorization = job.jobAuthorization();
      monitor.debug(
          () ->
              format(
                  "Starting copy job, id: %s, source: %s, destination: %s",
                  jobId, job.exportService(), job.importService()));
      String scheme = jobAuthorization.encryptionScheme();
      AuthDataDecryptService decryptService = getAuthDecryptService(scheme);
      if (decryptService == null) {
        monitor.severe(
            () ->
                format(
                    "No auth decrypter found for scheme %s while processing job: %s",
                    scheme, jobId));
        return;
      }
      String encrypted = jobAuthorization.encryptedAuthData();
      byte[] encodedPrivateKey = JobMetadata.getPrivateKey();
      AuthDataPair pair = decryptService.decrypt(encrypted, encodedPrivateKey);
      AuthData exportAuthData = objectMapper.readValue(pair.getExportAuthData(), AuthData.class);
      AuthData importAuthData = objectMapper.readValue(pair.getImportAuthData(), AuthData.class);
      String exportInfoStr = job.exportInformation();
      Optional<ExportInformation> exportInfo = Optional.empty();
      if (!Strings.isNullOrEmpty(exportInfoStr)) {
        exportInfo = Optional.of(objectMapper.readValue(exportInfoStr, ExportInformation.class));
      }
      dtpInternalMetricRecorder.startedJob(
          JobMetadata.getDataType(),
          JobMetadata.getExportService(),
          JobMetadata.getImportService());
      JobMetadata.getStopWatch().start();
      errors = copier.copy(exportAuthData, importAuthData, jobId, exportInfo);
      final int numErrors = errors.size();
      monitor.debug(
          () -> format("Finished copy for jobId: %s with %d error(s).", jobId, numErrors));
      success = errors.isEmpty();
    } catch (CopyExceptionWithFailureReason e) {
      String failureReason = e.getFailureReason();
      if (failureReason.contains(FailureReasons.DESTINATION_FULL.toString())) {
        monitor.info(() -> "The remaining storage in the user's account is not enough to perform this operation.", e);
      } else if (failureReason.contains(FailureReasons.INVALID_TOKEN.toString())  ||
              failureReason.contains(FailureReasons.SESSION_INVALIDATED.toString())  ||
              failureReason.contains(FailureReasons.UNCONFIRMED_USER.toString())  ||
              failureReason.contains(FailureReasons.USER_CHECKPOINTED.toString())) {
        monitor.info(() -> "Got token error", e);
      } else {
        monitor.severe(
                () ->
                        format(
                                "Error with failure code '%s' while processing jobId: %s",
                                failureReason, jobId),
                e,
                EventCode.WORKER_JOB_ERRORED);
      }
      addFailureReasonToJob(jobId, failureReason);
    } catch (IOException | CopyException | RuntimeException e) {
      monitor.severe(() -> "Error processing jobId: " + jobId, e, EventCode.WORKER_JOB_ERRORED);
    } finally {
      monitor.debug(() -> "Finished processing jobId: " + jobId, EventCode.WORKER_JOB_FINISHED);
      addErrorsAndMarkJobFinished(jobId, success, errors);
      hooks.jobFinished(jobId, success);
      dtpInternalMetricRecorder.finishedJob(
          JobMetadata.getDataType(),
          JobMetadata.getExportService(),
          JobMetadata.getImportService(),
          success,
          JobMetadata.getStopWatch().elapsed());
      monitor.flushLogs();
      JobMetadata.reset();
    }
  }
void processJob ( ) { boolean success = false ; UUID jobId = JobMetadata . getJobId ( ) ; monitor . debug ( ( ) -> format ( "Begin processing jobId: %s" , jobId ) , EventCode . WORKER_JOB_STARTED ) ; Collection < ErrorDetail > errors = null ; try { markJobStarted ( jobId ) ; hooks . jobStarted ( jobId ) ; PortabilityJob job = store . findJob ( jobId ) ; JobAuthorization jobAuthorization = job . jobAuthorization ( ) ; monitor . debug ( ( ) -> format ( "Starting copy job, id: %s, source: %s, destination: %s" , jobId , job . exportService ( ) , job . importService ( ) ) ) ; String scheme = jobAuthorization . encryptionScheme ( ) ; AuthDataDecryptService decryptService = getAuthDecryptService ( scheme ) ; if ( decryptService == null ) { monitor . severe ( ( ) -> format ( "No auth decrypter found for scheme %s while processing job: %s" , scheme , jobId ) ) ; return ; } String encrypted = jobAuthorization . encryptedAuthData ( ) ; byte [ ] encodedPrivateKey = JobMetadata . getPrivateKey ( ) ; AuthDataPair pair = decryptService . decrypt ( encrypted , encodedPrivateKey ) ; AuthData exportAuthData = objectMapper . readValue ( pair . getExportAuthData ( ) , AuthData . class ) ; AuthData importAuthData = objectMapper . readValue ( pair . getImportAuthData ( ) , AuthData . class ) ; String exportInfoStr = job . exportInformation ( ) ; Optional < ExportInformation > exportInfo = Optional . empty ( ) ; if ( ! Strings . isNullOrEmpty ( exportInfoStr ) ) { exportInfo = Optional . of ( objectMapper . readValue ( exportInfoStr , ExportInformation . class ) ) ; } dtpInternalMetricRecorder . startedJob ( JobMetadata . getDataType ( ) , JobMetadata . getExportService ( ) , JobMetadata . getImportService ( ) ) ; JobMetadata . getStopWatch ( ) . start ( ) ; errors = copier . copy ( exportAuthData , importAuthData , jobId , exportInfo ) ; final int numErrors = errors . size ( ) ; monitor . debug ( ( ) -> format ( "Finished copy for jobId: %s with %d error(s)." , jobId , numErrors ) ) ; success = errors . isEmpty ( ) ; } catch ( CopyExceptionWithFailureReason e ) { String failureReason = e . getFailureReason ( ) ; if ( failureReason . contains ( FailureReasons . DESTINATION_FULL . toString ( ) ) ) { monitor . info ( ( ) -> "The remaining storage in the user's account is not enough to perform this operation." , e ) ; } else if ( failureReason . contains ( FailureReasons . INVALID_TOKEN . toString ( ) ) || failureReason . contains ( FailureReasons . SESSION_INVALIDATED . toString ( ) ) || failureReason . contains ( FailureReasons . UNCONFIRMED_USER . toString ( ) ) || failureReason . contains ( FailureReasons . USER_CHECKPOINTED . toString ( ) ) ) { monitor . info ( ( ) -> "Got token error" , e ) ; } else { monitor . severe ( ( ) -> format ( "Error with failure code '%s' while processing jobId: %s" , failureReason , jobId ) , e , EventCode . WORKER_JOB_ERRORED ) ; } addFailureReasonToJob ( jobId , failureReason ) ; } catch ( IOException | CopyException | RuntimeException e ) { monitor . severe ( ( ) -> "Error processing jobId: " + jobId , e , EventCode . WORKER_JOB_ERRORED ) ; } finally { monitor . debug ( ( ) -> "Finished processing jobId: " + jobId , EventCode . WORKER_JOB_FINISHED ) ; addErrorsAndMarkJobFinished ( jobId , success , errors ) ; hooks . jobFinished ( jobId , success ) ; dtpInternalMetricRecorder . finishedJob ( JobMetadata . getDataType ( ) , JobMetadata . getExportService ( ) , JobMetadata . getImportService ( ) , success , JobMetadata . getStopWatch ( ) . elapsed ( ) ) ; monitor . flushLogs ( ) ; JobMetadata . reset ( ) ; } }
Process our job, whose metadata is available via {@link JobMetadata}.
Process our job whose metadata is available via { @link JobMetadata } .
// 根据jobId获取一个可迁移的job任务 // 获取导入授权信息认证 // Copy the data A DTP job started // TODO: 2022/2/18 In-memory Copier interface, 拷贝数据总入口