Getting "No fields found using key and value schemas for table" on JDBC SInk Connector with MemSQL

Nishant Verma

unread,

May 3, 2017, 11:32:15 PM5/3/17

to Confluent Platform

We are using Kafka Connect JDBC Connector. We have MemSQL as the JDBC sink. Below is the JSON which we have produced on a topic in kafka. It is the sample JSON as mentioned on confluent website for JDBC sink connector

Architecture:

2 broker node kafka cluster.

Kafka Connect in one of these servers.

Master of MemSQL on a third server.

Leaf nodes of MemSQL on other servers.

{"id": 999, "product": "foo", "quantity": 100, "price": 50}

Below is the connect-standalone.properties file:

bootstrap.servers=10.2.3.4:9092,10.5.6.7:9092

key.converter=org.apache.kafka.connect.json.JsonConverter

value.converter=org.apache.kafka.connect.json.JsonConverter

key.converter.schemas.enable=false

value.converter.schemas.enable=false

internal.key.converter=org.apache.kafka.connect.json.JsonConverter

internal.value.converter=org.apache.kafka.connect.json.JsonConverter

internal.key.converter.schemas.enable=false

internal.value.converter.schemas.enable=false

offset.flush.interval.ms=10000

offset.storage.file.filename=/media/data/confluent-3.2.0/connect.offsets

Below is the sink-quickstart-sqlite.properties file:

name=test-memsql

connector.class=io.confluent.connect.jdbc.JdbcSinkConnector

tasks.max=1

topics=test4

connection.url=jdbc:mysql://1.2.3.4:3306/ABCD

auto.create=true

I have created table test4 with ID, FNAME and LNAME columns. I have placed following jars in share/java/kafka-connect-jdbc:- mssql-jdbc-6.1.0.jre8.jar,

mysql-connector-java-5.1.30.jar and jdbc-mssql.jar.

I am using below command to start kafka Connect JDBC Connector:

./bin/connect-standalone etc/kafka/connect-standalone.properties etc/kafka-connect-jdbc/sink-quickstart-sqlite.properties

I cant see any record inserted in the table while doing "select * from ABCD.test4" .

In the logs, I see error as :

[2017-05-03 08:37:03,952] ERROR Task memsql-0 threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerSinkTask:449)

org.apache.kafka.connect.errors.ConnectException: No fields found using key and value schemas for table: test4

at io.confluent.connect.jdbc.sink.metadata.FieldsMetadata.extract(FieldsMetadata.java:190)

at io.confluent.connect.jdbc.sink.metadata.FieldsMetadata.extract(FieldsMetadata.java:58)

at io.confluent.connect.jdbc.sink.BufferedRecords.add(BufferedRecords.java:64)

at io.confluent.connect.jdbc.sink.JdbcDbWriter.write(JdbcDbWriter.java:62)

at io.confluent.connect.jdbc.sink.JdbcSinkTask.put(JdbcSinkTask.java:66)

at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:429)

at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:250)

at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:179)

at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:148)

at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:139)

at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:182)

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:1142)

at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)

at java.lang.Thread.run(Thread.java:745)

[2017-05-03 08:37:03,954] ERROR Task is being killed and will not recover until manually restarted

Kafka Version - 0.10.2.0

Kafka Connect Version - 3.2.0

MemSQL Version - 5.7.1

I am attaching the nohup.out after I executed kafka connect.

For testing purposes, I am using the sample JSON as mentioned on confluent website:

{"id": 999, "product": "foo", "quantity": 100, "price": 50}

nohup.out

Nishant Verma

unread,

May 5, 2017, 7:08:03 AM5/5/17

to Confluent Platform

Hi,

Can anyone provide inputs on this issue please?

Thanks

Nishant

Nishant Verma

unread,

May 8, 2017, 10:00:20 AM5/8/17

to Confluent Platform

I am using below in connect-standalone.properties:

key.converter=com.qubole.streamx.ByteArrayConverter

value.converter=com.qubole.streamx.ByteArrayConverter

Getting an error as:

[2017-05-08 09:56:38,399] ERROR Task test-sink-0 threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerSinkTask:449)

org.apache.kafka.connect.errors.ConnectException: Value schema must be of type Struct

at io.confluent.connect.jdbc.sink.metadata.FieldsMetadata.extract(FieldsMetadata.java:71)

at io.confluent.connect.jdbc.sink.metadata.FieldsMetadata.extract(FieldsMetadata.java:58)

at io.confluent.connect.jdbc.sink.BufferedRecords.add(BufferedRecords.java:65)

at io.confluent.connect.jdbc.sink.JdbcDbWriter.write(JdbcDbWriter.java:64)