Raise Embulk Error

lib/embulk/input/elasticsearch/connection.rb

@@ -73,8 +73,8 @@ def search_with_retry
               sleep 2**retries
               retry
             end
-            Embulk.logger.error "Could not search to Elasticsearch after #{retries} retries. #{e.message}"
-            return nil
+            msg = "Could not search to Elasticsearch after #{retries} retries. #{e.message}"
+            raise Elasticsearch::ConfigError(e, msg)
           end
         end
 

lib/embulk/input/elasticsearch/converter.rb

@@ -40,7 +40,7 @@ def self.convert_value(value, field)
           when "json"
             value
           else
-            raise "Unsupported type #{field['type']}"
+            raise Elasticsearch::TypecastError.new "Unsupported type #{field['type']}"
           end
         end
       end

lib/embulk/input/elasticsearch/error.rb

@@ -0,0 +1,36 @@
+module Embulk
+  module Input
+
+    class Elasticsearch < InputPlugin
+
+      module Traceable
+        def initialize(e, more_msg = nil)
+          message = e.is_a?(String) ? '' : "(#{e.class}) "
+          message << "#{e}#{more_msg}\n"
+          message << "\tat #{e.backtrace.join("\n\tat ")}\n" if e.respond_to?(:backtrace)
+
+          while e.respond_to?(:cause) and e.cause
+            # Java Exception cannot follow the JRuby causes.
+            message << "Caused by (#{e.cause.class}) #{e.cause}\n"
+            message << "\tat #{e.cause.backtrace.join("\n\tat ")}\n" if e.cause.respond_to?(:backtrace)
+            e = e.cause
+          end
+
+          super(message)
+        end
+      end
+
+      class ConfigError < ::Embulk::ConfigError
+        include Traceable
+      end
+
+      class DataError < ::Embulk::DataError
+        include Traceable
+      end
+
+      class TypecastError < DataError
+      end
+
+    end
+  end
+end