Skip to content

Managed Streaming Should Handle Throttling Events Correctly #601

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 3 commits into from
Closed

Managed Streaming Should Handle Throttling Events Correctly #601

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
854e6d6
In managed streaming, exhausted retries with throttling errors should…
yogilad Nov 9, 2025
8d07ce2
fix linter issues
yogilad Nov 9, 2025
cd390c5
add Gov SG and Delos cloud DNS support
yogilad Nov 25, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 22 additions & 0 deletions azure-kusto-data/azure/kusto/data/wellKnownKustoEndpoints.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -104,6 +104,28 @@
"adx.loganalytics.azure.fr",
"adx.monitor.azure.fr"
]
},
"https://login.sovcloud-identity.de": {
"AllowedKustoSuffixes": [
".kusto.sovcloud-api.de",
".kustomfa.sovcloud-api.de"
],
"AllowedKustoHostnames": [
"adx.applicationinsights.azure.de",
"adx.loganalytics.azure.de",
"adx.monitor.azure.de"
]
},
"https://login.sovcloud-identity.sg": {
"AllowedKustoSuffixes": [
".kusto.sovcloud-api.sg",
".kustomfa.sovcloud-api.sg"
],
"AllowedKustoHostnames": [
"adx.applicationinsights.azure.sg",
"adx.loganalytics.azure.sg",
"adx.monitor.azure.sg"
]
}
}
}
7 changes: 3 additions & 4 deletions azure-kusto-data/tests/test_well_known_kusto_endpoints.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -126,10 +126,9 @@ def test_well_known_kusto_endpoints_national_clouds():
"https://kustofrbwrznltavls.kusto.usgovcloudapi.net,https://login.microsoftonline.us",
"https://kusto7j53clqswr4he.kusto.core.eaglex.ic.gov,https://login.microsoftonline.eaglex.ic.gov",
"https://rpe2e0422132101fct2.kusto.core.microsoft.scloud,https://login.microsoftonline.microsoft.scloud",
"https://kustozszokb5yrauyq.kusto.chinacloudapi.cn,{0}".format(CHINA_CLOUD_LOGIN),
"https://kustofrbwrznltavls.kusto.usgovcloudapi.net,https://login.microsoftonline.us",
"https://kusto7j53clqswr4he.kusto.core.eaglex.ic.gov,https://login.microsoftonline.eaglex.ic.gov",
"https://rpe2e0422132101fct2.kusto.core.microsoft.scloud,https://login.microsoftonline.microsoft.scloud",
"https://abc.kusto.sovcloud-api.fr,https://login.sovcloud-identity.fr",
"https://efg.kusto.sovcloud-api.de,https://login.sovcloud-identity.de",
"https://hij.kusto.sovcloud-api.sg,https://login.sovcloud-identity.sg"
Copy link
Contributor

@github-actions github-actions bot Nov 25, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
"https://hij.kusto.sovcloud-api.sg,https://login.sovcloud-identity.sg"
"https://hij.kusto.sovcloud-api.sg,https://login.sovcloud-identity.sg",

]:
cluster_and_login_endpoint = c.split(",")
_validate_endpoint(cluster_and_login_endpoint[0], cluster_and_login_endpoint[1])
Expand Down
6 changes: 5 additions & 1 deletion azure-kusto-ingest/azure/kusto/ingest/managed_streaming_ingest_client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@
from azure.core.tracing import SpanKind

from azure.kusto.data import KustoConnectionStringBuilder
from azure.kusto.data.exceptions import KustoApiError, KustoClosedError
from azure.kusto.data.exceptions import KustoApiError, KustoClosedError, KustoThrottlingError
from azure.kusto.data._telemetry import MonitoredActivity

from . import BlobDescriptor, FileDescriptor, IngestionProperties, StreamDescriptor
Expand Down Expand Up @@ -99,6 +99,8 @@ def ingest_from_stream(self, stream_descriptor: Union[StreamDescriptor, IO[AnySt
if error.permanent:
raise
buffered_stream.seek(0, SEEK_SET)
except KustoThrottlingError:
_ = buffered_stream.seek(0, SEEK_SET)

return self.queued_client.ingest_from_stream(stream_descriptor, ingestion_properties)

Expand Down Expand Up @@ -127,6 +129,8 @@ def ingest_from_blob(self, blob_descriptor: BlobDescriptor, ingestion_properties
error = ex.get_api_error()
if error.permanent:
raise
except KustoThrottlingError:
pass

return self.queued_client.ingest_from_blob(blob_descriptor, ingestion_properties)

Expand Down
Loading