DBZ-538 Improve invalid DDL statement reporting

debezium-connector-mysql/src/main/java/io/debezium/connector/mysql/MySqlSchema.java

@@ -290,7 +290,6 @@ public boolean applyDdl(SourceInfo source, String databaseName, String ddlStatem
             if (skipUnparseableDDL) {
                 logger.warn("Ignoring unparseable DDL statement '{}': {}", ddlStatements);
             } else {
-                logger.error("Error parsing DDL statement and updating tables: {}", ddlStatements);
                 throw e;
             }
         } finally {

debezium-core/src/main/java/io/debezium/relational/ddl/DdlParser.java

@@ -289,7 +289,7 @@ public final void parse(TokenStream ddlContent, Tables databaseTables) throws Pa
             }
         } catch (ParsingException e) {
             ddlContent.rewind(marker);
-            throw e;
+            throw new ParsingException(e.getPosition(), "Failed to parse statement '" + ddlContent.getInputString() + "'", e);
         } catch (Throwable t) {
             parsingFailed(ddlContent.nextPosition(), "Unexpected exception (" + t.getMessage() + ") parsing", t);
         }

debezium-core/src/main/java/io/debezium/relational/history/AbstractDatabaseHistory.java

@@ -74,7 +74,6 @@ public final void recover(Map<String, ?> source, Map<String, ?> position, Tables
                         if (skipUnparseableDDL) {
                             logger.warn("Ignoring unparseable statements '{}' stored in database history: {}", ddl, e);
                         } else {
-                            logger.error("Cannot parse DDL statements {} stored in history, exiting", ddl);
                             throw e;
                         }
                     }

debezium-core/src/main/java/io/debezium/text/TokenStream.java

@@ -2691,4 +2691,11 @@ public void tokenize(CharacterStream input,
             }
         }
     }
+
+    /**
+     * @return a string to be parsed
+     */
+    public String getInputString() {
+        return inputString;
+    }
 }