Managed streaming - better estimation for decision for streaming vs queuing

CHANGELOG.md

@@ -4,6 +4,18 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## Unknown
+### Added
+- A new heuristic for choosing between queuing and streaming in Managed streaming client, the default behavior is using
+  an estimation against the 4mb limit after dividing or multiplying by some factor described by the consts:
+
+  This will also allow users to stream bigger than 4mb non-compressed data
+    - disableRetries option to client options - default is true only for streaming clients as if stream is not repeatable it
+      will simply fail - better to let the user deal with it (alternative considered was to always create repeatable streams
+      like in managed streaming, but this PR also tries to avoid that)
+### Fixed
+- Some better error messages
+
 ## [5.2.0] - 2024-08-27
 ### Fixed
 - Used Msal user prompt old code which is deprecated in the new version coming from last bom update resulted in method not found exception.

data/src/main/java/com/microsoft/azure/kusto/data/ClientFactory.java

@@ -4,9 +4,11 @@
 package com.microsoft.azure.kusto.data;
 
 import com.microsoft.azure.kusto.data.auth.ConnectionStringBuilder;
+import com.microsoft.azure.kusto.data.http.HttpClientFactory;
 import org.apache.http.impl.client.CloseableHttpClient;
 
 import java.net.URISyntaxException;
+import java.util.Optional;
 
 public class ClientFactory {
     private ClientFactory() {
@@ -73,7 +75,9 @@ public static StreamingClient createStreamingClient(ConnectionStringBuilder csb)
      * @throws URISyntaxException if the cluster URL is invalid
      */
     public static StreamingClient createStreamingClient(ConnectionStringBuilder csb, HttpClientProperties properties) throws URISyntaxException {
-        return new ClientImpl(csb, properties);
+        HttpClientProperties httpClientProperties = Optional.ofNullable(properties)
+                .orElse(HttpClientProperties.builder().disableRetries().build());
+        return new ClientImpl(csb, HttpClientFactory.create(httpClientProperties), false);
     }
 
     /**

data/src/main/java/com/microsoft/azure/kusto/data/ClientImpl.java

@@ -14,6 +14,7 @@
 import com.microsoft.azure.kusto.data.exceptions.DataServiceException;
 import com.microsoft.azure.kusto.data.exceptions.KustoClientInvalidConnectionStringException;
 import com.microsoft.azure.kusto.data.exceptions.KustoServiceQueryError;
+import com.microsoft.azure.kusto.data.exceptions.ExceptionsUtils;
 import com.microsoft.azure.kusto.data.http.HttpClientFactory;
 import com.microsoft.azure.kusto.data.http.HttpPostUtils;
 import com.microsoft.azure.kusto.data.http.UncloseableStream;
@@ -176,9 +177,9 @@ private KustoOperationResult executeImpl(String database, String command, Client
             return new KustoOperationResult(response, clusterEndpoint.endsWith("v2/rest/query") ? "v2" : "v1");
         } catch (KustoServiceQueryError e) {
             throw new DataServiceException(clusterEndpoint,
-                    "Error found while parsing json response as KustoOperationResult:" + e.getMessage(), e, e.isPermanent());
+                    "Error found while parsing json response as KustoOperationResult:" + e, e, e.isPermanent());
         } catch (Exception e) {
-            throw new DataClientException(clusterEndpoint, e.getMessage(), e);
+            throw new DataClientException(clusterEndpoint, ExceptionsUtils.getMessageEx(e), e);
         }
     }
 

data/src/main/java/com/microsoft/azure/kusto/data/HttpClientProperties.java

@@ -16,6 +16,7 @@ public class HttpClientProperties {
     private final HttpHost proxy;
     private final HttpRoutePlanner routePlanner;
     private final String[] supportedProtocols;
+    private final boolean disableRetries;
 
     private HttpClientProperties(HttpClientPropertiesBuilder builder) {
         this.maxIdleTime = builder.maxIdleTime;
@@ -26,6 +27,7 @@ private HttpClientProperties(HttpClientPropertiesBuilder builder) {
         this.proxy = builder.proxy;
         this.routePlanner = builder.routePlanner;
         this.supportedProtocols = builder.supportedProtocols;
+        this.disableRetries = builder.disableRetries;
     }
 
     /**
@@ -48,7 +50,7 @@ public Integer maxIdleTime() {
     }
 
     /**
-     * Indicates whether a custom connection keep-alive time should be used. If set to {@code false}, the HTTP
+     * Indicates whether or not a custom connection keep-alive time should be used. If set to {@code false}, the HTTP
      * client will use the default connection keep-alive strategy, which is to use only the server instructions
      * (if any) set in the {@code Keep-Alive} response header.
      * If set to {@code true}, the HTTP client will use a custom connection keep-alive strategy which uses the
@@ -111,6 +113,10 @@ public String[] supportedProtocols() {
         return supportedProtocols;
     }
 
+    public boolean isDisableRetries() {
+        return disableRetries;
+    }
+
     public static class HttpClientPropertiesBuilder {
 
         private Integer maxIdleTime = 120;
@@ -121,6 +127,7 @@ public static class HttpClientPropertiesBuilder {
         private HttpHost proxy = null;
         private HttpRoutePlanner routePlanner = null;
         private String[] supportedProtocols = null;
+        private boolean disableRetries;
 
         private HttpClientPropertiesBuilder() {
         }
@@ -226,9 +233,18 @@ public HttpClientPropertiesBuilder supportedProtocols(String[] tlsProtocols) {
             return this;
         }
 
+        /**
+         * Disable all http client internal retries.
+         *
+         * @return the builder instance
+         */
+        public HttpClientPropertiesBuilder disableRetries() {
+            this.disableRetries = true;
+            return this;
+        }
+
         public HttpClientProperties build() {
             return new HttpClientProperties(this);
         }
-
     }
 }

data/src/main/java/com/microsoft/azure/kusto/data/IngestionSourceStorage.java

@@ -2,7 +2,6 @@
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
-import com.microsoft.azure.kusto.data.http.HttpPostUtils;
 
 class IngestionSourceStorage {
     public String sourceUri;

data/src/main/java/com/microsoft/azure/kusto/data/auth/CallbackTokenProvider.java

@@ -5,6 +5,7 @@
 
 import com.microsoft.azure.kusto.data.exceptions.DataClientException;
 
+import com.microsoft.azure.kusto.data.exceptions.ExceptionsUtils;
 import org.apache.http.client.HttpClient;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
@@ -32,7 +33,7 @@ protected String acquireAccessTokenImpl() throws DataClientException {
         try {
             return tokenProvider.apply(httpClient);
         } catch (Exception e) {
-            throw new DataClientException(clusterUrl, e.getMessage(), e);
+            throw new DataClientException(clusterUrl, ExceptionsUtils.getMessageEx(e), e);
         }
     }
 

data/src/main/java/com/microsoft/azure/kusto/data/auth/CloudInfo.java

@@ -6,6 +6,7 @@
 import com.microsoft.azure.kusto.data.ExponentialRetry;
 import com.microsoft.azure.kusto.data.Utils;
 import com.microsoft.azure.kusto.data.exceptions.DataClientException;
+import com.microsoft.azure.kusto.data.exceptions.ExceptionsUtils;
 import com.microsoft.azure.kusto.data.http.HttpClientFactory;
 import com.microsoft.azure.kusto.data.instrumentation.SupplierOneException;
 import com.microsoft.azure.kusto.data.UriUtils;
@@ -104,7 +105,8 @@ public static CloudInfo retrieveCloudInfoForCluster(String clusterUrl,
                     throw new DataServiceException(clusterUrl, "URISyntaxException when trying to retrieve cluster metadata:" + e.getMessage(), e, true);
                 } catch (IOException ex) {
                     if (!Utils.isRetriableIOException(ex)) {
-                        throw new DataServiceException(clusterUrl, "IOException when trying to retrieve cluster metadata:" + ex.getMessage(), ex,
+                        throw new DataServiceException(clusterUrl, "IOException when trying to retrieve cluster metadata:" + ExceptionsUtils.getMessageEx(ex),
+                                ex,
                                 Utils.isRetriableIOException(ex));
                     }
                 } catch (DataServiceException e) {

data/src/main/java/com/microsoft/azure/kusto/data/auth/endpoints/WellKnownKustoEndpointsData.java

@@ -44,8 +44,7 @@ private static WellKnownKustoEndpointsData readInstance() {
                 return objectMapper.readValue(resourceAsStream, WellKnownKustoEndpointsData.class);
             }
         } catch (Exception ex) {
-            ex.printStackTrace();
+            throw new RuntimeException("Failed to read WellKnownKustoEndpoints.json", ex);
         }
-        return null;
     }
 }

data/src/main/java/com/microsoft/azure/kusto/data/exceptions/ExceptionsUtils.java

@@ -0,0 +1,8 @@
+package com.microsoft.azure.kusto.data.exceptions;
+
+public class ExceptionsUtils {
+    // Useful in IOException, where message might not propagate to the base IOException
+    public static String getMessageEx(Exception e) {
+        return (e.getMessage() == null && e.getCause() != null) ? e.getCause().getMessage() : e.getMessage();
+    }
+}

data/src/main/java/com/microsoft/azure/kusto/data/http/HttpClientFactory.java

@@ -44,6 +44,9 @@ public static CloseableHttpClient create(HttpClientProperties providedProperties
                 .evictIdleConnections(properties.maxIdleTime(), TimeUnit.SECONDS)
                 .disableRedirectHandling();
 
+        if (properties.isDisableRetries()) {
+            httpClientBuilder.disableAutomaticRetries();
+        }
         if (properties.isKeepAlive()) {
             final ConnectionKeepAliveStrategy keepAliveStrategy = new CustomConnectionKeepAliveStrategy(properties.maxKeepAliveTime());
             httpClientBuilder.setKeepAliveStrategy(keepAliveStrategy);

data/src/main/java/com/microsoft/azure/kusto/data/http/HttpPostUtils.java

@@ -74,7 +74,7 @@ public static String post(CloseableHttpClient httpClient, String urlStr, Abstrac
                 }
             }
         } catch (IOException e) {
-            throw new DataServiceException(urlStr, "IOException in post request:" + e.getMessage(), !Utils.isRetriableIOException(e));
+            throw new DataServiceException(urlStr, "IOException in post request:" + ExceptionsUtils.getMessageEx(e), e, !Utils.isRetriableIOException(e));
         }
 
         return null;

ingest/pom.xml

@@ -106,6 +106,7 @@
                     </ignoredUnusedDeclaredDependencies>
                     <ignoredUsedUndeclaredDependencies>
                         <dependency>io.projectreactor:reactor-core:jar</dependency>
+                        <dependency>com.fasterxml.jackson.core:jackson-core:jar</dependency>
                     </ignoredUsedUndeclaredDependencies>
                     <ignoreNonCompile>true</ignoreNonCompile>
                 </configuration>

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

@@ -1,11 +1,11 @@
 package com.microsoft.azure.kusto.ingest;
 
+import com.microsoft.azure.kusto.data.exceptions.ExceptionsUtils;
 import com.microsoft.azure.kusto.ingest.source.CompressionType;
 import org.apache.http.conn.util.InetAddressUtils;
 
-import java.net.InetAddress;
+import java.io.IOException;
 import java.net.URI;
-import java.net.UnknownHostException;
 import com.microsoft.azure.kusto.data.instrumentation.SupplierTwoExceptions;
 import com.microsoft.azure.kusto.data.instrumentation.TraceableAttributes;
 import com.microsoft.azure.kusto.data.instrumentation.MonitoredActivity;
@@ -130,14 +130,20 @@ public IngestionResult ingestFromResultSet(ResultSetSourceInfo resultSetSourceIn
      * @see IngestionProperties
      */
     protected abstract IngestionResult ingestFromStreamImpl(StreamSourceInfo streamSourceInfo, IngestionProperties ingestionProperties)
-            throws IngestionClientException, IngestionServiceException;
+            throws IngestionClientException, IngestionServiceException, IOException;
 
     public IngestionResult ingestFromStream(StreamSourceInfo streamSourceInfo, IngestionProperties ingestionProperties)
             throws IngestionClientException, IngestionServiceException {
         // trace ingestFromStream
         return MonitoredActivity.invoke(
-                (SupplierTwoExceptions<IngestionResult, IngestionClientException, IngestionServiceException>) () -> ingestFromStreamImpl(streamSourceInfo,
-                        ingestionProperties),
+                (SupplierTwoExceptions<IngestionResult, IngestionClientException, IngestionServiceException>) () -> {
+                    try {
+                        return ingestFromStreamImpl(streamSourceInfo,
+                                ingestionProperties);
+                    } catch (IOException e) {
+                        throw new IngestionServiceException(ExceptionsUtils.getMessageEx(e), e);
+                    }
+                },
                 getClientType().concat(".ingestFromStream"));
     }
 

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

@@ -431,6 +431,10 @@ public IngestionMapping.IngestionMappingKind getIngestionMappingKind() {
         public boolean isCompressible() {
             return compressible;
         }
+
+        public boolean isJsonFormat() {
+            return this.equals(JSON) || this.equals(MULTIJSON) || this.equals(SINGLEJSON);
+        }
     }
 
     public enum IngestionReportLevel {