@Test public void bulkCreateFromWorkspaceWithRetriesBlockerFailure() throws Exception { String workspaceContainer = "workspaceContainer"; List<String> workspaceUris = Arrays.asList("uri1", "uri2", "uri3"); List<String> objectNames = Arrays.asList("obj1", "obj2", "obj3"); List<String> offerIds = Arrays.asList("offer1", "offer2"); List<ObjectInfo> objectInfos = Arrays.asList( new ObjectInfo("obj1", "digest1", 1L), new ObjectInfo("obj2", "digest2", 2L), new ObjectInfo("obj3", "digest3", 3L) ); ImmutableMap<String, OfferBulkPutStatus> statusByOfferIds1 = ImmutableMap.of( "offer1", OfferBulkPutStatus.OK, "offer2", OfferBulkPutStatus.BLOCKER ); when(bulkPutTransferManager.bulkSendDataToOffers(workspaceContainer, TENANT_ID, DATA_CATEGORY, offerIds, driverMap, storageOfferMap, workspaceUris, objectNames) ).thenReturn(new BulkPutResult(objectInfos, statusByOfferIds1)); assertThatThrownBy(() -> instance.bulkCreateFromWorkspaceWithRetries( TENANT_ID, offerIds, driverMap, storageOfferMap, DATA_CATEGORY, workspaceContainer, workspaceUris, objectNames, REQUESTER )) .isInstanceOf(StorageException.class); checkResult(objectInfos, null, "offer1 attempt 1 : OK, offer2 attempt 1 : BLOCKER", "KO"); }
public Map<String, String> bulkCreateFromWorkspaceWithRetries(int tenantId, List<String> allOfferIds, Map<String, Driver> storageDrivers, Map<String, StorageOffer> storageOffers, DataCategory dataCategory, String workspaceContainerGUID, List<String> workspaceObjectURIs, List<String> objectIds, String requester) throws StorageException { List<String> remainingOfferIds = new ArrayList<>(allOfferIds); Map<String, ObjectInfo> objectInfos = null; List<String> events = new ArrayList<>(); try { for (int attempt = 1; attempt <= nbReties; attempt++) { BulkPutResult bulkOutResult = bulkPutTransferManager.bulkSendDataToOffers(workspaceContainerGUID, tenantId, dataCategory, remainingOfferIds, storageDrivers, storageOffers, workspaceObjectURIs, objectIds); if (bulkOutResult.getObjectInfos() != null) { objectInfos = bulkOutResult.getObjectInfos().stream() .collect(Collectors.toMap(ObjectInfo::getObjectId, objectInfo -> objectInfo)); } for (Map.Entry<String, OfferBulkPutStatus> entry : bulkOutResult.getStatusByOfferIds().entrySet()) { String offerId = entry.getKey(); OfferBulkPutStatus status = entry.getValue(); events.add(offerId + ATTEMPT + attempt + " : " + status); if (status == OfferBulkPutStatus.OK) { remainingOfferIds.remove(offerId); } } boolean hasBlockerError = bulkOutResult.getStatusByOfferIds().values().contains(OfferBulkPutStatus.BLOCKER); if (hasBlockerError) { throw new StorageInconsistentStateException("A fatal error occurred during bulk object storage"); } if (remainingOfferIds.isEmpty()) { return objectInfos.entrySet().stream() .collect(Collectors.toMap(Map.Entry::getKey, entry -> entry.getValue().getDigest())); } } throw new StorageException("Could not proceed bulk put operation after " + nbReties + " attempts"); } finally { logStorageEvents(tenantId, dataCategory, objectIds, requester, remainingOfferIds, objectInfos, events); } }
BulkStorageDistribution { public Map<String, String> bulkCreateFromWorkspaceWithRetries(int tenantId, List<String> allOfferIds, Map<String, Driver> storageDrivers, Map<String, StorageOffer> storageOffers, DataCategory dataCategory, String workspaceContainerGUID, List<String> workspaceObjectURIs, List<String> objectIds, String requester) throws StorageException { List<String> remainingOfferIds = new ArrayList<>(allOfferIds); Map<String, ObjectInfo> objectInfos = null; List<String> events = new ArrayList<>(); try { for (int attempt = 1; attempt <= nbReties; attempt++) { BulkPutResult bulkOutResult = bulkPutTransferManager.bulkSendDataToOffers(workspaceContainerGUID, tenantId, dataCategory, remainingOfferIds, storageDrivers, storageOffers, workspaceObjectURIs, objectIds); if (bulkOutResult.getObjectInfos() != null) { objectInfos = bulkOutResult.getObjectInfos().stream() .collect(Collectors.toMap(ObjectInfo::getObjectId, objectInfo -> objectInfo)); } for (Map.Entry<String, OfferBulkPutStatus> entry : bulkOutResult.getStatusByOfferIds().entrySet()) { String offerId = entry.getKey(); OfferBulkPutStatus status = entry.getValue(); events.add(offerId + ATTEMPT + attempt + " : " + status); if (status == OfferBulkPutStatus.OK) { remainingOfferIds.remove(offerId); } } boolean hasBlockerError = bulkOutResult.getStatusByOfferIds().values().contains(OfferBulkPutStatus.BLOCKER); if (hasBlockerError) { throw new StorageInconsistentStateException("A fatal error occurred during bulk object storage"); } if (remainingOfferIds.isEmpty()) { return objectInfos.entrySet().stream() .collect(Collectors.toMap(Map.Entry::getKey, entry -> entry.getValue().getDigest())); } } throw new StorageException("Could not proceed bulk put operation after " + nbReties + " attempts"); } finally { logStorageEvents(tenantId, dataCategory, objectIds, requester, remainingOfferIds, objectInfos, events); } } }
BulkStorageDistribution { public Map<String, String> bulkCreateFromWorkspaceWithRetries(int tenantId, List<String> allOfferIds, Map<String, Driver> storageDrivers, Map<String, StorageOffer> storageOffers, DataCategory dataCategory, String workspaceContainerGUID, List<String> workspaceObjectURIs, List<String> objectIds, String requester) throws StorageException { List<String> remainingOfferIds = new ArrayList<>(allOfferIds); Map<String, ObjectInfo> objectInfos = null; List<String> events = new ArrayList<>(); try { for (int attempt = 1; attempt <= nbReties; attempt++) { BulkPutResult bulkOutResult = bulkPutTransferManager.bulkSendDataToOffers(workspaceContainerGUID, tenantId, dataCategory, remainingOfferIds, storageDrivers, storageOffers, workspaceObjectURIs, objectIds); if (bulkOutResult.getObjectInfos() != null) { objectInfos = bulkOutResult.getObjectInfos().stream() .collect(Collectors.toMap(ObjectInfo::getObjectId, objectInfo -> objectInfo)); } for (Map.Entry<String, OfferBulkPutStatus> entry : bulkOutResult.getStatusByOfferIds().entrySet()) { String offerId = entry.getKey(); OfferBulkPutStatus status = entry.getValue(); events.add(offerId + ATTEMPT + attempt + " : " + status); if (status == OfferBulkPutStatus.OK) { remainingOfferIds.remove(offerId); } } boolean hasBlockerError = bulkOutResult.getStatusByOfferIds().values().contains(OfferBulkPutStatus.BLOCKER); if (hasBlockerError) { throw new StorageInconsistentStateException("A fatal error occurred during bulk object storage"); } if (remainingOfferIds.isEmpty()) { return objectInfos.entrySet().stream() .collect(Collectors.toMap(Map.Entry::getKey, entry -> entry.getValue().getDigest())); } } throw new StorageException("Could not proceed bulk put operation after " + nbReties + " attempts"); } finally { logStorageEvents(tenantId, dataCategory, objectIds, requester, remainingOfferIds, objectInfos, events); } } BulkStorageDistribution(int nbReties, WorkspaceClientFactory workspaceClientFactory,
        StorageLog storageLogService, TransfertTimeoutHelper transfertTimeoutHelper); @VisibleForTesting  BulkStorageDistribution(int nbReties, StorageLog storageLogService,
        DigestType digestType, BulkPutTransferManager bulkPutTransferManager); }
BulkStorageDistribution { public Map<String, String> bulkCreateFromWorkspaceWithRetries(int tenantId, List<String> allOfferIds, Map<String, Driver> storageDrivers, Map<String, StorageOffer> storageOffers, DataCategory dataCategory, String workspaceContainerGUID, List<String> workspaceObjectURIs, List<String> objectIds, String requester) throws StorageException { List<String> remainingOfferIds = new ArrayList<>(allOfferIds); Map<String, ObjectInfo> objectInfos = null; List<String> events = new ArrayList<>(); try { for (int attempt = 1; attempt <= nbReties; attempt++) { BulkPutResult bulkOutResult = bulkPutTransferManager.bulkSendDataToOffers(workspaceContainerGUID, tenantId, dataCategory, remainingOfferIds, storageDrivers, storageOffers, workspaceObjectURIs, objectIds); if (bulkOutResult.getObjectInfos() != null) { objectInfos = bulkOutResult.getObjectInfos().stream() .collect(Collectors.toMap(ObjectInfo::getObjectId, objectInfo -> objectInfo)); } for (Map.Entry<String, OfferBulkPutStatus> entry : bulkOutResult.getStatusByOfferIds().entrySet()) { String offerId = entry.getKey(); OfferBulkPutStatus status = entry.getValue(); events.add(offerId + ATTEMPT + attempt + " : " + status); if (status == OfferBulkPutStatus.OK) { remainingOfferIds.remove(offerId); } } boolean hasBlockerError = bulkOutResult.getStatusByOfferIds().values().contains(OfferBulkPutStatus.BLOCKER); if (hasBlockerError) { throw new StorageInconsistentStateException("A fatal error occurred during bulk object storage"); } if (remainingOfferIds.isEmpty()) { return objectInfos.entrySet().stream() .collect(Collectors.toMap(Map.Entry::getKey, entry -> entry.getValue().getDigest())); } } throw new StorageException("Could not proceed bulk put operation after " + nbReties + " attempts"); } finally { logStorageEvents(tenantId, dataCategory, objectIds, requester, remainingOfferIds, objectInfos, events); } } BulkStorageDistribution(int nbReties, WorkspaceClientFactory workspaceClientFactory,
        StorageLog storageLogService, TransfertTimeoutHelper transfertTimeoutHelper); @VisibleForTesting  BulkStorageDistribution(int nbReties, StorageLog storageLogService,
        DigestType digestType, BulkPutTransferManager bulkPutTransferManager); Map<String, String> bulkCreateFromWorkspaceWithRetries(int tenantId,
        List<String> allOfferIds, Map<String, Driver> storageDrivers,
        Map<String, StorageOffer> storageOffers,
        DataCategory dataCategory, String workspaceContainerGUID,
        List<String> workspaceObjectURIs, List<String> objectIds, String requester); }
BulkStorageDistribution { public Map<String, String> bulkCreateFromWorkspaceWithRetries(int tenantId, List<String> allOfferIds, Map<String, Driver> storageDrivers, Map<String, StorageOffer> storageOffers, DataCategory dataCategory, String workspaceContainerGUID, List<String> workspaceObjectURIs, List<String> objectIds, String requester) throws StorageException { List<String> remainingOfferIds = new ArrayList<>(allOfferIds); Map<String, ObjectInfo> objectInfos = null; List<String> events = new ArrayList<>(); try { for (int attempt = 1; attempt <= nbReties; attempt++) { BulkPutResult bulkOutResult = bulkPutTransferManager.bulkSendDataToOffers(workspaceContainerGUID, tenantId, dataCategory, remainingOfferIds, storageDrivers, storageOffers, workspaceObjectURIs, objectIds); if (bulkOutResult.getObjectInfos() != null) { objectInfos = bulkOutResult.getObjectInfos().stream() .collect(Collectors.toMap(ObjectInfo::getObjectId, objectInfo -> objectInfo)); } for (Map.Entry<String, OfferBulkPutStatus> entry : bulkOutResult.getStatusByOfferIds().entrySet()) { String offerId = entry.getKey(); OfferBulkPutStatus status = entry.getValue(); events.add(offerId + ATTEMPT + attempt + " : " + status); if (status == OfferBulkPutStatus.OK) { remainingOfferIds.remove(offerId); } } boolean hasBlockerError = bulkOutResult.getStatusByOfferIds().values().contains(OfferBulkPutStatus.BLOCKER); if (hasBlockerError) { throw new StorageInconsistentStateException("A fatal error occurred during bulk object storage"); } if (remainingOfferIds.isEmpty()) { return objectInfos.entrySet().stream() .collect(Collectors.toMap(Map.Entry::getKey, entry -> entry.getValue().getDigest())); } } throw new StorageException("Could not proceed bulk put operation after " + nbReties + " attempts"); } finally { logStorageEvents(tenantId, dataCategory, objectIds, requester, remainingOfferIds, objectInfos, events); } } BulkStorageDistribution(int nbReties, WorkspaceClientFactory workspaceClientFactory,
        StorageLog storageLogService, TransfertTimeoutHelper transfertTimeoutHelper); @VisibleForTesting  BulkStorageDistribution(int nbReties, StorageLog storageLogService,
        DigestType digestType, BulkPutTransferManager bulkPutTransferManager); Map<String, String> bulkCreateFromWorkspaceWithRetries(int tenantId,
        List<String> allOfferIds, Map<String, Driver> storageDrivers,
        Map<String, StorageOffer> storageOffers,
        DataCategory dataCategory, String workspaceContainerGUID,
        List<String> workspaceObjectURIs, List<String> objectIds, String requester);  }