Change: Make ManagedStreamingQueuingPolicy internal, expose just a factor #413

ohadbitt · 2025-01-27T14:16:23Z

Change:

Make ManagedStreamingQueuingPolicy internal, expose just a factor
Dont allow users to pass raw data, provide it only if we have it

Expose just a factor instead of the whole managed policy

github-actions · 2025-01-28T11:24:35Z

Test Results

149 tests - 183 149 ✅ - 174 6s ⏱️ - 3m 25s
14 suites - 13 0 💤 - 9
14 files - 13 0 ❌ ± 0

Results for commit 49888c1. ± Comparison against base commit 9884e0a.

This pull request removes 183 tests.

com.microsoft.azure.kusto.ingest.AzureStorageClientTest ‑ UploadStreamToBlob_NullInputStream_IllegalArgumentException
com.microsoft.azure.kusto.ingest.AzureStorageClientTest ‑ compressAndStream_NullBlob_IllegalArgumentException
com.microsoft.azure.kusto.ingest.AzureStorageClientTest ‑ compressAndStream_NullStream_IllegalArgumentException
com.microsoft.azure.kusto.ingest.AzureStorageClientTest ‑ compressAndUploadFileToBlob_NullBlob_IllegalArgumentException
com.microsoft.azure.kusto.ingest.AzureStorageClientTest ‑ compressAndUploadFileToBlob_NullFilePath_IllegalArgumentException
com.microsoft.azure.kusto.ingest.AzureStorageClientTest ‑ postMessageToQueue_NullContent_IllegalArgumentException
com.microsoft.azure.kusto.ingest.AzureStorageClientTest ‑ postMessageToQueue_NullEntity_IllegalArgumentException
com.microsoft.azure.kusto.ingest.AzureStorageClientTest ‑ postMessageToQueue_NullQueuePath_IllegalArgumentException
com.microsoft.azure.kusto.ingest.AzureStorageClientTest ‑ postMessageToQueue_NullTableUri_IllegalArgumentException
com.microsoft.azure.kusto.ingest.AzureStorageClientTest ‑ uploadFileToBlob_NullBlob_IllegalArgumentException
…

♻️ This comment has been updated with latest results.

AsafMah · 2025-01-29T08:46:26Z

ingest/src/main/java/com/microsoft/azure/kusto/ingest/source/BlobSourceInfo.java

+    /*
+     * For internal usage, adding blobExactSize
+     */
+    public static BlobSourceInfo fromFile(String blobPath, FileSourceInfo fileSourceInfo, CompressionType sourceCompressionType, boolean gotCompressed) {


If it's for internal use - can we lower the visibility?

no because its used outside the pacakge

Ok, so maybe we want to document it like a public function?

Not sure what the right call is ehre

AsafMah · 2025-01-29T08:47:21Z

ingest/src/main/java/com/microsoft/azure/kusto/ingest/source/BlobSourceInfo.java

+    /*
+     * For internal usage, adding blobExactSize
+     */
+    public static BlobSourceInfo fromFile(String blobPath, FileSourceInfo fileSourceInfo, CompressionType sourceCompressionType, boolean gotCompressed) {


The param on the outside is "shouldCompress", but here it's called "gotCompressed" - any reason?

Client should care if it should compress
. file naming should care only that it was compressed .

AsafMah · 2025-01-29T08:48:37Z

ingest/src/main/java/com/microsoft/azure/kusto/ingest/source/FileSourceInfo.java

@@ -23,14 +23,8 @@ public FileSourceInfo(String filePath) {
        this.filePath = filePath;
    }

-    // An estimation of the raw (uncompressed, un-indexed) size of the data, for binary formatted files - use only if known
-    public FileSourceInfo(String filePath, long rawSizeInBytes) {
+    public FileSourceInfo(String filePath, UUID sourceId) {


In V2 SourceID is not UUID, so maybe we already want to treat it as a string?

its not v2 here

Yes, but just not to break the interface when we move

ingest/src/main/java/com/microsoft/azure/kusto/ingest/source/FileSourceInfo.java

AsafMah · 2025-01-29T08:49:55Z

ingest/src/main/java/com/microsoft/azure/kusto/ingest/utils/IngestionUtils.java

@@ -39,22 +39,7 @@ public static StreamSourceInfo fileToStream(FileSourceInfo fileSourceInfo, boole
        }

        CompressionType compression = getCompression(filePath);


We no longer rely on format.isCompressible?

not here - in the policy

ingest/src/main/java/com/microsoft/azure/kusto/ingest/AzureStorageClient.java

AsafMah · 2025-01-29T08:51:01Z

ingest/src/main/java/com/microsoft/azure/kusto/ingest/AzureStorageClient.java

        }
    }

-    private void copyStream(InputStream inputStream, OutputStream outputStream, int bufferSize) throws IOException {
+    // Returns original stream size
+    private int copyStream(InputStream inputStream, OutputStream outputStream, int bufferSize) throws IOException {
        byte[] buffer = new byte[bufferSize];


I think there is an helper in apache for copy stream, you should check.

ingest/src/main/java/com/microsoft/azure/kusto/ingest/AzureStorageClient.java

AsafMah · 2025-01-29T08:53:02Z

ingest/src/main/java/com/microsoft/azure/kusto/ingest/ManagedStreamingIngestClient.java

+     * {@link ManagedStreamingQueuingPolicy.Default} which is created with no factor.
+     * @param factor - Default is 1.
+     **/
+    public void setQueuingPolicyFactor(double factor) {


Update CHANGELOG.md with all the changes, mention this specifically as a breaking change

AsafMah · 2025-01-29T08:54:31Z

ingest/src/main/java/com/microsoft/azure/kusto/ingest/ResourceAlgorithms.java

@@ -83,12 +84,13 @@ public static void postToQueueWithRetries(ResourceManager resourceManager, Azure
                Collections.singletonMap("blob", SecurityUtils.removeSecretsFromUrl(blob.getBlobPath())));
    }

-    public static String uploadStreamToBlobWithRetries(ResourceManager resourceManager, AzureStorageClient azureStorageClient, InputStream stream,
+    public static Pair<String, Integer> uploadStreamToBlobWithRetries(ResourceManager resourceManager, AzureStorageClient azureStorageClient,


With the previous comment, maybe you should just have an UploadResult type with the string and size, it's clearer for both here and the inner methods that return int

AsafMah · 2025-01-29T08:55:44Z

ingest/src/main/java/com/microsoft/azure/kusto/ingest/StreamingIngestClient.java

            @Nullable String clientRequestId)
            throws IngestionClientException, IngestionServiceException {
        String blobPath = blobSourceInfo.getBlobPath();
-        try {
-            // No need to check blob size if it was given to us that it's not empty
-            if (blobSourceInfo.getRawSizeInBytes() == 0 && cloudBlockBlob.getProperties().getBlobSize() == 0) {


Do we check for empty blob elsewhere?

AsafMah · 2025-01-29T08:56:12Z

ingest/src/test/java/com/microsoft/azure/kusto/ingest/ManagedStreamingTest.java

-        managedStreamingIngestClientSpy = spy(
-                new ManagedStreamingIngestClient(mock(StreamingIngestClient.class), queuedIngestClientMock, new ExponentialRetry(1)));
-    }
+    // @BeforeAll


Why is everything commented out

AsafMah

Some comments

Exact size instead of raw size

7219631

Expose just a factor instead of the whole managed policy

ohadbitt requested a review from AsafMah January 27, 2025 14:17

Merge branch 'master' into queuingPolicy-expose-factor

b981258

ohadbitt had a problem deploying to build January 28, 2025 11:23 — with GitHub Actions Failure

ohadbitt had a problem deploying to build January 28, 2025 11:23 — with GitHub Actions Error

fix and add tests

49888c1

ohadbitt had a problem deploying to build January 28, 2025 18:15 — with GitHub Actions Failure

ohadbitt had a problem deploying to build January 28, 2025 18:15 — with GitHub Actions Error

AsafMah reviewed Jan 29, 2025

View reviewed changes

ingest/src/main/java/com/microsoft/azure/kusto/ingest/source/FileSourceInfo.java Show resolved Hide resolved

AsafMah reviewed Jan 29, 2025

View reviewed changes

ingest/src/main/java/com/microsoft/azure/kusto/ingest/AzureStorageClient.java Show resolved Hide resolved

AsafMah reviewed Jan 29, 2025

View reviewed changes

ingest/src/main/java/com/microsoft/azure/kusto/ingest/AzureStorageClient.java Show resolved Hide resolved

AsafMah reviewed Jan 29, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Change: Make ManagedStreamingQueuingPolicy internal, expose just a factor #413

Change: Make ManagedStreamingQueuingPolicy internal, expose just a factor #413

ohadbitt commented Jan 27, 2025

github-actions bot commented Jan 28, 2025 •

edited

Loading

AsafMah Jan 29, 2025

ohadbitt Jan 30, 2025

AsafMah Feb 2, 2025

AsafMah Jan 29, 2025

ohadbitt Jan 30, 2025

AsafMah Jan 29, 2025

ohadbitt Jan 30, 2025

AsafMah Feb 2, 2025

AsafMah Jan 29, 2025

ohadbitt Jan 30, 2025

AsafMah Jan 29, 2025

AsafMah Jan 29, 2025

AsafMah Jan 29, 2025

AsafMah Jan 29, 2025

AsafMah Jan 29, 2025

AsafMah left a comment

		@@ -39,22 +39,7 @@ public static StreamSourceInfo fileToStream(FileSourceInfo fileSourceInfo, boole
		}

		CompressionType compression = getCompression(filePath);

Change: Make ManagedStreamingQueuingPolicy internal, expose just a factor #413

Are you sure you want to change the base?

Change: Make ManagedStreamingQueuingPolicy internal, expose just a factor #413

Conversation

ohadbitt commented Jan 27, 2025

github-actions bot commented Jan 28, 2025 • edited Loading

Test Results

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

AsafMah left a comment

Choose a reason for hiding this comment

github-actions bot commented Jan 28, 2025 •

edited

Loading