private void checkFlushException() {
        if (schemaUpdateExceptionPolicy.equals(SchemaUpdateExceptionPolicy.THROW_WITH_STOP)
                && !tableExceptionMap.isEmpty()) {
            String tableErr = "Writing table get failed, tables are:";
            for (Map.Entry<String, Exception> entry : tableExceptionMap.entrySet()) {
                LOG.error("Writing table:{} get err:{}", entry.getKey(), entry.getValue().getMessage());
                tableErr = tableErr + entry.getKey() + ",";
            }
            throw new RuntimeException(tableErr.substring(0, tableErr.length() - 1));
        }
    }
Use reflection to initialize TableMetricStatementExecutor, and replace the original executor
 or upsertExecutor to calculate metrics.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/jdbc/src/main/java/org/apache/inlong/sort/jdbc/internal/JdbcMultiBatchingOutputFormat.java