PLUGIN-1823: Retrying all SQLTransientExceptions

[sgarg-CS]

PLUGIN-1823

Add Failsafe Retry poilcy to all the places in the database-plugins where SQLTransientException could be thrown.

Added three new properties (hidden from UI)

• Initial Retry Duration (Default: 5sec)
• Max Retry Duration (Default: 80 sec)
• Max Retry Count (Default: 5)

[google-cla]

Thanks for your pull request! It looks like this may be your first contribution to a Google open source project. Before we can look at your pull request, you'll need to sign a Contributor License Agreement (CLA).

View this failed invocation of the CLA check for more information.

For the most up to date status, view the checks section at the bottom of the pull request.

[itsankit-google]

Overall, this looks everything is getting wrapped within Failsafe where we might end up with having nested level retries, we need to ensure we add retries only where we are actually interacting with JDBC client and not top level functions.

For example adding retries to DriverManager.getConnection(connectionString, connectionProperties) makes sense because you are actually interacting with the source db but adding retries to whole loadSchema(Connection connection, String query) do not makes sense we need to be careful while adding such retries.

[itsankit-google]

nit: remove empty line

[sgarg-CS]

Removed

[itsankit-google]

If loadSchemaFromDB(Connection connection, String query) already includes retry logic, we don't need to add another layer here, as having nested retries can lead to undesirable behavior.

Also Failsafe will wrap any non-retryable exception or an exception after all retries are exhausted in a FailsafeException so where ever we are doing catch(SQLException) it won't work, you might need to catch FailsafeException & check cause.

[sgarg-CS]

Removed the nested retry logic. Also, handled the FailsafeException and the wrapped exceptions in a separate method : unwrapFailsafeException() of RetryUtils class as suggested in the other comment.

[itsankit-google]

Please note E2E should not be modified and not fail with these changes. Otherwise, we have done something wrong which does not give expected failure messages.

[itsankit-google]

we should throw cause instead of FailsafeException.

[sgarg-CS]

Handled the FailsafeException in a separate method : unwrapFailsafeException of RetryUtils class as suggested in the other comment.

[itsankit-google]

we should throw cause instead of FailsafeException.

[sgarg-CS]

Handled the FailsafeException in a separate method : unwrapFailsafeException of RetryUtils class as suggested in the other comment.

[itsankit-google]

we should throw cause instead of FailsafeException.

[sgarg-CS]

Handled the FailsafeException in a separate method : unwrapFailsafeException of RetryUtils class as suggested in the other comment.

[itsankit-google]

we should throw cause instead of FailsafeException.

[sgarg-CS]

Handled the FailsafeException in a separate method : unwrapFailsafeException of RetryUtils class as suggested in the other comment.

[itsankit-google]

I can see some level of duplication in both AbstractDBSource & AbstractDBSink, can we please move it to the common AbstractDBUtil class?

[itsankit-google]

remove extra empty line

[sgarg-CS]

Removed

[itsankit-google]

we should not throw FailsafeException

[sgarg-CS]

Handled the FailsafeException in a separate method : unwrapFailsafeException of RetryUtils class as suggested in the other comment.

[itsankit-google]

public final class RetryUtils {

  public static Connection createConnectionWithRetry(RetryPolicy<?> retryPolicy, String connectionString,
                                                     Properties connectionProperties, String externalDocumentationLink) throws Exception {
    try {
      return Failsafe.with(retryPolicy).get(() ->
        DriverManager.getConnection(connectionString, connectionProperties)
      );
    } catch (Exception e) {
      throw unwrapFailsafeException(e, externalDocumentationLink);
    }
  }

  public static Statement createStatementWithRetry(RetryPolicy<?> retryPolicy,
                                                   Connection connection, String externalDocumentationLink) throws Exception {
    try {
      return Failsafe.with(retryPolicy).get(connection::createStatement);
    } catch (Exception e) {
      throw unwrapFailsafeException(e, externalDocumentationLink);
    }
  }

  public static PreparedStatement prepareStatementWithRetry(RetryPolicy<?> retryPolicy,
                                                            Connection connection,
                                                            String sqlQuery, String externalDocumentationLink) throws Exception {
    try {
      return Failsafe.with(retryPolicy).get(() ->
        connection.prepareStatement(sqlQuery)
      );
    } catch (Exception e) {
      throw unwrapFailsafeException(e, externalDocumentationLink);
    }
  }

 public static ResultSet executeWithRetry(RetryPolicy<?> retryPolicy,
                                                            Connection connection,
                                                            String sqlQuery, String externalDocumentationLink) throws Exception {
        try {
            return Failsafe.with(retryPolicy).get(() -> connection.createStatement().executeQuery(sqlQuery));
        } catch (Exception e) {
            throw unwrapFailsafeException(e, externalDocumentationLink);
        }
    }

 private static Exception unwrapFailsafeException(Exception e) {
    if (e instanceof FailsafeException && e.getCause() instanceof Exception) {
        if (e instanceOf SQLException) {
           return programFailureException(e, externalDocumentationLink);
        } else {
          return (Exception) e.getCause();
       }
    }
    return e;
  }
  
private static ProgramFailureException programFailureException(SQLException e, String externalDocumentationLink) {
    // wrap exception to ensure SQLException-child instances not exposed to contexts without jdbc
    // driver in classpath
    String errorMessage =
      String.format("SQL Exception occurred: [Message='%s', SQLState='%s', ErrorCode='%s'].",
        e.getMessage(), e.getSQLState(), e.getErrorCode());
    String errorMessageWithDetails = String.format("Error occurred while trying to" +
      " get schema from database." + "Error message: '%s'. Error code: '%s'. SQLState: '%s'", e.getMessage(),
        e.getErrorCode(), e.getSQLState());
 
    if (!Strings.isNullOrEmpty(externalDocumentationLink)) {
      if (!errorMessage.endsWith(".")) {
        errorMessage = errorMessage + ".";
      }
      errorMessage = String.format("%s For more details, see %s", errorMessage, externalDocumentationLink);
    }
    return ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategory.ErrorCategoryEnum.PLUGIN),
      errorMessage, errorMessageWithDetails, ErrorType.USER, false, ErrorCodeType.SQLSTATE, e.getSQLState(),
        externalDocumentationLink, e);
  }
}

You can create a RetryUtils like above which accepts connection params.

[itsankit-google]

remove extra empty line

[sgarg-CS]

Removed

[sgarg-CS]

Overall, this looks everything is getting wrapped within Failsafe where we might end up with having nested level retries, we need to ensure we add retries only where we are actually interacting with JDBC client and not top level functions.

For example adding retries to DriverManager.getConnection(connectionString, connectionProperties) makes sense because you are actually interacting with the source db but adding retries to whole loadSchema(Connection connection, String query) do not makes sense we need to be careful while adding such retries.

Refactored the code to add the retry logic only for the methods interacting with the JDBC client.

[itsankit-google]

where this method used now?

[sgarg-CS]

it's unused now. Removed it. fc78d9c

[itsankit-google]

this cannot be USER always, we should check based on SQLState and dependency should be true. Not sure how was this missed earlier.

[sgarg-CS]

Can you give some pointers where to fetch all possible SQL States?

Should it be something like this?

String sqlState = e.getSQLState();
ErrorType errorType = getErrorTypeFromSQLState(sqlState);

private static ErrorType getErrorTypeFromSQLState(String sqlState) {
    if (sqlState != null && sqlState.length() >= 2) {
        String classCode = sqlState.substring(0, 2);
        switch (classCode) {
            case "22": // Data Exception
            case "23": // Integrity Constraint Violation
            case "28": // Invalid Authorization
            case "42": // Syntax Error or Access Rule Violation
                return ErrorType.USER;
            case "08": // Connection Exception
            case "53": // Insufficient Resources
            case "57": // Operator Intervention
            case "58": // System Error
                return ErrorType.SYSTEM;
            default:
                // Check for the timeout codes
                if ("HYT00".equals(sqlState) || "HYT01".equals(sqlState)) {
                    return ErrorType.SYSTEM;
                }
                break;
        }
    }
    return ErrorType.UNKNOWN;
}

[itsankit-google]

you can directly use DBErrorDetailsProvider#getException#getProgramFailureException : https://github.com/cdapio/hydrator-plugins/blob/b9723d1b24068347c60ad2b0489064902f335de9/hydrator-common/src/main/java/io/cdap/plugin/common/db/DBErrorDetailsProvider.java#L184

create an object of DBErrorDetailsProvider in RetryUtils

[sgarg-CS]

If we directly use this method, then we wouldn't be able to pass the externalDocumentationLink as an argument in the method call to DBErrorDetailsProvider#getException#getProgramFailureException unless we create another overloaded method in the DBErrorDetailsProvider class to accept this new parameter.

[sgarg-CS]

Did some more code analysis to find out that there's an alternate approach we can try where we'd need to create a method to return the instance of DBErrorDetailsProvider in AbstractSource/AbstractSink classes and then override this method in the respective connectors' subclasses to return the instance of sub-classes of DBErrorDetailsProvider and then call the below methods for

1. getExternalDocumentationLink()
2. getErrorTypeFromErrorCodeAndSqlState() and
3. getErrorCategoryFromSqlState()
   using the returned instance

AbstractDBSource

protected DBErrorDetailsProvider getDBErrorDetailsProvider() {
    return new DBErrorDetailsProvider();
}

PostgresSource

@Override
protected PostgresErrorDetailsProvider getDBErrorDetailsProvider() {
    return new PostgresErrorDetailsProvider();
}

To go with this approach, we'd need to make the access-modifier for the 3 methods mentioned above, from protected to public in the DBErrorDetailsProvider class because Java uses static types at compile time. (Even though we can increase the visibility for the overriden methods from protected to public and JVM will resolve to the sub-class overridden methods for respective connectors at the run time)

Now, if we need to avoid this change for the access modifier, it will additionally require us to create methods for all the 3 methods mentioned above, in AbstractSource/AbstractSink and then override them with the connector specific implementation such as this

AbstractDBSource

protected String getExternalDocumentationLink() {
    return "https://en.wikipedia.org/wiki/SQLSTATE"; //Default
}

PostgresSource

@Override 
protected String getExternalDocumentationLink() {
    return getDBErrorDetailsProvider().getExternalDocumentationLink();
}

This will be a significant change and may stretch this PR further.

[itsankit-google]

move all of this to RetryPolicyUtil and use it everywhere.

[sgarg-CS]

Moved the constants from AbstractDBConnectorConfig and ConnectionConfig classes to RetryUtils class 037a0d4

[itsankit-google]

when it will be thrown if RetryUtils will always throw ProgramFailureException for every SQLException?

[sgarg-CS]

The try-with-resources automatically closes any resources that implement AutoCloseable (like Connection, Statement, or ResultSet)

This SQLException is thrown from the auto-closeable resource when we perform any SQL related operation such as creating a connection or creating statement in the try-with-resources block. If we remove the try-wth-resources, then the SQLException is not thrown

In case we want to keep the try-with-resources, then either we need to add throws SQLException (or Exception) in the enclosing method signature OR we need to catch the SQLException and handle it.

[itsankit-google]

can we use try with resources here?

[sgarg-CS]

Yes, added try-with-resource. 037a0d4

[itsankit-google]

How does it affect the UI if validateSchema fails for DB sinks? Can we test before/after the change?

[sgarg-CS]

How does it affect the UI if validateSchema fails for DB sinks? Can we test before/after the change?

Yes. Will check and update the behaviour here.

I've reverted the changes done to handle SQLException thrown by validateSchema() method call from AbstractDBSink.configurePipeline class. Still, I see a change in the error message on the UI. This is probably due to catching SQLException and then wrapping it to throw ProgramFailureException. While this wrapping was done for inferSchema() but not for validateSchema() in AbstractSink class earlier.

Test Scenario: Validate the schema in PostgreSQL Sink Plugin, if connection is not active. (Postgres DB is down)

[BEFORE CHANGES]

Error message on the UI:
Exception while trying to validate schema of database table '"users2"' for connection 'jdbc:postgresql://localhost:5433/postgres' with Connection to localhost:5433 refused. Check that the hostname and port are correct and that the postmaster is accepting TCP/IP connections.

[AFTER CHANGES]

Error message on the UI:
Error encountered while configuring the stage: 'Error occurred while trying to get schema from database. Error message: 'Connection to localhost:5433 refused. Check that the hostname and port are correct and that the postmaster is accepting TCP/IP connections.'. Error code: '0'. SQLState: '08001''

Is the new error message acceptable or should we revert the changes done to throw ProgramFailureException for validateSchema() method?

[itsankit-google]

How does it affect the UI if validateSchema fails for DB sinks? Can we test before/after the change?

Yes. Will check and update the behaviour here.

I've reverted the changes done to handle SQLException thrown by validateSchema() method call from AbstractDBSink.configurePipeline class. Still, I see a change in the error message on the UI. This is probably due to catching SQLException and then wrapping it to throw ProgramFailureException. While this wrapping was done for inferSchema() but not for validateSchema() in AbstractSink class earlier.

Test Scenario: Validate the schema in PostgreSQL Sink Plugin, if connection is not active. (Postgres DB is down)

[BEFORE CHANGES]

Error message on the UI: Exception while trying to validate schema of database table '"users2"' for connection 'jdbc:postgresql://localhost:5433/postgres' with Connection to localhost:5433 refused. Check that the hostname and port are correct and that the postmaster is accepting TCP/IP connections.

[AFTER CHANGES]

Error message on the UI: Error encountered while configuring the stage: 'Error occurred while trying to get schema from database. Error message: 'Connection to localhost:5433 refused. Check that the hostname and port are correct and that the postmaster is accepting TCP/IP connections.'. Error code: '0'. SQLState: '08001''

Is the new error message acceptable or should we revert the changes done to throw ProgramFailureException for validateSchema() method?

LGTM

[itsankit-google]

this is not meant to be copied and maintained at two places.

We can just create an object of DBErrorDetailsProvider

[sgarg-CS]

Those maps are private and there are not public getter methods for them in the DBErrorDetailsProvider. Are you saying that we can modify the class and raise a PR in thehydrator-plugins repo?

[sgarg-CS]

This class:

database-plugins/postgresql-plugin/src/main/java/io/cdap/plugin/postgres/PostgresErrorDetailsProvider.java

Line 35 in efa54d4

static {

does make use of some of the error codes

[itsankit-google]

Those maps are private and there are not public getter methods for them in the DBErrorDetailsProvider. Are you saying that we can modify the class and raise a PR in thehydrator-plugins repo?

I specifically pointed out the method which can be called in earlier comment : https://github.com/cdapio/hydrator-plugins/blob/b9723d1b24068347c60ad2b0489064902f335de9/hydrator-common/src/main/java/io/cdap/plugin/common/db/DBErrorDetailsProvider.java#L184

[itsankit-google]

why is externalDocumentationLink passed as null, this is not correct.

You can create a method at top level which returns generic link and others can override it.

See how we do it in OracleErrorDetailsProvider, MysqlErrorDetailsProvider, etc.

[itsankit-google]

We can use this link as default one "https://en.wikipedia.org/wiki/SQLSTATE" as it is being used by DBErrorDetailsProvider. similar comment for AbstractDBSource & AbstractDBSink.

[sgarg-CS]

Added the default value for external documentation link in the above mentioned classes. 35537d7

[itsankit-google]

We are not always fetching schema from database in above calls.

[itsankit-google]

As I said before this is not needed, we could just create an object of DBErrorDetailsProvider and use DBErrorDetailsProvider#getProgramFailureException to do all of this stuff.