Skip to content

Commit 23af441

Browse files
psainicspsainics
authored
Merge pull request #523 from cloudsufi/fem/mysql
[PLUGIN-1824] ErrorDetailsProvider - MySql Source/Sink plugin
2 parents 91d0ff7 + 26dc98e commit 23af441

File tree

13 files changed

+250
-9
lines changed

13 files changed

+250
-9
lines changed

‎database-commons/src/main/java/io/cdap/plugin/db/DBErrorDetailsProvider.java‎

Lines changed: 124 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,124 @@
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.db;
18+
19+
import com.google.common.base.Strings;
20+
import com.google.common.base.Throwables;
21+
import io.cdap.cdap.api.exception.ErrorCategory;
22+
import io.cdap.cdap.api.exception.ErrorType;
23+
import io.cdap.cdap.api.exception.ErrorUtils;
24+
import io.cdap.cdap.api.exception.ProgramFailureException;
25+
import io.cdap.cdap.etl.api.exception.ErrorContext;
26+
import io.cdap.cdap.etl.api.exception.ErrorDetailsProvider;
27+
28+
import java.sql.SQLException;
29+
import java.util.List;
30+
31+
/**
32+
* A custom ErrorDetailsProvider for Database plugins.
33+
*/
34+
public class DBErrorDetailsProvider implements ErrorDetailsProvider {
35+
36+
public ProgramFailureException getExceptionDetails(Exception e, ErrorContext errorContext) {
37+
List<Throwable> causalChain = Throwables.getCausalChain(e);
38+
for (Throwable t : causalChain) {
39+
if (t instanceof ProgramFailureException) {
40+
// if causal chain already has program failure exception, return null to avoid double wrap.
41+
return null;
42+
}
43+
if (t instanceof SQLException) {
44+
return getProgramFailureException((SQLException) t, errorContext);
45+
}
46+
if (t instanceof IllegalArgumentException) {
47+
return getProgramFailureException((IllegalArgumentException) t, errorContext);
48+
}
49+
if (t instanceof IllegalStateException) {
50+
return getProgramFailureException((IllegalStateException) t, errorContext);
51+
}
52+
}
53+
return null;
54+
}
55+
56+
/**
57+
* Get a ProgramFailureException with the given error
58+
* information from {@link SQLException}.
59+
*
60+
* @param e The SQLException to get the error information from.
61+
* @return A ProgramFailureException with the given error information.
62+
*/
63+
private ProgramFailureException getProgramFailureException(SQLException e, ErrorContext errorContext) {
64+
String errorMessage = e.getMessage();
65+
String sqlState = e.getSQLState();
66+
int errorCode = e.getErrorCode();
67+
String errorMessageWithDetails = String.format(
68+
"Error occurred in the phase: '%s'. Error message: '%s'. Error code: '%s'. sqlState: '%s'",
69+
errorContext.getPhase(), errorMessage, errorCode, sqlState);
70+
String externalDocumentationLink = getExternalDocumentationLink();
71+
if (!Strings.isNullOrEmpty(externalDocumentationLink)) {
72+
if (!errorMessageWithDetails.endsWith(".")) {
73+
errorMessageWithDetails = errorMessageWithDetails + ".";
74+
}
75+
errorMessageWithDetails = String.format("%s For more details, see %s", errorMessageWithDetails,
76+
externalDocumentationLink);
77+
}
78+
return ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN),
79+
errorMessage, errorMessageWithDetails, getErrorTypeFromErrorCode(errorCode), false, e);
80+
}
81+
82+
/**
83+
* Get a ProgramFailureException with the given error
84+
* information from {@link IllegalArgumentException}.
85+
*
86+
* @param e The IllegalArgumentException to get the error information from.
87+
* @return A ProgramFailureException with the given error information.
88+
*/
89+
private ProgramFailureException getProgramFailureException(IllegalArgumentException e, ErrorContext errorContext) {
90+
String errorMessage = e.getMessage();
91+
String errorMessageFormat = "Error occurred in the phase: '%s'. Error message: %s";
92+
return ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN),
93+
errorMessage,
94+
String.format(errorMessageFormat, errorContext.getPhase(), errorMessage), ErrorType.USER, false, e);
95+
}
96+
97+
/**
98+
* Get a ProgramFailureException with the given error
99+
* information from {@link IllegalStateException}.
100+
*
101+
* @param e The IllegalStateException to get the error information from.
102+
* @return A ProgramFailureException with the given error information.
103+
*/
104+
private ProgramFailureException getProgramFailureException(IllegalStateException e, ErrorContext errorContext) {
105+
String errorMessage = e.getMessage();
106+
String errorMessageFormat = "Error occurred in the phase: '%s'. Error message: %s";
107+
return ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN),
108+
errorMessage,
109+
String.format(errorMessageFormat, errorContext.getPhase(), errorMessage), ErrorType.SYSTEM, false, e);
110+
}
111+
112+
/**
113+
* Get the external documentation link for the client errors if available.
114+
*
115+
* @return The external documentation link as a {@link String}.
116+
*/
117+
protected String getExternalDocumentationLink() {
118+
return null;
119+
}
120+
121+
protected ErrorType getErrorTypeFromErrorCode(int errorCode) {
122+
return ErrorType.UNKNOWN;
123+
}
124+
}

‎database-commons/src/main/java/io/cdap/plugin/db/DBRecord.java‎

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,9 @@
2121
import io.cdap.cdap.api.common.Bytes;
2222
import io.cdap.cdap.api.data.format.StructuredRecord;
2323
import io.cdap.cdap.api.data.schema.Schema;
24+
import io.cdap.cdap.api.exception.ErrorCategory;
25+
import io.cdap.cdap.api.exception.ErrorType;
26+
import io.cdap.cdap.api.exception.ErrorUtils;
2427
import io.cdap.plugin.util.DBUtils;
2528
import io.cdap.plugin.util.Lazy;
2629
import org.apache.hadoop.conf.Configurable;
@@ -305,7 +308,10 @@ protected void updateOperation(PreparedStatement stmt) throws SQLException {
305308
* @throws SQLException
306309
*/
307310
protected void upsertOperation(PreparedStatement stmt) throws SQLException {
308-
throw new UnsupportedOperationException();
311+
String errorMessage = "Upsert operation is not supported for this plugin.";
312+
throw ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN),
313+
errorMessage, errorMessage, ErrorType.SYSTEM, false, new UnsupportedOperationException(errorMessage));
314+
309315
}
310316

311317
private boolean fillUpdateParams(List<String> updatedKeyList, ColumnType columnType) {
@@ -366,7 +372,9 @@ private void writeToDataOut(DataOutput out, Schema.Field field) throws IOExcepti
366372
out.write((byte[]) fieldValue);
367373
break;
368374
default:
369-
throw new IOException(String.format("Unsupported datatype: %s with value: %s.", fieldType, fieldValue));
375+
String errorMessage = String.format("Unsupported datatype: %s with value: %s.", fieldType, fieldValue);
376+
throw ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN),
377+
errorMessage, errorMessage, ErrorType.USER, false, new IOException(errorMessage));
370378
}
371379
}
372380

‎database-commons/src/main/java/io/cdap/plugin/db/sink/AbstractDBSink.java‎

Lines changed: 14 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,7 @@
3232
import io.cdap.cdap.etl.api.StageConfigurer;
3333
import io.cdap.cdap.etl.api.batch.BatchRuntimeContext;
3434
import io.cdap.cdap.etl.api.batch.BatchSinkContext;
35+
import io.cdap.cdap.etl.api.exception.ErrorDetailsProviderSpec;
3536
import io.cdap.cdap.etl.api.validation.InvalidStageException;
3637
import io.cdap.plugin.common.LineageRecorder;
3738
import io.cdap.plugin.common.ReferenceBatchSink;
@@ -42,6 +43,7 @@
4243
import io.cdap.plugin.db.ConnectionConfig;
4344
import io.cdap.plugin.db.ConnectionConfigAccessor;
4445
import io.cdap.plugin.db.DBConfig;
46+
import io.cdap.plugin.db.DBErrorDetailsProvider;
4547
import io.cdap.plugin.db.DBRecord;
4648
import io.cdap.plugin.db.Operation;
4749
import io.cdap.plugin.db.SchemaReader;
@@ -163,6 +165,16 @@ public void validateOperations(FailureCollector collector, T dbSinkConfig, @Null
163165
}
164166
}
165167

168+
/**
169+
* Returns the ErrorDetailsProvider class name.
170+
* Override this method to provide a custom ErrorDetailsProvider class name.
171+
*
172+
* @return ErrorDetailsProvider class name
173+
*/
174+
protected String getErrorDetailsProviderClassName() {
175+
return DBErrorDetailsProvider.class.getName();
176+
}
177+
166178
@Override
167179
public void prepareRun(BatchSinkContext context) {
168180
String connectionString = dbSinkConfig.getConnectionString();
@@ -227,7 +239,8 @@ public void prepareRun(BatchSinkContext context) {
227239
configuration.set(ETLDBOutputFormat.COMMIT_BATCH_SIZE,
228240
context.getArguments().get(ETLDBOutputFormat.COMMIT_BATCH_SIZE));
229241
}
230-
242+
// set error details provider
243+
context.setErrorDetailsProvider(new ErrorDetailsProviderSpec(getErrorDetailsProviderClassName()));
231244
addOutputContext(context);
232245
}
233246
protected void addOutputContext(BatchSinkContext context) {

‎database-commons/src/main/java/io/cdap/plugin/db/source/AbstractDBSource.java‎

Lines changed: 25 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -25,13 +25,17 @@
2525
import io.cdap.cdap.api.data.format.StructuredRecord;
2626
import io.cdap.cdap.api.data.schema.Schema;
2727
import io.cdap.cdap.api.dataset.lib.KeyValue;
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.cdap.api.plugin.PluginConfig;
2932
import io.cdap.cdap.etl.api.Emitter;
3033
import io.cdap.cdap.etl.api.FailureCollector;
3134
import io.cdap.cdap.etl.api.PipelineConfigurer;
3235
import io.cdap.cdap.etl.api.StageConfigurer;
3336
import io.cdap.cdap.etl.api.batch.BatchRuntimeContext;
3437
import io.cdap.cdap.etl.api.batch.BatchSourceContext;
38+
import io.cdap.cdap.etl.api.exception.ErrorDetailsProviderSpec;
3539
import io.cdap.cdap.internal.io.SchemaTypeAdapter;
3640
import io.cdap.plugin.common.LineageRecorder;
3741
import io.cdap.plugin.common.ReferenceBatchSource;
@@ -41,6 +45,7 @@
4145
import io.cdap.plugin.db.ConnectionConfig;
4246
import io.cdap.plugin.db.ConnectionConfigAccessor;
4347
import io.cdap.plugin.db.DBConfig;
48+
import io.cdap.plugin.db.DBErrorDetailsProvider;
4449
import io.cdap.plugin.db.DBRecord;
4550
import io.cdap.plugin.db.SchemaReader;
4651
import io.cdap.plugin.db.TransactionIsolationLevel;
@@ -119,8 +124,9 @@ public void configurePipeline(PipelineConfigurer pipelineConfigurer) {
119124
collector.addFailure("Unable to instantiate JDBC driver: " + e.getMessage(), null)
120125
.withStacktrace(e.getStackTrace());
121126
} catch (SQLException e) {
122-
collector.addFailure("SQL error while getting query schema: " + e.getMessage(), null)
123-
.withStacktrace(e.getStackTrace());
127+
String details = String.format("SQL error while getting query schema: Error: %s, SQLState: %s, ErrorCode: %s",
128+
e.getMessage(), e.getSQLState(), e.getErrorCode());
129+
collector.addFailure(details, null).withStacktrace(e.getStackTrace());
124130
} catch (Exception e) {
125131
collector.addFailure(e.getMessage(), null).withStacktrace(e.getStackTrace());
126132
}
@@ -194,7 +200,11 @@ private Schema loadSchemaFromDB(Class<? extends Driver> driverClass)
194200

195201
} catch (SQLException e) {
196202
// wrap exception to ensure SQLException-child instances not exposed to contexts without jdbc driver in classpath
197-
throw new SQLException(e.getMessage(), e.getSQLState(), e.getErrorCode());
203+
String errorMessageWithDetails = String.format("Error occurred while trying to get schema from database." +
204+
"Error message: '%s'. Error code: '%s'. SQLState: '%s'", e.getMessage(), e.getErrorCode(), e.getSQLState());
205+
throw ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN),
206+
e.getMessage(), errorMessageWithDetails, ErrorType.USER, false, new SQLException(e.getMessage(),
207+
e.getSQLState(), e.getErrorCode()));
198208
} finally {
199209
driverCleanup.destroy();
200210
}
@@ -212,6 +222,16 @@ protected SchemaReader getSchemaReader() {
212222
return new CommonSchemaReader();
213223
}
214224

225+
/**
226+
* Returns the ErrorDetailsProvider class name.
227+
* Override this method to provide a custom ErrorDetailsProvider class name.
228+
*
229+
* @return ErrorDetailsProvider class name
230+
*/
231+
protected String getErrorDetailsProviderClassName() {
232+
return DBErrorDetailsProvider.class.getName();
233+
}
234+
215235
private DriverCleanup loadPluginClassAndGetDriver(Class<? extends Driver> driverClass)
216236
throws IllegalAccessException, InstantiationException, SQLException {
217237

@@ -268,6 +288,8 @@ public void prepareRun(BatchSourceContext context) throws Exception {
268288
lineageRecorder.recordRead("Read", "Read from database plugin",
269289
schema.getFields().stream().map(Schema.Field::getName).collect(Collectors.toList()));
270290
}
291+
// set error details provider
292+
context.setErrorDetailsProvider(new ErrorDetailsProviderSpec(getErrorDetailsProviderClassName()));
271293
context.setInput(Input.of(sourceConfig.getReferenceName(), new SourceInputFormatProvider(
272294
DataDrivenETLDBInputFormat.class, connectionConfigAccessor.getConfiguration())));
273295
}

‎mysql-plugin/src/e2e-test/features/mysqlsink/RunTimeWithMacros.feature‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -100,6 +100,7 @@ Feature: MySQL Sink - Run time scenarios (macro)
100100
Then Wait till pipeline is in running state
101101
Then Open and capture logs
102102
And Verify the pipeline status is "Failed"
103+
And Close the pipeline logs
103104

104105
@BQ_SOURCE_TEST @MYSQL_TARGET_TABLE @Mysql_Required
105106
Scenario: Verify that the pipeline fails when user provides invalid Credentials for connection with Macros
@@ -135,4 +136,6 @@ Feature: MySQL Sink - Run time scenarios (macro)
135136
And Enter runtime argument value "invalid.password" for key "password"
136137
And Run the Pipeline in Runtime with runtime arguments
137138
And Wait till pipeline is in running state
139+
And Open and capture logs
138140
And Verify the pipeline status is "Failed"
141+
And Close the pipeline logs

‎mysql-plugin/src/e2e-test/features/mysqlsource/RunTime.feature‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -143,6 +143,7 @@ Feature: MySQL Source - Run time scenarios
143143
Then Save the pipeline
144144
Then Preview and run the pipeline
145145
Then Wait till pipeline preview is in running state
146+
Then Open and capture pipeline preview logs
146147
Then Verify the preview run status of pipeline in the logs is "failed"
147148

148149
@MYSQL_SOURCE_TEST @MYSQL_TARGET_TEST @Mysql_Required

‎mysql-plugin/src/e2e-test/features/mysqlsource/RunTimeWithMacros.feature‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -201,7 +201,9 @@ Feature: MySQL Source - Run time scenarios (macro)
201201
And Enter runtime argument value "invalid.query" for key "importQuery"
202202
And Run the Pipeline in Runtime with runtime arguments
203203
And Wait till pipeline is in running state
204+
And Open and capture logs
204205
And Verify the pipeline status is "Failed"
206+
And Close the pipeline logs
205207

206208
@MYSQL_SOURCE_TEST @MYSQL_TARGET_TEST @Mysql_Required
207209
Scenario: Verify that pipeline fails when user provides invalid Credentials for connection with Macros
@@ -241,4 +243,6 @@ Feature: MySQL Source - Run time scenarios (macro)
241243
And Enter runtime argument value "invalid.password" for key "Password"
242244
And Run the Pipeline in Runtime with runtime arguments
243245
And Wait till pipeline is in running state
246+
And Open and capture logs
244247
And Verify the pipeline status is "Failed"
248+
And Close the pipeline logs

‎mysql-plugin/src/main/java/io/cdap/plugin/mysql/MysqlConstants.java‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,12 +16,18 @@
1616

1717
package io.cdap.plugin.mysql;
1818

19+
import io.cdap.cdap.api.exception.ErrorCategory;
20+
import io.cdap.cdap.api.exception.ErrorType;
21+
import io.cdap.cdap.api.exception.ErrorUtils;
22+
1923
/**
2024
* MySQL Constants.
2125
*/
2226
public final class MysqlConstants {
2327
private MysqlConstants() {
24-
throw new AssertionError("Should not instantiate static utility class.");
28+
String errorMessage = "Should not instantiate static utility class.";
29+
throw ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN),
30+
errorMessage, errorMessage, ErrorType.SYSTEM, false, new AssertionError(errorMessage));
2531
}
2632

2733
public static final String PLUGIN_NAME = "Mysql";

‎mysql-plugin/src/main/java/io/cdap/plugin/mysql/MysqlErrorDetailsProvider.java‎

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,45 @@
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.mysql;
18+
19+
import io.cdap.cdap.api.exception.ErrorType;
20+
import io.cdap.plugin.db.DBErrorDetailsProvider;
21+
22+
/**
23+
* A custom ErrorDetailsProvider for MySQL plugins.
24+
*/
25+
public class MysqlErrorDetailsProvider extends DBErrorDetailsProvider {
26+
27+
@Override
28+
protected String getExternalDocumentationLink() {
29+
return "https://dev.mysql.com/doc/mysql-errors/9.0/en/";
30+
}
31+
32+
@Override
33+
protected ErrorType getErrorTypeFromErrorCode(int errorCode) {
34+
// https://dev.mysql.com/doc/refman/9.0/en/error-message-elements.html#error-code-ranges
35+
if (errorCode >= 1000 && errorCode <= 5999) {
36+
return ErrorType.USER;
37+
} else if (errorCode >= 10000 && errorCode <= 51999) {
38+
// SYSTEM errors: Enterprise and user-defined custom error messages
39+
return ErrorType.SYSTEM;
40+
} else {
41+
// UNKNOWN errors: Anything outside defined range
42+
return ErrorType.UNKNOWN;
43+
}
44+
}
45+
}

‎mysql-plugin/src/main/java/io/cdap/plugin/mysql/MysqlSink.java‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -109,6 +109,11 @@ String getDbColumns() {
109109
return dbColumns;
110110
}
111111

112+
@Override
113+
protected String getErrorDetailsProviderClassName() {
114+
return MysqlErrorDetailsProvider.class.getName();
115+
}
116+
112117
/**
113118
* MySQL action configuration.
114119
*/

0 commit comments

Comments
 (0)