@Test void IngestFromFile_NullFileSourceInfo_IllegalArgumentException() { assertThrows( IllegalArgumentException.class, () -> queuedIngestClient.ingestFromFile(null, ingestionProperties)); }
@Override public IngestionResult ingestFromFile(FileSourceInfo fileSourceInfo, IngestionProperties ingestionProperties) throws IngestionClientException, IngestionServiceException { Ensure.argIsNotNull(fileSourceInfo, "fileSourceInfo"); Ensure.argIsNotNull(ingestionProperties, "ingestionProperties"); fileSourceInfo.validate(); ingestionProperties.validate(); try { String filePath = fileSourceInfo.getFilePath(); Ensure.fileExists(filePath); CompressionType sourceCompressionType = AzureStorageClient.getCompression(filePath); boolean shouldCompress = AzureStorageClient.shouldCompress(sourceCompressionType, ingestionProperties.getDataFormat()); File file = new File(filePath); String blobName = genBlobName( file.getName(), ingestionProperties.getDatabaseName(), ingestionProperties.getTableName(), ingestionProperties.getDataFormat(), shouldCompress ? CompressionType.gz : sourceCompressionType); CloudBlockBlob blob = azureStorageClient.uploadLocalFileToBlob(fileSourceInfo.getFilePath(), blobName, resourceManager.getIngestionResource(ResourceManager.ResourceType.TEMP_STORAGE), shouldCompress); String blobPath = azureStorageClient.getBlobPathWithSas(blob); long rawDataSize = fileSourceInfo.getRawSizeInBytes() > 0L ? fileSourceInfo.getRawSizeInBytes() : estimateFileRawSize(filePath, IngestionProperties.DATA_FORMAT.valueOf(ingestionProperties.getDataFormat())); BlobSourceInfo blobSourceInfo = new BlobSourceInfo(blobPath, rawDataSize, fileSourceInfo.getSourceId()); return ingestFromBlob(blobSourceInfo, ingestionProperties); } catch (StorageException e) { throw new IngestionServiceException("Failed to ingest from file", e); } catch (IOException | URISyntaxException e) { throw new IngestionClientException("Failed to ingest from file", e); } catch (IngestionServiceException e) { validateEndpointServiceType(); throw e; } }
QueuedIngestClient implements IngestClient { @Override public IngestionResult ingestFromFile(FileSourceInfo fileSourceInfo, IngestionProperties ingestionProperties) throws IngestionClientException, IngestionServiceException { Ensure.argIsNotNull(fileSourceInfo, "fileSourceInfo"); Ensure.argIsNotNull(ingestionProperties, "ingestionProperties"); fileSourceInfo.validate(); ingestionProperties.validate(); try { String filePath = fileSourceInfo.getFilePath(); Ensure.fileExists(filePath); CompressionType sourceCompressionType = AzureStorageClient.getCompression(filePath); boolean shouldCompress = AzureStorageClient.shouldCompress(sourceCompressionType, ingestionProperties.getDataFormat()); File file = new File(filePath); String blobName = genBlobName( file.getName(), ingestionProperties.getDatabaseName(), ingestionProperties.getTableName(), ingestionProperties.getDataFormat(), shouldCompress ? CompressionType.gz : sourceCompressionType); CloudBlockBlob blob = azureStorageClient.uploadLocalFileToBlob(fileSourceInfo.getFilePath(), blobName, resourceManager.getIngestionResource(ResourceManager.ResourceType.TEMP_STORAGE), shouldCompress); String blobPath = azureStorageClient.getBlobPathWithSas(blob); long rawDataSize = fileSourceInfo.getRawSizeInBytes() > 0L ? fileSourceInfo.getRawSizeInBytes() : estimateFileRawSize(filePath, IngestionProperties.DATA_FORMAT.valueOf(ingestionProperties.getDataFormat())); BlobSourceInfo blobSourceInfo = new BlobSourceInfo(blobPath, rawDataSize, fileSourceInfo.getSourceId()); return ingestFromBlob(blobSourceInfo, ingestionProperties); } catch (StorageException e) { throw new IngestionServiceException("Failed to ingest from file", e); } catch (IOException | URISyntaxException e) { throw new IngestionClientException("Failed to ingest from file", e); } catch (IngestionServiceException e) { validateEndpointServiceType(); throw e; } } }
QueuedIngestClient implements IngestClient { @Override public IngestionResult ingestFromFile(FileSourceInfo fileSourceInfo, IngestionProperties ingestionProperties) throws IngestionClientException, IngestionServiceException { Ensure.argIsNotNull(fileSourceInfo, "fileSourceInfo"); Ensure.argIsNotNull(ingestionProperties, "ingestionProperties"); fileSourceInfo.validate(); ingestionProperties.validate(); try { String filePath = fileSourceInfo.getFilePath(); Ensure.fileExists(filePath); CompressionType sourceCompressionType = AzureStorageClient.getCompression(filePath); boolean shouldCompress = AzureStorageClient.shouldCompress(sourceCompressionType, ingestionProperties.getDataFormat()); File file = new File(filePath); String blobName = genBlobName( file.getName(), ingestionProperties.getDatabaseName(), ingestionProperties.getTableName(), ingestionProperties.getDataFormat(), shouldCompress ? CompressionType.gz : sourceCompressionType); CloudBlockBlob blob = azureStorageClient.uploadLocalFileToBlob(fileSourceInfo.getFilePath(), blobName, resourceManager.getIngestionResource(ResourceManager.ResourceType.TEMP_STORAGE), shouldCompress); String blobPath = azureStorageClient.getBlobPathWithSas(blob); long rawDataSize = fileSourceInfo.getRawSizeInBytes() > 0L ? fileSourceInfo.getRawSizeInBytes() : estimateFileRawSize(filePath, IngestionProperties.DATA_FORMAT.valueOf(ingestionProperties.getDataFormat())); BlobSourceInfo blobSourceInfo = new BlobSourceInfo(blobPath, rawDataSize, fileSourceInfo.getSourceId()); return ingestFromBlob(blobSourceInfo, ingestionProperties); } catch (StorageException e) { throw new IngestionServiceException("Failed to ingest from file", e); } catch (IOException | URISyntaxException e) { throw new IngestionClientException("Failed to ingest from file", e); } catch (IngestionServiceException e) { validateEndpointServiceType(); throw e; } } QueuedIngestClient(ConnectionStringBuilder csb);   QueuedIngestClient(ResourceManager resourceManager);   QueuedIngestClient(ResourceManager resourceManager, AzureStorageClient azureStorageClient); }
QueuedIngestClient implements IngestClient { @Override public IngestionResult ingestFromFile(FileSourceInfo fileSourceInfo, IngestionProperties ingestionProperties) throws IngestionClientException, IngestionServiceException { Ensure.argIsNotNull(fileSourceInfo, "fileSourceInfo"); Ensure.argIsNotNull(ingestionProperties, "ingestionProperties"); fileSourceInfo.validate(); ingestionProperties.validate(); try { String filePath = fileSourceInfo.getFilePath(); Ensure.fileExists(filePath); CompressionType sourceCompressionType = AzureStorageClient.getCompression(filePath); boolean shouldCompress = AzureStorageClient.shouldCompress(sourceCompressionType, ingestionProperties.getDataFormat()); File file = new File(filePath); String blobName = genBlobName( file.getName(), ingestionProperties.getDatabaseName(), ingestionProperties.getTableName(), ingestionProperties.getDataFormat(), shouldCompress ? CompressionType.gz : sourceCompressionType); CloudBlockBlob blob = azureStorageClient.uploadLocalFileToBlob(fileSourceInfo.getFilePath(), blobName, resourceManager.getIngestionResource(ResourceManager.ResourceType.TEMP_STORAGE), shouldCompress); String blobPath = azureStorageClient.getBlobPathWithSas(blob); long rawDataSize = fileSourceInfo.getRawSizeInBytes() > 0L ? fileSourceInfo.getRawSizeInBytes() : estimateFileRawSize(filePath, IngestionProperties.DATA_FORMAT.valueOf(ingestionProperties.getDataFormat())); BlobSourceInfo blobSourceInfo = new BlobSourceInfo(blobPath, rawDataSize, fileSourceInfo.getSourceId()); return ingestFromBlob(blobSourceInfo, ingestionProperties); } catch (StorageException e) { throw new IngestionServiceException("Failed to ingest from file", e); } catch (IOException | URISyntaxException e) { throw new IngestionClientException("Failed to ingest from file", 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 ingestFromFile(FileSourceInfo fileSourceInfo, IngestionProperties ingestionProperties) throws IngestionClientException, IngestionServiceException { Ensure.argIsNotNull(fileSourceInfo, "fileSourceInfo"); Ensure.argIsNotNull(ingestionProperties, "ingestionProperties"); fileSourceInfo.validate(); ingestionProperties.validate(); try { String filePath = fileSourceInfo.getFilePath(); Ensure.fileExists(filePath); CompressionType sourceCompressionType = AzureStorageClient.getCompression(filePath); boolean shouldCompress = AzureStorageClient.shouldCompress(sourceCompressionType, ingestionProperties.getDataFormat()); File file = new File(filePath); String blobName = genBlobName( file.getName(), ingestionProperties.getDatabaseName(), ingestionProperties.getTableName(), ingestionProperties.getDataFormat(), shouldCompress ? CompressionType.gz : sourceCompressionType); CloudBlockBlob blob = azureStorageClient.uploadLocalFileToBlob(fileSourceInfo.getFilePath(), blobName, resourceManager.getIngestionResource(ResourceManager.ResourceType.TEMP_STORAGE), shouldCompress); String blobPath = azureStorageClient.getBlobPathWithSas(blob); long rawDataSize = fileSourceInfo.getRawSizeInBytes() > 0L ? fileSourceInfo.getRawSizeInBytes() : estimateFileRawSize(filePath, IngestionProperties.DATA_FORMAT.valueOf(ingestionProperties.getDataFormat())); BlobSourceInfo blobSourceInfo = new BlobSourceInfo(blobPath, rawDataSize, fileSourceInfo.getSourceId()); return ingestFromBlob(blobSourceInfo, ingestionProperties); } catch (StorageException e) { throw new IngestionServiceException("Failed to ingest from file", e); } catch (IOException | URISyntaxException e) { throw new IngestionClientException("Failed to ingest from file", 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; }