Merge pull request #15850 from cdapio/fix-gcs-uploa

[CDAP-21123] fix error messages while gcs upload
cdapio · Feb 4, 2025 · 4f004e1 · 4f004e1
2 parents 33bdca1 + 998de4a
commit 4f004e1
Showing 1 changed file with 16 additions and 5 deletions.
diff --git a/...dataproc/src/main/java/io/cdap/cdap/runtime/spi/runtimejob/DataprocRuntimeJobManager.java b/...dataproc/src/main/java/io/cdap/cdap/runtime/spi/runtimejob/DataprocRuntimeJobManager.java
@@ -416,8 +416,13 @@ public Optional<RuntimeJobDetail> getDetail(ProgramRunInfo programRunInfo) throw
           getJobStatusDetails(job)));
     } catch (ApiException e) {
       if (e.getStatusCode().getCode() != StatusCode.Code.NOT_FOUND) {
-        throw new Exception(String.format("Error while getting details for job %s on cluster %s.",
-            jobId, clusterName), e);
+        int code = e.getStatusCode().getCode().getHttpStatusCode();
+        ActionErrorPair pair = ErrorUtils.getActionErrorByStatusCode(code);
+        String errorReason = String.format("%s Unable to get details for job %s on cluster %s. %s",
+            code, jobId, clusterName, pair.getCorrectiveAction());
+        throw ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategoryEnum.OTHERS),
+            errorReason, e.getMessage(), pair.getErrorType(), true, ErrorCodeType.HTTP,
+            String.valueOf(code), null, e);
       }
       // Status is not found if job is finished or manually deleted by the user
       LOG.debug("Dataproc job {} does not exist in project {}, region {}.", jobId, projectId,
@@ -560,7 +565,13 @@ private LocalFile uploadCacheableFile(String bucket, String targetFilePath,
               Storage.BlobWriteOption.metagenerationMatch());
         } catch (StorageException e) {
           if (e.getCode() != HttpURLConnection.HTTP_PRECON_FAILED) {
-            throw e;
+            ActionErrorPair pair = ErrorUtils.getActionErrorByStatusCode(e.getCode());
+            String errorReason = String.format("%s Unable to upload file %s to GCS bucket gs://%s."
+                    + " %s", e.getCode(), localFile.getURI(), bucket, pair.getCorrectiveAction());
+            throw ErrorUtils.getProgramFailureException(
+                new ErrorCategory(ErrorCategoryEnum.STARTING), errorReason, e.getMessage(),
+                pair.getErrorType(), true, ErrorCodeType.HTTP, String.valueOf(e.getCode()),
+                GCS_DOC_URL, e);
           }
           // Precondition failed means file has already been replaced, hence ignore it.
           LOG.debug("Skip uploading file {} to gs://{}/{} because it exists.",
@@ -633,8 +644,8 @@ public LocalFile uploadFile(String bucket, String targetFilePath,
     } catch (StorageException e) {
       if (e.getCode() != HttpURLConnection.HTTP_PRECON_FAILED) {
         ActionErrorPair pair = ErrorUtils.getActionErrorByStatusCode(e.getCode());
-        String errorReason = String.format("%s Unable to access GCS bucket %s. %s", e.getCode(),
-            bucket, pair.getCorrectiveAction());
+        String errorReason = String.format("%s Unable to upload file %s to GCS bucket gs://%s. %s",
+            e.getCode(), localFile.getURI(), bucket, pair.getCorrectiveAction());
         throw ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategoryEnum.STARTING),
             errorReason, e.getMessage(), pair.getErrorType(), true,
             ErrorCodeType.HTTP, String.valueOf(e.getCode()), GCS_DOC_URL, e);