Skip to content

Commit 3c06ecb

Browse files
psainicspsainics
committed
TEST
1 parent 601f62b commit 3c06ecb

File tree

4 files changed

+68
-22
lines changed

4 files changed

+68
-22
lines changed

‎src/main/java/io/cdap/plugin/gcp/bigquery/source/BigQueryAvroToStructuredTransformer.java‎

Lines changed: 9 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,9 @@
2020
import io.cdap.cdap.api.data.format.StructuredRecord;
2121
import io.cdap.cdap.api.data.format.UnexpectedFormatException;
2222
import io.cdap.cdap.api.data.schema.Schema;
23+
import io.cdap.cdap.api.exception.ErrorCategory;
24+
import io.cdap.cdap.api.exception.ErrorType;
25+
import io.cdap.cdap.api.exception.ErrorUtils;
2326
import io.cdap.plugin.common.RecordConverter;
2427
import org.apache.avro.generic.GenericRecord;
2528

@@ -90,11 +93,10 @@ protected Object convertField(Object field, Schema fieldSchema) throws IOExcepti
9093
try {
9194
LocalDateTime.parse(field.toString());
9295
} catch (DateTimeParseException exception) {
93-
throw new UnexpectedFormatException(
94-
String.format("Datetime field with value '%s' is not in ISO-8601 format.",
95-
fieldSchema.getDisplayName(),
96-
field.toString()),
97-
exception);
96+
String errorMessage = String.format("Datetime field %s with value '%s' is not in ISO-8601 format.",
97+
fieldSchema.getDisplayName(), field);
98+
throw ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN),
99+
errorMessage, exception.getMessage(), ErrorType.UNKNOWN, true, exception);
98100
}
99101
//If properly formatted return the string
100102
return field.toString();
@@ -110,7 +112,8 @@ protected Object convertField(Object field, Schema fieldSchema) throws IOExcepti
110112
}
111113
}
112114
} catch (ArithmeticException e) {
113-
throw new IOException("Field type %s has value that is too large." + fieldType);
115+
throw ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN),
116+
"Field type %s has value that is too large.", e.getMessage(), ErrorType.UNKNOWN, true, e);
114117
}
115118

116119
// Complex types like maps and unions are not supported in BigQuery plugins.

‎src/main/java/io/cdap/plugin/gcp/bigquery/source/BigQuerySource.java‎

Lines changed: 47 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,7 @@
4848
import io.cdap.cdap.etl.api.batch.BatchSourceContext;
4949
import io.cdap.cdap.etl.api.connector.Connector;
5050
import io.cdap.cdap.etl.api.engine.sql.SQLEngineInput;
51+
import io.cdap.cdap.etl.api.exception.ErrorDetailsProviderSpec;
5152
import io.cdap.cdap.etl.api.validation.ValidationFailure;
5253
import io.cdap.plugin.common.Asset;
5354
import io.cdap.plugin.common.LineageRecorder;
@@ -57,6 +58,7 @@
5758
import io.cdap.plugin.gcp.bigquery.util.BigQueryConstants;
5859
import io.cdap.plugin.gcp.bigquery.util.BigQueryUtil;
5960
import io.cdap.plugin.gcp.common.CmekUtils;
61+
import io.cdap.plugin.gcp.common.GCPErrorDetailsProvider;
6062
import io.cdap.plugin.gcp.common.GCPUtils;
6163
import org.apache.avro.generic.GenericData;
6264
import org.apache.hadoop.conf.Configuration;
@@ -135,7 +137,17 @@ public void prepareRun(BatchSourceContext context) throws Exception {
135137

136138
// Create BigQuery client
137139
String serviceAccount = config.getServiceAccount();
138-
Credentials credentials = BigQuerySourceUtils.getCredentials(config.getConnection());
140+
Credentials credentials = null;
141+
try {
142+
credentials = BigQuerySourceUtils.getCredentials(config.getConnection());
143+
} catch (Exception e) {
144+
String errorReason = "Unable to load service account credentials.";
145+
collector.addFailure(String.format("%s %s", errorReason, e.getMessage()), null)
146+
.withStacktrace(e.getStackTrace());
147+
collector.getOrThrowException();
148+
}
149+
150+
139151
BigQuery bigQuery = GCPUtils.getBigQuery(config.getProject(), credentials, null);
140152
Dataset dataset = bigQuery.getDataset(DatasetId.of(config.getDatasetProject(), config.getDataset()));
141153
Storage storage = GCPUtils.getStorage(config.getProject(), credentials);
@@ -144,19 +156,30 @@ public void prepareRun(BatchSourceContext context) throws Exception {
144156
bucketPath = UUID.randomUUID().toString();
145157
CryptoKeyName cmekKeyName = CmekUtils.getCmekKey(config.cmekKey, context.getArguments().asMap(), collector);
146158
collector.getOrThrowException();
147-
configuration = BigQueryUtil.getBigQueryConfig(serviceAccount, config.getProject(), cmekKeyName,
148-
config.getServiceAccountType());
159+
try {
160+
configuration = BigQueryUtil.getBigQueryConfig(serviceAccount, config.getProject(), cmekKeyName,
161+
config.getServiceAccountType());
162+
} catch (Exception e) {
163+
String errorReason = "Failed to create BigQuery configuration.";
164+
collector.addFailure(String.format("%s %s", errorReason, e.getMessage()), null)
165+
.withStacktrace(e.getStackTrace());
166+
collector.getOrThrowException();
167+
}
149168

150169
String bucketName = BigQueryUtil.getStagingBucketName(context.getArguments().asMap(), null,
151170
dataset, config.getBucket());
152171

153172
// Configure GCS Bucket to use
154-
String bucket = BigQuerySourceUtils.getOrCreateBucket(configuration,
155-
storage,
156-
bucketName,
157-
dataset,
158-
bucketPath,
159-
cmekKeyName);
173+
String bucket = null;
174+
try {
175+
bucket = BigQuerySourceUtils.getOrCreateBucket(configuration, storage, bucketName, dataset, bucketPath,
176+
cmekKeyName);
177+
} catch (Exception e) {
178+
String errorReason = "Failed to create bucket.";
179+
collector.addFailure(String.format("%s %s", errorReason, e.getMessage()), null)
180+
.withStacktrace(e.getStackTrace());
181+
collector.getOrThrowException();
182+
}
160183

161184
// Configure Service account credentials
162185
BigQuerySourceUtils.configureServiceAccount(configuration, config.getConnection());
@@ -166,10 +189,17 @@ public void prepareRun(BatchSourceContext context) throws Exception {
166189

167190
// Configure BigQuery input format.
168191
String temporaryGcsPath = BigQuerySourceUtils.getTemporaryGcsPath(bucket, bucketPath, bucketPath);
169-
BigQuerySourceUtils.configureBigQueryInput(configuration,
170-
DatasetId.of(config.getDatasetProject(), config.getDataset()),
171-
config.getTable(),
172-
temporaryGcsPath);
192+
try {
193+
BigQuerySourceUtils.configureBigQueryInput(configuration,
194+
DatasetId.of(config.getDatasetProject(), config.getDataset()),
195+
config.getTable(),
196+
temporaryGcsPath);
197+
} catch (Exception e) {
198+
String errorReason = "Failed to configure BigQuery input.";
199+
collector.addFailure(String.format("%s %s", errorReason, e.getMessage()), null)
200+
.withStacktrace(e.getStackTrace());
201+
collector.getOrThrowException();
202+
}
173203

174204
// Both emitLineage and setOutputFormat internally try to create an external dataset if it does not already exists.
175205
// We call emitLineage before since it creates the dataset with schema.
@@ -178,6 +208,10 @@ public void prepareRun(BatchSourceContext context) throws Exception {
178208
.setFqn(BigQueryUtil.getFQN(config.getDatasetProject(), config.getDataset(), config.getTable()))
179209
.setLocation(dataset.getLocation())
180210
.build();
211+
212+
// set error details provider
213+
context.setErrorDetailsProvider(new ErrorDetailsProviderSpec(GCPErrorDetailsProvider.class.getName()));
214+
181215
emitLineage(context, configuredSchema, sourceTableType, config.getTable(), asset);
182216
setInputFormat(context, configuredSchema);
183217
}

‎src/main/java/io/cdap/plugin/gcp/bigquery/source/BigQuerySourceUtils.java‎

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,9 @@
2525
import com.google.cloud.kms.v1.CryptoKeyName;
2626
import com.google.cloud.storage.Storage;
2727
import com.google.cloud.storage.StorageException;
28+
import io.cdap.cdap.api.exception.ErrorCategory;
29+
import io.cdap.cdap.api.exception.ErrorType;
30+
import io.cdap.cdap.api.exception.ErrorUtils;
2831
import io.cdap.plugin.gcp.bigquery.connector.BigQueryConnectorConfig;
2932
import io.cdap.plugin.gcp.bigquery.util.BigQueryConstants;
3033
import io.cdap.plugin.gcp.bigquery.util.BigQueryUtil;
@@ -93,10 +96,12 @@ public static String getOrCreateBucket(Configuration configuration,
9396
// Ignore this and move on, since all that matters is that the bucket exists.
9497
return bucket;
9598
}
96-
throw new IOException(String.format("Unable to create Cloud Storage bucket '%s' in the same " +
99+
String errorMessage = String.format("Unable to create Cloud Storage bucket '%s' in the same " +
97100
"location ('%s') as BigQuery dataset '%s'. " + "Please use a bucket " +
98101
"that is in the same location as the dataset.",
99-
bucket, dataset.getLocation(), dataset.getDatasetId().getDataset()), e);
102+
bucket, dataset.getLocation(), dataset.getDatasetId().getDataset());
103+
throw ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN),
104+
errorMessage, e.getMessage(), ErrorType.UNKNOWN, true, e);
100105
}
101106
}
102107

‎src/main/java/io/cdap/plugin/gcp/bigquery/source/PartitionedBigQueryInputFormat.java‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,9 @@
3737
import com.google.common.annotations.VisibleForTesting;
3838
import com.google.common.base.Preconditions;
3939
import com.google.common.base.Strings;
40+
import io.cdap.cdap.api.exception.ErrorCategory;
41+
import io.cdap.cdap.api.exception.ErrorType;
42+
import io.cdap.cdap.api.exception.ErrorUtils;
4043
import io.cdap.plugin.gcp.bigquery.util.BigQueryConstants;
4144
import io.cdap.plugin.gcp.bigquery.util.BigQueryUtil;
4245
import io.cdap.plugin.gcp.common.GCPUtils;
@@ -110,7 +113,8 @@ private void processQuery(JobContext context) throws IOException, InterruptedExc
110113
try {
111114
bigQueryHelper = getBigQueryHelper(configuration);
112115
} catch (GeneralSecurityException gse) {
113-
throw new IOException("Failed to create BigQuery client", gse);
116+
throw ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN),
117+
"Failed to create BigQuery client", gse.getMessage(), ErrorType.UNKNOWN, true, gse);
114118
}
115119

116120
List<HadoopConfigurationProperty<?>> hadoopConfigurationProperties = new ArrayList<>(

0 commit comments

Comments
 (0)