logstash jdbc syn oracle table to kafka every second

logstash_jdbc_oracle.conf

input { 
  jdbc {
    jdbc_driver_library => "D:\logstash-7.12.0\ojdbc8.jar"
    jdbc_driver_class => "Java::oracle.jdbc.driver.OracleDriver"
    jdbc_connection_string => "jdbc:oracle:thin:@//127.0.0.1:1521/servicename"
    jdbc_user => "user"
    jdbc_password => "password"
    tracking_column => "record_sn"
    use_column_value => true
    tracking_column_type => "numeric"
    statement => "select * from matching where record_sn > ?"
    prepared_statement_bind_values => [":sql_last_value"]
    prepared_statement_name => "SelectfromMatchingWhereRecordSN_BigThan"
    use_prepared_statements => true
    last_run_metadata_path => "D:\logstash-7.12.0\_LOGSTASH_JDBC_LAST_RUN"
    schedule => "* * * * * *"
  }
}

filter {
}

output {
  # stdout { codec => rubydebug }
  # stdout { codec => dots }
 kafka {
    bootstrap_servers => "127.0.0.1:9092"
    codec => json
    topic_id => "matching"
    compression_type => "snappy"
    acks => "0"
  }
}

PS D:\logstash-7.12.0> .\bin\nssm.exe install logstash
PS D:\logstash-7.12.0> .\bin\nssm.exe edit logstash

matching表每天清空重置，用以下脚本重启logstash

echo "will RESET logstash reord_sn..."
timeout /t 10
net stop logstash
timeout /t 5
del D:\logstash-7.12.0\_LOGSTASH_JDBC_LAST_RUN
net start logstash
timeout /t 15