title: Logstash input jdbc is duplicating results
body: <p>I'm using logstash input jdbc plugin to read  two (or more) databases and send the data to elasticsearch, and using kibana 4 to vizualize these data.</p>
  
  <p>This is my logstash config:</p>
  
  <pre><code>input {
    jdbc {
      type =&gt; "A"
      jdbc_driver_library =&gt; "C:\DEV\elasticsearch-1.7.1\plugins\elasticsearch-jdbc-1.7.1.0\lib\jtds-1.3.1.jar"
      jdbc_driver_class =&gt; "Java::net.sourceforge.jtds.jdbc.Driver"
      jdbc_connection_string =&gt; "jdbc:jtds:sqlserver://dev_data_base_server:1433/dbApp1;domain=CORPDOMAIN;useNTLMv2=true"
      jdbc_user =&gt; "user"
      jdbc_password =&gt; "pass"
      schedule =&gt; "5 * * * *"
      statement =&gt; "SELECT id, date, content, status from test_table"
    }
  
  jdbc {
      type =&gt; "B"
      jdbc_driver_library =&gt; "C:\DEV\elasticsearch-1.7.1\plugins\elasticsearch-jdbc-1.7.1.0\lib\jtds-1.3.1.jar"
      jdbc_driver_class =&gt; "Java::net.sourceforge.jtds.jdbc.Driver"
      jdbc_connection_string =&gt; "jdbc:jtds:sqlserver://dev_data_base_server:1433/dbApp2;domain=CORPDOMAIN;useNTLMv2=true"
      jdbc_user =&gt; "user"
      jdbc_password =&gt; "pass"
      schedule =&gt; "5 * * * *"
      statement =&gt; "SELECT id, date, content, status from test_table"
    }
  }
  filter {
  
  }
  output {
  
      if [type] == "A" {
          elasticsearch {
              host =&gt; "localhost"
              protocol =&gt; http
              index =&gt; "logstash-servera-%{+YYYY.MM.dd}"
          }    
      }
      if [type] == "B" {
          elasticsearch {
              host =&gt; "localhost"
              protocol =&gt; http
              index =&gt; "logstash-serverb-%{+YYYY.MM.dd}"
          }    
      }
  
    stdout { codec =&gt; rubydebug }
  }
  </code></pre>
  
  <p>The problem is that every time run the logstash, it starts to save all data that is already in elastic search.</p>
  
  <p>After run with the where clause = date > '2015-09-10' I'd stoped the logstash and run again (with --debug) with the 'special parameter' :sql_last_date. After the logstash startup It starts to show this in the log:</p>
  
  <pre><code>←[36mExecuting JDBC query {:statement=&gt;"SELECT \n\tSUBSTRING(R.RECEBEDOR, 1, 2)
  AS 'DDD',\nCASE WHEN R.STATUS &lt;&gt;  'RCON' AND R.COD_RESPOSTA in (428,429,230,425,
  430,427,418,422,415,424,214,433,435,207,426) THEN 'REGRA DE NEGÓCIO'  \n       W
  HEN R.STATUS = 'RCON' THEN 'SUCESSO'\n\t   ELSE 'ERRO'\n   END AS 'TIPO_MENSAGEM
  ',\nAP.ALIAS as 'CANAL', R.ID_RECARGA, R.VALOR, R.STATUS, R.COD_RESPOSTA, R.DESC
  _RESPOSTA, R.DT_RECARGA as '@timestamp', R.ID_CLIENTE, R.ID_DEPENDENTE, R.ID_APL
  ICACAO, RECEBEDOR, R.ID_OPERADORA, R.TIPO_PRODUTO \n\nFROM RECARGA R (NOLOCK)\nJ
  OIN APLICACAO AP ON R.ID_APLICACAO = AP.ID_APLICACAO \nwhere R.DT_RECARGA &gt; :sql
  _last_start\nORDER BY R.DT_RECARGA ASC", :parameters=&gt;{:sql_last_start=&gt;2015-09-
  10 18:48:00 UTC}, :level=&gt;:debug, :file=&gt;"/DEV/logstash-1.5.4/vendor/bundle/jrub
  y/1.9/gems/logstash-input-jdbc-1.0.0/lib/logstash/plugin_mixins/jdbc.rb", :line=
  &gt;"107", :method=&gt;"execute_statement"}←[0m
  </code></pre>
  
  <p>This time i ran with the 'real' statement that is:</p>
  
  <pre><code>SELECT 
      SUBSTRING(R.RECEBEDOR, 1, 2) AS 'DDD',
  CASE WHEN R.STATUS &lt;&gt;  'RCON' AND R.COD_RESPOSTA in (428,429,230,425,430,427,418,422,415,424,214,433,435,207,426) THEN 'REGRA DE NEGÓCIO'  
         WHEN R.STATUS = 'RCON' THEN 'SUCESSO'
         ELSE 'ERRO'
     END AS 'TIPO_MENSAGEM',
  AP.ALIAS as 'CANAL', R.ID_RECARGA, R.VALOR, R.STATUS, R.COD_RESPOSTA, R.DESC_RESPOSTA, R.DT_RECARGA as '@timestamp', R.ID_CLIENTE, R.ID_DEPENDENTE, R.ID_APLICACAO, RECEBEDOR, R.ID_OPERADORA
  
  FROM RECARGA R (NOLOCK)
  JOIN APLICACAO AP ON R.ID_APLICACAO = AP.ID_APLICACAO 
  where R.DT_RECARGA &gt; :sql_last_start
  ORDER BY R.DT_RECARGA ASC
  </code></pre>
  
  <p>Anyone knows how to solve it?</p>
  
  <p>Thanks!</p>