@Test @RunWithCustomExecutor public void whenDistributeKindLargeFileOK() throws Exception { File file = PropertiesUtils.getResourceFile(FILE_WITH_GUIDS); givenWorkspaceClientReturnsFileContent(file, operationId, FILE_WITH_GUIDS); ItemStatus itemStatus = processDistributor .distribute(workerParameters, getStep(DistributionKind.LIST_IN_JSONL_FILE, FILE_WITH_GUIDS), operationId, PauseRecover.NO_RECOVER); assertNotNull(itemStatus); assertTrue(StatusCode.OK.equals(itemStatus.getGlobalStatus())); Map<String, ItemStatus> imap = itemStatus.getItemsStatus(); assertNotNull(imap); assertFalse(imap.isEmpty()); }
@Override public ItemStatus distribute(WorkerParameters workParams, Step step, String operationId, PauseRecover pauseRecover) { ParametersChecker.checkParameter("WorkParams is a mandatory parameter", workParams); ParametersChecker.checkParameter("Step is a mandatory parameter", step); ParametersChecker.checkParameter("workflowId is a mandatory parameter", operationId); ParametersChecker.checkParameter("pauseRecover is a mandatory parameter", pauseRecover); boolean useDistributorIndex = PauseOrCancelAction.ACTION_RECOVER.equals(step.getPauseOrCancelAction()); final int tenantId = VitamThreadUtils.getVitamSession().getTenantId(); step.setStepResponses(new ItemStatus(step.getStepName())); try (MetaDataClient metadataClient = metaDataClientFactory.getClient()) { try { metadataClient.refreshUnits(); metadataClient.refreshObjectGroups(); } catch (MetaDataClientServerException e) { LOGGER.error("Error while refresh metadata indexes", e); return step.getStepResponses().increment(StatusCode.FATAL); } } try { List<String> objectsList = new ArrayList<>(); if (step.getDistribution().getKind().equals(DistributionKind.LIST_ORDERING_IN_FILE)) { try (final WorkspaceClient workspaceClient = workspaceClientFactory.getClient()) { if (DistributionType.Units == step.getDistribution().getType()) { Response response = null; InputStream levelFile = null; final JsonNode levelFileJson; try { response = workspaceClient.getObject(workParams.getContainerName(), step.getDistribution().getElement()); levelFile = (InputStream) response.getEntity(); levelFileJson = JsonHandler.getFromInputStream(levelFile); } finally { StreamUtils.closeSilently(levelFile); workspaceClient.consumeAnyEntityAndClose(response); } final Iterator<Entry<String, JsonNode>> iteratorLevelFile = levelFileJson.fields(); while (iteratorLevelFile.hasNext()) { final Entry<String, JsonNode> guidFieldList = iteratorLevelFile.next(); final String level = guidFieldList.getKey(); final JsonNode guid = guidFieldList.getValue(); if (guid != null && guid.size() > 0) { for (final JsonNode _idGuid : guid) { objectsList.add(_idGuid.asText() + JSON_EXTENSION); } boolean distributorIndexUsed = distributeOnList(workParams, step, level, objectsList, useDistributorIndex, tenantId); if (useDistributorIndex && distributorIndexUsed) { useDistributorIndex = false; } objectsList.clear(); if (step.getStepResponses().getGlobalStatus().isGreaterOrEqualToFatal()) { break; } } } } } } else if (step.getDistribution().getKind().equals(DistributionKind.LIST_IN_DIRECTORY)) { try (final WorkspaceClient workspaceClient = workspaceClientFactory.getClient()) { final List<URI> objectsListUri = JsonHandler.getFromStringAsTypeReference( workspaceClient.getListUriDigitalObjectFromFolder(workParams.getContainerName(), step.getDistribution().getElement()) .toJsonNode().get("$results").get(0).toString(), LIST_URI_TYPE_REFERENCE); for (URI uri : objectsListUri) { objectsList.add(uri.getPath()); } workParams.setObjectMetadataList(Collections.emptyList()); distributeOnList(workParams, step, NOLEVEL, objectsList, useDistributorIndex, tenantId); } } else if (step.getDistribution().getKind().equals(DistributionKind.LIST_IN_FILE)) { try (final WorkspaceClient workspaceClient = workspaceClientFactory.getClient()) { Response response = null; final JsonNode ogIdList; try { response = workspaceClient .getObject(workParams.getContainerName(), step.getDistribution().getElement()); ogIdList = JsonHandler.getFromInputStream((InputStream) response.getEntity()); } finally { workspaceClient.consumeAnyEntityAndClose(response); } if (ogIdList.isArray()) { for (JsonNode node : ogIdList) { objectsList.add(node.textValue()); } } distributeOnList(workParams, step, NOLEVEL, objectsList, useDistributorIndex, tenantId); } } else if (step.getDistribution().getKind().equals(DistributionKind.LIST_IN_JSONL_FILE)) { File tmpDirectory = new File(VitamConfiguration.getVitamTmpFolder()); try (InputStream inputStream = new WorkspaceBufferingInputStream(workspaceClientFactory, workParams.getContainerName(), step.getDistribution().getElement(), serverConfiguration.getMaxDistributionOnDiskBufferSize(), serverConfiguration.getMaxDistributionInMemoryBufferSize(), tmpDirectory); BufferedReader br = new BufferedReader( new InputStreamReader(inputStream, StandardCharsets.UTF_8))) { distributeOnStream(workParams, step, br, useDistributorIndex, tenantId); } } else { if (step.getDistribution().getElement() == null || step.getDistribution().getElement().trim().isEmpty()) { objectsList.add(workParams.getContainerName()); } else { objectsList.add(step.getDistribution().getElement()); } distributeOnList(workParams, step, NOLEVEL, objectsList, useDistributorIndex, tenantId); } } catch (final PauseCancelException e) { LOGGER.debug("Operation Cancel", e); return step.getStepResponses(); } catch (final Exception e) { step.getStepResponses().increment(StatusCode.FATAL); LOGGER.error(EXCEPTION_MESSAGE, e); } if (step.getStepResponses().getGlobalStatus().isGreaterOrEqualToFatal()) { return step.getStepResponses(); } else { return step.setPauseOrCancelAction(PauseOrCancelAction.ACTION_COMPLETE).getStepResponses(); } }
ProcessDistributorImpl implements ProcessDistributor { @Override public ItemStatus distribute(WorkerParameters workParams, Step step, String operationId, PauseRecover pauseRecover) { ParametersChecker.checkParameter("WorkParams is a mandatory parameter", workParams); ParametersChecker.checkParameter("Step is a mandatory parameter", step); ParametersChecker.checkParameter("workflowId is a mandatory parameter", operationId); ParametersChecker.checkParameter("pauseRecover is a mandatory parameter", pauseRecover); boolean useDistributorIndex = PauseOrCancelAction.ACTION_RECOVER.equals(step.getPauseOrCancelAction()); final int tenantId = VitamThreadUtils.getVitamSession().getTenantId(); step.setStepResponses(new ItemStatus(step.getStepName())); try (MetaDataClient metadataClient = metaDataClientFactory.getClient()) { try { metadataClient.refreshUnits(); metadataClient.refreshObjectGroups(); } catch (MetaDataClientServerException e) { LOGGER.error("Error while refresh metadata indexes", e); return step.getStepResponses().increment(StatusCode.FATAL); } } try { List<String> objectsList = new ArrayList<>(); if (step.getDistribution().getKind().equals(DistributionKind.LIST_ORDERING_IN_FILE)) { try (final WorkspaceClient workspaceClient = workspaceClientFactory.getClient()) { if (DistributionType.Units == step.getDistribution().getType()) { Response response = null; InputStream levelFile = null; final JsonNode levelFileJson; try { response = workspaceClient.getObject(workParams.getContainerName(), step.getDistribution().getElement()); levelFile = (InputStream) response.getEntity(); levelFileJson = JsonHandler.getFromInputStream(levelFile); } finally { StreamUtils.closeSilently(levelFile); workspaceClient.consumeAnyEntityAndClose(response); } final Iterator<Entry<String, JsonNode>> iteratorLevelFile = levelFileJson.fields(); while (iteratorLevelFile.hasNext()) { final Entry<String, JsonNode> guidFieldList = iteratorLevelFile.next(); final String level = guidFieldList.getKey(); final JsonNode guid = guidFieldList.getValue(); if (guid != null && guid.size() > 0) { for (final JsonNode _idGuid : guid) { objectsList.add(_idGuid.asText() + JSON_EXTENSION); } boolean distributorIndexUsed = distributeOnList(workParams, step, level, objectsList, useDistributorIndex, tenantId); if (useDistributorIndex && distributorIndexUsed) { useDistributorIndex = false; } objectsList.clear(); if (step.getStepResponses().getGlobalStatus().isGreaterOrEqualToFatal()) { break; } } } } } } else if (step.getDistribution().getKind().equals(DistributionKind.LIST_IN_DIRECTORY)) { try (final WorkspaceClient workspaceClient = workspaceClientFactory.getClient()) { final List<URI> objectsListUri = JsonHandler.getFromStringAsTypeReference( workspaceClient.getListUriDigitalObjectFromFolder(workParams.getContainerName(), step.getDistribution().getElement()) .toJsonNode().get("$results").get(0).toString(), LIST_URI_TYPE_REFERENCE); for (URI uri : objectsListUri) { objectsList.add(uri.getPath()); } workParams.setObjectMetadataList(Collections.emptyList()); distributeOnList(workParams, step, NOLEVEL, objectsList, useDistributorIndex, tenantId); } } else if (step.getDistribution().getKind().equals(DistributionKind.LIST_IN_FILE)) { try (final WorkspaceClient workspaceClient = workspaceClientFactory.getClient()) { Response response = null; final JsonNode ogIdList; try { response = workspaceClient .getObject(workParams.getContainerName(), step.getDistribution().getElement()); ogIdList = JsonHandler.getFromInputStream((InputStream) response.getEntity()); } finally { workspaceClient.consumeAnyEntityAndClose(response); } if (ogIdList.isArray()) { for (JsonNode node : ogIdList) { objectsList.add(node.textValue()); } } distributeOnList(workParams, step, NOLEVEL, objectsList, useDistributorIndex, tenantId); } } else if (step.getDistribution().getKind().equals(DistributionKind.LIST_IN_JSONL_FILE)) { File tmpDirectory = new File(VitamConfiguration.getVitamTmpFolder()); try (InputStream inputStream = new WorkspaceBufferingInputStream(workspaceClientFactory, workParams.getContainerName(), step.getDistribution().getElement(), serverConfiguration.getMaxDistributionOnDiskBufferSize(), serverConfiguration.getMaxDistributionInMemoryBufferSize(), tmpDirectory); BufferedReader br = new BufferedReader( new InputStreamReader(inputStream, StandardCharsets.UTF_8))) { distributeOnStream(workParams, step, br, useDistributorIndex, tenantId); } } else { if (step.getDistribution().getElement() == null || step.getDistribution().getElement().trim().isEmpty()) { objectsList.add(workParams.getContainerName()); } else { objectsList.add(step.getDistribution().getElement()); } distributeOnList(workParams, step, NOLEVEL, objectsList, useDistributorIndex, tenantId); } } catch (final PauseCancelException e) { LOGGER.debug("Operation Cancel", e); return step.getStepResponses(); } catch (final Exception e) { step.getStepResponses().increment(StatusCode.FATAL); LOGGER.error(EXCEPTION_MESSAGE, e); } if (step.getStepResponses().getGlobalStatus().isGreaterOrEqualToFatal()) { return step.getStepResponses(); } else { return step.setPauseOrCancelAction(PauseOrCancelAction.ACTION_COMPLETE).getStepResponses(); } } }
ProcessDistributorImpl implements ProcessDistributor { @Override public ItemStatus distribute(WorkerParameters workParams, Step step, String operationId, PauseRecover pauseRecover) { ParametersChecker.checkParameter("WorkParams is a mandatory parameter", workParams); ParametersChecker.checkParameter("Step is a mandatory parameter", step); ParametersChecker.checkParameter("workflowId is a mandatory parameter", operationId); ParametersChecker.checkParameter("pauseRecover is a mandatory parameter", pauseRecover); boolean useDistributorIndex = PauseOrCancelAction.ACTION_RECOVER.equals(step.getPauseOrCancelAction()); final int tenantId = VitamThreadUtils.getVitamSession().getTenantId(); step.setStepResponses(new ItemStatus(step.getStepName())); try (MetaDataClient metadataClient = metaDataClientFactory.getClient()) { try { metadataClient.refreshUnits(); metadataClient.refreshObjectGroups(); } catch (MetaDataClientServerException e) { LOGGER.error("Error while refresh metadata indexes", e); return step.getStepResponses().increment(StatusCode.FATAL); } } try { List<String> objectsList = new ArrayList<>(); if (step.getDistribution().getKind().equals(DistributionKind.LIST_ORDERING_IN_FILE)) { try (final WorkspaceClient workspaceClient = workspaceClientFactory.getClient()) { if (DistributionType.Units == step.getDistribution().getType()) { Response response = null; InputStream levelFile = null; final JsonNode levelFileJson; try { response = workspaceClient.getObject(workParams.getContainerName(), step.getDistribution().getElement()); levelFile = (InputStream) response.getEntity(); levelFileJson = JsonHandler.getFromInputStream(levelFile); } finally { StreamUtils.closeSilently(levelFile); workspaceClient.consumeAnyEntityAndClose(response); } final Iterator<Entry<String, JsonNode>> iteratorLevelFile = levelFileJson.fields(); while (iteratorLevelFile.hasNext()) { final Entry<String, JsonNode> guidFieldList = iteratorLevelFile.next(); final String level = guidFieldList.getKey(); final JsonNode guid = guidFieldList.getValue(); if (guid != null && guid.size() > 0) { for (final JsonNode _idGuid : guid) { objectsList.add(_idGuid.asText() + JSON_EXTENSION); } boolean distributorIndexUsed = distributeOnList(workParams, step, level, objectsList, useDistributorIndex, tenantId); if (useDistributorIndex && distributorIndexUsed) { useDistributorIndex = false; } objectsList.clear(); if (step.getStepResponses().getGlobalStatus().isGreaterOrEqualToFatal()) { break; } } } } } } else if (step.getDistribution().getKind().equals(DistributionKind.LIST_IN_DIRECTORY)) { try (final WorkspaceClient workspaceClient = workspaceClientFactory.getClient()) { final List<URI> objectsListUri = JsonHandler.getFromStringAsTypeReference( workspaceClient.getListUriDigitalObjectFromFolder(workParams.getContainerName(), step.getDistribution().getElement()) .toJsonNode().get("$results").get(0).toString(), LIST_URI_TYPE_REFERENCE); for (URI uri : objectsListUri) { objectsList.add(uri.getPath()); } workParams.setObjectMetadataList(Collections.emptyList()); distributeOnList(workParams, step, NOLEVEL, objectsList, useDistributorIndex, tenantId); } } else if (step.getDistribution().getKind().equals(DistributionKind.LIST_IN_FILE)) { try (final WorkspaceClient workspaceClient = workspaceClientFactory.getClient()) { Response response = null; final JsonNode ogIdList; try { response = workspaceClient .getObject(workParams.getContainerName(), step.getDistribution().getElement()); ogIdList = JsonHandler.getFromInputStream((InputStream) response.getEntity()); } finally { workspaceClient.consumeAnyEntityAndClose(response); } if (ogIdList.isArray()) { for (JsonNode node : ogIdList) { objectsList.add(node.textValue()); } } distributeOnList(workParams, step, NOLEVEL, objectsList, useDistributorIndex, tenantId); } } else if (step.getDistribution().getKind().equals(DistributionKind.LIST_IN_JSONL_FILE)) { File tmpDirectory = new File(VitamConfiguration.getVitamTmpFolder()); try (InputStream inputStream = new WorkspaceBufferingInputStream(workspaceClientFactory, workParams.getContainerName(), step.getDistribution().getElement(), serverConfiguration.getMaxDistributionOnDiskBufferSize(), serverConfiguration.getMaxDistributionInMemoryBufferSize(), tmpDirectory); BufferedReader br = new BufferedReader( new InputStreamReader(inputStream, StandardCharsets.UTF_8))) { distributeOnStream(workParams, step, br, useDistributorIndex, tenantId); } } else { if (step.getDistribution().getElement() == null || step.getDistribution().getElement().trim().isEmpty()) { objectsList.add(workParams.getContainerName()); } else { objectsList.add(step.getDistribution().getElement()); } distributeOnList(workParams, step, NOLEVEL, objectsList, useDistributorIndex, tenantId); } } catch (final PauseCancelException e) { LOGGER.debug("Operation Cancel", e); return step.getStepResponses(); } catch (final Exception e) { step.getStepResponses().increment(StatusCode.FATAL); LOGGER.error(EXCEPTION_MESSAGE, e); } if (step.getStepResponses().getGlobalStatus().isGreaterOrEqualToFatal()) { return step.getStepResponses(); } else { return step.setPauseOrCancelAction(PauseOrCancelAction.ACTION_COMPLETE).getStepResponses(); } } ProcessDistributorImpl(IWorkerManager workerManager, ServerConfiguration serverConfiguration); @VisibleForTesting  ProcessDistributorImpl(IWorkerManager workerManager, ServerConfiguration serverConfiguration,
        ProcessDataManagement processDataManagement, WorkspaceClientFactory workspaceClientFactory,
        MetaDataClientFactory metaDataClientFactory, WorkerClientFactory workerClientFactory); }
ProcessDistributorImpl implements ProcessDistributor { @Override public ItemStatus distribute(WorkerParameters workParams, Step step, String operationId, PauseRecover pauseRecover) { ParametersChecker.checkParameter("WorkParams is a mandatory parameter", workParams); ParametersChecker.checkParameter("Step is a mandatory parameter", step); ParametersChecker.checkParameter("workflowId is a mandatory parameter", operationId); ParametersChecker.checkParameter("pauseRecover is a mandatory parameter", pauseRecover); boolean useDistributorIndex = PauseOrCancelAction.ACTION_RECOVER.equals(step.getPauseOrCancelAction()); final int tenantId = VitamThreadUtils.getVitamSession().getTenantId(); step.setStepResponses(new ItemStatus(step.getStepName())); try (MetaDataClient metadataClient = metaDataClientFactory.getClient()) { try { metadataClient.refreshUnits(); metadataClient.refreshObjectGroups(); } catch (MetaDataClientServerException e) { LOGGER.error("Error while refresh metadata indexes", e); return step.getStepResponses().increment(StatusCode.FATAL); } } try { List<String> objectsList = new ArrayList<>(); if (step.getDistribution().getKind().equals(DistributionKind.LIST_ORDERING_IN_FILE)) { try (final WorkspaceClient workspaceClient = workspaceClientFactory.getClient()) { if (DistributionType.Units == step.getDistribution().getType()) { Response response = null; InputStream levelFile = null; final JsonNode levelFileJson; try { response = workspaceClient.getObject(workParams.getContainerName(), step.getDistribution().getElement()); levelFile = (InputStream) response.getEntity(); levelFileJson = JsonHandler.getFromInputStream(levelFile); } finally { StreamUtils.closeSilently(levelFile); workspaceClient.consumeAnyEntityAndClose(response); } final Iterator<Entry<String, JsonNode>> iteratorLevelFile = levelFileJson.fields(); while (iteratorLevelFile.hasNext()) { final Entry<String, JsonNode> guidFieldList = iteratorLevelFile.next(); final String level = guidFieldList.getKey(); final JsonNode guid = guidFieldList.getValue(); if (guid != null && guid.size() > 0) { for (final JsonNode _idGuid : guid) { objectsList.add(_idGuid.asText() + JSON_EXTENSION); } boolean distributorIndexUsed = distributeOnList(workParams, step, level, objectsList, useDistributorIndex, tenantId); if (useDistributorIndex && distributorIndexUsed) { useDistributorIndex = false; } objectsList.clear(); if (step.getStepResponses().getGlobalStatus().isGreaterOrEqualToFatal()) { break; } } } } } } else if (step.getDistribution().getKind().equals(DistributionKind.LIST_IN_DIRECTORY)) { try (final WorkspaceClient workspaceClient = workspaceClientFactory.getClient()) { final List<URI> objectsListUri = JsonHandler.getFromStringAsTypeReference( workspaceClient.getListUriDigitalObjectFromFolder(workParams.getContainerName(), step.getDistribution().getElement()) .toJsonNode().get("$results").get(0).toString(), LIST_URI_TYPE_REFERENCE); for (URI uri : objectsListUri) { objectsList.add(uri.getPath()); } workParams.setObjectMetadataList(Collections.emptyList()); distributeOnList(workParams, step, NOLEVEL, objectsList, useDistributorIndex, tenantId); } } else if (step.getDistribution().getKind().equals(DistributionKind.LIST_IN_FILE)) { try (final WorkspaceClient workspaceClient = workspaceClientFactory.getClient()) { Response response = null; final JsonNode ogIdList; try { response = workspaceClient .getObject(workParams.getContainerName(), step.getDistribution().getElement()); ogIdList = JsonHandler.getFromInputStream((InputStream) response.getEntity()); } finally { workspaceClient.consumeAnyEntityAndClose(response); } if (ogIdList.isArray()) { for (JsonNode node : ogIdList) { objectsList.add(node.textValue()); } } distributeOnList(workParams, step, NOLEVEL, objectsList, useDistributorIndex, tenantId); } } else if (step.getDistribution().getKind().equals(DistributionKind.LIST_IN_JSONL_FILE)) { File tmpDirectory = new File(VitamConfiguration.getVitamTmpFolder()); try (InputStream inputStream = new WorkspaceBufferingInputStream(workspaceClientFactory, workParams.getContainerName(), step.getDistribution().getElement(), serverConfiguration.getMaxDistributionOnDiskBufferSize(), serverConfiguration.getMaxDistributionInMemoryBufferSize(), tmpDirectory); BufferedReader br = new BufferedReader( new InputStreamReader(inputStream, StandardCharsets.UTF_8))) { distributeOnStream(workParams, step, br, useDistributorIndex, tenantId); } } else { if (step.getDistribution().getElement() == null || step.getDistribution().getElement().trim().isEmpty()) { objectsList.add(workParams.getContainerName()); } else { objectsList.add(step.getDistribution().getElement()); } distributeOnList(workParams, step, NOLEVEL, objectsList, useDistributorIndex, tenantId); } } catch (final PauseCancelException e) { LOGGER.debug("Operation Cancel", e); return step.getStepResponses(); } catch (final Exception e) { step.getStepResponses().increment(StatusCode.FATAL); LOGGER.error(EXCEPTION_MESSAGE, e); } if (step.getStepResponses().getGlobalStatus().isGreaterOrEqualToFatal()) { return step.getStepResponses(); } else { return step.setPauseOrCancelAction(PauseOrCancelAction.ACTION_COMPLETE).getStepResponses(); } } ProcessDistributorImpl(IWorkerManager workerManager, ServerConfiguration serverConfiguration); @VisibleForTesting  ProcessDistributorImpl(IWorkerManager workerManager, ServerConfiguration serverConfiguration,
        ProcessDataManagement processDataManagement, WorkspaceClientFactory workspaceClientFactory,
        MetaDataClientFactory metaDataClientFactory, WorkerClientFactory workerClientFactory); @Override ItemStatus distribute(WorkerParameters workParams, Step step, String operationId,
        PauseRecover pauseRecover); @Override ProcessDataManagement getProcessDataManagement(); }
ProcessDistributorImpl implements ProcessDistributor { @Override public ItemStatus distribute(WorkerParameters workParams, Step step, String operationId, PauseRecover pauseRecover) { ParametersChecker.checkParameter("WorkParams is a mandatory parameter", workParams); ParametersChecker.checkParameter("Step is a mandatory parameter", step); ParametersChecker.checkParameter("workflowId is a mandatory parameter", operationId); ParametersChecker.checkParameter("pauseRecover is a mandatory parameter", pauseRecover); boolean useDistributorIndex = PauseOrCancelAction.ACTION_RECOVER.equals(step.getPauseOrCancelAction()); final int tenantId = VitamThreadUtils.getVitamSession().getTenantId(); step.setStepResponses(new ItemStatus(step.getStepName())); try (MetaDataClient metadataClient = metaDataClientFactory.getClient()) { try { metadataClient.refreshUnits(); metadataClient.refreshObjectGroups(); } catch (MetaDataClientServerException e) { LOGGER.error("Error while refresh metadata indexes", e); return step.getStepResponses().increment(StatusCode.FATAL); } } try { List<String> objectsList = new ArrayList<>(); if (step.getDistribution().getKind().equals(DistributionKind.LIST_ORDERING_IN_FILE)) { try (final WorkspaceClient workspaceClient = workspaceClientFactory.getClient()) { if (DistributionType.Units == step.getDistribution().getType()) { Response response = null; InputStream levelFile = null; final JsonNode levelFileJson; try { response = workspaceClient.getObject(workParams.getContainerName(), step.getDistribution().getElement()); levelFile = (InputStream) response.getEntity(); levelFileJson = JsonHandler.getFromInputStream(levelFile); } finally { StreamUtils.closeSilently(levelFile); workspaceClient.consumeAnyEntityAndClose(response); } final Iterator<Entry<String, JsonNode>> iteratorLevelFile = levelFileJson.fields(); while (iteratorLevelFile.hasNext()) { final Entry<String, JsonNode> guidFieldList = iteratorLevelFile.next(); final String level = guidFieldList.getKey(); final JsonNode guid = guidFieldList.getValue(); if (guid != null && guid.size() > 0) { for (final JsonNode _idGuid : guid) { objectsList.add(_idGuid.asText() + JSON_EXTENSION); } boolean distributorIndexUsed = distributeOnList(workParams, step, level, objectsList, useDistributorIndex, tenantId); if (useDistributorIndex && distributorIndexUsed) { useDistributorIndex = false; } objectsList.clear(); if (step.getStepResponses().getGlobalStatus().isGreaterOrEqualToFatal()) { break; } } } } } } else if (step.getDistribution().getKind().equals(DistributionKind.LIST_IN_DIRECTORY)) { try (final WorkspaceClient workspaceClient = workspaceClientFactory.getClient()) { final List<URI> objectsListUri = JsonHandler.getFromStringAsTypeReference( workspaceClient.getListUriDigitalObjectFromFolder(workParams.getContainerName(), step.getDistribution().getElement()) .toJsonNode().get("$results").get(0).toString(), LIST_URI_TYPE_REFERENCE); for (URI uri : objectsListUri) { objectsList.add(uri.getPath()); } workParams.setObjectMetadataList(Collections.emptyList()); distributeOnList(workParams, step, NOLEVEL, objectsList, useDistributorIndex, tenantId); } } else if (step.getDistribution().getKind().equals(DistributionKind.LIST_IN_FILE)) { try (final WorkspaceClient workspaceClient = workspaceClientFactory.getClient()) { Response response = null; final JsonNode ogIdList; try { response = workspaceClient .getObject(workParams.getContainerName(), step.getDistribution().getElement()); ogIdList = JsonHandler.getFromInputStream((InputStream) response.getEntity()); } finally { workspaceClient.consumeAnyEntityAndClose(response); } if (ogIdList.isArray()) { for (JsonNode node : ogIdList) { objectsList.add(node.textValue()); } } distributeOnList(workParams, step, NOLEVEL, objectsList, useDistributorIndex, tenantId); } } else if (step.getDistribution().getKind().equals(DistributionKind.LIST_IN_JSONL_FILE)) { File tmpDirectory = new File(VitamConfiguration.getVitamTmpFolder()); try (InputStream inputStream = new WorkspaceBufferingInputStream(workspaceClientFactory, workParams.getContainerName(), step.getDistribution().getElement(), serverConfiguration.getMaxDistributionOnDiskBufferSize(), serverConfiguration.getMaxDistributionInMemoryBufferSize(), tmpDirectory); BufferedReader br = new BufferedReader( new InputStreamReader(inputStream, StandardCharsets.UTF_8))) { distributeOnStream(workParams, step, br, useDistributorIndex, tenantId); } } else { if (step.getDistribution().getElement() == null || step.getDistribution().getElement().trim().isEmpty()) { objectsList.add(workParams.getContainerName()); } else { objectsList.add(step.getDistribution().getElement()); } distributeOnList(workParams, step, NOLEVEL, objectsList, useDistributorIndex, tenantId); } } catch (final PauseCancelException e) { LOGGER.debug("Operation Cancel", e); return step.getStepResponses(); } catch (final Exception e) { step.getStepResponses().increment(StatusCode.FATAL); LOGGER.error(EXCEPTION_MESSAGE, e); } if (step.getStepResponses().getGlobalStatus().isGreaterOrEqualToFatal()) { return step.getStepResponses(); } else { return step.setPauseOrCancelAction(PauseOrCancelAction.ACTION_COMPLETE).getStepResponses(); } } ProcessDistributorImpl(IWorkerManager workerManager, ServerConfiguration serverConfiguration); @VisibleForTesting  ProcessDistributorImpl(IWorkerManager workerManager, ServerConfiguration serverConfiguration,
        ProcessDataManagement processDataManagement, WorkspaceClientFactory workspaceClientFactory,
        MetaDataClientFactory metaDataClientFactory, WorkerClientFactory workerClientFactory); @Override ItemStatus distribute(WorkerParameters workParams, Step step, String operationId,
        PauseRecover pauseRecover); @Override ProcessDataManagement getProcessDataManagement();  }