title: How kafka fetch data only when new row insert or update old row from mysql database using kafka streaming
body: <p>I am using confluent platform JDBC connector, it is streaming data from mysql to kafka consumer. Through an application inserting data into another database for reporting purpose.
  Here problem it is only streaming all data again and again within some interval of time. Actually a want only those data which is newly inserted or any update of previous record.</p>
  <p>According to timestamp can't do because table does not contain any time column. And as per increment id also not possible. Please share any solution.</p>
  <p>I have sample configuration file.</p>
  <pre><code>demo.json
  {
      &quot;name&quot;:&quot;mysql-connector-demo&quot;,
      &quot;config&quot;:{
                  &quot;connector.class&quot;: &quot;io.confluent.connect.jdbc.JdbcSourceConnector&quot;,
                  &quot;key.converter&quot;:&quot;org.apache.kafka.connect.json.JsonConverter&quot;,
                  &quot;value.converter&quot;:&quot;org.apache.kafka.connect.json.JsonConverter&quot;,
                  &quot;connection.url&quot;: &quot;jdbc:mysql://localhost:3306/test&quot;,
                  &quot;connection.user&quot;: &quot;root&quot;,
                  &quot;connection.password&quot;: &quot;1234&quot;,
                  &quot;topic.prefix&quot;: &quot;test&quot;,
                  &quot;catalog.pattern&quot;:&quot;test&quot;,
                  &quot;mode&quot;: &quot;bulk&quot;,
                  &quot;validate.non.null&quot;: false,
                  &quot;query&quot;: &quot;select * from test &quot;,
                  &quot;table.types&quot;: &quot;TABLE&quot;,
                  &quot;topic.prefix&quot;: &quot;test-jdbc-&quot;,
                  &quot;poll.interval.ms&quot;: 10000
                  &quot;schema.ignore&quot;: true,
                  &quot;key.converter.schemas.enable&quot;: &quot;false&quot;,
                  &quot;value.converter.schemas.enable&quot;: &quot;false&quot;
      }
  } 
  </code></pre>
  <p>But here new inserted record and newly updated record not affected to kafka consumer.</p>