Repackage #200 and Release 4.2.1

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 4.2.1
+ - Wrap queries in rolled back transactions to work better with PostgreSQL
+
 ## 4.2.0
  - Automatically reconnect on connection issues
  - Fix test failures

lib/logstash/plugin_mixins/jdbc.rb

@@ -143,6 +143,7 @@ def open_jdbc_connection
     require "java"
     require "sequel"
     require "sequel/adapters/jdbc"
+    require "sequel/adapters/jdbc/transactions"
     load_drivers(@jdbc_driver_library.split(",")) if @jdbc_driver_library
 
     begin
@@ -157,6 +158,7 @@ def open_jdbc_connection
       raise LogStash::ConfigurationError, "#{e}. #{message}"
     end
     @database = jdbc_connect()
+    @database.extend(Sequel::JDBC::Transactions)
     @database.extension(:pagination)
     if @jdbc_default_timezone
       @database.extension(:named_timezones)
@@ -218,26 +220,31 @@ def execute_statement(statement, parameters)
       query = @database[statement, parameters]
       sql_last_value = @use_column_value ? @sql_last_value : Time.now.utc
       @tracking_column_warning_sent = false
-      @logger.debug? and @logger.debug("Executing JDBC query", :statement => statement, :parameters => parameters, :count => query.count)
-
-      if @jdbc_paging_enabled
-        query.each_page(@jdbc_page_size) do |paged_dataset|
-          paged_dataset.each do |row|
+      @logger.debug? and @logger.debug("Executing JDBC query", :statement => statement, :parameters => parameters)
+
+      # Execute query in transaction cause PG driver require autocommit off for set fetch count
+      # See: https://jdbc.postgresql.org/documentation/head/query.html
+      @database.transaction do
+        if @jdbc_paging_enabled
+          query.each_page(@jdbc_page_size) do |paged_dataset|
+            paged_dataset.each do |row|
+              sql_last_value = get_column_value(row) if @use_column_value
+              if @tracking_column_type=="timestamp" and @use_column_value and sql_last_value.is_a?(DateTime)
+                sql_last_value=Time.parse(sql_last_value.to_s) # Coerce the timestamp to a `Time`
+              end
+              yield extract_values_from(row)
+            end
+          end
+        else
+          query.each do |row|
             sql_last_value = get_column_value(row) if @use_column_value
             if @tracking_column_type=="timestamp" and @use_column_value and sql_last_value.is_a?(DateTime)
               sql_last_value=Time.parse(sql_last_value.to_s) # Coerce the timestamp to a `Time`
             end
             yield extract_values_from(row)
           end
         end
-      else
-        query.each do |row|
-          sql_last_value = get_column_value(row) if @use_column_value
-          if @tracking_column_type=="timestamp" and @use_column_value and sql_last_value.is_a?(DateTime)
-            sql_last_value=Time.parse(sql_last_value.to_s) # Coerce the timestamp to a `Time`
-          end
-          yield extract_values_from(row)
-        end
+        raise Sequel::Rollback
       end
       success = true
     rescue Sequel::DatabaseConnectionError, Sequel::DatabaseError => e

logstash-input-jdbc.gemspec

@@ -1,6 +1,6 @@
 Gem::Specification.new do |s|
   s.name = 'logstash-input-jdbc'
-  s.version         = '4.2.0'
+  s.version         = '4.2.1'
   s.licenses = ['Apache License (2.0)']
   s.summary = "This example input streams a string at a definable interval."
   s.description     = "This gem is a Logstash plugin required to be installed on top of the Logstash core pipeline using $LS_HOME/bin/logstash-plugin install gemname. This gem is not a stand-alone program"

spec/inputs/jdbc_spec.rb

@@ -833,7 +833,7 @@
       {
         "statement" => "SELECT * FROM test_table",
         "jdbc_pool_timeout" => 0,
-        "jdbc_connection_string" => 'mock://localhost:1527/db',
+        "jdbc_connection_string" => 'jdbc:derby:memory:testdb;create=true',
         "sequel_opts" => {
           "max_connections" => 1
         }
@@ -889,7 +889,7 @@
     end
 
     it "should report the statements to logging" do
-      expect(plugin.logger).to receive(:debug).once
+      expect(plugin.logger).to receive(:debug).thrice
       plugin.run(queue)
     end
   end