jdbc.rb in logstash-input-jdbc-4.2.0

- old
+ new

@@ -136,12 +136,12 @@
         @logger.error("Failed to load #{driver}", :exception => e)
       end
     end
   end
 
-  public
-  def prepare_jdbc_connection
+  private
+  def open_jdbc_connection
     require "java"
     require "sequel"
     require "sequel/adapters/jdbc"
     load_drivers(@jdbc_driver_library.split(",")) if @jdbc_driver_library
 
@@ -168,20 +168,31 @@
     end
     @database.fetch_size = @jdbc_fetch_size unless @jdbc_fetch_size.nil?
     begin
       @database.test_connection
     rescue Sequel::DatabaseConnectionError => e
+      @logger.warn("Failed test_connection.")
+      @database.close_jdbc_connection
+
       #TODO return false and let the plugin raise a LogStash::ConfigurationError
       raise e
     end
+
     @database.sql_log_level = @sql_log_level.to_sym
     @database.logger = @logger
+
+    @database.extension :identifier_mangling
+
     if @lowercase_column_names
       @database.identifier_output_method = :downcase
     else
       @database.identifier_output_method = :to_s
     end
+  end
+
+  public
+  def prepare_jdbc_connection
     if @use_column_value
       case @tracking_column_type
         when "numeric"
           @sql_last_value = 0
         when "timestamp"
@@ -193,17 +204,19 @@
   end # def prepare_jdbc_connection
 
   public
   def close_jdbc_connection
     @database.disconnect if @database
+    @database = nil
   end
 
   public
   def execute_statement(statement, parameters)
     success = false
     begin
       parameters = symbolized_params(parameters)
+      open_jdbc_connection if @database == nil
       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)
 
@@ -227,9 +240,12 @@
         end
       end
       success = true
     rescue Sequel::DatabaseConnectionError, Sequel::DatabaseError => e
       @logger.warn("Exception when executing JDBC query", :exception => e)
+      @logger.warn("Attempt reconnection.")
+      close_jdbc_connection()
+      open_jdbc_connection()
     else
       @sql_last_value = sql_last_value
     end
     return success
   end