Skip to content

Commit 554f6f5

Browse files
Amit-CloudSufiAmit-CloudSufi
committed
Error management for BigQuery Argument Setter and Execute plugin
1 parent 77e0ac8 commit 554f6f5

File tree

7 files changed

+247
-47
lines changed

7 files changed

+247
-47
lines changed

‎src/main/java/io/cdap/plugin/gcp/bigquery/action/BigQueryArgumentSetter.java

Lines changed: 51 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919
import com.google.auth.Credentials;
2020
import com.google.cloud.StringEnumValue;
2121
import com.google.cloud.bigquery.BigQuery;
22+
import com.google.cloud.bigquery.BigQueryException;
2223
import com.google.cloud.bigquery.Field;
2324
import com.google.cloud.bigquery.FieldValue;
2425
import com.google.cloud.bigquery.FieldValueList;
@@ -33,8 +34,12 @@
3334
import io.cdap.cdap.api.annotation.Description;
3435
import io.cdap.cdap.api.annotation.Name;
3536
import io.cdap.cdap.api.annotation.Plugin;
37+
import io.cdap.cdap.api.exception.ErrorCategory;
38+
import io.cdap.cdap.api.exception.ErrorType;
39+
import io.cdap.cdap.api.exception.ErrorUtils;
3640
import io.cdap.cdap.etl.api.action.Action;
3741
import io.cdap.cdap.etl.api.action.ActionContext;
42+
import io.cdap.plugin.gcp.bigquery.common.BigQueryErrorUtil;
3843
import io.cdap.plugin.gcp.common.GCPUtils;
3944
import org.slf4j.Logger;
4045
import org.slf4j.LoggerFactory;
@@ -70,34 +75,70 @@ public AbstractBigQueryActionConfig getConfig() {
7075
}
7176

7277
@Override
73-
public void run(ActionContext context) throws Exception {
78+
public void run(ActionContext context) {
7479
config.validate(context.getFailureCollector());
7580

7681
QueryJobConfiguration queryConfig = config.getQueryJobConfiguration(context.getFailureCollector());
7782
JobId jobId = JobId.newBuilder().setRandomJob().build();
7883

7984
// API request - starts the query.
80-
Credentials credentials = config.getServiceAccount() == null ?
81-
null : GCPUtils.loadServiceAccountCredentials(config.getServiceAccount(),
82-
config.isServiceAccountFilePath());
85+
Credentials credentials = null;
86+
try {
87+
credentials = config.getServiceAccount() == null ? null :
88+
GCPUtils.loadServiceAccountCredentials(config.getServiceAccount(), config.isServiceAccountFilePath());
89+
} catch (Exception e) {
90+
context.getFailureCollector().addFailure(
91+
String.format("Failed to load service account credentials, %s: %s",
92+
e.getClass().getName(), e.getMessage()), null).withStacktrace(e.getStackTrace());
93+
context.getFailureCollector().getOrThrowException();
94+
}
8395
BigQuery bigQuery = GCPUtils.getBigQuery(config.getProject(), credentials, null);
8496
Job queryJob = bigQuery.create(JobInfo.newBuilder(queryConfig).setJobId(jobId).build());
8597

8698
LOG.info("Executing SQL as job {}.", jobId.getJob());
8799
LOG.debug("The BigQuery SQL {}", queryConfig.getQuery());
88100

89101
// Wait for the query to complete
90-
queryJob.waitFor();
102+
try {
103+
queryJob.waitFor();
104+
} catch (BigQueryException | InterruptedException e) {
105+
String errorMessage = String.format("The query job was interrupted, %s: %s",
106+
e.getClass().getName(), e.getMessage());
107+
if (e instanceof BigQueryException) {
108+
throw BigQueryErrorUtil.getProgramFailureException(errorMessage,
109+
((BigQueryException) e).getReason(), e);
110+
}
111+
throw ErrorUtils.getProgramFailureException(
112+
new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN), errorMessage, errorMessage,
113+
ErrorType.UNKNOWN, true, e);
114+
}
91115

92116
// Check for errors
93117
if (queryJob.getStatus().getError() != null) {
94-
throw new RuntimeException(queryJob.getStatus().getExecutionErrors().toString());
118+
String error = queryJob.getStatus().getExecutionErrors().toString();
119+
ErrorType type = BigQueryErrorUtil.getErrorType(queryJob.getStatus().getError().getReason());
120+
throw ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN),
121+
error, error, type, true, null);
122+
}
123+
TableResult queryResults;
124+
try {
125+
queryResults = queryJob.getQueryResults();
126+
} catch (BigQueryException | InterruptedException e) {
127+
String errorMessage = String.format("The bigquery query job failed, %s: %s",
128+
e.getClass().getName(), e.getMessage());
129+
if (e instanceof BigQueryException) {
130+
throw BigQueryErrorUtil.getProgramFailureException(errorMessage,
131+
((BigQueryException) e).getReason(), e);
132+
}
133+
throw ErrorUtils.getProgramFailureException(
134+
new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN), errorMessage, errorMessage,
135+
ErrorType.UNKNOWN, false, e);
95136
}
96-
97-
TableResult queryResults = queryJob.getQueryResults();
98137
if (queryResults.getTotalRows() == 0 || queryResults.getTotalRows() > 1) {
99-
throw new RuntimeException(String.format("The query result total rows should be \"1\" but is \"%d\"",
100-
queryResults.getTotalRows()));
138+
String error = String.format("The query result total rows should be \"1\" but is \"%d\"",
139+
queryResults.getTotalRows());
140+
throw ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN),
141+
error, error, ErrorType.USER, false, null);
101142
}
102143

103144
Schema schema = queryResults.getSchema();

‎src/main/java/io/cdap/plugin/gcp/bigquery/action/BigQueryArgumentSetterConfig.java

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,9 @@
2727
import io.cdap.cdap.api.annotation.Description;
2828
import io.cdap.cdap.api.annotation.Macro;
2929
import io.cdap.cdap.api.annotation.Name;
30+
import io.cdap.cdap.api.exception.ErrorCategory;
31+
import io.cdap.cdap.api.exception.ErrorType;
32+
import io.cdap.cdap.api.exception.ErrorUtils;
3033
import io.cdap.cdap.etl.api.FailureCollector;
3134
import io.cdap.plugin.common.ConfigUtil;
3235
import io.cdap.plugin.gcp.bigquery.source.BigQuerySource;
@@ -224,9 +227,11 @@ private void checkIfArgumentsColumnsExitsInSource(Map<String, String> argumentCo
224227
String nonExistingColumnNames = argumentConditionMap.keySet().stream()
225228
.filter(columnName -> !argumentConditionFields.containsKey(columnName))
226229
.collect(Collectors.joining(" ,"));
227-
throw new RuntimeException(String.format(
230+
String error = String.format(
228231
"Columns: \" %s \"do not exist in table. Argument selections columns must exist in table.",
229-
nonExistingColumnNames));
232+
nonExistingColumnNames);
233+
throw ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN),
234+
error, error, ErrorType.USER, false, null);
230235
}
231236

232237
static void checkIfArgumentsColumnsListExistsInSource(

‎src/main/java/io/cdap/plugin/gcp/bigquery/action/BigQueryExecute.java

Lines changed: 88 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -44,14 +44,19 @@
4444
import io.cdap.cdap.api.annotation.Macro;
4545
import io.cdap.cdap.api.annotation.Name;
4646
import io.cdap.cdap.api.annotation.Plugin;
47+
import io.cdap.cdap.api.exception.ErrorCategory;
48+
import io.cdap.cdap.api.exception.ErrorType;
49+
import io.cdap.cdap.api.exception.ErrorUtils;
4750
import io.cdap.cdap.etl.api.FailureCollector;
4851
import io.cdap.cdap.etl.api.action.Action;
4952
import io.cdap.cdap.etl.api.action.ActionContext;
5053
import io.cdap.cdap.etl.common.Constants;
54+
import io.cdap.plugin.gcp.bigquery.common.BigQueryErrorUtil;
5155
import io.cdap.plugin.gcp.bigquery.exception.BigQueryJobExecutionException;
5256
import io.cdap.plugin.gcp.bigquery.sink.BigQuerySinkUtils;
5357
import io.cdap.plugin.gcp.bigquery.util.BigQueryUtil;
5458
import io.cdap.plugin.gcp.common.CmekUtils;
59+
import io.cdap.plugin.gcp.common.GCPErrorDetailsProviderUtil;
5560
import io.cdap.plugin.gcp.common.GCPUtils;
5661
import org.slf4j.Logger;
5762
import org.slf4j.LoggerFactory;
@@ -93,7 +98,7 @@ public final class BigQueryExecute extends AbstractBigQueryAction {
9398
}
9499

95100
@Override
96-
public void run(ActionContext context) throws Exception {
101+
public void run(ActionContext context) {
97102
FailureCollector collector = context.getFailureCollector();
98103
config.validate(collector, context.getArguments().asMap());
99104
QueryJobConfiguration.Builder builder = QueryJobConfiguration.newBuilder(config.getSql());
@@ -125,9 +130,16 @@ public void run(ActionContext context) throws Exception {
125130
builder.setUseLegacySql(config.isLegacySQL());
126131

127132
// API request - starts the query.
128-
Credentials credentials = config.getServiceAccount() == null ?
129-
null : GCPUtils.loadServiceAccountCredentials(config.getServiceAccount(),
130-
config.isServiceAccountFilePath());
133+
Credentials credentials = null;
134+
try {
135+
credentials = config.getServiceAccount() == null ?
136+
null : GCPUtils.loadServiceAccountCredentials(config.getServiceAccount(),
137+
config.isServiceAccountFilePath());
138+
} catch (IOException e) {
139+
collector.addFailure(String.format("Failed to load service account credentials, %s: %s",
140+
e.getClass().getName(), e.getMessage()), null).withStacktrace(e.getStackTrace());
141+
collector.getOrThrowException();
142+
}
131143
BigQuery bigQuery = GCPUtils.getBigQuery(config.getProject(), credentials, config.getReadTimeout());
132144
//create dataset to store the results if not exists
133145
if (config.getStoreResults() && !Strings.isNullOrEmpty(datasetName) &&
@@ -152,23 +164,46 @@ public void run(ActionContext context) throws Exception {
152164
try {
153165
executeQueryWithExponentialBackoff(bigQuery, queryConfig, context);
154166
} catch (Throwable e) {
155-
throw new RuntimeException(e);
167+
String errorMessage = String.format(
168+
"Failed to execute query with exponential backoff, %s: %s", e.getClass().getName(),
169+
e.getMessage());
170+
if (e instanceof BigQueryException) {
171+
throw BigQueryErrorUtil.getProgramFailureException(errorMessage,
172+
((BigQueryException) e).getReason(), (Exception) e);
173+
}
174+
throw ErrorUtils.getProgramFailureException(
175+
new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN), errorMessage, errorMessage,
176+
ErrorType.UNKNOWN, true, e);
156177
}
157178
} else {
158-
executeQuery(bigQuery, queryConfig, context);
179+
try {
180+
executeQuery(bigQuery, queryConfig, context);
181+
} catch (Exception e) {
182+
String errorMessage = String.format("Failed to execute query, %s: %s",
183+
e.getClass().getName(), e.getMessage());
184+
String errorReason = null;
185+
if (e instanceof BigQueryException) {
186+
errorReason = ((BigQueryException) e).getReason();
187+
}
188+
throw BigQueryErrorUtil.getProgramFailureException(errorMessage, errorReason, e);
189+
}
159190
}
160191
}
161192

162193
protected void executeQueryWithExponentialBackoff(BigQuery bigQuery,
163-
QueryJobConfiguration queryConfig, ActionContext context)
164-
throws Throwable {
194+
QueryJobConfiguration queryConfig, ActionContext context) {
165195
try {
166196
Failsafe.with(getRetryPolicy()).run(() -> executeQuery(bigQuery, queryConfig, context));
167197
} catch (FailsafeException e) {
198+
String errorReason = String.format("Failed to execute query with message: %s",
199+
e.getMessage());
168200
if (e.getCause() != null) {
169-
throw e.getCause();
201+
errorReason = String.format("Failed to execute query with message: %s",
202+
e.getCause().getMessage());
170203
}
171-
throw e;
204+
throw GCPErrorDetailsProviderUtil.getHttpResponseExceptionDetailsFromChain(
205+
e == null ? e : e.getCause(), errorReason, ErrorType.UNKNOWN, true,
206+
GCPUtils.BQ_SUPPORTED_DOC_URL);
172207
}
173208
}
174209

@@ -185,7 +220,7 @@ private RetryPolicy<Object> getRetryPolicy() {
185220
}
186221

187222
private void executeQuery(BigQuery bigQuery, QueryJobConfiguration queryConfig, ActionContext context)
188-
throws InterruptedException, BigQueryJobExecutionException {
223+
throws BigQueryJobExecutionException {
189224
// Location must match that of the dataset(s) referenced in the query.
190225
JobId jobId = JobId.newBuilder().setRandomJob().setLocation(config.getLocation()).build();
191226
Job queryJob;
@@ -198,12 +233,22 @@ private void executeQuery(BigQuery bigQuery, QueryJobConfiguration queryConfig,
198233

199234
// Wait for the query to complete
200235
queryJob = queryJob.waitFor();
201-
} catch (BigQueryException e) {
202-
LOG.error("The query job {} failed. Error: {}", jobId.getJob(), e.getError().getMessage());
203-
if (RETRY_ON_REASON.contains(e.getError().getReason())) {
204-
throw new BigQueryJobExecutionException(e.getError().getMessage(), e);
236+
} catch (BigQueryException | InterruptedException e) {
237+
String errorMessage = String.format("Failed to execute query, %s: %s", e.getClass().getName(),
238+
e.getMessage());
239+
if (e instanceof BigQueryException) {
240+
LOG.error("The query job {} failed. Error: {}", jobId.getJob(),
241+
((BigQueryException) e).getError().getMessage());
242+
if (RETRY_ON_REASON.contains(((BigQueryException) e).getError().getReason())) {
243+
throw new BigQueryJobExecutionException(((BigQueryException) e).getError().getMessage(),
244+
e);
245+
}
246+
throw BigQueryErrorUtil.getProgramFailureException(errorMessage,
247+
((BigQueryException) e).getReason(), e);
205248
}
206-
throw new RuntimeException(e);
249+
throw ErrorUtils.getProgramFailureException(
250+
new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN), errorMessage, errorMessage,
251+
ErrorType.UNKNOWN, true, e);
207252
}
208253

209254
// Check for errors
@@ -214,10 +259,29 @@ private void executeQuery(BigQuery bigQuery, QueryJobConfiguration queryConfig,
214259
if (RETRY_ON_REASON.contains(queryJob.getStatus().getError().getReason())) {
215260
throw new BigQueryJobExecutionException(queryJob.getStatus().getError().getMessage());
216261
}
217-
throw new RuntimeException(queryJob.getStatus().getError().getMessage());
262+
String error = String.format("Failed to execute query with reason: %s and message: %s",
263+
queryJob.getStatus().getError().getReason(),
264+
queryJob.getStatus().getError().getMessage());
265+
ErrorType type = BigQueryErrorUtil.getErrorType(queryJob.getStatus().getError().getReason());
266+
throw ErrorUtils.getProgramFailureException(
267+
new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN), error, error, type, true,
268+
null);
218269
}
219270

220-
TableResult queryResults = queryJob.getQueryResults();
271+
TableResult queryResults;
272+
try {
273+
queryResults = queryJob.getQueryResults();
274+
} catch (BigQueryException | InterruptedException e) {
275+
String errorMessage = String.format("Failed to retrieve query result, %s: %s",
276+
e.getClass().getName(), e.getMessage());
277+
if (e instanceof BigQueryException) {
278+
throw BigQueryErrorUtil.getProgramFailureException(errorMessage,
279+
((BigQueryException) e).getReason(), e);
280+
}
281+
throw ErrorUtils.getProgramFailureException(
282+
new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN), errorMessage, errorMessage,
283+
ErrorType.UNKNOWN, true, e);
284+
}
221285
long rows = queryResults.getTotalRows();
222286

223287
if (config.shouldSetAsArguments()) {
@@ -659,11 +723,12 @@ public void validateSQLSyntax(FailureCollector failureCollector, BigQuery bigQue
659723
bigQuery.create(JobInfo.of(queryJobConfiguration));
660724
} catch (BigQueryException e) {
661725
final String errorMessage;
662-
if (e.getCode() == ERROR_CODE_NOT_FOUND) {
663-
errorMessage = String.format("Resource was not found. Please verify the resource name. If the resource " +
664-
"will be created at runtime, then update to use a macro for the resource name. Error message received " +
665-
"was: %s", e.getMessage());
666-
} else {
726+
if (e.getCode() == ERROR_CODE_NOT_FOUND) {
727+
errorMessage = String.format(
728+
"Resource was not found. Please verify the resource name. If the resource will be "
729+
+ "created at runtime, then update to use a macro for the resource name. "
730+
+ "Error message received was %s, %s", e.getClass().getName(), e.getMessage());
731+
} else {
667732
errorMessage = e.getMessage();
668733
}
669734
failureCollector.addFailure(String.format("%s. Error code: %s.", errorMessage, e.getCode()),

0 commit comments

Comments
 (0)