@Test void IngestFromBlob_IngestionReportMethodIsTable_NotEmptyIngestionStatus() throws Exception { BlobSourceInfo blobSourceInfo = new BlobSourceInfo("http: ingestionProperties.setReportMethod(IngestionProperties.IngestionReportMethod.Table); IngestionResult result = queuedIngestClient.ingestFromBlob(blobSourceInfo, ingestionProperties); assert result.getIngestionStatusesLength() != 0; }
@Override public IngestionResult ingestFromBlob(BlobSourceInfo blobSourceInfo, IngestionProperties ingestionProperties) throws IngestionClientException, IngestionServiceException { Ensure.argIsNotNull(blobSourceInfo, "blobSourceInfo"); Ensure.argIsNotNull(ingestionProperties, "ingestionProperties"); blobSourceInfo.validate(); ingestionProperties.validate(); try { ingestionProperties.setAuthorizationContextToken(resourceManager.getIdentityToken()); List<IngestionStatusInTableDescription> tableStatuses = new LinkedList<>(); IngestionBlobInfo ingestionBlobInfo = new IngestionBlobInfo(blobSourceInfo.getBlobPath(), ingestionProperties.getDatabaseName(), ingestionProperties.getTableName()); String urlWithoutSecrets = SecurityUtils.removeSecretsFromUrl(blobSourceInfo.getBlobPath()); if (blobSourceInfo.getRawSizeInBytes() > 0L) { ingestionBlobInfo.rawDataSize = blobSourceInfo.getRawSizeInBytes(); } else { log.warn("Blob '{}' was sent for ingestion without specifying its raw data size", urlWithoutSecrets); } ingestionBlobInfo.reportLevel = ingestionProperties.getReportLevel(); ingestionBlobInfo.reportMethod = ingestionProperties.getReportMethod(); ingestionBlobInfo.flushImmediately = ingestionProperties.getFlushImmediately(); ingestionBlobInfo.additionalProperties = ingestionProperties.getIngestionProperties(); if (blobSourceInfo.getSourceId() != null) { ingestionBlobInfo.id = blobSourceInfo.getSourceId(); } if (ingestionProperties.getReportMethod() != IngestionProperties.IngestionReportMethod.Queue) { String tableStatusUri = resourceManager .getIngestionResource(ResourceManager.ResourceType.INGESTIONS_STATUS_TABLE); ingestionBlobInfo.IngestionStatusInTable = new IngestionStatusInTableDescription(); ingestionBlobInfo.IngestionStatusInTable.TableConnectionString = tableStatusUri; ingestionBlobInfo.IngestionStatusInTable.RowKey = ingestionBlobInfo.id.toString(); ingestionBlobInfo.IngestionStatusInTable.PartitionKey = ingestionBlobInfo.id.toString(); IngestionStatus status = new IngestionStatus(ingestionBlobInfo.id); status.database = ingestionProperties.getDatabaseName(); status.table = ingestionProperties.getTableName(); status.status = OperationStatus.Pending; status.updatedOn = Date.from(Instant.now()); status.ingestionSourceId = ingestionBlobInfo.id; status.setIngestionSourcePath(urlWithoutSecrets); azureStorageClient.azureTableInsertEntity(tableStatusUri, status); tableStatuses.add(ingestionBlobInfo.IngestionStatusInTable); } ObjectMapper objectMapper = new ObjectMapper(); String serializedIngestionBlobInfo = objectMapper.writeValueAsString(ingestionBlobInfo); azureStorageClient.postMessageToQueue( resourceManager .getIngestionResource(ResourceManager.ResourceType.SECURED_READY_FOR_AGGREGATION_QUEUE) , serializedIngestionBlobInfo); return new TableReportIngestionResult(tableStatuses); } catch (StorageException e) { throw new IngestionServiceException("Failed to ingest from blob", e); } catch (IOException | URISyntaxException e) { throw new IngestionClientException("Failed to ingest from blob", e); } catch (IngestionServiceException e) { validateEndpointServiceType(); throw e; } }
QueuedIngestClient implements IngestClient { @Override public IngestionResult ingestFromBlob(BlobSourceInfo blobSourceInfo, IngestionProperties ingestionProperties) throws IngestionClientException, IngestionServiceException { Ensure.argIsNotNull(blobSourceInfo, "blobSourceInfo"); Ensure.argIsNotNull(ingestionProperties, "ingestionProperties"); blobSourceInfo.validate(); ingestionProperties.validate(); try { ingestionProperties.setAuthorizationContextToken(resourceManager.getIdentityToken()); List<IngestionStatusInTableDescription> tableStatuses = new LinkedList<>(); IngestionBlobInfo ingestionBlobInfo = new IngestionBlobInfo(blobSourceInfo.getBlobPath(), ingestionProperties.getDatabaseName(), ingestionProperties.getTableName()); String urlWithoutSecrets = SecurityUtils.removeSecretsFromUrl(blobSourceInfo.getBlobPath()); if (blobSourceInfo.getRawSizeInBytes() > 0L) { ingestionBlobInfo.rawDataSize = blobSourceInfo.getRawSizeInBytes(); } else { log.warn("Blob '{}' was sent for ingestion without specifying its raw data size", urlWithoutSecrets); } ingestionBlobInfo.reportLevel = ingestionProperties.getReportLevel(); ingestionBlobInfo.reportMethod = ingestionProperties.getReportMethod(); ingestionBlobInfo.flushImmediately = ingestionProperties.getFlushImmediately(); ingestionBlobInfo.additionalProperties = ingestionProperties.getIngestionProperties(); if (blobSourceInfo.getSourceId() != null) { ingestionBlobInfo.id = blobSourceInfo.getSourceId(); } if (ingestionProperties.getReportMethod() != IngestionProperties.IngestionReportMethod.Queue) { String tableStatusUri = resourceManager .getIngestionResource(ResourceManager.ResourceType.INGESTIONS_STATUS_TABLE); ingestionBlobInfo.IngestionStatusInTable = new IngestionStatusInTableDescription(); ingestionBlobInfo.IngestionStatusInTable.TableConnectionString = tableStatusUri; ingestionBlobInfo.IngestionStatusInTable.RowKey = ingestionBlobInfo.id.toString(); ingestionBlobInfo.IngestionStatusInTable.PartitionKey = ingestionBlobInfo.id.toString(); IngestionStatus status = new IngestionStatus(ingestionBlobInfo.id); status.database = ingestionProperties.getDatabaseName(); status.table = ingestionProperties.getTableName(); status.status = OperationStatus.Pending; status.updatedOn = Date.from(Instant.now()); status.ingestionSourceId = ingestionBlobInfo.id; status.setIngestionSourcePath(urlWithoutSecrets); azureStorageClient.azureTableInsertEntity(tableStatusUri, status); tableStatuses.add(ingestionBlobInfo.IngestionStatusInTable); } ObjectMapper objectMapper = new ObjectMapper(); String serializedIngestionBlobInfo = objectMapper.writeValueAsString(ingestionBlobInfo); azureStorageClient.postMessageToQueue( resourceManager .getIngestionResource(ResourceManager.ResourceType.SECURED_READY_FOR_AGGREGATION_QUEUE) , serializedIngestionBlobInfo); return new TableReportIngestionResult(tableStatuses); } catch (StorageException e) { throw new IngestionServiceException("Failed to ingest from blob", e); } catch (IOException | URISyntaxException e) { throw new IngestionClientException("Failed to ingest from blob", e); } catch (IngestionServiceException e) { validateEndpointServiceType(); throw e; } } }
QueuedIngestClient implements IngestClient { @Override public IngestionResult ingestFromBlob(BlobSourceInfo blobSourceInfo, IngestionProperties ingestionProperties) throws IngestionClientException, IngestionServiceException { Ensure.argIsNotNull(blobSourceInfo, "blobSourceInfo"); Ensure.argIsNotNull(ingestionProperties, "ingestionProperties"); blobSourceInfo.validate(); ingestionProperties.validate(); try { ingestionProperties.setAuthorizationContextToken(resourceManager.getIdentityToken()); List<IngestionStatusInTableDescription> tableStatuses = new LinkedList<>(); IngestionBlobInfo ingestionBlobInfo = new IngestionBlobInfo(blobSourceInfo.getBlobPath(), ingestionProperties.getDatabaseName(), ingestionProperties.getTableName()); String urlWithoutSecrets = SecurityUtils.removeSecretsFromUrl(blobSourceInfo.getBlobPath()); if (blobSourceInfo.getRawSizeInBytes() > 0L) { ingestionBlobInfo.rawDataSize = blobSourceInfo.getRawSizeInBytes(); } else { log.warn("Blob '{}' was sent for ingestion without specifying its raw data size", urlWithoutSecrets); } ingestionBlobInfo.reportLevel = ingestionProperties.getReportLevel(); ingestionBlobInfo.reportMethod = ingestionProperties.getReportMethod(); ingestionBlobInfo.flushImmediately = ingestionProperties.getFlushImmediately(); ingestionBlobInfo.additionalProperties = ingestionProperties.getIngestionProperties(); if (blobSourceInfo.getSourceId() != null) { ingestionBlobInfo.id = blobSourceInfo.getSourceId(); } if (ingestionProperties.getReportMethod() != IngestionProperties.IngestionReportMethod.Queue) { String tableStatusUri = resourceManager .getIngestionResource(ResourceManager.ResourceType.INGESTIONS_STATUS_TABLE); ingestionBlobInfo.IngestionStatusInTable = new IngestionStatusInTableDescription(); ingestionBlobInfo.IngestionStatusInTable.TableConnectionString = tableStatusUri; ingestionBlobInfo.IngestionStatusInTable.RowKey = ingestionBlobInfo.id.toString(); ingestionBlobInfo.IngestionStatusInTable.PartitionKey = ingestionBlobInfo.id.toString(); IngestionStatus status = new IngestionStatus(ingestionBlobInfo.id); status.database = ingestionProperties.getDatabaseName(); status.table = ingestionProperties.getTableName(); status.status = OperationStatus.Pending; status.updatedOn = Date.from(Instant.now()); status.ingestionSourceId = ingestionBlobInfo.id; status.setIngestionSourcePath(urlWithoutSecrets); azureStorageClient.azureTableInsertEntity(tableStatusUri, status); tableStatuses.add(ingestionBlobInfo.IngestionStatusInTable); } ObjectMapper objectMapper = new ObjectMapper(); String serializedIngestionBlobInfo = objectMapper.writeValueAsString(ingestionBlobInfo); azureStorageClient.postMessageToQueue( resourceManager .getIngestionResource(ResourceManager.ResourceType.SECURED_READY_FOR_AGGREGATION_QUEUE) , serializedIngestionBlobInfo); return new TableReportIngestionResult(tableStatuses); } catch (StorageException e) { throw new IngestionServiceException("Failed to ingest from blob", e); } catch (IOException | URISyntaxException e) { throw new IngestionClientException("Failed to ingest from blob", e); } catch (IngestionServiceException e) { validateEndpointServiceType(); throw e; } } QueuedIngestClient(ConnectionStringBuilder csb);   QueuedIngestClient(ResourceManager resourceManager);   QueuedIngestClient(ResourceManager resourceManager, AzureStorageClient azureStorageClient); }
QueuedIngestClient implements IngestClient { @Override public IngestionResult ingestFromBlob(BlobSourceInfo blobSourceInfo, IngestionProperties ingestionProperties) throws IngestionClientException, IngestionServiceException { Ensure.argIsNotNull(blobSourceInfo, "blobSourceInfo"); Ensure.argIsNotNull(ingestionProperties, "ingestionProperties"); blobSourceInfo.validate(); ingestionProperties.validate(); try { ingestionProperties.setAuthorizationContextToken(resourceManager.getIdentityToken()); List<IngestionStatusInTableDescription> tableStatuses = new LinkedList<>(); IngestionBlobInfo ingestionBlobInfo = new IngestionBlobInfo(blobSourceInfo.getBlobPath(), ingestionProperties.getDatabaseName(), ingestionProperties.getTableName()); String urlWithoutSecrets = SecurityUtils.removeSecretsFromUrl(blobSourceInfo.getBlobPath()); if (blobSourceInfo.getRawSizeInBytes() > 0L) { ingestionBlobInfo.rawDataSize = blobSourceInfo.getRawSizeInBytes(); } else { log.warn("Blob '{}' was sent for ingestion without specifying its raw data size", urlWithoutSecrets); } ingestionBlobInfo.reportLevel = ingestionProperties.getReportLevel(); ingestionBlobInfo.reportMethod = ingestionProperties.getReportMethod(); ingestionBlobInfo.flushImmediately = ingestionProperties.getFlushImmediately(); ingestionBlobInfo.additionalProperties = ingestionProperties.getIngestionProperties(); if (blobSourceInfo.getSourceId() != null) { ingestionBlobInfo.id = blobSourceInfo.getSourceId(); } if (ingestionProperties.getReportMethod() != IngestionProperties.IngestionReportMethod.Queue) { String tableStatusUri = resourceManager .getIngestionResource(ResourceManager.ResourceType.INGESTIONS_STATUS_TABLE); ingestionBlobInfo.IngestionStatusInTable = new IngestionStatusInTableDescription(); ingestionBlobInfo.IngestionStatusInTable.TableConnectionString = tableStatusUri; ingestionBlobInfo.IngestionStatusInTable.RowKey = ingestionBlobInfo.id.toString(); ingestionBlobInfo.IngestionStatusInTable.PartitionKey = ingestionBlobInfo.id.toString(); IngestionStatus status = new IngestionStatus(ingestionBlobInfo.id); status.database = ingestionProperties.getDatabaseName(); status.table = ingestionProperties.getTableName(); status.status = OperationStatus.Pending; status.updatedOn = Date.from(Instant.now()); status.ingestionSourceId = ingestionBlobInfo.id; status.setIngestionSourcePath(urlWithoutSecrets); azureStorageClient.azureTableInsertEntity(tableStatusUri, status); tableStatuses.add(ingestionBlobInfo.IngestionStatusInTable); } ObjectMapper objectMapper = new ObjectMapper(); String serializedIngestionBlobInfo = objectMapper.writeValueAsString(ingestionBlobInfo); azureStorageClient.postMessageToQueue( resourceManager .getIngestionResource(ResourceManager.ResourceType.SECURED_READY_FOR_AGGREGATION_QUEUE) , serializedIngestionBlobInfo); return new TableReportIngestionResult(tableStatuses); } catch (StorageException e) { throw new IngestionServiceException("Failed to ingest from blob", e); } catch (IOException | URISyntaxException e) { throw new IngestionClientException("Failed to ingest from blob", e); } catch (IngestionServiceException e) { validateEndpointServiceType(); throw e; } } QueuedIngestClient(ConnectionStringBuilder csb);   QueuedIngestClient(ResourceManager resourceManager);   QueuedIngestClient(ResourceManager resourceManager, AzureStorageClient azureStorageClient); @Override IngestionResult ingestFromBlob(BlobSourceInfo blobSourceInfo, IngestionProperties ingestionProperties); @Override IngestionResult ingestFromFile(FileSourceInfo fileSourceInfo, IngestionProperties ingestionProperties); @Override IngestionResult ingestFromStream(StreamSourceInfo streamSourceInfo, IngestionProperties ingestionProperties); @Override IngestionResult ingestFromResultSet(ResultSetSourceInfo resultSetSourceInfo, IngestionProperties ingestionProperties); @Override void close(); }
QueuedIngestClient implements IngestClient { @Override public IngestionResult ingestFromBlob(BlobSourceInfo blobSourceInfo, IngestionProperties ingestionProperties) throws IngestionClientException, IngestionServiceException { Ensure.argIsNotNull(blobSourceInfo, "blobSourceInfo"); Ensure.argIsNotNull(ingestionProperties, "ingestionProperties"); blobSourceInfo.validate(); ingestionProperties.validate(); try { ingestionProperties.setAuthorizationContextToken(resourceManager.getIdentityToken()); List<IngestionStatusInTableDescription> tableStatuses = new LinkedList<>(); IngestionBlobInfo ingestionBlobInfo = new IngestionBlobInfo(blobSourceInfo.getBlobPath(), ingestionProperties.getDatabaseName(), ingestionProperties.getTableName()); String urlWithoutSecrets = SecurityUtils.removeSecretsFromUrl(blobSourceInfo.getBlobPath()); if (blobSourceInfo.getRawSizeInBytes() > 0L) { ingestionBlobInfo.rawDataSize = blobSourceInfo.getRawSizeInBytes(); } else { log.warn("Blob '{}' was sent for ingestion without specifying its raw data size", urlWithoutSecrets); } ingestionBlobInfo.reportLevel = ingestionProperties.getReportLevel(); ingestionBlobInfo.reportMethod = ingestionProperties.getReportMethod(); ingestionBlobInfo.flushImmediately = ingestionProperties.getFlushImmediately(); ingestionBlobInfo.additionalProperties = ingestionProperties.getIngestionProperties(); if (blobSourceInfo.getSourceId() != null) { ingestionBlobInfo.id = blobSourceInfo.getSourceId(); } if (ingestionProperties.getReportMethod() != IngestionProperties.IngestionReportMethod.Queue) { String tableStatusUri = resourceManager .getIngestionResource(ResourceManager.ResourceType.INGESTIONS_STATUS_TABLE); ingestionBlobInfo.IngestionStatusInTable = new IngestionStatusInTableDescription(); ingestionBlobInfo.IngestionStatusInTable.TableConnectionString = tableStatusUri; ingestionBlobInfo.IngestionStatusInTable.RowKey = ingestionBlobInfo.id.toString(); ingestionBlobInfo.IngestionStatusInTable.PartitionKey = ingestionBlobInfo.id.toString(); IngestionStatus status = new IngestionStatus(ingestionBlobInfo.id); status.database = ingestionProperties.getDatabaseName(); status.table = ingestionProperties.getTableName(); status.status = OperationStatus.Pending; status.updatedOn = Date.from(Instant.now()); status.ingestionSourceId = ingestionBlobInfo.id; status.setIngestionSourcePath(urlWithoutSecrets); azureStorageClient.azureTableInsertEntity(tableStatusUri, status); tableStatuses.add(ingestionBlobInfo.IngestionStatusInTable); } ObjectMapper objectMapper = new ObjectMapper(); String serializedIngestionBlobInfo = objectMapper.writeValueAsString(ingestionBlobInfo); azureStorageClient.postMessageToQueue( resourceManager .getIngestionResource(ResourceManager.ResourceType.SECURED_READY_FOR_AGGREGATION_QUEUE) , serializedIngestionBlobInfo); return new TableReportIngestionResult(tableStatuses); } catch (StorageException e) { throw new IngestionServiceException("Failed to ingest from blob", e); } catch (IOException | URISyntaxException e) { throw new IngestionClientException("Failed to ingest from blob", e); } catch (IngestionServiceException e) { validateEndpointServiceType(); throw e; } } QueuedIngestClient(ConnectionStringBuilder csb);   QueuedIngestClient(ResourceManager resourceManager);   QueuedIngestClient(ResourceManager resourceManager, AzureStorageClient azureStorageClient); @Override IngestionResult ingestFromBlob(BlobSourceInfo blobSourceInfo, IngestionProperties ingestionProperties); @Override IngestionResult ingestFromFile(FileSourceInfo fileSourceInfo, IngestionProperties ingestionProperties); @Override IngestionResult ingestFromStream(StreamSourceInfo streamSourceInfo, IngestionProperties ingestionProperties); @Override IngestionResult ingestFromResultSet(ResultSetSourceInfo resultSetSourceInfo, IngestionProperties ingestionProperties); @Override void close(); static final String INGEST_PREFIX; static final String EXPECTED_SERVICE_TYPE; }