mysql> create table test_sync(id int not null auto_increment, name varchar(32), description varchar(64), create_time timestamp, update_time timestamp, primary key(id));

Query OK, 0 rows affected (0.04 sec)

mysql> insert into test_sync (name, description, create_time, update_time) values('test1', '', now(), now());

Query OK, 1 row affected (0.02 sec)

mysql> select * from test_sync;

+----+-------+-------------+---------------------+---------------------+

| id | name  | description | create_time         | update_time         |

+----+-------+-------------+---------------------+---------------------+

|  1 | test1 | 1           | 2019-03-13 10:45:49 | 2019-03-13 10:45:49 |

+----+-------+-------------+---------------------+---------------------+

1 row in set (0.00 sec)

# bin/kafka-topics.sh --zookeeper $zk:2181/kafka -create --topic test_sync --partitions 2 --replication-factor 2

WARNING: Due to limitations in metric names, topics with a period ('.') or underscore ('_') could collide. To avoid issues it is best to use either, but not both.

Created topic "test_sync".

# bin/kafka-topics.sh --zookeeper $zk:2181/kafka -describe --topic test_sync

Topic:test_sync PartitionCount:2        ReplicationFactor:2     Configs:

        Topic: test_sync        Partition: 0    Leader: 112     Replicas: 112,111       Isr: 112,111

        Topic: test_sync        Partition: 1    Leader: 110     Replicas: 110,112       Isr: 110,112

input {

  jdbc {

  jdbc_driver_library => "/path/to/mysql-connector-java-5.1.36-bin.jar"

  jdbc_driver_class => "com.mysql.jdbc.Driver"

  jdbc_connection_string => "jdbc:mysql://localhost:3306/database"

  jdbc_user => "DBUSERNAME"

  jdbc_password => "DBPASSWORD"
  jdbc_paging_enabled => "true"
  jdbc_page_size => "10000"
  jdbc_default_timezone => "UTC"

  statement => "select * from TABLE where id > :sql_last_value"

  tracking_column => id
  tracking_column_type => "numeric"

  use_column_value => true
  record_last_run => true
  clean_run => false
  #I have scheduled this to run in every second

  schedule => "* * * * * * Asia/Shanghai"

  }

}

output {
  kafka {
    bootstrap_servers => "192.168.0.1:9092,192.168.0.2:9092"
    topic_id => "test_sync"
    codec => json
  }
  stdout {codec => json}
}

  statement => "select * from table where update_time > :sql_last_value;"

  tracking_column => "update_time"

  tracking_column_type => "timestamp"