Odtnhj

We have a 3 node confluent enterprise kafka cluster (linux on-prem) and one node has kafka connect service running . We want to ingest data into kafka topic using mysql .

Tried following stuff -

1.Installed mysql in my local windows desktop,created db ,table and inserted some data in it.

2.Created a source-quickstart-mysql.properties file with below details-

connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
tasks.max=1
connection.url=jdbc:mysql://<IPAddressOfLocalMachine>:3306/test_db?user=root&password=pwd
tables.whitelist=emp
mode=incrementing
incrementing.column.name=empid
topic.prefix=test-mysql-jdbc-

The connect-standalone.properties has this info:

bootstrap.servers=IPaddressOfKCnode:9092
plugin.path=/usr/share/java

3. Restarted kafka connect service




4. Tried to submit kafka connect service a request to connect to my sql -

curl -X POST -H "Accept:application/json" -H "Content-Type:application/json" IPaddressOfKCnode:8083/connectors/ -d '"name": "emp-connector", "config": "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector", "tasks.max": "1", "connection.url": "jdbc:mysql://IPaddressOfLocalMachine:3306/test_db?user=root&password=pwd","table.whitelist": "emp","mode": "timestamp","topic.prefix": "mysql-" '


Getting following error here:

"error_code":400,"message":"Connector configuration is invalid and contains the following 2 error(s):nInvalid value java.sql.SQLException: No suitable driver found for jdbc:mysql://X.X.X.X:3306/test_db?user=root&password=pwd for configuration Couldn't open connection to jdbc:mysql://X.X.X.X:3306/test_db?user=root&password=pwdnInvalid value java.sql.SQLException: No suitable driver found for jdbc:mysql://X.X.X.X:3306/test_db?user=root&password=admin for configuration Couldn't open connection to jdbc:mysql://X.X.X.X:3306/test_db?user=root&password=pwdnYou can also find the above list of errors at the endpoint `/connectorType/config/validate`"

I also tried following things like-

a.Stopped kafka connect service and ran manually -

systemctl stop confluent-kafka-connect

b. Ran the connect like this

/usr/bin/connect-standalone /etc/kafka/connect-standalone.properties /etc/kafka-connect-jdbc/source-quickstart-mysql.properties

This process starts successfully at the beginning but dies after some time . Here are the logs :

[2018-11-10 19:42:53,027] INFO Kafka Connect started (org.apache.kafka.connect.runtime.Connect:55)
[2018-11-10 19:42:53,048] INFO AbstractConfig values:
 batch.max.rows = 100
 catalog.pattern = null
 connection.attempts = 3
 connection.backoff.ms = 10000
 connection.password = null
 connection.url = jdbc:mysql://X.X.X.X:3306/test_db?user=root&password=pwd
 connection.user = null
 dialect.name =
 incrementing.column.name = empid
 mode = incrementing
 numeric.mapping = null
 numeric.precision.mapping = false
 poll.interval.ms = 5000
 query =
 schema.pattern = null
 table.blacklist = 
 table.poll.interval.ms = 60000
 table.types = [TABLE]
 table.whitelist = 
 timestamp.column.name = 
 timestamp.delay.interval.ms = 0
 topic.prefix = test-mysql-jdbc-
 validate.non.null = true
 (org.apache.kafka.common.config.AbstractConfig:279)
[2018-11-10 19:45:00,439] INFO AbstractConfig values:
 batch.max.rows = 100
 catalog.pattern = null
 connection.attempts = 3
 connection.backoff.ms = 10000
 connection.password = null
 connection.url = jdbc:mysql://X.X.X.X:3306/test_db?user=root&password=admin
 connection.user = null
 dialect.name =
 incrementing.column.name = empid
 mode = incrementing
 numeric.mapping = null
 numeric.precision.mapping = false
 poll.interval.ms = 5000
 query =
 schema.pattern = null
 table.blacklist = 
 table.poll.interval.ms = 60000
 table.types = [TABLE]
 table.whitelist = 
 timestamp.column.name = 
 timestamp.delay.interval.ms = 0
 topic.prefix = test-mysql-jdbc-
 validate.non.null = true
 (org.apache.kafka.common.config.AbstractConfig:279)
[2018-11-10 19:47:07,666] ERROR Failed to create job for /etc/kafka-connect-jdbc/source-quickstart-mysql.properties (org.apache.kafka.connect.cli.ConnectStandalone:102)
[2018-11-10 19:47:07,668] ERROR Stopping after connector error (org.apache.kafka.connect.cli.ConnectStandalone:113)
java.util.concurrent.ExecutionException: org.apache.kafka.connect.runtime.rest.errors.BadRequestException: Connector configuration is invalid and contains the following 2 error(s):
Invalid value com.mysql.jdbc.exceptions.jdbc4.CommunicationsException: Communications link failure

The last packet sent successfully to the server was 0 milliseconds ago. The driver has not received any packets from the server. for configuration Couldn't open connection to jdbc:mysql://192.168.178.14:3306/test_db?user=root&password=admin
Invalid value com.mysql.jdbc.exceptions.jdbc4.CommunicationsException: Communications link failure

The last packet sent successfully to the server was 0 milliseconds ago. The driver has not received any packets from the server. for configuration Couldn't open connection to jdbc:mysql://192.168.178.14:3306/test_db?user=root&password=admin
You can also find the above list of errors at the endpoint `/connectorType/config/validate`
 at org.apache.kafka.connect.util.ConvertingFutureCallback.result(ConvertingFutureCallback.java:79)
 at org.apache.kafka.connect.util.ConvertingFutureCallback.get(ConvertingFutureCallback.java:66)
 at org.apache.kafka.connect.cli.ConnectStandalone.main(ConnectStandalone.java:110)
Caused by: org.apache.kafka.connect.runtime.rest.errors.BadRequestException: Connector configuration is invalid and contains the following 2 error(s):
Invalid value com.mysql.jdbc.exceptions.jdbc4.CommunicationsException: Communications link failure

The last packet sent successfully to the server was 0 milliseconds ago. The driver has not received any packets from the server. for configuration Couldn't open connection to jdbc:mysql://192.168.178.14:3306/test_db?user=root&password=admin
Invalid value com.mysql.jdbc.exceptions.jdbc4.CommunicationsException: Communications link failure

The last packet sent successfully to the server was 0 milliseconds ago. The driver has not received any packets from the server. for configuration Couldn't open connection to jdbc:mysql://192.168.178.14:3306/test_db?user=root&password=admin
You can also find the above list of errors at the endpoint `/connectorType/config/validate`
 at org.apache.kafka.connect.runtime.AbstractHerder.maybeAddConfigErrors(AbstractHerder.java:415)
 at org.apache.kafka.connect.runtime.standalone.StandaloneHerder.putConnectorConfig(StandaloneHerder.java:189)
 at org.apache.kafka.connect.cli.ConnectStandalone.main(ConnectStandalone.java:107)
[2018-11-10 19:47:07,669] INFO Kafka Connect stopping (org.apache.kafka.connect.runtime.Connect:65)

I could not find a smooth and proper documentation on confluent website to use Kafka Connect service for its various connectors , configurations etc .
Please help in getting the correct steps to implement the data ingestion pipeline: mySQL--kafkaconnect--kafka

In the end I am expecting that inserts in mysql table produce data in kafka topic and the kafka consumer to display those records .
This ingestion seems to be simple but I am missing out some basic connection properties :(

Thanks!

It seems to be a problem with the JDBC connector. What MySQL version are you running?
To solve the issue, you need to:

1. Download the Connector/J 8.0.13 if you are running MySQL 8 or 5.1.47 for older versions.


2. Place the jar file under /usr/share/java/kafka-connect-jdbc/.


3. Restart Kafka Connect and start your mysql connector.

In the first instance, your error was returned to you from the curl command output:

Connector configuration is invalid and contains the following 2 error(s)
java.sql.SQLException: No suitable driver found for jdbc:mysql://X.X.X.X:3306/test_db?user=root&password=pwd

So, you were missing the JDBC driver for MySQL from your Kafka Connect path.

The second error is in the output you posted:

Connector configuration is invalid and contains the following 2 error(s):
Invalid value com.mysql.jdbc.exceptions.jdbc4.CommunicationsException: Communications link failure.
The last packet sent successfully to the server was 0 milliseconds ago. The driver has not received any packets from the server. for configuration Couldn't open connection to jdbc:mysql://192.168.178.14:3306/test_db?user=root&password=admin

this suggests a problem with Kafka Connect being able to reach your MySQL machine.

Where are you running Confluent Platform from, is it in Docker, a machine local to MySQL, etc? Is 192.168.178.14 the address for your MySQL server and can it be reached from the host where Kafka Connect is running?

You can find a couple of examples of setting up MySQL with Kafka:

• 
  https://www.confluent.io/blog/simplest-useful-kafka-connect-data-pipeline-world-thereabouts-part-1/ - this uses the JDBC connector, as you are doing


• 
  https://rmoff.net/2018/03/24/streaming-data-from-mysql-into-kafka-with-kafka-connect-and-debezium/ - this uses Debezium, a log-based CDC tool

For pros and cons of JDBC connector vs log-based CDC see https://www.confluent.io/blog/no-more-silos-how-to-integrate-your-databases-with-apache-kafka-and-cdc.

Disclaimer: I wrote the above blog posts.

Thank you Robin & Giorgos for your answers ! It helped a lot.
This problem had to do with a few things -
1. Missing MySQL JDBC connector jar .
We have to place the MySQL Connector/J 8.0.13 under /usr/share/java/kafka-connect-jdbc/.

2.The connection issue was causing because the MySQL user to which kafka connect was trying to connect had insufficient privileges to connect to remote connect service .
For this, I created a new mySQL user with full permission and access to remote server(Kafka connect) .

After the above steps , restarted the kafka-connect and the ingestion pipeline started working .