title: MySQL to Confluent Enterprise Kafka data ingestion
body: <p>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 .</p>
  
  <p>Tried following stuff -</p>
  
  <p>1.Installed mysql in my local windows desktop,created db ,table and inserted some data in it.</p>
  
  <p>2.Created a <code>source-quickstart-mysql.properties</code> file with below details-</p>
  
  <pre><code>connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
  tasks.max=1
  connection.url=jdbc:mysql://&lt;IPAddressOfLocalMachine&gt;:3306/test_db?user=root&amp;password=pwd
  tables.whitelist=emp
  mode=incrementing
  incrementing.column.name=empid
  topic.prefix=test-mysql-jdbc-
  </code></pre>
  
  <p>The <code>connect-standalone.properties</code> has this info:</p>
  
  <pre><code>bootstrap.servers=IPaddressOfKCnode:9092
  plugin.path=/usr/share/java
  </code></pre>
  
  <ol start="3">
  <li><p>Restarted kafka connect service</p></li>
  <li><p>Tried to submit kafka connect service a request to connect to my sql -</p></li>
  </ol>
  
  <p><code>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&amp;password=pwd","table.whitelist": "emp","mode": "timestamp","topic.prefix": "mysql-" } }'
  </code></p>
  
  <p>Getting following error here:</p>
  
  <pre><code>{"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&amp;password=pwd for configuration Couldn't open connection to jdbc:mysql://X.X.X.X:3306/test_db?user=root&amp;password=pwd\nInvalid value java.sql.SQLException: No suitable driver found for jdbc:mysql://X.X.X.X:3306/test_db?user=root&amp;password=admin for configuration Couldn't open connection to jdbc:mysql://X.X.X.X:3306/test_db?user=root&amp;password=pwd\nYou can also find the above list of errors at the endpoint `/{connectorType}/config/validate`"}
  </code></pre>
  
  <p>I also tried following things like-</p>
  
  <p>a.Stopped kafka connect service and ran manually -</p>
  
  <pre><code>systemctl stop confluent-kafka-connect
  </code></pre>
  
  <p>b. Ran the connect like this</p>
  
  <pre><code>/usr/bin/connect-standalone /etc/kafka/connect-standalone.properties /etc/kafka-connect-jdbc/source-quickstart-mysql.properties
  </code></pre>
  
  <p>This process starts successfully at the beginning but dies after some time . Here are the logs :</p>
  
  <pre><code>[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&amp;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&amp;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&amp;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&amp;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&amp;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&amp;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)
  </code></pre>
  
  <p>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  </p>
  
  <p>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 :(</p>
  
  <p>Thanks!</p>