@Test @RunWithCustomExecutor public void bulkSendDataToOfferTestDeadlockOfferFailureTransferThreadShutdown() throws Exception { List<String> offerIds = Arrays.asList("offer1", "offer2"); Map<String, StorageOffer> storageOfferMap = ImmutableMap.of( "offer1", mock(StorageOffer.class), "offer2", mock(StorageOffer.class)); Map<String, Driver> driverMap = ImmutableMap.of( "offer1", driver1, "offer2", driver2); List<String> workspaceObjectURIs = Arrays.asList("uri1"); List<String> objectIds = Arrays.asList("obj1"); long longFileSize = 10_000_000L; File veryLargeFile = folder.newFile(GUIDFactory.newGUID().toString()); Digest digest = new Digest(DigestType.SHA512); try(OutputStream os = new FileOutputStream(veryLargeFile)) { MultiplexedStreamWriter multiplexedStreamWriter = new MultiplexedStreamWriter(os); NullInputStream bigFileInputStream = new NullInputStream(longFileSize); multiplexedStreamWriter.appendEntry(longFileSize, digest.getDigestInputStream(bigFileInputStream)); multiplexedStreamWriter.appendEndOfFile(); } doReturn(Response.ok(new FileInputStream(veryLargeFile)) .header(VitamHttpHeader.X_CONTENT_LENGTH.getName(), veryLargeFile.length()).build()) .when(workspaceClient).bulkGetObjects(WORKSPACE_CONTAINER, workspaceObjectURIs); StorageBulkPutResult storageBulkPutResult = new StorageBulkPutResult(Arrays.asList( new StorageBulkPutResultEntry("obj1", digest.digestHex(), longFileSize) )); doReturn(storageBulkPutResult).when(connection1).bulkPutObjects(any()); doThrow(new StorageDriverException("driver", "ko", true)).when(connection2).bulkPutObjects(any()); BulkPutResult bulkPutResult = bulkPutTransferManager.bulkSendDataToOffers(WORKSPACE_CONTAINER, TENANT_ID, DATA_CATEGORY, offerIds, driverMap, storageOfferMap, workspaceObjectURIs, objectIds); assertThat(bulkPutResult.getObjectInfos()).hasSize(1); assertThat(bulkPutResult.getObjectInfos().get(0).getObjectId()).isEqualTo("obj1"); assertThat(bulkPutResult.getObjectInfos().get(0).getDigest()).isEqualTo(digest.digestHex()); assertThat(bulkPutResult.getObjectInfos().get(0).getSize()).isEqualTo(longFileSize); assertThat(bulkPutResult.getStatusByOfferIds()).isEqualTo(ImmutableMap.of( "offer1", OfferBulkPutStatus.OK, "offer2", OfferBulkPutStatus.KO )); }
BulkPutResult bulkSendDataToOffers(String workspaceContainerGUID, int tenantId, DataCategory dataCategory, List<String> offerIds, Map<String, Driver> storageDrivers, Map<String, StorageOffer> storageOffers, List<String> workspaceObjectURIs, List<String> objectIds) { ResultOrError<StreamAndInfo, BulkPutResult> streamWithInfoFromWorkspace = null; ResultOrError<StreamAndInfo, BulkPutResult> prependedStreamWithInfo = null; MultiplePipedInputStream streams = null; List<Future<StorageBulkPutResult>> transferThreadFutures = null; Future<List<ObjectInfo>> digestListenerFuture = null; try { streamWithInfoFromWorkspace = retrieveBulkDataFromWorkspace(workspaceContainerGUID, workspaceObjectURIs, offerIds); if (streamWithInfoFromWorkspace.hasError()) { return streamWithInfoFromWorkspace.getError(); } prependedStreamWithInfo = prependWithObjectIdsHeaderEntry(objectIds, streamWithInfoFromWorkspace.getResult(), offerIds); if (prependedStreamWithInfo.hasError()) { return prependedStreamWithInfo.getError(); } streams = new MultiplePipedInputStream(prependedStreamWithInfo.getResult().getStream(), offerIds.size() + 1); transferThreadFutures = startTransferThreads(tenantId, dataCategory, objectIds, offerIds, storageDrivers, storageOffers, streams, prependedStreamWithInfo.getResult().getSize()); digestListenerFuture = startDigestComputeThread(offerIds, streams, objectIds); long finalTimeout = transfertTimeoutHelper.getTransferTimeout( prependedStreamWithInfo.getResult().getSize()); TimeoutStopwatch timeoutStopwatch = new TimeoutStopwatch(finalTimeout); ResultOrError<List<ObjectInfo>, BulkPutResult> objectInfos = awaitDigestListenerThread(digestListenerFuture, timeoutStopwatch, offerIds); if (objectInfos.hasError()) { return objectInfos.getError(); } HashMap<String, OfferBulkPutStatus> statusByOfferIds = new HashMap<>(); for (int rank = 0; rank < offerIds.size(); rank++) { String offerId = offerIds.get(rank); Future<StorageBulkPutResult> transferThreadFuture = transferThreadFutures.get(rank); OfferBulkPutStatus status = awaitTransferThread(timeoutStopwatch, objectInfos.getResult(), offerId, transferThreadFuture); statusByOfferIds.put(offerId, status); } return new BulkPutResult(objectInfos.getResult(), statusByOfferIds); } finally { if (transferThreadFutures != null) { for (Future<StorageBulkPutResult> transferThreadFuture : transferThreadFutures) { transferThreadFuture.cancel(true); } } if (digestListenerFuture != null) { digestListenerFuture.cancel(true); } if (streams != null) { streams.close(); } if (prependedStreamWithInfo != null && prependedStreamWithInfo.hasResult()) { prependedStreamWithInfo.getResult().close(); } if (streamWithInfoFromWorkspace != null && streamWithInfoFromWorkspace.hasResult()) { streamWithInfoFromWorkspace.getResult().close(); } } }
BulkPutTransferManager { BulkPutResult bulkSendDataToOffers(String workspaceContainerGUID, int tenantId, DataCategory dataCategory, List<String> offerIds, Map<String, Driver> storageDrivers, Map<String, StorageOffer> storageOffers, List<String> workspaceObjectURIs, List<String> objectIds) { ResultOrError<StreamAndInfo, BulkPutResult> streamWithInfoFromWorkspace = null; ResultOrError<StreamAndInfo, BulkPutResult> prependedStreamWithInfo = null; MultiplePipedInputStream streams = null; List<Future<StorageBulkPutResult>> transferThreadFutures = null; Future<List<ObjectInfo>> digestListenerFuture = null; try { streamWithInfoFromWorkspace = retrieveBulkDataFromWorkspace(workspaceContainerGUID, workspaceObjectURIs, offerIds); if (streamWithInfoFromWorkspace.hasError()) { return streamWithInfoFromWorkspace.getError(); } prependedStreamWithInfo = prependWithObjectIdsHeaderEntry(objectIds, streamWithInfoFromWorkspace.getResult(), offerIds); if (prependedStreamWithInfo.hasError()) { return prependedStreamWithInfo.getError(); } streams = new MultiplePipedInputStream(prependedStreamWithInfo.getResult().getStream(), offerIds.size() + 1); transferThreadFutures = startTransferThreads(tenantId, dataCategory, objectIds, offerIds, storageDrivers, storageOffers, streams, prependedStreamWithInfo.getResult().getSize()); digestListenerFuture = startDigestComputeThread(offerIds, streams, objectIds); long finalTimeout = transfertTimeoutHelper.getTransferTimeout( prependedStreamWithInfo.getResult().getSize()); TimeoutStopwatch timeoutStopwatch = new TimeoutStopwatch(finalTimeout); ResultOrError<List<ObjectInfo>, BulkPutResult> objectInfos = awaitDigestListenerThread(digestListenerFuture, timeoutStopwatch, offerIds); if (objectInfos.hasError()) { return objectInfos.getError(); } HashMap<String, OfferBulkPutStatus> statusByOfferIds = new HashMap<>(); for (int rank = 0; rank < offerIds.size(); rank++) { String offerId = offerIds.get(rank); Future<StorageBulkPutResult> transferThreadFuture = transferThreadFutures.get(rank); OfferBulkPutStatus status = awaitTransferThread(timeoutStopwatch, objectInfos.getResult(), offerId, transferThreadFuture); statusByOfferIds.put(offerId, status); } return new BulkPutResult(objectInfos.getResult(), statusByOfferIds); } finally { if (transferThreadFutures != null) { for (Future<StorageBulkPutResult> transferThreadFuture : transferThreadFutures) { transferThreadFuture.cancel(true); } } if (digestListenerFuture != null) { digestListenerFuture.cancel(true); } if (streams != null) { streams.close(); } if (prependedStreamWithInfo != null && prependedStreamWithInfo.hasResult()) { prependedStreamWithInfo.getResult().close(); } if (streamWithInfoFromWorkspace != null && streamWithInfoFromWorkspace.hasResult()) { streamWithInfoFromWorkspace.getResult().close(); } } } }
BulkPutTransferManager { BulkPutResult bulkSendDataToOffers(String workspaceContainerGUID, int tenantId, DataCategory dataCategory, List<String> offerIds, Map<String, Driver> storageDrivers, Map<String, StorageOffer> storageOffers, List<String> workspaceObjectURIs, List<String> objectIds) { ResultOrError<StreamAndInfo, BulkPutResult> streamWithInfoFromWorkspace = null; ResultOrError<StreamAndInfo, BulkPutResult> prependedStreamWithInfo = null; MultiplePipedInputStream streams = null; List<Future<StorageBulkPutResult>> transferThreadFutures = null; Future<List<ObjectInfo>> digestListenerFuture = null; try { streamWithInfoFromWorkspace = retrieveBulkDataFromWorkspace(workspaceContainerGUID, workspaceObjectURIs, offerIds); if (streamWithInfoFromWorkspace.hasError()) { return streamWithInfoFromWorkspace.getError(); } prependedStreamWithInfo = prependWithObjectIdsHeaderEntry(objectIds, streamWithInfoFromWorkspace.getResult(), offerIds); if (prependedStreamWithInfo.hasError()) { return prependedStreamWithInfo.getError(); } streams = new MultiplePipedInputStream(prependedStreamWithInfo.getResult().getStream(), offerIds.size() + 1); transferThreadFutures = startTransferThreads(tenantId, dataCategory, objectIds, offerIds, storageDrivers, storageOffers, streams, prependedStreamWithInfo.getResult().getSize()); digestListenerFuture = startDigestComputeThread(offerIds, streams, objectIds); long finalTimeout = transfertTimeoutHelper.getTransferTimeout( prependedStreamWithInfo.getResult().getSize()); TimeoutStopwatch timeoutStopwatch = new TimeoutStopwatch(finalTimeout); ResultOrError<List<ObjectInfo>, BulkPutResult> objectInfos = awaitDigestListenerThread(digestListenerFuture, timeoutStopwatch, offerIds); if (objectInfos.hasError()) { return objectInfos.getError(); } HashMap<String, OfferBulkPutStatus> statusByOfferIds = new HashMap<>(); for (int rank = 0; rank < offerIds.size(); rank++) { String offerId = offerIds.get(rank); Future<StorageBulkPutResult> transferThreadFuture = transferThreadFutures.get(rank); OfferBulkPutStatus status = awaitTransferThread(timeoutStopwatch, objectInfos.getResult(), offerId, transferThreadFuture); statusByOfferIds.put(offerId, status); } return new BulkPutResult(objectInfos.getResult(), statusByOfferIds); } finally { if (transferThreadFutures != null) { for (Future<StorageBulkPutResult> transferThreadFuture : transferThreadFutures) { transferThreadFuture.cancel(true); } } if (digestListenerFuture != null) { digestListenerFuture.cancel(true); } if (streams != null) { streams.close(); } if (prependedStreamWithInfo != null && prependedStreamWithInfo.hasResult()) { prependedStreamWithInfo.getResult().close(); } if (streamWithInfoFromWorkspace != null && streamWithInfoFromWorkspace.hasResult()) { streamWithInfoFromWorkspace.getResult().close(); } } } BulkPutTransferManager(WorkspaceClientFactory workspaceClientFactory, TransfertTimeoutHelper transfertTimeoutHelper); @VisibleForTesting  BulkPutTransferManager(WorkspaceClientFactory workspaceClientFactory, DigestType digestType,
        AlertService alertService, ExecutorService executor, TransfertTimeoutHelper transfertTimeoutHelper); }
BulkPutTransferManager { BulkPutResult bulkSendDataToOffers(String workspaceContainerGUID, int tenantId, DataCategory dataCategory, List<String> offerIds, Map<String, Driver> storageDrivers, Map<String, StorageOffer> storageOffers, List<String> workspaceObjectURIs, List<String> objectIds) { ResultOrError<StreamAndInfo, BulkPutResult> streamWithInfoFromWorkspace = null; ResultOrError<StreamAndInfo, BulkPutResult> prependedStreamWithInfo = null; MultiplePipedInputStream streams = null; List<Future<StorageBulkPutResult>> transferThreadFutures = null; Future<List<ObjectInfo>> digestListenerFuture = null; try { streamWithInfoFromWorkspace = retrieveBulkDataFromWorkspace(workspaceContainerGUID, workspaceObjectURIs, offerIds); if (streamWithInfoFromWorkspace.hasError()) { return streamWithInfoFromWorkspace.getError(); } prependedStreamWithInfo = prependWithObjectIdsHeaderEntry(objectIds, streamWithInfoFromWorkspace.getResult(), offerIds); if (prependedStreamWithInfo.hasError()) { return prependedStreamWithInfo.getError(); } streams = new MultiplePipedInputStream(prependedStreamWithInfo.getResult().getStream(), offerIds.size() + 1); transferThreadFutures = startTransferThreads(tenantId, dataCategory, objectIds, offerIds, storageDrivers, storageOffers, streams, prependedStreamWithInfo.getResult().getSize()); digestListenerFuture = startDigestComputeThread(offerIds, streams, objectIds); long finalTimeout = transfertTimeoutHelper.getTransferTimeout( prependedStreamWithInfo.getResult().getSize()); TimeoutStopwatch timeoutStopwatch = new TimeoutStopwatch(finalTimeout); ResultOrError<List<ObjectInfo>, BulkPutResult> objectInfos = awaitDigestListenerThread(digestListenerFuture, timeoutStopwatch, offerIds); if (objectInfos.hasError()) { return objectInfos.getError(); } HashMap<String, OfferBulkPutStatus> statusByOfferIds = new HashMap<>(); for (int rank = 0; rank < offerIds.size(); rank++) { String offerId = offerIds.get(rank); Future<StorageBulkPutResult> transferThreadFuture = transferThreadFutures.get(rank); OfferBulkPutStatus status = awaitTransferThread(timeoutStopwatch, objectInfos.getResult(), offerId, transferThreadFuture); statusByOfferIds.put(offerId, status); } return new BulkPutResult(objectInfos.getResult(), statusByOfferIds); } finally { if (transferThreadFutures != null) { for (Future<StorageBulkPutResult> transferThreadFuture : transferThreadFutures) { transferThreadFuture.cancel(true); } } if (digestListenerFuture != null) { digestListenerFuture.cancel(true); } if (streams != null) { streams.close(); } if (prependedStreamWithInfo != null && prependedStreamWithInfo.hasResult()) { prependedStreamWithInfo.getResult().close(); } if (streamWithInfoFromWorkspace != null && streamWithInfoFromWorkspace.hasResult()) { streamWithInfoFromWorkspace.getResult().close(); } } } BulkPutTransferManager(WorkspaceClientFactory workspaceClientFactory, TransfertTimeoutHelper transfertTimeoutHelper); @VisibleForTesting  BulkPutTransferManager(WorkspaceClientFactory workspaceClientFactory, DigestType digestType,
        AlertService alertService, ExecutorService executor, TransfertTimeoutHelper transfertTimeoutHelper);  }
BulkPutTransferManager { BulkPutResult bulkSendDataToOffers(String workspaceContainerGUID, int tenantId, DataCategory dataCategory, List<String> offerIds, Map<String, Driver> storageDrivers, Map<String, StorageOffer> storageOffers, List<String> workspaceObjectURIs, List<String> objectIds) { ResultOrError<StreamAndInfo, BulkPutResult> streamWithInfoFromWorkspace = null; ResultOrError<StreamAndInfo, BulkPutResult> prependedStreamWithInfo = null; MultiplePipedInputStream streams = null; List<Future<StorageBulkPutResult>> transferThreadFutures = null; Future<List<ObjectInfo>> digestListenerFuture = null; try { streamWithInfoFromWorkspace = retrieveBulkDataFromWorkspace(workspaceContainerGUID, workspaceObjectURIs, offerIds); if (streamWithInfoFromWorkspace.hasError()) { return streamWithInfoFromWorkspace.getError(); } prependedStreamWithInfo = prependWithObjectIdsHeaderEntry(objectIds, streamWithInfoFromWorkspace.getResult(), offerIds); if (prependedStreamWithInfo.hasError()) { return prependedStreamWithInfo.getError(); } streams = new MultiplePipedInputStream(prependedStreamWithInfo.getResult().getStream(), offerIds.size() + 1); transferThreadFutures = startTransferThreads(tenantId, dataCategory, objectIds, offerIds, storageDrivers, storageOffers, streams, prependedStreamWithInfo.getResult().getSize()); digestListenerFuture = startDigestComputeThread(offerIds, streams, objectIds); long finalTimeout = transfertTimeoutHelper.getTransferTimeout( prependedStreamWithInfo.getResult().getSize()); TimeoutStopwatch timeoutStopwatch = new TimeoutStopwatch(finalTimeout); ResultOrError<List<ObjectInfo>, BulkPutResult> objectInfos = awaitDigestListenerThread(digestListenerFuture, timeoutStopwatch, offerIds); if (objectInfos.hasError()) { return objectInfos.getError(); } HashMap<String, OfferBulkPutStatus> statusByOfferIds = new HashMap<>(); for (int rank = 0; rank < offerIds.size(); rank++) { String offerId = offerIds.get(rank); Future<StorageBulkPutResult> transferThreadFuture = transferThreadFutures.get(rank); OfferBulkPutStatus status = awaitTransferThread(timeoutStopwatch, objectInfos.getResult(), offerId, transferThreadFuture); statusByOfferIds.put(offerId, status); } return new BulkPutResult(objectInfos.getResult(), statusByOfferIds); } finally { if (transferThreadFutures != null) { for (Future<StorageBulkPutResult> transferThreadFuture : transferThreadFutures) { transferThreadFuture.cancel(true); } } if (digestListenerFuture != null) { digestListenerFuture.cancel(true); } if (streams != null) { streams.close(); } if (prependedStreamWithInfo != null && prependedStreamWithInfo.hasResult()) { prependedStreamWithInfo.getResult().close(); } if (streamWithInfoFromWorkspace != null && streamWithInfoFromWorkspace.hasResult()) { streamWithInfoFromWorkspace.getResult().close(); } } } BulkPutTransferManager(WorkspaceClientFactory workspaceClientFactory, TransfertTimeoutHelper transfertTimeoutHelper); @VisibleForTesting  BulkPutTransferManager(WorkspaceClientFactory workspaceClientFactory, DigestType digestType,
        AlertService alertService, ExecutorService executor, TransfertTimeoutHelper transfertTimeoutHelper);   }