[opt] 优化impala异常处理

Author: FlechazoW

impala/impala-sink/src/main/java/com/dtstack/flink/sql/sink/impala/ImpalaOutputFormat.java

@@ -77,7 +77,7 @@ public class ImpalaOutputFormat extends AbstractDtRichOutputFormat<Tuple2<Boolea
     // cast(value as string) -> cast('value' as string)  cast(value as timestamp) -> cast('value' as timestamp)
     private static final Pattern TYPE_PATTERN = Pattern.compile("cast\\((.*) as (.*)\\)");
     //specific type which values need to be quoted
-    private static final String[] NEED_QUOTE_TYPE = {"string", "timestamp", "varchar"};
+    private static final String[] NEED_QUOTE_TYPE = {"timestamp", "varchar"};
 
     private static final Integer DEFAULT_CONN_TIME_OUT = 60;
     private static final int RECEIVE_DATA_PRINT_FREQUENCY = 1000;
@@ -162,7 +162,7 @@ public void open(int taskNumber, int numTasks) throws IOException {
             init();
             initMetric();
         } catch (Exception e) {
-            throw new RemoteException("impala output format open error!", e);
+            throw new RuntimeException("impala output format open error!", e);
         }
     }
 
@@ -189,19 +189,23 @@ private void init() throws SQLException {
     }
 
     private void initScheduledTask(Long batchWaitInterval) {
-        if (batchWaitInterval != 0) {
-            this.scheduler = new ScheduledThreadPoolExecutor(1,
-                    new DTThreadFactory("impala-upsert-output-format"));
-            this.scheduledFuture = this.scheduler.scheduleWithFixedDelay(() -> {
-                synchronized (ImpalaOutputFormat.this) {
-                    try {
-                        flush();
-                    } catch (Exception e) {
-                        LOG.error("Writing records to impala jdbc failed.", e);
-                        throw new RuntimeException("Writing records to impala jdbc failed.", e);
+        try {
+            if (batchWaitInterval != 0) {
+                this.scheduler = new ScheduledThreadPoolExecutor(1,
+                        new DTThreadFactory("impala-upsert-output-format"));
+                this.scheduledFuture = this.scheduler.scheduleWithFixedDelay(() -> {
+                    synchronized (ImpalaOutputFormat.this) {
+                        try {
+                            flush();
+                        } catch (Exception e) {
+                            LOG.error("Writing records to impala jdbc failed.", e);
+                            throw new RuntimeException("Writing records to impala jdbc failed.", e);
+                        }
                     }
-                }
-            }, batchWaitInterval, batchWaitInterval, TimeUnit.MILLISECONDS);
+                }, batchWaitInterval, batchWaitInterval, TimeUnit.MILLISECONDS);
+            }
+        } catch (Exception e) {
+            throw new RuntimeException(e);
         }
     }
 
@@ -235,7 +239,7 @@ private void openJdbc() {
         }
     }
 
-    private synchronized void flush() throws SQLException {
+    private void flush() throws Exception {
         if (batchCount > 0) {
             if (updateMode.equalsIgnoreCase(UPDATE_MODE)) {
                 executeUpdateBatch();
@@ -387,7 +391,7 @@ private String valueConditionAddQuotation(String valueCondition) {
     }
 
     @Override
-    public synchronized void writeRecord(Tuple2<Boolean, Row> record) throws IOException {
+    public void writeRecord(Tuple2<Boolean, Row> record) throws IOException {
         try {
             if (!record.f0) {
                 return;
@@ -433,7 +437,7 @@ public synchronized void writeRecord(Tuple2<Boolean, Row> record) throws IOExcep
             // Receive data
             outRecords.inc();
         } catch (Exception e) {
-            throw new RuntimeException("Writing records to impala failed.", e);
+            throw new IOException("Writing records to impala failed.", e);
         }
     }
 
@@ -469,7 +473,7 @@ public void close() throws IOException {
                 updateStatement.close();
             }
         } catch (SQLException e) {
-            throw new RemoteException("impala connection close failed!");
+            throw new RemoteException("impala connection close failed!", e);
         } finally {
             connection = null;
             statement = null;
@@ -489,17 +493,16 @@ public void close() throws IOException {
      * @param fieldNames      field name list
      * @param partitionFields partition fields
      * @param rowDataMap      row data map
-     * @throws SQLException throw sql exception
      */
-    private synchronized void executeBatchSql(Statement statement,
-                                              String tempSql,
-                                              String schema,
-                                              String tableName,
-                                              String storeType,
-                                              Boolean enablePartition,
-                                              List<String> fieldNames,
-                                              String partitionFields,
-                                              Map<String, ArrayList<String>> rowDataMap) throws SQLException {
+    private void executeBatchSql(Statement statement,
+                                 String tempSql,
+                                 String schema,
+                                 String tableName,
+                                 String storeType,
+                                 Boolean enablePartition,
+                                 List<String> fieldNames,
+                                 String partitionFields,
+                                 Map<String, ArrayList<String>> rowDataMap) {
         StringBuilder valuesCondition = new StringBuilder();
         StringBuilder partitionCondition = new StringBuilder();
         String tableFieldsCondition = buildTableFieldsCondition(fieldNames, partitionFields);
@@ -510,14 +513,18 @@ private synchronized void executeBatchSql(Statement statement,
 
         // kudu ${partitionCondition} is null
         if (storeType.equalsIgnoreCase(KUDU_TYPE) || !enablePartition) {
-            rowData = rowDataMap.get(NO_PARTITION);
-            rowData.forEach(row -> valuesCondition.append(row).append(", "));
-            String executeSql = tempSql.replace(VALUES_CONDITION, valuesCondition.toString())
-                    .replace(PARTITION_CONDITION, partitionCondition.toString())
-                    .replace(PARTITION_CONSTANT, "")
-                    .replace(TABLE_FIELDS_CONDITION, tableFieldsCondition);
-            String substring = executeSql.substring(0, executeSql.length() - 2);
-            statement.execute(substring);
+            try {
+                rowData = rowDataMap.get(NO_PARTITION);
+                rowData.forEach(row -> valuesCondition.append(row).append(", "));
+                String executeSql = tempSql.replace(VALUES_CONDITION, valuesCondition.toString())
+                        .replace(PARTITION_CONDITION, partitionCondition.toString())
+                        .replace(PARTITION_CONSTANT, "")
+                        .replace(TABLE_FIELDS_CONDITION, tableFieldsCondition);
+                String substring = executeSql.substring(0, executeSql.length() - 2);
+                statement.execute(substring);
+            } catch (Exception e) {
+                throw new RuntimeException("execute impala SQL error!", e);
+            }
             return;
         }
 
@@ -535,7 +542,7 @@ private synchronized void executeBatchSql(Statement statement,
                 statement.execute(executeSql);
                 partitionCondition.delete(0, partitionCondition.length());
             } catch (SQLException sqlException) {
-                throw new RuntimeException("execute impala partition SQL error! ", sqlException);
+                throw new RuntimeException("execute impala SQL error! ", sqlException);
             }
         });
     }
@@ -583,7 +590,7 @@ private String buildTableFieldsCondition(List<String> fieldNames, String partiti
     private String buildValuesCondition(List<String> fieldTypes, Row row) {
         String valuesCondition = fieldTypes.stream().map(
                 f -> {
-                    for(String item : NEED_QUOTE_TYPE) {
+                    for (String item : NEED_QUOTE_TYPE) {
                         if (f.toLowerCase().contains(item)) {
                             return String.format("cast(? as %s)", f.toLowerCase());
                         }