Merge branch 'MET-6544-flink-code-cleanup-in-the-area-of-exception-handling' into develop

Author: Ruufuus

client/src/main/java/eu/europeana/cloud/flink/client/JobExecutor.java

@@ -120,7 +120,7 @@ private JobDetails getProgressWithRetry(String jobId) throws InterruptedExceptio
           return getProgress(jobId);
         } catch (RestClientResponseException e) {
           if (e.getStatusCode() == HttpStatus.NOT_FOUND &&
-              e.getResponseBodyAsString().contains("org.apache.flink.runtime.rest.NotFoundException")) {
+                  e.getResponseBodyAsString().contains("org.apache.flink.runtime.rest.NotFoundException")) {
             throw new RuntimeException("There is no more job of the id: " + jobId + " on the server", e);
           }
           throw e;

commons/src/main/java/eu/europeana/processing/MetisJob.java

@@ -113,7 +113,7 @@ protected void generateTaskIdIfNeeded() {
                 }
             }
         }catch (Exception e){
-            throw new RuntimeException("Error while generating task id!",e);
+            throw new RuntimeException("Error while generating task id!", e);
         }
     }
 

commons/src/main/java/eu/europeana/processing/exception/FlinkWorkflowException.java

@@ -0,0 +1,10 @@
+package eu.europeana.processing.exception;
+
+public class FlinkWorkflowException extends Exception{
+    public FlinkWorkflowException(String message, Throwable cause) {
+        super(message, cause);
+    }
+    public FlinkWorkflowException(Throwable cause) {
+        super(cause);
+    }
+}

commons/src/main/java/eu/europeana/processing/exception/FlinkWorkflowRuntimeException.java

@@ -0,0 +1,10 @@
+package eu.europeana.processing.exception;
+
+public class FlinkWorkflowRuntimeException extends RuntimeException{
+    public FlinkWorkflowRuntimeException(String message, Throwable cause) {
+        super(message, cause);
+    }
+    public FlinkWorkflowRuntimeException(Throwable cause) {
+        super(cause);
+    }
+}

commons/src/main/java/eu/europeana/processing/exception/RecoverableException.java

@@ -0,0 +1,10 @@
+package eu.europeana.processing.exception;
+
+public class RecoverableException extends FlinkWorkflowRuntimeException {
+    public RecoverableException(String message, Throwable cause) {
+        super(message, cause);
+    }
+    public RecoverableException(Throwable cause) {
+        super(cause);
+    }
+}

commons/src/main/java/eu/europeana/processing/exception/UnrecoverableJobException.java

@@ -0,0 +1,12 @@
+package eu.europeana.processing.exception;
+
+
+
+public class UnrecoverableJobException extends FlinkWorkflowRuntimeException {
+    public UnrecoverableJobException(String message, Throwable cause) {
+        super(message, cause);
+    }
+    public UnrecoverableJobException(Throwable cause) {
+        super(cause);
+    }
+}

commons/src/main/java/eu/europeana/processing/exception/UnrecoverableRecordException.java

@@ -0,0 +1,12 @@
+package eu.europeana.processing.exception;
+
+
+
+public class UnrecoverableRecordException extends FlinkWorkflowException {
+    public UnrecoverableRecordException(String message, Throwable cause) {
+        super(message, cause);
+    }
+    public UnrecoverableRecordException(Throwable cause) {
+        super(cause);
+    }
+}

commons/src/main/java/eu/europeana/processing/exception/classifier/DatabaseExceptionClassifier.java

@@ -0,0 +1,35 @@
+package eu.europeana.processing.exception.classifier;
+
+
+import eu.europeana.processing.exception.RecoverableException;
+import eu.europeana.processing.exception.UnrecoverableJobException;
+import eu.europeana.processing.exception.UnrecoverableRecordException;
+
+import java.sql.SQLException;
+
+public class DatabaseExceptionClassifier {
+
+    private DatabaseExceptionClassifier() {}
+
+    public static void classifyAndThrow(SQLException sqlException) throws UnrecoverableRecordException {
+        String sqlState = sqlException.getSQLState();
+        if (sqlState == null || sqlState.length() < 2) {
+            throw new UnrecoverableJobException("Unknown database error", sqlException);
+        }
+
+//        08	Connection Exception	ConnectionException
+//        22	Data Exception (invalid data)	DataFormatException
+//        23	Integrity Constraint Violation	IntegrityConstraintException
+//        28	Invalid Authorization	AuthenticationException
+//        40	Transaction Rollback	TransactionException
+//        42	Syntax Error or Access Rule Violation	SqlSyntaxException
+
+        switch (sqlState.substring(0, 2)) {
+            case "08" -> throw new RecoverableException("Connection error", sqlException);
+            case "22" -> throw new UnrecoverableRecordException("Invalid data format", sqlException);
+            case "23", "28", "40", "42" -> throw new UnrecoverableJobException("Database error (" + sqlException.getSQLState() + ")", sqlException);
+            default -> throw new UnrecoverableJobException("Database error", sqlException);
+        }
+    }
+
+}

commons/src/main/java/eu/europeana/processing/repository/ExecutionRecordExceptionLogRepository.java

@@ -1,20 +1,22 @@
 package eu.europeana.processing.repository;
 
 import eu.europeana.processing.DbConnectionProvider;
+import eu.europeana.processing.exception.FlinkWorkflowException;
 import eu.europeana.processing.retryable.Retryable;
 import eu.europeana.processing.model.ExecutionRecord;
 import eu.europeana.processing.model.ExecutionRecordResult;
 import java.io.Serial;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.io.IOException;
 import java.io.Serializable;
 import java.sql.Connection;
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
 
+import static eu.europeana.processing.exception.classifier.DatabaseExceptionClassifier.classifyAndThrow;
+
 /**
  * Database repository responsible for <b>execution_record_exception_log</b> table
  */
@@ -56,8 +58,9 @@ public ExecutionRecordExceptionLogRepository(DbConnectionProvider dbConnectionPr
      *  In case of conflict (try to insert the same record twice) the method does nothing.
      *
      * @param executionRecordResult instance to be saved in the database
+     * @throws FlinkWorkflowException in case of any DB exception
      */
-    public void save(ExecutionRecordResult executionRecordResult) {
+    public void save(ExecutionRecordResult executionRecordResult) throws FlinkWorkflowException {
         try (Connection con = dbConnectionProvider.getConnection();
              PreparedStatement preparedStatement = con.prepareStatement(
                      "INSERT INTO \"batch-framework\".execution_record_exception_log (DATASET_ID,EXECUTION_ID,EXECUTION_NAME, RECORD_ID, exception)"
@@ -76,7 +79,7 @@ public void save(ExecutionRecordResult executionRecordResult) {
                 LOGGER.info("Record error log already existed in the DB: {}", executionRecord.getExecutionRecordKey());
             }
         } catch (SQLException e) {
-            throw new RuntimeException(e);
+            classifyAndThrow(e);
         }
     }
 
@@ -85,25 +88,24 @@ public void save(ExecutionRecordResult executionRecordResult) {
      * @param datasetId dataset identifier
      * @param executionId execution identifier
      * @return number of elements in <b>execution_record_exception_log</b> table for specified dataset and execution
-     * @throws IOException
+     * @throws FlinkWorkflowException in case of any DB exception
      */
-    public long countByDatasetIdAndExecutionId(String datasetId, String executionId) throws IOException {
+    public long countByDatasetIdAndExecutionId(String datasetId, String executionId) throws FlinkWorkflowException {
 
-        ResultSet resultSet;
+        long count = 0L;
         try (PreparedStatement preparedStatement = dbConnectionProvider.getConnection().prepareStatement(NO_OF_ELEMENTS)) {
             preparedStatement.setString(1, datasetId);
             preparedStatement.setString(2, executionId);
 
-            resultSet = preparedStatement.executeQuery();
+            ResultSet resultSet = preparedStatement.executeQuery();
 
             if (resultSet.next()) {
-                return resultSet.getLong("elements");
-            } else {
-                return 0L;
+                count = resultSet.getLong("elements");
             }
         } catch(SQLException e){
-            throw new IOException(e);
+            classifyAndThrow(e);
         }
+        return count;
     }
 
 }

commons/src/main/java/eu/europeana/processing/repository/ExecutionRecordRepository.java

@@ -1,6 +1,7 @@
 package eu.europeana.processing.repository;
 
 import eu.europeana.processing.DbConnectionProvider;
+import eu.europeana.processing.exception.FlinkWorkflowException;
 import eu.europeana.processing.model.ExecutionRecord;
 import eu.europeana.processing.model.ExecutionRecordKey;
 import eu.europeana.processing.model.ExecutionRecordResult;
@@ -9,7 +10,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.io.IOException;
 import java.io.Serializable;
 import java.sql.Connection;
 import java.sql.PreparedStatement;
@@ -18,6 +18,8 @@
 import java.util.ArrayList;
 import java.util.List;
 
+import static eu.europeana.processing.exception.classifier.DatabaseExceptionClassifier.classifyAndThrow;
+
 /**
  * Database repository responsible for <b>execution_record</b> table
  */
@@ -67,9 +69,9 @@ public ExecutionRecordRepository(DbConnectionProvider dbConnectionProvider) {
      * In case of conflict (try to insert the same record twice) the method does nothing.
      *
      * @param executionRecordResult instance to be saved in the database
-     * @throws IOException in case of any DB exception
+     * @throws FlinkWorkflowException in case of any DB exception
      */
-    public void save(ExecutionRecordResult executionRecordResult) throws IOException {
+    public void save(ExecutionRecordResult executionRecordResult) throws FlinkWorkflowException {
 
         try (Connection con = dbConnectionProvider.getConnection();
              PreparedStatement preparedStatement = con.prepareStatement(
@@ -88,7 +90,7 @@ public void save(ExecutionRecordResult executionRecordResult) throws IOException
                 LOGGER.info("Execution record already existed in the DB: {}", executionRecord.getExecutionRecordKey());
             }
         } catch (SQLException e) {
-            throw new IOException(e);
+            classifyAndThrow(e);
         }
     }
 
@@ -98,26 +100,25 @@ public void save(ExecutionRecordResult executionRecordResult) throws IOException
      * @param datasetId dataset identifier
      * @param executionId execution identifier
      * @return number of elements in <b>execution_record</b> table for specified dataset and execution
-     * @throws IOException in case of any DB exception
+     * @throws FlinkWorkflowException in case of any DB exception
      */
-    public long countByDatasetIdAndExecutionId(String datasetId, String executionId) throws IOException {
+    public long countByDatasetIdAndExecutionId(String datasetId, String executionId) throws FlinkWorkflowException {
 
-        ResultSet resultSet;
+        long count = 0L;
         try (Connection con = dbConnectionProvider.getConnection();
              PreparedStatement preparedStatement = con.prepareStatement(NO_OF_ELEMENTS)) {
             preparedStatement.setString(1, datasetId);
             preparedStatement.setString(2, executionId);
 
-            resultSet = preparedStatement.executeQuery();
+            ResultSet resultSet = preparedStatement.executeQuery();
 
             if (resultSet.next()) {
-                return resultSet.getLong("elements");
-            } else {
-                return 0L;
+                count = resultSet.getLong("elements");
             }
         } catch (SQLException e) {
-            throw new IOException(e);
+            classifyAndThrow(e);
         }
+        return count;
     }
 
     /**
@@ -127,14 +128,15 @@ public long countByDatasetIdAndExecutionId(String datasetId, String executionId)
      * @param offset dataset offset
      * @param limit dataset limit
      * @return list of all {@link ExecutionRecord} fulfilling provided criteria
-     * @throws IOException in case of any DB exception
+     * @throws FlinkWorkflowException in case of any DB exception
      */
     //TODO to be changed, returned list may be really big
     public List<ExecutionRecord> getByDatasetIdAndExecutionIdAndOffsetAndLimit(
         String datasetId,
         String executionId,
         long offset,
-        long limit) throws IOException {
+        long limit)  throws FlinkWorkflowException {
+        List<ExecutionRecord> result = new ArrayList<>();
         try (Connection con = dbConnectionProvider.getConnection();
              PreparedStatement preparedStatement = con.prepareStatement(LIMIT)) {
 
@@ -143,7 +145,6 @@ public List<ExecutionRecord> getByDatasetIdAndExecutionIdAndOffsetAndLimit(
             preparedStatement.setLong(3, offset);
             preparedStatement.setLong(4, limit);
 
-            List<ExecutionRecord> result = new ArrayList<>();
             ResultSet resultSet = preparedStatement.executeQuery();
             while (resultSet.next()) {
                 result.add(
@@ -159,9 +160,9 @@ public List<ExecutionRecord> getByDatasetIdAndExecutionIdAndOffsetAndLimit(
                                 .build()
                 );
             }
-            return result;
         } catch (SQLException e) {
-            throw new IOException(e);
+            classifyAndThrow(e);
         }
+        return result;
     }
 }