Kafka-connect-jdbc: JDBC source connector is not working with timestamp+incrementing mode

Created on 11 Jul 2017  ·  26Comments  ·  Source: confluentinc/kafka-connect-jdbc

Hi,
I am trying to use mode timestamp+incrementing with mysql, but it does not create any topic in my kafka queue when I do so and there is also no error log.

Here are the connector properties that I am using,

name=test-source-mysql-jdbc-autoincrement
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
tasks.max=1
connection.url=jdbc:mysql://127.0.0.1:3306/test?user=root&password=root
connection.user=root
connection.password=root

mode=incrementing

mode=timestamp+incrementing
timestamp.column.name=name
incrementing.column.name=name
topic.prefix=test-mysql-t3-

My database test contains a single table example(id int,name varchar(40)).

picture himani1

Most helpful comment

@himani1 that doc refers to the type of the configuration option. The timestamp.column.name field should be a string (the name of the column that is a timestamp column). If you look at the description for that option, it provides more information: The name of the timestamp column to use to detect new or modified rows. This column may not be nullable.

picture makearl on 14 Jul 2017
👍2

All 26 comments

@himani1 Can you provide any logs that might give us a better idea of what's going wrong?

ewencp picture ewencp on 12 Jul 2017

@himani1 incrementing.column.name should correspond to a strictly incrementing column in your table. In your example table, this would likely be the id field. timestamp.column.name should refer to a timestamp column which it doesn't look like your example table has. Both the timestamp and incrementing columns need to be not null.

makearl picture makearl on 13 Jul 2017

@makearl Are you sure? According to this doc timestamp.column.name should be of type String.

himani1 picture himani1 on 14 Jul 2017

@himani1 that doc refers to the type of the configuration option. The timestamp.column.name field should be a string (the name of the column that is a timestamp column). If you look at the description for that option, it provides more information: The name of the timestamp column to use to detect new or modified rows. This column may not be nullable.

makearl picture makearl on 14 Jul 2017
👍2

@makearl Thanks, it worked.

himani1 picture himani1 on 28 Jul 2017

Hi, i am facing the same issue and not been able to solve it for quite sometime now. Can anyone help?
Below are the details of jdbc-quickstart.properties:

name=test-source-oracle-jdbc-autoincrement
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
tasks.max=1
connection.url=jdbc:oracle:thin:ent_app/[email protected]:1521:xe
schema.pattern=ENT_APP
table.whitelist=ENT_TS_DEMO
mode=timestamp+incrementing
timestamp.column.name=UPDATED_DATE_TIME
incrementing.column.name=ID
topic.prefix=ns-
poll.interval.ms=1000
debug=true

And the table details are as below: (I am connecting to Oracle DB)

Name              Null     Type         
----------------- -------- ------------ 
ID                NOT NULL NUMBER(10)   
UPDATED_DATE_TIME NOT NULL TIMESTAMP(6) 
NAME              NOT NULL VARCHAR2(20)

My observations so far:

  • mode=incrementing is working fine.
  • mode=timestamp+incrementing is not working, at this mode, even inserts are not being pushed o kafka.
Sarvesh-D picture Sarvesh-D on 31 Jul 2017

@Sarvesh-D Do you have any logs? If the logs you're getting by default aren't sufficient, I'd suggest turning the log level up to DEBUG in your connect-log4j.properties configuration.

ewencp picture ewencp on 31 Jul 2017

@ewencp Thanks for quick reply. I inserted a record in table and updated same as well, and took the DEBUG level logs. On analysing i saw there is something strange in lines:

[2017-08-01 09:59:24,538] DEBUG executing query select CURRENT_TIMESTAMP from dual to get current time from database (io.confluent.connect.jdbc.util.JdbcUtils:198)
[2017-08-01 09:59:24,540] DEBUG Executing prepared statement with start time value = 1970-01-01 00:00:00.000 end time = 2017-08-01 04:29:23.944 and incrementing value = -1

As you see the end time = 2017-08-01 04:29:23.944 whereas the current time was [2017-08-01 09:59:24,540].

I Thought this would be some UTC issue as mentioned in issue-103. But even changing the DDL (as mentioned in the comments) did not help.

Pasting complete logs below for your analysis. (I made poll.interval.ms=5000 to avoid clutter)

2017-08-01 09:59:15,096] INFO Reflections took 12600 ms to scan 561 urls, producing 13343 keys and 87059 values  (org.reflections.Reflections:229)
[2017-08-01 09:59:19,524] DEBUG Checking for next block of results from TimestampIncrementingTableQuerier{name='ENT_TS_DEMO', query='null', topicPrefix='n-', timestampColumn='UPDATED_DATE_TIME', incrementingColumn='ID'} (io.confluent.connect.jdbc.source.JdbcSourceTask:199)
[2017-08-01 09:59:19,525] DEBUG TimestampIncrementingTableQuerier{name='ENT_TS_DEMO', query='null', topicPrefix='n-', timestampColumn='UPDATED_DATE_TIME', incrementingColumn='ID'} prepared SQL query: SELECT * FROM "ENT_TS_DEMO" WHERE "UPDATED_DATE_TIME" < ? AND (("UPDATED_DATE_TIME" = ? AND "ID" > ?) OR "UPDATED_DATE_TIME" > ?) ORDER BY "UPDATED_DATE_TIME","ID" ASC (io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier:145)
[2017-08-01 09:59:19,525] DEBUG executing query select CURRENT_TIMESTAMP from dual to get current time from database (io.confluent.connect.jdbc.util.JdbcUtils:198)
[2017-08-01 09:59:19,527] DEBUG Executing prepared statement with start time value = 1970-01-01 00:00:00.000 end time = 2017-08-01 04:29:18.930 and incrementing value = -1 (io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier:159)
[2017-08-01 09:59:19,528] DEBUG Resetting querier TimestampIncrementingTableQuerier{name='ENT_TS_DEMO', query='null', topicPrefix='n-', timestampColumn='UPDATED_DATE_TIME', incrementingColumn='ID'} (io.confluent.connect.jdbc.source.JdbcSourceTask:232)
[2017-08-01 09:59:24,532] DEBUG Checking for next block of results from TimestampIncrementingTableQuerier{name='ENT_TS_DEMO', query='null', topicPrefix='n-', timestampColumn='UPDATED_DATE_TIME', incrementingColumn='ID'} (io.confluent.connect.jdbc.source.JdbcSourceTask:199)
[2017-08-01 09:59:24,538] DEBUG TimestampIncrementingTableQuerier{name='ENT_TS_DEMO', query='null', topicPrefix='n-', timestampColumn='UPDATED_DATE_TIME', incrementingColumn='ID'} prepared SQL query: SELECT * FROM "ENT_TS_DEMO" WHERE "UPDATED_DATE_TIME" < ? AND (("UPDATED_DATE_TIME" = ? AND "ID" > ?) OR "UPDATED_DATE_TIME" > ?) ORDER BY "UPDATED_DATE_TIME","ID" ASC (io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier:145)
[2017-08-01 09:59:24,538] DEBUG executing query select CURRENT_TIMESTAMP from dual to get current time from database (io.confluent.connect.jdbc.util.JdbcUtils:198)
[2017-08-01 09:59:24,540] DEBUG Executing prepared statement with start time value = 1970-01-01 00:00:00.000 end time = 2017-08-01 04:29:23.944 and incrementing value = -1 (io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier:159)
[2017-08-01 09:59:24,541] DEBUG Resetting querier TimestampIncrementingTableQuerier{name='ENT_TS_DEMO', query='null', topicPrefix='n-', timestampColumn='UPDATED_DATE_TIME', incrementingColumn='ID'} (io.confluent.connect.jdbc.source.JdbcSourceTask:232)
[2017-08-01 09:59:29,544] DEBUG Checking for next block of results from TimestampIncrementingTableQuerier{name='ENT_TS_DEMO', query='null', topicPrefix='n-', timestampColumn='UPDATED_DATE_TIME', incrementingColumn='ID'} (io.confluent.connect.jdbc.source.JdbcSourceTask:199)
[2017-08-01 09:59:29,545] DEBUG TimestampIncrementingTableQuerier{name='ENT_TS_DEMO', query='null', topicPrefix='n-', timestampColumn='UPDATED_DATE_TIME', incrementingColumn='ID'} prepared SQL query: SELECT * FROM "ENT_TS_DEMO" WHERE "UPDATED_DATE_TIME" < ? AND (("UPDATED_DATE_TIME" = ? AND "ID" > ?) OR "UPDATED_DATE_TIME" > ?) ORDER BY "UPDATED_DATE_TIME","ID" ASC (io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier:145)
[2017-08-01 09:59:29,545] DEBUG executing query select CURRENT_TIMESTAMP from dual to get current time from database (io.confluent.connect.jdbc.util.JdbcUtils:198)
[2017-08-01 09:59:29,547] DEBUG Executing prepared statement with start time value = 1970-01-01 00:00:00.000 end time = 2017-08-01 04:29:28.950 and incrementing value = -1 (io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier:159)
[2017-08-01 09:59:29,548] DEBUG Resetting querier TimestampIncrementingTableQuerier{name='ENT_TS_DEMO', query='null', topicPrefix='n-', timestampColumn='UPDATED_DATE_TIME', incrementingColumn='ID'} (io.confluent.connect.jdbc.source.JdbcSourceTask:232)
[2017-08-01 09:59:32,626] DEBUG Scavenging sessions at 1501561772626 (org.eclipse.jetty.server.session:347)
[2017-08-01 09:59:34,552] DEBUG Checking for next block of results from TimestampIncrementingTableQuerier{name='ENT_TS_DEMO', query='null', topicPrefix='n-', timestampColumn='UPDATED_DATE_TIME', incrementingColumn='ID'} (io.confluent.connect.jdbc.source.JdbcSourceTask:199)
[2017-08-01 09:59:34,554] DEBUG TimestampIncrementingTableQuerier{name='ENT_TS_DEMO', query='null', topicPrefix='n-', timestampColumn='UPDATED_DATE_TIME', incrementingColumn='ID'} prepared SQL query: SELECT * FROM "ENT_TS_DEMO" WHERE "UPDATED_DATE_TIME" < ? AND (("UPDATED_DATE_TIME" = ? AND "ID" > ?) OR "UPDATED_DATE_TIME" > ?) ORDER BY "UPDATED_DATE_TIME","ID" ASC (io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier:145)
[2017-08-01 09:59:34,554] DEBUG executing query select CURRENT_TIMESTAMP from dual to get current time from database (io.confluent.connect.jdbc.util.JdbcUtils:198)
[2017-08-01 09:59:34,555] DEBUG Executing prepared statement with start time value = 1970-01-01 00:00:00.000 end time = 2017-08-01 04:29:33.959 and incrementing value = -1 (io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier:159)
[2017-08-01 09:59:34,557] DEBUG Resetting querier TimestampIncrementingTableQuerier{name='ENT_TS_DEMO', query='null', topicPrefix='n-', timestampColumn='UPDATED_DATE_TIME', incrementingColumn='ID'} (io.confluent.connect.jdbc.source.JdbcSourceTask:232)
[2017-08-01 09:59:39,561] DEBUG Checking for next block of results from TimestampIncrementingTableQuerier{name='ENT_TS_DEMO', query='null', topicPrefix='n-', timestampColumn='UPDATED_DATE_TIME', incrementingColumn='ID'} (io.confluent.connect.jdbc.source.JdbcSourceTask:199)
[2017-08-01 09:59:39,563] DEBUG TimestampIncrementingTableQuerier{name='ENT_TS_DEMO', query='null', topicPrefix='n-', timestampColumn='UPDATED_DATE_TIME', incrementingColumn='ID'} prepared SQL query: SELECT * FROM "ENT_TS_DEMO" WHERE "UPDATED_DATE_TIME" < ? AND (("UPDATED_DATE_TIME" = ? AND "ID" > ?) OR "UPDATED_DATE_TIME" > ?) ORDER BY "UPDATED_DATE_TIME","ID" ASC (io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier:145)
[2017-08-01 09:59:39,563] DEBUG executing query select CURRENT_TIMESTAMP from dual to get current time from database (io.confluent.connect.jdbc.util.JdbcUtils:198)
[2017-08-01 09:59:39,564] DEBUG Executing prepared statement with start time value = 1970-01-01 00:00:00.000 end time = 2017-08-01 04:29:38.968 and incrementing value = -1 (io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier:159)
[2017-08-01 09:59:39,566] DEBUG Resetting querier TimestampIncrementingTableQuerier{name='ENT_TS_DEMO', query='null', topicPrefix='n-', timestampColumn='UPDATED_DATE_TIME', incrementingColumn='ID'} (io.confluent.connect.jdbc.source.JdbcSourceTask:232)
[2017-08-01 09:59:44,568] DEBUG Checking for next block of results from TimestampIncrementingTableQuerier{name='ENT_TS_DEMO', query='null', topicPrefix='n-', timestampColumn='UPDATED_DATE_TIME', incrementingColumn='ID'} (io.confluent.connect.jdbc.source.JdbcSourceTask:199)
[2017-08-01 09:59:44,570] DEBUG TimestampIncrementingTableQuerier{name='ENT_TS_DEMO', query='null', topicPrefix='n-', timestampColumn='UPDATED_DATE_TIME', incrementingColumn='ID'} prepared SQL query: SELECT * FROM "ENT_TS_DEMO" WHERE "UPDATED_DATE_TIME" < ? AND (("UPDATED_DATE_TIME" = ? AND "ID" > ?) OR "UPDATED_DATE_TIME" > ?) ORDER BY "UPDATED_DATE_TIME","ID" ASC (io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier:145)
[2017-08-01 09:59:44,570] DEBUG executing query select CURRENT_TIMESTAMP from dual to get current time from database (io.confluent.connect.jdbc.util.JdbcUtils:198)
[2017-08-01 09:59:44,571] DEBUG Executing prepared statement with start time value = 1970-01-01 00:00:00.000 end time = 2017-08-01 04:29:43.975 and incrementing value = -1 (io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier:159)
[2017-08-01 09:59:44,573] DEBUG Resetting querier TimestampIncrementingTableQuerier{name='ENT_TS_DEMO', query='null', topicPrefix='n-', timestampColumn='UPDATED_DATE_TIME', incrementingColumn='ID'} (io.confluent.connect.jdbc.source.JdbcSourceTask:232)
[2017-08-01 09:59:49,578] DEBUG Checking for next block of results from TimestampIncrementingTableQuerier{name='ENT_TS_DEMO', query='null', topicPrefix='n-', timestampColumn='UPDATED_DATE_TIME', incrementingColumn='ID'} (io.confluent.connect.jdbc.source.JdbcSourceTask:199)
[2017-08-01 09:59:49,580] DEBUG TimestampIncrementingTableQuerier{name='ENT_TS_DEMO', query='null', topicPrefix='n-', timestampColumn='UPDATED_DATE_TIME', incrementingColumn='ID'} prepared SQL query: SELECT * FROM "ENT_TS_DEMO" WHERE "UPDATED_DATE_TIME" < ? AND (("UPDATED_DATE_TIME" = ? AND "ID" > ?) OR "UPDATED_DATE_TIME" > ?) ORDER BY "UPDATED_DATE_TIME","ID" ASC (io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier:145)
[2017-08-01 09:59:49,581] DEBUG executing query select CURRENT_TIMESTAMP from dual to get current time from database (io.confluent.connect.jdbc.util.JdbcUtils:198)
[2017-08-01 09:59:49,582] DEBUG Executing prepared statement with start time value = 1970-01-01 00:00:00.000 end time = 2017-08-01 04:29:48.986 and incrementing value = -1 (io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier:159)
[2017-08-01 09:59:49,583] DEBUG Resetting querier TimestampIncrementingTableQuerier{name='ENT_TS_DEMO', query='null', topicPrefix='n-', timestampColumn='UPDATED_DATE_TIME', incrementingColumn='ID'} (io.confluent.connect.jdbc.source.JdbcSourceTask:232)
[2017-08-01 09:59:54,588] DEBUG Checking for next block of results from TimestampIncrementingTableQuerier{name='ENT_TS_DEMO', query='null', topicPrefix='n-', timestampColumn='UPDATED_DATE_TIME', incrementingColumn='ID'} (io.confluent.connect.jdbc.source.JdbcSourceTask:199)
[2017-08-01 09:59:54,589] DEBUG TimestampIncrementingTableQuerier{name='ENT_TS_DEMO', query='null', topicPrefix='n-', timestampColumn='UPDATED_DATE_TIME', incrementingColumn='ID'} prepared SQL query: SELECT * FROM "ENT_TS_DEMO" WHERE "UPDATED_DATE_TIME" < ? AND (("UPDATED_DATE_TIME" = ? AND "ID" > ?) OR "UPDATED_DATE_TIME" > ?) ORDER BY "UPDATED_DATE_TIME","ID" ASC (io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier:145)
[2017-08-01 09:59:54,590] DEBUG executing query select CURRENT_TIMESTAMP from dual to get current time from database (io.confluent.connect.jdbc.util.JdbcUtils:198)
[2017-08-01 09:59:54,591] DEBUG Executing prepared statement with start time value = 1970-01-01 00:00:00.000 end time = 2017-08-01 04:29:53.995 and incrementing value = -1 (io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier:159)
[2017-08-01 09:59:54,592] DEBUG Resetting querier TimestampIncrementingTableQuerier{name='ENT_TS_DEMO', query='null', topicPrefix='n-', timestampColumn='UPDATED_DATE_TIME', incrementingColumn='ID'} (io.confluent.connect.jdbc.source.JdbcSourceTask:232)
[2017-08-01 09:59:59,595] DEBUG Checking for next block of results from TimestampIncrementingTableQuerier{name='ENT_TS_DEMO', query='null', topicPrefix='n-', timestampColumn='UPDATED_DATE_TIME', incrementingColumn='ID'} (io.confluent.connect.jdbc.source.JdbcSourceTask:199)
[2017-08-01 09:59:59,597] DEBUG TimestampIncrementingTableQuerier{name='ENT_TS_DEMO', query='null', topicPrefix='n-', timestampColumn='UPDATED_DATE_TIME', incrementingColumn='ID'} prepared SQL query: SELECT * FROM "ENT_TS_DEMO" WHERE "UPDATED_DATE_TIME" < ? AND (("UPDATED_DATE_TIME" = ? AND "ID" > ?) OR "UPDATED_DATE_TIME" > ?) ORDER BY "UPDATED_DATE_TIME","ID" ASC (io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier:145)
[2017-08-01 09:59:59,597] DEBUG executing query select CURRENT_TIMESTAMP from dual to get current time from database (io.confluent.connect.jdbc.util.JdbcUtils:198)
[2017-08-01 09:59:59,598] DEBUG Executing prepared statement with start time value = 1970-01-01 00:00:00.000 end time = 2017-08-01 04:29:59.002 and incrementing value = -1 (io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier:159)
[2017-08-01 09:59:59,600] DEBUG Resetting querier TimestampIncrementingTableQuerier{name='ENT_TS_DEMO', query='null', topicPrefix='n-', timestampColumn='UPDATED_DATE_TIME', incrementingColumn='ID'} (io.confluent.connect.jdbc.source.JdbcSourceTask:232)
[2017-08-01 10:00:02,627] DEBUG Scavenging sessions at 1501561802627 (org.eclipse.jetty.server.session:347)
[2017-08-01 10:00:04,213] DEBUG Got the following tables: [ENT_TS_DEMO] (io.confluent.connect.jdbc.source.TableMonitorThread:115)
[2017-08-01 10:00:04,266] DEBUG Committing offsets for WorkerSourceTask{id=test-source-oracle-jdbc-autoincrement-0} (org.apache.kafka.connect.runtime.SourceTaskOffsetCommitter:107)
[2017-08-01 10:00:04,266] DEBUG WorkerSourceTask{id=test-source-oracle-jdbc-autoincrement-0} Committing offsets (org.apache.kafka.connect.runtime.WorkerSourceTask:282)
[2017-08-01 10:00:04,266] DEBUG WorkerSourceTask{id=test-source-oracle-jdbc-autoincrement-0} flushing 0 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask:299)
[2017-08-01 10:00:04,266] DEBUG Finished WorkerSourceTask{id=test-source-oracle-jdbc-autoincrement-0} offset commitOffsets successfully in 0 ms (org.apache.kafka.connect.runtime.WorkerSourceTask:325)
[2017-08-01 10:00:04,603] DEBUG Checking for next block of results from TimestampIncrementingTableQuerier{name='ENT_TS_DEMO', query='null', topicPrefix='n-', timestampColumn='UPDATED_DATE_TIME', incrementingColumn='ID'} (io.confluent.connect.jdbc.source.JdbcSourceTask:199)
[2017-08-01 10:00:04,605] DEBUG TimestampIncrementingTableQuerier{name='ENT_TS_DEMO', query='null', topicPrefix='n-', timestampColumn='UPDATED_DATE_TIME', incrementingColumn='ID'} prepared SQL query: SELECT * FROM "ENT_TS_DEMO" WHERE "UPDATED_DATE_TIME" < ? AND (("UPDATED_DATE_TIME" = ? AND "ID" > ?) OR "UPDATED_DATE_TIME" > ?) ORDER BY "UPDATED_DATE_TIME","ID" ASC (io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier:145)
[2017-08-01 10:00:04,606] DEBUG executing query select CURRENT_TIMESTAMP from dual to get current time from database (io.confluent.connect.jdbc.util.JdbcUtils:198)
[2017-08-01 10:00:04,607] DEBUG Executing prepared statement with start time value = 1970-01-01 00:00:00.000 end time = 2017-08-01 04:30:04.012 and incrementing value = -1 (io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier:159)
[2017-08-01 10:00:04,609] DEBUG Resetting querier TimestampIncrementingTableQuerier{name='ENT_TS_DEMO', query='null', topicPrefix='n-', timestampColumn='UPDATED_DATE_TIME', incrementingColumn='ID'} (io.confluent.connect.jdbc.source.JdbcSourceTask:232)
[2017-08-01 10:00:09,613] DEBUG Checking for next block of results from TimestampIncrementingTableQuerier{name='ENT_TS_DEMO', query='null', topicPrefix='n-', timestampColumn='UPDATED_DATE_TIME', incrementingColumn='ID'} (io.confluent.connect.jdbc.source.JdbcSourceTask:199)
[2017-08-01 10:00:09,614] DEBUG TimestampIncrementingTableQuerier{name='ENT_TS_DEMO', query='null', topicPrefix='n-', timestampColumn='UPDATED_DATE_TIME', incrementingColumn='ID'} prepared SQL query: SELECT * FROM "ENT_TS_DEMO" WHERE "UPDATED_DATE_TIME" < ? AND (("UPDATED_DATE_TIME" = ? AND "ID" > ?) OR "UPDATED_DATE_TIME" > ?) ORDER BY "UPDATED_DATE_TIME","ID" ASC (io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier:145)
[2017-08-01 10:00:09,614] DEBUG executing query select CURRENT_TIMESTAMP from dual to get current time from database (io.confluent.connect.jdbc.util.JdbcUtils:198)
[2017-08-01 10:00:09,615] DEBUG Executing prepared statement with start time value = 1970-01-01 00:00:00.000 end time = 2017-08-01 04:30:09.020 and incrementing value = -1 (io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier:159)
[2017-08-01 10:00:09,616] DEBUG Resetting querier TimestampIncrementingTableQuerier{name='ENT_TS_DEMO', query='null', topicPrefix='n-', timestampColumn='UPDATED_DATE_TIME', incrementingColumn='ID'} (io.confluent.connect.jdbc.source.JdbcSourceTask:232)
[2017-08-01 10:00:14,621] DEBUG Checking for next block of results from TimestampIncrementingTableQuerier{name='ENT_TS_DEMO', query='null', topicPrefix='n-', timestampColumn='UPDATED_DATE_TIME', incrementingColumn='ID'} (io.confluent.connect.jdbc.source.JdbcSourceTask:199)
[2017-08-01 10:00:14,622] DEBUG TimestampIncrementingTableQuerier{name='ENT_TS_DEMO', query='null', topicPrefix='n-', timestampColumn='UPDATED_DATE_TIME', incrementingColumn='ID'} prepared SQL query: SELECT * FROM "ENT_TS_DEMO" WHERE "UPDATED_DATE_TIME" < ? AND (("UPDATED_DATE_TIME" = ? AND "ID" > ?) OR "UPDATED_DATE_TIME" > ?) ORDER BY "UPDATED_DATE_TIME","ID" ASC (io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier:145)
[2017-08-01 10:00:14,622] DEBUG executing query select CURRENT_TIMESTAMP from dual to get current time from database (io.confluent.connect.jdbc.util.JdbcUtils:198)
[2017-08-01 10:00:14,623] DEBUG Executing prepared statement with start time value = 1970-01-01 00:00:00.000 end time = 2017-08-01 04:30:14.027 and incrementing value = -1 (io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier:159)
[2017-08-01 10:00:14,624] DEBUG Resetting querier TimestampIncrementingTableQuerier{name='ENT_TS_DEMO', query='null', topicPrefix='n-', timestampColumn='UPDATED_DATE_TIME', incrementingColumn='ID'} (io.confluent.connect.jdbc.source.JdbcSourceTask:232)
[2017-08-01 10:00:19,629] DEBUG Checking for next block of results from TimestampIncrementingTableQuerier{name='ENT_TS_DEMO', query='null', topicPrefix='n-', timestampColumn='UPDATED_DATE_TIME', incrementingColumn='ID'} (io.confluent.connect.jdbc.source.JdbcSourceTask:199)
[2017-08-01 10:00:19,630] DEBUG TimestampIncrementingTableQuerier{name='ENT_TS_DEMO', query='null', topicPrefix='n-', timestampColumn='UPDATED_DATE_TIME', incrementingColumn='ID'} prepared SQL query: SELECT * FROM "ENT_TS_DEMO" WHERE "UPDATED_DATE_TIME" < ? AND (("UPDATED_DATE_TIME" = ? AND "ID" > ?) OR "UPDATED_DATE_TIME" > ?) ORDER BY "UPDATED_DATE_TIME","ID" ASC (io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier:145)
[2017-08-01 10:00:19,631] DEBUG executing query select CURRENT_TIMESTAMP from dual to get current time from database (io.confluent.connect.jdbc.util.JdbcUtils:198)
[2017-08-01 10:00:19,632] DEBUG Executing prepared statement with start time value = 1970-01-01 00:00:00.000 end time = 2017-08-01 04:30:19.036 and incrementing value = -1 (io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier:159)
[2017-08-01 10:00:19,633] DEBUG Resetting querier TimestampIncrementingTableQuerier{name='ENT_TS_DEMO', query='null', topicPrefix='n-', timestampColumn='UPDATED_DATE_TIME', incrementingColumn='ID'} (io.confluent.connect.jdbc.source.JdbcSourceTask:232)
Sarvesh-D picture Sarvesh-D on 1 Aug 2017

@ewencp did you get time to look at the logs?

Sarvesh-D picture Sarvesh-D on 4 Aug 2017
👍1

### Hi-We are using Kafka connect to pull the data from Mysql to Kafka topic. for Incremenatl mode it working fine but when changed the mode>timestmp+Incrementing everytime it's loading all records from mySQL table to kafka topic wheather it has been changed or not.

MySql.Properties

name=test-postgres-jdbc
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
tasks.max=1
connection.url=jdbc:mysql://10.40.11.135:4343/emsworkflow?user=webuser&password=EmsDev@209
table.whitelist=country_kafka
mode=timestamp+incrementing
incrementing.column.name=CountryID
timestamp.column.name =ProcessedDatetime
topic.prefix=test_jdbc_t3_

timestamp.delay.interval.ms =10

would someone from kafka connected please help me out.....

sharendra picture sharendra on 2 Sep 2017

I face with the same problems, with mode bulk, incrementing, connect worked fine. If I add timestamp configuration, there is no nothing to push to kafka

    batch.max.rows = 100
    connection.password = [hidden]
    connection.url = jdbc:postgresql://mydb.com:5432/test_sync_db
    connection.user = dbuser
    incrementing.column.name = id
    mode = timestamp+incrementing
    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 = updated_at
    timestamp.delay.interval.ms = 0
    topic.prefix = source-postgresql-test-
    validate.non.null = true

my db:

select * from users limit 10;
 id |  name  | age |         updated_at
----+--------+-----+----------------------------
  1 | john   |  26 | 2017-11-07 16:15:16.264088
  2 | jane   |  24 | 2017-11-07 16:15:16.348562
  3 | julia  |  25 | 2017-11-07 16:15:16.441743
  4 | jamie  |  22 | 2017-11-07 16:15:16.580902
  5 | jenny  |  27 | 2017-11-07 16:15:16.709461
  6 | joahn  |  26 | 2017-11-07 16:28:25.57042
  7 | jaane  |  24 | 2017-11-07 16:28:25.673931
  8 | jualia |  25 | 2017-11-07 16:28:25.798361
  9 | jaamie |  22 | 2017-11-07 16:28:25.90579
 10 | jeanny |  27 | 2017-11-07 16:28:26.021805
(10 rows)
jindov picture jindov on 7 Nov 2017

@Sarvesh-D Hi, did you find the solution finally?
Thanks!

HanzhiWang1105 picture HanzhiWang1105 on 17 Apr 2018

Hi @HanzhiWang1105,

After investigating, I figured out the solution.
You have to save data of timestamp column by UTC time. If possible, you could pull this and copy the DateTimeUtils.class in utils package. It help you have the right data

Thanks!

tungntt picture tungntt on 18 Apr 2018

@tungntt Thank you a lot! finally it works. :)

HanzhiWang1105 picture HanzhiWang1105 on 19 Apr 2018

@HanzhiWang1105 MySQL有解决办法吗？

suiziguo picture suiziguo on 21 Apr 2018

@tungntt Is there a MySQL solution？

suiziguo picture suiziguo on 22 Apr 2018

@suiziguo I test in Postgres. However, I think it also work with MySQL. Because kafka-connector-jdbc use PrepareStatement to recognize created and modified records in database.
You can replace INFO log by DEBUG in $CONFLUENT_PATH/etc/kafka/connect-log4j.properties to monitor the query generated by kafka-jdbc-connector.

tungntt picture tungntt on 23 Apr 2018

@tungntt Thank you very much！
I have some questions about UTC time in MySQL.
MySQL converts TIMESTAMP values from the current time zone to UTC for storage, and back from UTC to the current time zone for retrieval.
As mentioned above，it has already saved data of timestamp column by UTC time？

suiziguo picture suiziguo on 23 Apr 2018

Hi @suiziguo ,
I have not ever tested with MySQL so I not clarify about MySQL mechanism as you mention. However I think the timestamp column just saved UTC time in database. That is ok for jdbc-connector work.

tungntt picture tungntt on 24 Apr 2018

Hi @tungntt They are normal when they work independently，but the timestamp+incrementing mode only record updates，miss insertion operation。So I have to give it up，how regrettable it is……

suiziguo picture suiziguo on 25 Apr 2018

I am facing the same issue with an Oracle DB and a timestamp field.
According to the log bellow:
[2018-08-27 15:26:30,511] DEBUG executing query select CURRENT_TIMESTAMP from dual to get current time from database (io.confluent.connect.jdbc.dialect.OracleDatabaseDialect:462)
[2018-08-27 15:26:30,512] DEBUG Executing prepared statement with start time value = 2018-08-27 13:07:09.901 end time = 2018-08-27 13:26:53.900 and incrementing value = 227 (io.confluent.connect.jdbc.source.TimestampIncrementingCriteria:138)

It seems that the jdbc connector with Oracle dialects returns a UTC timestamp which it uses to compare local time (My test was made at 15:26 and the current timestmap set in the connector was 13:26).

I've tries to change the ORacle session time zone with environment variables (export ORA_SDTZ='Europe/Brussels') but it seems to make no difference.

any idea how t change that?

julienmmm picture julienmmm on 27 Aug 2018

Hi All
I am still facing issue with the mode timestamp in jdbc kafka source connector. All of the above recommendations have been implemented by me. I have converted the timestamp column to UTC format even using

alter table TEMP add UPDATED_DATE TIMESTAMP(6) default SYS_EXTRACT_UTC(CURRENT_TIMESTAMP) not null;

Following is the connector config

name=TRANSACTION
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
tasks.max=10
connection.url=jdbc:oracle:thin:@SERVICE:1701:CONNECTOR
connection.user=TRANSACTION
connection.password=txn#789
mode=timestamp
timestamp.column.name=UPDATED_DATE
validate.non.null=false

incrementing.column.name=ID

query=select * from temp inner join tempdata on temp.id=tempdata.id

topic.prefix=test
batch.max.rows=3
timestamp.delay.interval.ms=3000
table.whitelist=TEMP
table.poll.interval.ms=3000

Issues being faced

  1. Kafka connector not polling data and not pushing to topic when I use timestamp mode or timestamp+incrementing mode , however only incrementing mode works fine.
    2.I am not able to use multiple tables to poll out data from them by mentioning them in table.whitelist separated by comma like this

table.whitelist=TEMP,TEMPDATA
connector always picks up second table and creates a topic with prefix 'test'
and fullname testTEMPDATA. In case i mention auto.create.topics.enable=true , then nothing happens in logs and no error.

Please help over these issues as I might be missing on something. @ewencp Urgent help is required .

Thanks in Advance!

nik09541 picture nik09541 on 1 Nov 2018

db
My db:

nik09541 picture nik09541 on 1 Nov 2018

Hi all

I have same problem with timestamp mode. bulk mode and incrementing mode works as expected.

...with this config:

{ "name": "quickstart-jdbc-source", "config": { "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector", "tasks.max": 1, "dialect.name": "SybaseDatabaseDialect", "connection.url": "jdbc:sybase:Tds:172.16.30.158:2638?ServiceName=promis&USER=DBA&PASSWORD=DBA", "mode": "timestamp","timestamp.column.name": "modified","incrementing.column.name": "id", "topic.prefix": "quickstart-jdbc-", "table.whitelist":"promis.nis.c_pacient", "poll.interval.ms": 5000, "debug": true, "table.types": "TABLE, VIEW", "db.timezone": "Europe/Bratislava" } }

timespan mode does not fetch any record. Here is the log:

[2018-11-23 11:25:19,200] INFO TimestampIncrementingTableQuerier{table="promis"."nis"."c_pacient", query='null', topicPrefix='quickstart-jdbc-', incrementingColumn='', timestampColumns=[modified]} prepared SQL query: SELECT * FROM "promis"."nis"."c_pacient" WHERE "promis"."nis"."c_pacient"."modified" > ? AND "promis"."nis"."c_pacient"."modified" < ? ORDER BY "promis"."nis"."c_pacient"."modified" ASC (io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier)
[2018-11-23 11:25:19,208] DEBUG executing query select getdate() to get current time from database (io.confluent.connect.jdbc.dialect.SybaseDatabaseDialect)
[2018-11-23 11:25:19,212] DEBUG Executing prepared statement with timestamp value = 1970-01-01 00:00:00.000 end time = 2018-11-23 12:25:19.206 (io.confluent.connect.jdbc.source.TimestampIncrementingCriteria)
[2018-11-23 11:25:19,244] DEBUG Resetting querier TimestampIncrementingTableQuerier{table="promis"."nis"."c_pacient", query='null', topicPrefix='quickstart-jdbc-', incrementingColumn='', timestampColumns=[modified]} (io.confluent.connect.jdbc.source.JdbcSourceTask)

query looks fine, timestamp values looks fine. but at the end, when I turn DB profiling on, query was executed with this variables:

hostvariable 0 2018-11-23 12:25:19.206
hostvariable 1 2018-11-23 12:25:19.206

so the begin timestamp was same as end. And the result of this query is 0 records.

source is SQL Anywhere17 with jconn4-7.0 driver. I use confluentinc/cp-kafka-connect:latest image

Thanks in Advance!

toniiiik picture toniiiik on 23 Nov 2018

Hi @ewencp , i am using mysql jdbc connection and details of my source table.
Table size 2.6 GB,No of rows 2994498,No of columns 124,max_allowed_packet 1073741824

Here are the connector properties that I am using,
{
"name": "jdbc_source_mysql_reqistrations_local_bkp",
"config": {
"connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
"key.converter": "io.confluent.connect.avro.AvroConverter",
"key.converter.schema.registry.url": "http://localhost:8081",
"value.converter": "io.confluent.connect.avro.AvroConverter",
"value.converter.schema.registry.url": "http://localhost:8081",
"connection.url": "jdbc:mysql://localhost:3306/prokafka?zeroDateTimeBehavior=ROUND&user=kotesh&password=kotesh",
"table.whitelist": "XXXXXXXX",
"mode": "timestamp",
"timestamp.column.name": "DateUpdated",
"validate.non.null": "false",
"topic.prefix": "mysql-local-bkp-"
}
}

Unable to consume data due to Java Heap Size.
Error :
ERROR Failed to run query for table TimestampIncrementingTableQuerier{table="prokafka"."XXXXXXXX", query='null', topicPrefix='mysql-registrations', incrementingColumn='', timestampColumns=[DateUpdated]}: {} (io.confluent.connect.jdbc.source.JdbcSourceTask:328)
java.sql.SQLException: Java heap space
at com.mysql.cj.jdbc.exceptions.SQLError.createSQLException(SQLError.java:129)
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.ClientPreparedStatement.executeInternal(ClientPreparedStatement.java:975)
at com.mysql.cj.jdbc.ClientPreparedStatement.executeQuery(ClientPreparedStatement.java:1025)
at io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier.executeQuery(TimestampIncrementingTableQuerier.java:168)
at io.confluent.connect.jdbc.source.TableQuerier.maybeStartQuery(TableQuerier.java:

memoryerror2

bkotesh picture bkotesh on 27 Nov 2018

Stumbled across similar issue. In my case, even the topic didn't create. Found out that timezone for connect worker must be same as that of DB. Setting db.timezone property in the connect worker's property file correctly based on this list (under SHORT_IDS section) makes it work:

db.timezone=Asia/Kolkata

swapnil-saxena picture swapnil-saxena on 3 Aug 2020
Was this page helpful?
0 / 5 - 0 ratings

Related issues

bruwozniak picture bruwozniak  ·  4Comments
abo picture abo  ·  8Comments
rmoff picture rmoff  ·  8Comments
rdinkel picture rdinkel  ·  8Comments
bli24 picture bli24  ·  6Comments