.NET InsertMediaUpload 将 CSV 上传到 BigQuery 时出现奇怪的问题

Question

我正在使用 .NET 客户端 API：IUploadProgress progress = insertMediaUpload.Upload() 将 csv 上传到 BigQuery。基本上我所做的是：

1.提交上传作业，
2. 获取作业状态（待处理、正在运行、已完成..）
3. 如果 BigQuery 给出任何错误，则打印出来并抛出异常以便进一步处理。

下面的代码并不是我想要的，我希望有人可以帮助我改进它。

具体来说，发生了几个奇怪的代码行为：
1. 即使在相同的 CSV 上运行相同的代码（故意失败），在 UploadOnResponseReceived() 中解析出的 BQ 错误消息也会在某些调用中打印出来，但在其他调用中不会打印出来。为什么？
2. IUploadProgress 值似乎与 UploadOnResponseReceived() 行为有关：如果我在 UploadOnResponseReceived 中什么都不做，那么 progress.status 将始终为“已完成”，如果 UploadOnResponseReceived 抛出异常，则 progress.status 将失败。
3.progress.status失败时，无法获取UploadOnResponseReceived抛出的异常。我实际上确实需要获得异常，我该怎么办？

 public bool ExecuteUploadJobToTable(string dataset, string tableId, string filePath, TableSchema schema, string createDisposition, char delimiter)
    {

        TableReference destTable = new TableReference { ProjectId = _account.ProjectId, DatasetId = dataset, TableId = tableId };

        JobConfigurationLoad configLoad = new JobConfigurationLoad 
        {
            Schema = schema,
            DestinationTable = destTable,
            Encoding = "ISO-8859-1",
            CreateDisposition = "CREATE_IF_NEEDED",
            WriteDisposition = createDisposition,
            FieldDelimiter = delimiter.ToString(),
            AllowJaggedRows = true,
            SourceFormat = "CSV"
        };

        JobConfiguration config = new JobConfiguration {Load = configLoad};

        Job job = new Job {Configuration = config};

        //set job reference (mainly job id)
        JobReference jobRef = new JobReference
        {
            JobId = GenerateJobID("Upload"),
            ProjectId = _account.ProjectId
        };
        job.JobReference = jobRef;

        bool isSuccess = true;
        using (var fileStream = new FileStream(filePath, FileMode.Open))
        {
            JobsResource.InsertMediaUpload insertMediaUpload = new JobsResource.InsertMediaUpload(BigQueryService, job, job.JobReference.ProjectId, stream: fileStream, contentType: "application/octet-stream");
            insertMediaUpload.ProgressChanged += UploadOnProgressChanged;
            insertMediaUpload.ResponseReceived += UploadOnResponseReceived;

            Console.WriteLine(string.Format("start {0}",jobRef.JobId));
            IUploadProgress progress = insertMediaUpload.Upload();
            if (progress.Status.ToString().Contains("Fail"))
            {
                isSuccess = false;
            }
        }
        Console.WriteLine(isSuccess);
        return isSuccess;
    }

    private void UploadOnProgressChanged(IUploadProgress process)
    {
        Console.WriteLine(process.Status + " " + process.BytesSent);
    }

    //thowring an exception will make IUploadProgress "Failed", otherwise, IUploadProgress will be "Completed"
    private void UploadOnResponseReceived(Job job)
    {
        try
        {
            job = PollUntilJobDone(job.JobReference, 5);
        }
        catch(Exception e)
        {
            Console.WriteLine("Unexcepted unretryable exception happens when poll job status");
            throw new BigQueryException("Unexcepted unretryable exception happens when poll job status",e);
        }

        StringBuilder errorMessageBuilder = new StringBuilder();
        ErrorProto fatalError = job.Status.ErrorResult;
        IList<ErrorProto> errors = job.Status.Errors;
        if (fatalError != null)
        {
            errorMessageBuilder.AppendLine("Job failed while writing to Bigquery. " + fatalError.Reason + ": " + fatalError.Message +
                      " at " + fatalError.Location);
        }
        if (errors != null)
        {
            foreach (ErrorProto error in errors)
            {
                errorMessageBuilder.AppendLine("Error: [REASON] " + error.Reason + " [MESSAGE] " + error.Message +
                                               " [LOCATION] " + error.Location);
            }

        }
        if (errorMessageBuilder.Length>0)//fatalError != null || errors != null  
        {
            Console.WriteLine(errorMessageBuilder.ToString());
            throw new BigQueryException(errorMessageBuilder.ToString());
        }
        Console.WriteLine("upload should be successful");
    }

    private Job PollUntilJobDone(JobReference jobReference, int pauseSeconds)
    {
        int backoff = 1000;//backoff starts from 1 sec + random

        for(int i = 0; i < 10; i++)
        {
            try
            {
                var pollJob = BigQueryService.Jobs.Get(jobReference.ProjectId, jobReference.JobId).Execute();
                Console.WriteLine(jobReference.JobId + ": " + pollJob.Status.State);
                if (pollJob.Status.State.Equals("DONE"))
                {
                    return pollJob;
                }
                // Pause execution for pauseSeconds before polling job status again,
                // to reduce unnecessary calls to the BigQuery API and lower overall
                // application bandwidth.
                Thread.Sleep(pauseSeconds * 1000);
            }
            catch (Exception e)
            {
                BigQueryException exception = new BigQueryException(e.Message,e);
                if (exception.IsTemporary)
                {
                    int sleep = backoff + Random.Next(1000);
                    Console.WriteLine("pollUntilJobDone job execute failed. Sleeping {0} ms before retry", sleep);
                    Thread.Sleep(sleep);
                }
                else
                {
                    throw;
                }
            }
            backoff *= 2;
        }
        return null;
    }

Answer 1

关于您的“我如何捕获异常”问题，回调似乎在另一个线程上异步发生。如果您抛出异常，它将被调用回调的任何框架捕获。

搜索类似的问题，我找到了这些可能对您有帮助的答案：Catching an exception thrown in an asynchronous callback，而这个答案显示了如何根据后台线程中收到的上传进度更新另一个线程中的 UI：Tracking upload progress of WebClient