Skip to content

Commit 283824a

Browse files
itsankit-googleitsankit-google
authored
Merge pull request #1452 from data-integrations/PLUGIN-1807-3
[PLUGIN-1807] Implement error details provider to get more information about exceptions from GCP plugins
2 parents a1a4049 + 77bdf7e commit 283824a

File tree

6 files changed

+253
-52
lines changed

6 files changed

+253
-52
lines changed

‎src/main/java/io/cdap/plugin/gcp/common/GCPErrorDetailsProvider.java‎

Lines changed: 84 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,84 @@
1+
/*
2+
* Copyright © 2024 Cask Data, Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License"); you may not
5+
* use this file except in compliance with the License. You may obtain a copy of
6+
* the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12+
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13+
* License for the specific language governing permissions and limitations under
14+
* the License.
15+
*/
16+
17+
package io.cdap.plugin.gcp.common;
18+
19+
import com.google.api.client.googleapis.json.GoogleJsonResponseException;
20+
import com.google.api.client.http.HttpResponseException;
21+
import com.google.common.base.Throwables;
22+
import io.cdap.cdap.api.exception.ErrorCategory;
23+
import io.cdap.cdap.api.exception.ErrorCategory.ErrorCategoryEnum;
24+
import io.cdap.cdap.api.exception.ErrorUtils;
25+
import io.cdap.cdap.api.exception.ProgramFailureException;
26+
import io.cdap.cdap.etl.api.exception.ErrorContext;
27+
import io.cdap.cdap.etl.api.exception.ErrorDetailsProvider;
28+
29+
import java.util.List;
30+
31+
/**
32+
* A custom ErrorDetailsProvider for GCP plugins.
33+
*/
34+
public class GCPErrorDetailsProvider implements ErrorDetailsProvider {
35+
36+
/**
37+
* Get a ProgramFailureException with the given error
38+
* information from generic exceptions like {@link java.io.IOException}.
39+
*
40+
* @param e The Throwable to get the error information from.
41+
* @return A ProgramFailureException with the given error information, otherwise null.
42+
*/
43+
@Override
44+
public ProgramFailureException getExceptionDetails(Exception e, ErrorContext errorContext) {
45+
List<Throwable> causalChain = Throwables.getCausalChain(e);
46+
for (Throwable t : causalChain) {
47+
if (t instanceof ProgramFailureException) {
48+
// if causal chain already has program failure exception, return null to avoid double wrap.
49+
return null;
50+
}
51+
if (t instanceof HttpResponseException) {
52+
return getProgramFailureException((HttpResponseException) t, errorContext);
53+
}
54+
}
55+
return null;
56+
}
57+
58+
/**
59+
* Get a ProgramFailureException with the given error
60+
* information from {@link HttpResponseException}.
61+
*
62+
* @param e The HttpResponseException to get the error information from.
63+
* @return A ProgramFailureException with the given error information.
64+
*/
65+
private ProgramFailureException getProgramFailureException(HttpResponseException e,
66+
ErrorContext errorContext) {
67+
Integer statusCode = e.getStatusCode();
68+
ErrorUtils.ActionErrorPair pair = ErrorUtils.getActionErrorByStatusCode(statusCode);
69+
String errorReason = String.format("%s %s %s", e.getStatusCode(), e.getStatusMessage(),
70+
pair.getCorrectiveAction());
71+
String errorMessageFormat = "Error occurred in the phase: '%s'. Error message: %s";
72+
73+
String errorMessage = e.getMessage();
74+
if (e instanceof GoogleJsonResponseException) {
75+
GoogleJsonResponseException exception = (GoogleJsonResponseException) e;
76+
errorMessage = exception.getDetails() != null ? exception.getDetails().getMessage() :
77+
exception.getMessage();
78+
}
79+
80+
return ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategoryEnum.PLUGIN),
81+
errorReason, String.format(errorMessageFormat, errorContext.getPhase(), errorMessage),
82+
pair.getErrorType(), true, e);
83+
}
84+
}

‎src/main/java/io/cdap/plugin/gcp/gcs/ServiceAccountAccessTokenProvider.java‎

Lines changed: 13 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,10 @@
2121
import com.google.bigtable.repackaged.com.google.gson.Gson;
2222
import com.google.cloud.hadoop.util.AccessTokenProvider;
2323
import com.google.cloud.hadoop.util.CredentialFactory;
24+
import io.cdap.cdap.api.exception.ErrorCategory;
25+
import io.cdap.cdap.api.exception.ErrorCategory.ErrorCategoryEnum;
26+
import io.cdap.cdap.api.exception.ErrorType;
27+
import io.cdap.cdap.api.exception.ErrorUtils;
2428
import io.cdap.plugin.gcp.common.GCPUtils;
2529
import org.apache.hadoop.conf.Configuration;
2630

@@ -50,13 +54,20 @@ public AccessToken getAccessToken() {
5054
}
5155
return new AccessToken(token.getTokenValue(), token.getExpirationTime().getTime());
5256
} catch (IOException e) {
53-
throw new RuntimeException(e);
57+
throw ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategoryEnum.PLUGIN),
58+
"Unable to get service account access token.", e.getMessage(), ErrorType.UNKNOWN, true, e);
5459
}
5560
}
5661

5762
@Override
5863
public void refresh() throws IOException {
59-
getCredentials().refresh();
64+
try {
65+
getCredentials().refresh();
66+
} catch (IOException e) {
67+
throw ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategoryEnum.PLUGIN),
68+
"Unable to refresh service account access token.", e.getMessage(),
69+
ErrorType.UNKNOWN, true, e);
70+
}
6071
}
6172

6273
private GoogleCredentials getCredentials() throws IOException {

‎src/main/java/io/cdap/plugin/gcp/gcs/sink/GCSBatchSink.java‎

Lines changed: 52 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,7 @@
3939
import io.cdap.cdap.etl.api.batch.BatchSink;
4040
import io.cdap.cdap.etl.api.batch.BatchSinkContext;
4141
import io.cdap.cdap.etl.api.connector.Connector;
42+
import io.cdap.cdap.etl.api.exception.ErrorDetailsProviderSpec;
4243
import io.cdap.cdap.etl.api.validation.ValidatingOutputFormat;
4344
import io.cdap.plugin.common.Asset;
4445
import io.cdap.plugin.common.ConfigUtil;
@@ -51,6 +52,7 @@
5152
import io.cdap.plugin.format.plugin.FileSinkProperties;
5253
import io.cdap.plugin.gcp.common.CmekUtils;
5354
import io.cdap.plugin.gcp.common.GCPConnectorConfig;
55+
import io.cdap.plugin.gcp.common.GCPErrorDetailsProvider;
5456
import io.cdap.plugin.gcp.common.GCPUtils;
5557
import io.cdap.plugin.gcp.gcs.Formats;
5658
import io.cdap.plugin.gcp.gcs.GCSPath;
@@ -121,30 +123,50 @@ public void prepareRun(BatchSinkContext context) throws Exception {
121123
collector.addFailure("Service account type is undefined.",
122124
"Must be `filePath` or `JSON`");
123125
collector.getOrThrowException();
124-
return;
125126
}
126-
Credentials credentials = config.connection.getServiceAccount() == null ?
127-
null : GCPUtils.loadServiceAccountCredentials(config.connection.getServiceAccount(), isServiceAccountFilePath);
127+
128+
Credentials credentials = null;
129+
try {
130+
credentials = config.connection.getServiceAccount() == null ?
131+
null : GCPUtils.loadServiceAccountCredentials(config.connection.getServiceAccount(),
132+
isServiceAccountFilePath);
133+
} catch (Exception e) {
134+
String errorReason = "Unable to load service account credentials.";
135+
collector.addFailure(String.format("%s %s", errorReason, e.getMessage()), null)
136+
.withStacktrace(e.getStackTrace());
137+
collector.getOrThrowException();
138+
}
139+
140+
String bucketName = config.getBucket(collector);
128141
Storage storage = GCPUtils.getStorage(config.connection.getProject(), credentials);
142+
String errorReasonFormat = "Error code: %s, Unable to read or access GCS bucket.";
143+
String correctiveAction = "Ensure you entered the correct bucket path and "
144+
+ "have permissions for it.";
129145
Bucket bucket;
130-
String location;
146+
String location = null;
131147
try {
132-
bucket = storage.get(config.getBucket());
148+
bucket = storage.get(bucketName);
149+
if (bucket != null) {
150+
location = bucket.getLocation();
151+
} else {
152+
location = config.getLocation();
153+
GCPUtils.createBucket(storage, bucketName, location, cmekKeyName);
154+
}
133155
} catch (StorageException e) {
134-
throw new RuntimeException(
135-
String.format("Unable to access or create bucket %s. ", config.getBucket())
136-
+ "Ensure you entered the correct bucket path and have permissions for it.", e);
137-
}
138-
if (bucket != null) {
139-
location = bucket.getLocation();
140-
} else {
141-
GCPUtils.createBucket(storage, config.getBucket(), config.getLocation(), cmekKeyName);
142-
location = config.getLocation();
156+
String errorReason = String.format(errorReasonFormat, e.getCode());
157+
collector.addFailure(String.format("%s %s", errorReason, e.getMessage()), correctiveAction)
158+
.withStacktrace(e.getStackTrace());
159+
collector.getOrThrowException();
143160
}
161+
144162
this.outputPath = getOutputDir(context);
145163
// create asset for lineage
146164
asset = Asset.builder(config.getReferenceName())
147165
.setFqn(GCSPath.getFQN(config.getPath())).setLocation(location).build();
166+
167+
// set error details provider
168+
context.setErrorDetailsProvider(
169+
new ErrorDetailsProviderSpec(GCPErrorDetailsProvider.class.getName()));
148170

149171
// super is called down here to avoid instantiating the lineage recorder with a null asset
150172
super.prepareRun(context);
@@ -532,8 +554,20 @@ public void validateContentType(FailureCollector failureCollector) {
532554
}
533555
}
534556

535-
public String getBucket() {
536-
return GCSPath.from(path).getBucket();
557+
/**
558+
* Get the bucket name from the path.
559+
* @param collector failure collector
560+
* @return bucket name as {@link String} if found, otherwise null.
561+
*/
562+
public String getBucket(FailureCollector collector) {
563+
try {
564+
return GCSPath.from(path).getBucket();
565+
} catch (IllegalArgumentException e) {
566+
collector.addFailure(e.getMessage(), null)
567+
.withStacktrace(e.getStackTrace());
568+
collector.getOrThrowException();
569+
}
570+
return null;
537571
}
538572

539573
@Override
@@ -718,8 +752,8 @@ public Builder setCustomContentType(@Nullable String customContentType) {
718752
return this;
719753
}
720754

721-
public GCSBatchSink.GCSBatchSinkConfig build() {
722-
return new GCSBatchSink.GCSBatchSinkConfig(
755+
public GCSBatchSinkConfig build() {
756+
return new GCSBatchSinkConfig(
723757
referenceName,
724758
project,
725759
fileSystemProperties,

‎src/main/java/io/cdap/plugin/gcp/gcs/sink/GCSMultiBatchSink.java‎

Lines changed: 34 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -41,10 +41,12 @@
4141
import io.cdap.cdap.etl.api.batch.BatchSink;
4242
import io.cdap.cdap.etl.api.batch.BatchSinkContext;
4343
import io.cdap.cdap.etl.api.connector.Connector;
44+
import io.cdap.cdap.etl.api.exception.ErrorDetailsProviderSpec;
4445
import io.cdap.cdap.etl.api.validation.ValidatingOutputFormat;
4546
import io.cdap.plugin.common.batch.sink.SinkOutputFormatProvider;
4647
import io.cdap.plugin.format.FileFormat;
4748
import io.cdap.plugin.gcp.common.CmekUtils;
49+
import io.cdap.plugin.gcp.common.GCPErrorDetailsProvider;
4850
import io.cdap.plugin.gcp.common.GCPUtils;
4951
import io.cdap.plugin.gcp.gcs.connector.GCSConnector;
5052
import org.apache.hadoop.io.NullWritable;
@@ -129,33 +131,52 @@ public void prepareRun(BatchSinkContext context) throws IOException, Instantiati
129131
config.validate(collector, context.getArguments().asMap());
130132
collector.getOrThrowException();
131133

132-
Map<String, String> baseProperties = GCPUtils.getFileSystemProperties(config.connection,
133-
config.getPath(), new HashMap<>());
134134
Map<String, String> argumentCopy = new HashMap<>(context.getArguments().asMap());
135135

136136
CryptoKeyName cmekKeyName = CmekUtils.getCmekKey(config.cmekKey, context.getArguments().asMap(), collector);
137137
collector.getOrThrowException();
138+
138139
Boolean isServiceAccountFilePath = config.connection.isServiceAccountFilePath();
139140
if (isServiceAccountFilePath == null) {
140-
context.getFailureCollector().addFailure("Service account type is undefined.",
141+
collector.addFailure("Service account type is undefined.",
141142
"Must be `filePath` or `JSON`");
142-
context.getFailureCollector().getOrThrowException();
143-
return;
143+
collector.getOrThrowException();
144144
}
145-
Credentials credentials = config.connection.getServiceAccount() == null ?
146-
null : GCPUtils.loadServiceAccountCredentials(config.connection.getServiceAccount(), isServiceAccountFilePath);
145+
146+
Credentials credentials = null;
147+
try {
148+
credentials = config.connection.getServiceAccount() == null ?
149+
null : GCPUtils.loadServiceAccountCredentials(config.connection.getServiceAccount(),
150+
isServiceAccountFilePath);
151+
} catch (Exception e) {
152+
String errorReason = "Unable to load service account credentials.";
153+
collector.addFailure(String.format("%s %s", errorReason, e.getMessage()), null)
154+
.withStacktrace(e.getStackTrace());
155+
collector.getOrThrowException();
156+
}
157+
158+
String bucketName = config.getBucket(collector);
147159
Storage storage = GCPUtils.getStorage(config.connection.getProject(), credentials);
160+
String errorReasonFormat = "Error code: %s, Unable to read or access GCS bucket.";
161+
String correctiveAction = "Ensure you entered the correct bucket path and "
162+
+ "have permissions for it.";
148163
try {
149-
if (storage.get(config.getBucket()) == null) {
150-
GCPUtils.createBucket(storage, config.getBucket(), config.getLocation(), cmekKeyName);
164+
if (storage.get(bucketName) == null) {
165+
GCPUtils.createBucket(storage, bucketName, config.getLocation(), cmekKeyName);
151166
}
152167
} catch (StorageException e) {
153-
// Add more descriptive error message
154-
throw new RuntimeException(
155-
String.format("Unable to access or create bucket %s. ", config.getBucket())
156-
+ "Ensure you entered the correct bucket path and have permissions for it.", e);
168+
String errorReason = String.format(errorReasonFormat, e.getCode());
169+
collector.addFailure(String.format("%s %s", errorReason, e.getMessage()), correctiveAction)
170+
.withStacktrace(e.getStackTrace());
171+
collector.getOrThrowException();
157172
}
158173

174+
// set error details provider
175+
context.setErrorDetailsProvider(
176+
new ErrorDetailsProviderSpec(GCPErrorDetailsProvider.class.getName()));
177+
178+
Map<String, String> baseProperties = GCPUtils.getFileSystemProperties(config.connection,
179+
config.getPath(), new HashMap<>());
159180
if (config.getAllowFlexibleSchema()) {
160181
//Configure MultiSink with support for flexible schemas.
161182
configureSchemalessMultiSink(context, baseProperties, argumentCopy);

0 commit comments

Comments
 (0)