Repository: corona-warn-app/cwa-server
File: services/download/src/main/java/app/coronawarn/server/services/download/FederationBatchProcessor.java (Java)

Code before:

    while (!unprocessedBatches.isEmpty()) {
      FederationBatchInfo currentBatchInfo = unprocessedBatches.remove();
      seenBatches.add(currentBatchInfo.getBatchTag());
      processBatchAndReturnNextBatchId(currentBatchInfo, ERROR)
          .ifPresent(nextBatchTag -> {
            if (isEfgsEnforceDateBasedDownloadAndNotSeen(nextBatchTag)) {
              unprocessedBatches.add(new FederationBatchInfo(nextBatchTag, currentBatchInfo.getDate()));
            }
          });
    }
  }

  private boolean isEfgsEnforceDateBasedDownloadAndNotSeen(String batchTag) {
    return config.getEfgsEnforceDateBasedDownload() && !seenBatches.contains(batchTag);
  }

  private Optional<String> processBatchAndReturnNextBatchId(
      FederationBatchInfo batchInfo, FederationBatchStatus errorStatus) throws FatalFederationGatewayException {
    LocalDate date = batchInfo.getDate();
    String batchTag = batchInfo.getBatchTag();
    logger.info("Processing batch for date {} and batchTag {}.", date, batchTag);
    try {
      BatchDownloadResponse response = federationGatewayDownloadService.downloadBatch(batchTag, date);
      if (config.isBatchAuditEnabled()) {
        federationGatewayDownloadService.auditBatch(batchTag, date);
      }
      AtomicBoolean batchContainsInvalidKeys = new AtomicBoolean(false);
      response.getDiagnosisKeyBatch().ifPresentOrElse(batch -> {
        logger.info("Downloaded {} keys for date {} and batchTag {}.", batch.getKeysCount(), date, batchTag);
        List<DiagnosisKey> validDiagnosisKeys = extractValidDiagnosisKeysFromBatch(batch);
        int numOfInvalidKeys = batch.getKeysCount() - validDiagnosisKeys.size();
        if (numOfInvalidKeys > 0) {
          batchContainsInvalidKeys.set(true);
          logger.info("{} keys failed validation and were skipped.", numOfInvalidKeys);
        }
        int insertedKeys = diagnosisKeyService.saveDiagnosisKeys(validDiagnosisKeys);
        logger.info("Successfully inserted {} keys for date {} and batchTag {}.", insertedKeys, date, batchTag);
      }, () -> logger.info("Batch for date {} and batchTag {} did not contain any keys.", date, batchTag));
      batchInfoService.updateStatus(batchInfo, batchContainsInvalidKeys.get() ? PROCESSED_WITH_ERROR : PROCESSED);
      return response.getNextBatchTag();
    } catch (BatchDownloadException e) {
      logger.error(
          "Federation batch processing for date " + date + " and batchTag " + batchTag + " failed. Status set to "
              + errorStatus.name() + ".",
          e);
      batchInfoService.updateStatus(batchInfo, errorStatus);
      return Optional.empty();
    } catch (FatalFederationGatewayException e) {
      throw e;
    } catch (Exception e) {

Review comment (bug):
Noticed that there are two separate catch-clauses for this block for `BatchDownloadException` and `Exception`. They seem to be identical, maybe it's worth to merge them together. It may be worth to also mention the `BatchAuditException` in a catch-clause in order to make clear what might go wrong?

Code after:

    while (!unprocessedBatches.isEmpty()) {
      FederationBatchInfo currentBatchInfo = unprocessedBatches.remove();
      seenBatches.add(currentBatchInfo.getBatchTag());
      processBatchAndReturnNextBatchId(currentBatchInfo, ERROR)
          .ifPresent(nextBatchTag -> {
            if (isEfgsEnforceDateBasedDownloadAndNotSeen(nextBatchTag)) {
              unprocessedBatches.add(new FederationBatchInfo(nextBatchTag, currentBatchInfo.getDate()));
            }
          });
    }
  }

  private boolean isEfgsEnforceDateBasedDownloadAndNotSeen(String batchTag) {
    return config.getEfgsEnforceDateBasedDownload() && !seenBatches.contains(batchTag);
  }

  private Optional<String> processBatchAndReturnNextBatchId(
      FederationBatchInfo batchInfo, FederationBatchStatus errorStatus) throws FatalFederationGatewayException {
    LocalDate date = batchInfo.getDate();
    String batchTag = batchInfo.getBatchTag();
    logger.info("Processing batch for date {} and batchTag {}.", date, batchTag);
    try {
      BatchDownloadResponse response = federationGatewayDownloadService.downloadBatch(batchTag, date);
      if (config.isBatchAuditEnabled()) {
        federationGatewayDownloadService.auditBatch(batchTag, date);
      }
      AtomicBoolean batchContainsInvalidKeys = new AtomicBoolean(false);
      response.getDiagnosisKeyBatch().ifPresentOrElse(batch -> {
        logger.info("Downloaded {} keys for date {} and batchTag {}.", batch.getKeysCount(), date, batchTag);
        List<DiagnosisKey> validDiagnosisKeys = extractValidDiagnosisKeysFromBatch(batch);
        int numOfInvalidKeys = batch.getKeysCount() - validDiagnosisKeys.size();
        if (numOfInvalidKeys > 0) {
          batchContainsInvalidKeys.set(true);
          logger.info("{} keys failed validation and were skipped.", numOfInvalidKeys);
        }
        int insertedKeys = diagnosisKeyService.saveDiagnosisKeys(validDiagnosisKeys);
        logger.info("Successfully inserted {} keys for date {} and batchTag {}.", insertedKeys, date, batchTag);
      }, () -> logger.info("Batch for date {} and batchTag {} did not contain any keys.", date, batchTag));
      batchInfoService.updateStatus(batchInfo, batchContainsInvalidKeys.get() ? PROCESSED_WITH_ERROR : PROCESSED);
      return response.getNextBatchTag();
    } catch (FatalFederationGatewayException e) {
      throw e;
    } catch (Exception e) {
      logger.error(
          "Federation batch processing for date " + date + " and batchTag " + batchTag + " failed. Status set to "
              + errorStatus.name() + ".", e);
      batchInfoService.updateStatus(batchInfo, errorStatus);
      return Optional.empty();
    }
  }