Skip to content

/ kafka-connect-jdbc

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Mysql Sink : unknown table X in information_schema Exception #573

Open
rdinkel opened this issue Jan 17, 2019 · 8 comments
Open

Mysql Sink : unknown table X in information_schema Exception #573

rdinkel opened this issue Jan 17, 2019 · 8 comments
Labels
mysql

Comments

@rdinkel
Copy link

@rdinkel rdinkel commented Jan 17, 2019 •
edited

Hello,
the connector is able to read from the mysql db log and write it to kafka. I can see it via KSQL Server. But the sink is not working.
If the table "thermostat_data" is not there it will be created in the demodb schema.

{
  "name": "mysql-sink-thermostat-data",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "tasks.max": "1",
    "topics": "thermostat_data",
    "connection.url": "jdbc:mysql://127.0.0.1:3306/demodb?user=connector&password=XXX",
    "auto.create": "true",
    "insert.mode": "insert",
    "key.converter.schema.registry.url": "http://schema-registry-service:8081",
    "value.converter.schema.registry.url": "http://schema-registry-service:8081",
    "transforms": "unwrap",
    "transforms.unwrap.type": "io.debezium.transforms.UnwrapFromEnvelope"
  }
}

After that the connector crashes with

2019-01-17 10:32:06,074 ERROR  ||  WorkerSinkTask{id=mysql-sink-thermostat-data-0} RetriableException from SinkTask:   [org.apache.kafka.connect.runtime.WorkerSinkTask]
org.apache.kafka.connect.errors.RetriableException: java.sql.SQLException: java.sql.SQLSyntaxErrorException: Unknown table 'thermostat_data' in information_schema

	at io.confluent.connect.jdbc.sink.JdbcSinkTask.put(JdbcSinkTask.java:92)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:565)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:323)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:226)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:194)
	at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:175)
	at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:219)
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)
Caused by: java.sql.SQLException: java.sql.SQLSyntaxErrorException: Unknown table 'thermostat_data' in information_schema

in use:
mysql db server 5.7
mysql connector driver 8.0.11, 8.0.13 (tried both)
kafka-connect-jdbc-5.1.0
docker container: debezium/connect:0.9.0.Beta2

Please help
@rdinkel

This comment has been minimized.

Sign in to view
Copy link
Author

@rdinkel rdinkel commented Jan 17, 2019

Downgrading the mysql driver to 5.1.17 fixed the issue for me! :) the 6er version of the driver causes the same issue.
I hope that helps!
@rmoff rmoff added the mysql label Feb 5, 2019
@eyesmoker

This comment has been minimized.

Sign in to view
Copy link

@eyesmoker eyesmoker commented Feb 11, 2019

I have the same issue. if I downgrade the mysql connector java from 8.x to 5.17, sink connector works but mysql source connector fails. I needs this urgent. Any help?

Caused by: java.lang.NoClassDefFoundError: com/mysql/cj/jdbc/Driver at io.debezium.connector.mysql.MySqlConnectorConfig.<clinit>(MySqlConnectorConfig.java:695) at io.debezium.connector.mysql.MySqlConnector.validate(MySqlConnector.java:79) at org.apache.kafka.connect.runtime.AbstractHerder.validateConnectorConfig(AbstractHerder.java:282) at org.apache.kafka.connect.runtime.distributed.DistributedHerder$6.call(DistributedHerder.java:538) at org.apache.kafka.connect.runtime.distributed.DistributedHerder$6.call(DistributedHerder.java:535) at org.apache.kafka.connect.runtime.distributed.DistributedHerder.tick(DistributedHerder.java:271) at org.apache.kafka.connect.runtime.distributed.DistributedHerder.run(DistributedHerder.java:220) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ... 1 more

@heaje

This comment has been minimized.

Sign in to view
Copy link

@heaje heaje commented Mar 25, 2019

We are seeing the same issue with the Connect JDBC version 5.0.1 and MySQL JDBC driver 8.0.15. Downgrading the JDBC driver to 5.1.25 fixed the issue.
@gdtroszak

This comment has been minimized.

Sign in to view
Copy link

@gdtroszak gdtroszak commented Mar 27, 2019

We're indeed seeing the same issue, but downgrading our MySQL driver is not an option. Is there a timeline for a fix?
@rmoff rmoff changed the title Mysql Sink unkown table X in information_schema Exception Mysql Sink : unknown table X in information_schema Exception Apr 8, 2019
@rmoff

This comment has been minimized.

Sign in to view
Copy link

@rmoff rmoff commented Apr 8, 2019

Server version: 5.7.25 MySQL Community Server (GPL)
Kafka version: 2.2.0-cp2
MySQL driver: mysql-connector-java-8.0.13

curl -X POST http://localhost:8083/connectors -H "Content-Type: application/json" -d '
  { "name": "jdbc_sink_01",
    "config": {
     "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
     "tasks.max": "1",
     "topics": "mysql-00-customers",
     "table.name.format": "sink_${topic}",
                    "connection.url": "jdbc:mysql://mysql:3306/demo",
    "connection.user": "connect_user",
    "connection.password": "asgard",
     "auto.create": "true",
     "auto.evolve":"true",
     "pk.mode":"none"
    }
  }'

🔴 Fails:

[2019-04-08 09:34:12,120] INFO Checking MySql dialect for existence of table "sink_mysql-00-customers" (io.confluent.connect.jdbc.dialect.MySqlDatabaseDialect)
[2019-04-08 09:34:12,125] INFO Using MySql dialect table "sink_mysql-00-customers" present (io.confluent.connect.jdbc.dialect.MySqlDatabaseDialect)
[2019-04-08 09:34:12,129] WARN Write of 5 records failed, remainingRetries=10 (io.confluent.connect.jdbc.sink.JdbcSinkTask)
java.sql.SQLSyntaxErrorException: Unknown table 'sink_mysql-00-customers' in information_schema
   at com.mysql.cj.jdbc.exceptions.SQLError.createSQLException(SQLError.java:120)
   at com.mysql.cj.jdbc.exceptions.SQLError.createSQLException(SQLError.java:97)
   at com.mysql.cj.jdbc.exceptions.SQLExceptionsMapping.translateException(SQLExceptionsMapping.java:122)
   at com.mysql.cj.jdbc.StatementImpl.executeQuery(StatementImpl.java:1218)
   at com.mysql.cj.jdbc.DatabaseMetaData$7.forEach(DatabaseMetaData.java:2965)
   at com.mysql.cj.jdbc.DatabaseMetaData$7.forEach(DatabaseMetaData.java:2953)
   at com.mysql.cj.jdbc.IterateBlock.doForAll(IterateBlock.java:56)
   at com.mysql.cj.jdbc.DatabaseMetaData.getPrimaryKeys(DatabaseMetaData.java:3006)
   at io.confluent.connect.jdbc.dialect.GenericDatabaseDialect.primaryKeyColumns(GenericDatabaseDialect.java:717)
   at io.confluent.connect.jdbc.dialect.GenericDatabaseDialect.describeColumns(GenericDatabaseDialect.java:554)
   at io.confluent.connect.jdbc.dialect.GenericDatabaseDialect.describeTable(GenericDatabaseDialect.java:753)
   at io.confluent.connect.jdbc.util.TableDefinitions.get(TableDefinitions.java:62)
   at io.confluent.connect.jdbc.sink.DbStructure.createOrAmendIfNecessary(DbStructure.java:58)
   at io.confluent.connect.jdbc.sink.BufferedRecords.add(BufferedRecords.java:85)
   at io.confluent.connect.jdbc.sink.JdbcDbWriter.write(JdbcDbWriter.java:66)
   at io.confluent.connect.jdbc.sink.JdbcSinkTask.put(JdbcSinkTask.java:74)
   at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:538)
   at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:321)
   at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:224)
   at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:192)
   at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:175)
   at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:219)
   at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
   at java.util.concurrent.FutureTask.run(FutureTask.java:266)
   at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
   at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
   at java.lang.Thread.run(Thread.java:748)
[2019-04-08 09:34:12,129] INFO Closing connection #1 to MySql (io.confluent.connect.jdbc.util.CachedConnectionProvider)
[2019-04-08 09:34:12,131] INFO Initializing writer using SQL dialect: MySqlDatabaseDialect (io.confluent.connect.jdbc.sink.JdbcSinkTask)
[2019-04-08 09:34:12,132] ERROR WorkerSinkTask{id=jdbc_sink_01-0} RetriableException from SinkTask: (org.apache.kafka.connect.runtime.WorkerSinkTask)
org.apache.kafka.connect.errors.RetriableException: java.sql.SQLException: java.sql.SQLSyntaxErrorException: Unknown table 'sink_mysql-00-customers' in information_schema

   at io.confluent.connect.jdbc.sink.JdbcSinkTask.put(JdbcSinkTask.java:93)
   at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:538)
   at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:321)
   at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:224)
   at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:192)
   at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:175)
   at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:219)
   at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
   at java.util.concurrent.FutureTask.run(FutureTask.java:266)
   at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
   at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
   at java.lang.Thread.run(Thread.java:748)
Caused by: java.sql.SQLException: java.sql.SQLSyntaxErrorException: Unknown table 'sink_mysql-00-customers' in information_schema

   ... 12 more

✅ Same config but with MySQL driver: mysql-connector-java-5.1.47 - works fine, records written to MySQL.

Looking at the DEBUG for the connector in both configurations I see:

DEBUG Querying MySql dialect column metadata for catalog:null schema:demo table:sink_mysql-00-customers (io.confluent.connect.jdbc.dialect.MySqlDatabaseDialect)

With the v5 driver we then get:

INFO Setting metadata for table "demo"."sink_mysql-00-customers" to Table{name='"demo"."sink_mysql-00-customers"', columns=[Column{'gender', isPrimaryKey=false, allowsNull=true, sqlType=VARCHAR}, Column{'id', isPrimaryKey=false, allowsNull=false, sqlType=INT}, Column{'email', isPrimaryKey=false, allowsNull=true, sqlType=VARCHAR}, Column{'UPDATE_TS', isPrimaryKey=false, allowsNull=false, sqlType=DATETIME}, Column{'comments', isPrimaryKey=false, allowsNull=true, sqlType=VARCHAR}, Column{'first_name', isPrimaryKey=false, allowsNull=true, sqlType=VARCHAR}, Column{'last_name', isPrimaryKey=false, allowsNull=true, sqlType=VARCHAR}]} (io.confluent.connect.jdbc.util.TableDefinitions)

Whilst with the v8 drivers the failure is logged:

[2019-04-08 09:55:21,303] WARN Write of 5 records failed, remainingRetries=10 (io.confluent.connect.jdbc.sink.JdbcSinkTask)
java.sql.SQLSyntaxErrorException: Unknown table 'sink_mysql-00-customers' in information_schema
        at com.mysql.cj.jdbc.exceptions.SQLError.createSQLException(SQLError.java:120)
        at com.mysql.cj.jdbc.exceptions.SQLError.createSQLException(SQLError.java:97)
        at com.mysql.cj.jdbc.exceptions.SQLExceptionsMapping.translateException(SQLExceptionsMapping.java:122)
        at com.mysql.cj.jdbc.StatementImpl.executeQuery(StatementImpl.java:1218)
        at com.mysql.cj.jdbc.DatabaseMetaData$7.forEach(DatabaseMetaData.java:2965)
        at com.mysql.cj.jdbc.DatabaseMetaData$7.forEach(DatabaseMetaData.java:2953)
        at com.mysql.cj.jdbc.IterateBlock.doForAll(IterateBlock.java:56)
        at com.mysql.cj.jdbc.DatabaseMetaData.getPrimaryKeys(DatabaseMetaData.java:3006)
        at io.confluent.connect.jdbc.dialect.GenericDatabaseDialect.primaryKeyColumns(GenericDatabaseDialect.java:717)
        at io.confluent.connect.jdbc.dialect.GenericDatabaseDialect.describeColumns(GenericDatabaseDialect.java:554)
        at io.confluent.connect.jdbc.dialect.GenericDatabaseDialect.describeTable(GenericDatabaseDialect.java:753)
        at io.confluent.connect.jdbc.util.TableDefinitions.get(TableDefinitions.java:62)
        at io.confluent.connect.jdbc.sink.DbStructure.createOrAmendIfNecessary(DbStructure.java:58)
        at io.confluent.connect.jdbc.sink.BufferedRecords.add(BufferedRecords.java:85)
        at io.confluent.connect.jdbc.sink.JdbcDbWriter.write(JdbcDbWriter.java:66)
        at io.confluent.connect.jdbc.sink.JdbcSinkTask.put(JdbcSinkTask.java:74)
        at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:538)
        at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:321)
        at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:224)
        at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:192)
        at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:175)
        at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:219)
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
        at java.util.concurrent.FutureTask.run(FutureTask.java:266)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)
[2019-04-08 09:55:21,304] INFO Closing connection #1 to MySql (io.confluent.connect.jdbc.util.CachedConnectionProvider)
@rmoff

This comment has been minimized.

Sign in to view
Copy link

@rmoff rmoff commented Apr 8, 2019

@eyesmoker this issue is about the JDBCSink - please raise another issue and reference this one if you are seeing issues with the JDBCSource that you think are related.
@gdtroszak

This comment has been minimized.

Sign in to view
Copy link

@gdtroszak gdtroszak commented Apr 8, 2019

We've managed to find a workaround that doesn't involve downgrading the MySQL JDBC driver. Appending the nullCatalogMeansCurrent=true query param to the JDBC connection url appears to fix this problem for us.

For example, "connection.url": "jdbc:mysql://mysql:3306/demo?nullCatalogMeansCurrent=true"
@zymergen-glonkar

This comment has been minimized.

Sign in to view
Copy link

@zymergen-glonkar zymergen-glonkar commented Apr 8, 2019 •
edited

@rmoff some context and then request for guidance on next steps..
context..
the issue i believe is because of this:
https://dev.mysql.com/doc/connector-j/8.0/en/connector-j-properties-changed.html

which lists at the very end that the default value for nullCatalogMeansCurrent is now false..
so this affects anyone who is upgrading to use mysql driver 8.X. Anyone who is using Debezium on Kafka Connect 5.X i think uses the latest version of debezium which is using the mysql driver 8.X . We have added this JAR to our connect build image. After the update to our connect image use the Mysql 8.X driver the JDBC sink stopped working.. until above change fixed the issue.
Hence I think the next steps i think would be..

a) either update the JDBC Connector documentation to specify this so that others know how to fix. (BTW, Confluent documentation has be exceptional !!!)
b) or update the JDBC Connector source code to check above condition (i.e. driver version) and add the flag automatically ?
c) something else ?

happy to help with either, though I do not know about the exact steps per se and will need guidance :)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Assignees
No one assigned
Labels
mysql
Projects
None yet
Milestone
No milestone
Linked pull requests

Successfully merging a pull request may close this issue.

None yet
6 participants
@rdinkel @eyesmoker @heaje @gdtroszak @rmoff @zymergen-glonkar
You can’t perform that action at this time.