@Test @RunWithCustomExecutor public void should_retry_when_transfer_storage_failed_one_times() throws Exception { VitamThreadUtils.getVitamSession().setContextId("contextTest"); VitamThreadUtils.getVitamSession().setContractId("contractTest"); VitamThreadUtils.getVitamSession().setRequestId("requestId"); VitamThreadUtils.getVitamSession().setApplicationSessionId("MyApplicationId"); String guid = "aeaaaaaaaaasqm2gaak5wak7uvv55tqaaaaq"; ByteArrayInputStream entity = new ByteArrayInputStream(new byte[] {1, 2, 3, 4}); given(storageClient.getContainerAsync(eq("other_strategy"), eq(guid), eq(DataCategory.OBJECT), eq(AccessLogUtils.getNoLogAccessLog()))) .willThrow(new StorageServerClientException("transfer failed")) .willReturn(new ServerResponse(entity, 200, new Headers<>())); DefaultWorkerParameters param = WorkerParametersFactory.newWorkerParameters(); param.setObjectName(guid); ItemStatus itemStatus = putBinaryOnWorkspace.execute(param, handlerIO); assertThat(itemStatus.getGlobalStatus()).isEqualTo(StatusCode.OK); verify(handlerIO) .transferInputStreamToWorkspace("Content/aeaaaaaaaaasqm2gaak5wak7uvv55tqaaaaq", entity, null, false); }
@Override public ItemStatus execute(WorkerParameters param, HandlerIO handler) throws ProcessingException { final ItemStatus itemStatus = new ItemStatus(PUT_BINARY_ON_WORKSPACE); Map<String, Object> guidToPath; try (InputStream inputStream = new FileInputStream((File) handler.getInput(GUID_TO_INFO_RANK))) { guidToPath = JsonHandler.getMapFromInputStream(inputStream); } catch (IOException | InvalidParseOperationException e) { itemStatus.increment(StatusCode.FATAL); return new ItemStatus(PUT_BINARY_ON_WORKSPACE).setItemsStatus(PUT_BINARY_ON_WORKSPACE, itemStatus); } for (int i = 0; i < NUMBER_OF_RETRY; i++) { try { transferFile(param, handler, guidToPath); itemStatus.increment(StatusCode.OK); return new ItemStatus(PUT_BINARY_ON_WORKSPACE).setItemsStatus(PUT_BINARY_ON_WORKSPACE, itemStatus); } catch (StorageNotFoundException | StorageServerClientException | ProcessingException e) { LOGGER.error(format("unable to transfer file from offer to workspace, retry: %d", i), e); } } itemStatus.increment(StatusCode.FATAL); return new ItemStatus(PUT_BINARY_ON_WORKSPACE).setItemsStatus(PUT_BINARY_ON_WORKSPACE, itemStatus); }
PutBinaryOnWorkspace extends ActionHandler { @Override public ItemStatus execute(WorkerParameters param, HandlerIO handler) throws ProcessingException { final ItemStatus itemStatus = new ItemStatus(PUT_BINARY_ON_WORKSPACE); Map<String, Object> guidToPath; try (InputStream inputStream = new FileInputStream((File) handler.getInput(GUID_TO_INFO_RANK))) { guidToPath = JsonHandler.getMapFromInputStream(inputStream); } catch (IOException | InvalidParseOperationException e) { itemStatus.increment(StatusCode.FATAL); return new ItemStatus(PUT_BINARY_ON_WORKSPACE).setItemsStatus(PUT_BINARY_ON_WORKSPACE, itemStatus); } for (int i = 0; i < NUMBER_OF_RETRY; i++) { try { transferFile(param, handler, guidToPath); itemStatus.increment(StatusCode.OK); return new ItemStatus(PUT_BINARY_ON_WORKSPACE).setItemsStatus(PUT_BINARY_ON_WORKSPACE, itemStatus); } catch (StorageNotFoundException | StorageServerClientException | ProcessingException e) { LOGGER.error(format("unable to transfer file from offer to workspace, retry: %d", i), e); } } itemStatus.increment(StatusCode.FATAL); return new ItemStatus(PUT_BINARY_ON_WORKSPACE).setItemsStatus(PUT_BINARY_ON_WORKSPACE, itemStatus); } }
PutBinaryOnWorkspace extends ActionHandler { @Override public ItemStatus execute(WorkerParameters param, HandlerIO handler) throws ProcessingException { final ItemStatus itemStatus = new ItemStatus(PUT_BINARY_ON_WORKSPACE); Map<String, Object> guidToPath; try (InputStream inputStream = new FileInputStream((File) handler.getInput(GUID_TO_INFO_RANK))) { guidToPath = JsonHandler.getMapFromInputStream(inputStream); } catch (IOException | InvalidParseOperationException e) { itemStatus.increment(StatusCode.FATAL); return new ItemStatus(PUT_BINARY_ON_WORKSPACE).setItemsStatus(PUT_BINARY_ON_WORKSPACE, itemStatus); } for (int i = 0; i < NUMBER_OF_RETRY; i++) { try { transferFile(param, handler, guidToPath); itemStatus.increment(StatusCode.OK); return new ItemStatus(PUT_BINARY_ON_WORKSPACE).setItemsStatus(PUT_BINARY_ON_WORKSPACE, itemStatus); } catch (StorageNotFoundException | StorageServerClientException | ProcessingException e) { LOGGER.error(format("unable to transfer file from offer to workspace, retry: %d", i), e); } } itemStatus.increment(StatusCode.FATAL); return new ItemStatus(PUT_BINARY_ON_WORKSPACE).setItemsStatus(PUT_BINARY_ON_WORKSPACE, itemStatus); } PutBinaryOnWorkspace(); @VisibleForTesting  PutBinaryOnWorkspace(StorageClientFactory storageClientFactory); }
PutBinaryOnWorkspace extends ActionHandler { @Override public ItemStatus execute(WorkerParameters param, HandlerIO handler) throws ProcessingException { final ItemStatus itemStatus = new ItemStatus(PUT_BINARY_ON_WORKSPACE); Map<String, Object> guidToPath; try (InputStream inputStream = new FileInputStream((File) handler.getInput(GUID_TO_INFO_RANK))) { guidToPath = JsonHandler.getMapFromInputStream(inputStream); } catch (IOException | InvalidParseOperationException e) { itemStatus.increment(StatusCode.FATAL); return new ItemStatus(PUT_BINARY_ON_WORKSPACE).setItemsStatus(PUT_BINARY_ON_WORKSPACE, itemStatus); } for (int i = 0; i < NUMBER_OF_RETRY; i++) { try { transferFile(param, handler, guidToPath); itemStatus.increment(StatusCode.OK); return new ItemStatus(PUT_BINARY_ON_WORKSPACE).setItemsStatus(PUT_BINARY_ON_WORKSPACE, itemStatus); } catch (StorageNotFoundException | StorageServerClientException | ProcessingException e) { LOGGER.error(format("unable to transfer file from offer to workspace, retry: %d", i), e); } } itemStatus.increment(StatusCode.FATAL); return new ItemStatus(PUT_BINARY_ON_WORKSPACE).setItemsStatus(PUT_BINARY_ON_WORKSPACE, itemStatus); } PutBinaryOnWorkspace(); @VisibleForTesting  PutBinaryOnWorkspace(StorageClientFactory storageClientFactory); @Override ItemStatus execute(WorkerParameters param, HandlerIO handler); @Override void checkMandatoryIOParameter(HandlerIO handler); }
PutBinaryOnWorkspace extends ActionHandler { @Override public ItemStatus execute(WorkerParameters param, HandlerIO handler) throws ProcessingException { final ItemStatus itemStatus = new ItemStatus(PUT_BINARY_ON_WORKSPACE); Map<String, Object> guidToPath; try (InputStream inputStream = new FileInputStream((File) handler.getInput(GUID_TO_INFO_RANK))) { guidToPath = JsonHandler.getMapFromInputStream(inputStream); } catch (IOException | InvalidParseOperationException e) { itemStatus.increment(StatusCode.FATAL); return new ItemStatus(PUT_BINARY_ON_WORKSPACE).setItemsStatus(PUT_BINARY_ON_WORKSPACE, itemStatus); } for (int i = 0; i < NUMBER_OF_RETRY; i++) { try { transferFile(param, handler, guidToPath); itemStatus.increment(StatusCode.OK); return new ItemStatus(PUT_BINARY_ON_WORKSPACE).setItemsStatus(PUT_BINARY_ON_WORKSPACE, itemStatus); } catch (StorageNotFoundException | StorageServerClientException | ProcessingException e) { LOGGER.error(format("unable to transfer file from offer to workspace, retry: %d", i), e); } } itemStatus.increment(StatusCode.FATAL); return new ItemStatus(PUT_BINARY_ON_WORKSPACE).setItemsStatus(PUT_BINARY_ON_WORKSPACE, itemStatus); } PutBinaryOnWorkspace(); @VisibleForTesting  PutBinaryOnWorkspace(StorageClientFactory storageClientFactory); @Override ItemStatus execute(WorkerParameters param, HandlerIO handler); @Override void checkMandatoryIOParameter(HandlerIO handler); static final int NUMBER_OF_RETRY; }