Fix managed ingestion to poll for final status on queued fallback

When managed streaming ingestion falls back to queued, the immediate
status is Queued/Pending. Previously this was treated as success,
silently dropping data when permissions were wrong.

Now polls for final status (1 min timeout, 5 sec interval) to catch
permission errors and ingestion failures. Moved PollIngestionStatus
to base class for reuse by both managed and queued paths.

Author: monishkadas-ms

src/Services/IKustoIngestionService.cs

@@ -47,6 +47,30 @@ public static KustoIngestionProperties GetKustoIngestionProperties(DataSourceFor
             }
             return kustoIngestProperties;
         }
+
+        /// <summary>
+        /// Polls the ingestion status until a terminal state is reached or timeout occurs.
+        /// Used by both queued ingestion and managed ingestion when it falls back to queued.
+        /// </summary>
+        protected static async Task<IngestionStatus> PollIngestionStatus(IKustoIngestionResult queuedIngestResult, Guid sourceId, int ingestionTimeoutMinutes, int pollIntervalSeconds, CancellationToken cancellationToken)
+        {
+            var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
+            cts.CancelAfter(TimeSpan.FromMinutes(ingestionTimeoutMinutes));
+            IngestionStatus ingestionStatus = null;
+            while (!cancellationToken.IsCancellationRequested)
+            {
+                ingestionStatus = queuedIngestResult.GetIngestionStatusBySourceId(sourceId);
+                if (ingestionStatus.Status == Status.Succeeded
+                    || ingestionStatus.Status == Status.Skipped
+                    || ingestionStatus.Status == Status.PartiallySucceeded
+                    || ingestionStatus.Status == Status.Failed)
+                {
+                    break;
+                }
+                await Task.Delay(TimeSpan.FromSeconds(pollIntervalSeconds), cancellationToken);
+            }
+            return ingestionStatus;
+        }
     }
 
     internal class KustoManagedIngestionService : IKustoIngestionService
@@ -69,6 +93,13 @@ public override async Task<IngestionStatus> IngestData(DataSourceFormat dataForm
             {
                 this._logger.LogDebug($"Ingestion status for sourceId {streamSourceOptions.SourceId} is {managedIngestionStatus.Status}");
             }
+            // When managed streaming ingestion falls back to queued, the immediate status is Queued/Pending.
+            // In this case, poll for the final status to ensure we catch permission and ingestion errors.
+            if (managedIngestionStatus.Status == Status.Queued || managedIngestionStatus.Status == Status.Pending)
+            {
+                this._logger.LogDebug($"Managed ingestion fell back to queued for sourceId {streamSourceOptions.SourceId}. Polling for final status.");
+                return await PollIngestionStatus(ingestionResult, streamSourceOptions.SourceId, 1, 5, cancellationToken);
+            }
             return managedIngestionStatus;
         }
     }
@@ -123,27 +154,5 @@ public override async Task<IngestionStatus> IngestData(DataSourceFormat dataForm
             }
             return await PollIngestionStatus(ingestionResult, streamSourceOptions.SourceId, pollTimeoutMinutes, pollIntervalSeconds, cancellationToken);
         }
-
-        private static async Task<IngestionStatus> PollIngestionStatus(IKustoIngestionResult queuedIngestResult, Guid sourceId, int ingestionTimeoutMinutes, int pollIntervalSeconds, CancellationToken cancellationToken)
-        {
-            var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
-            cts.CancelAfter(TimeSpan.FromMinutes(ingestionTimeoutMinutes));
-            IngestionStatus ingestionStatus = null;
-            while (!cancellationToken.IsCancellationRequested)
-            {
-                ingestionStatus = queuedIngestResult.GetIngestionStatusBySourceId(sourceId);
-                // Check if the ingestion status indicates completion
-                if (ingestionStatus.Status == Status.Succeeded
-                    || ingestionStatus.Status == Status.Skipped // The ingestion was skipped because it was already ingested 
-                    || ingestionStatus.Status == Status.PartiallySucceeded // Some of the records were ingested 
-                    || ingestionStatus.Status == Status.Failed)
-                {
-                    break;
-                }
-                // Wait for a specified interval before polling again
-                await Task.Delay(TimeSpan.FromSeconds(pollIntervalSeconds), cancellationToken);
-            }
-            return ingestionStatus;
-        }
     }
 }

test/IntegrationTests/KustoBindingE2EIntegrationTests.cs

@@ -115,7 +115,6 @@ public async Task KustoFunctionsE2E()
                 Assert.True(isError, readPrivilegeException.GetBaseException().Message);
 
                 // Fail scenario for no ingest privileges
-
                 string[] testsNoPrivilegesExecute = { nameof(KustoEndToEndTestClass.OutputFailForUserWithNoReadPrivileges) };
                 // , nameof(KustoEndToEndTestClass.OutputQueuedFailForUserWithNoReadPrivileges) 
                 foreach (string testNoPrivilegesExecute in testsNoPrivilegesExecute)