fixing client connection settings, simplifying ingest handling

jprante · Dec 23, 2015 · 4c8ae3c · 4c8ae3c
1 parent 224ddc3
commit 4c8ae3c
Show file tree

Hide file tree

Showing 12 changed files with 306 additions and 552 deletions.
diff --git a/README.md b/README.md
@@ -40,8 +40,7 @@ about what happened.
 
 | Release date | JDBC Importer version | Elasticsearch version |
 | -------------| ----------------------| ----------------------|
-| Dec 23 2015  | 2.1.1.1               | 2.1.1                 |
-| Dec 22 2015  | 2.1.1.0               | 2.1.1                 |
+| Dec 23 2015  | 2.1.1.2               | 2.1.1                 |
 | Nov 29 2015  | 2.1.0.0               | 2.1.0                 |
 | Oct 29 2015  | 2.0.0.1               | 2.0.0                 |
 | Oct 28 2015  | 2.0.0.0               | 2.0.0                 |
@@ -54,9 +53,9 @@ about what happened.
 
 ## Quick links
 
-JDBC importer 2.1.1.1
+JDBC importer 2.1.1.2
 
-`http://xbib.org/repository/org/xbib/elasticsearch/importer/elasticsearch-jdbc/2.1.1.1/elasticsearch-jdbc-2.1.1.1-dist.zip`
+`http://xbib.org/repository/org/xbib/elasticsearch/importer/elasticsearch-jdbc/2.1.1.2/elasticsearch-jdbc-2.1.1.2-dist.zip`
 
 ## Installation
 

diff --git a/pom.xml b/pom.xml
@@ -7,7 +7,7 @@
 
     <groupId>org.xbib.elasticsearch.importer</groupId>
     <artifactId>elasticsearch-jdbc</artifactId>
-    <version>2.1.1.1</version>
+    <version>2.1.1.2</version>
 
     <packaging>jar</packaging>
 

diff --git a/src/main/java/org/xbib/elasticsearch/jdbc/strategy/Sink.java b/src/main/java/org/xbib/elasticsearch/jdbc/strategy/Sink.java
@@ -51,26 +51,6 @@ public interface Sink<C extends Context> {
      */
     Sink<C> setContext(C context);
 
-    Sink setIngestFactory(IngestFactory ingestFactory);
-
-    IngestFactory getIngestFactory();
-
-    /**
-     * Set index settings
-     *
-     * @param indexSettings the index settings
-     * @return this sink
-     */
-    Sink setIndexSettings(Settings indexSettings);
-
-    /**
-     * Set index type mappings
-     *
-     * @param typeMapping the index type mappings
-     * @return this sink
-     */
-    Sink setTypeMapping(Map<String, String> typeMapping);
-
     /**
      * Executed before source fetch
      *

diff --git a/src/main/java/org/xbib/elasticsearch/jdbc/strategy/standard/StandardContext.java b/src/main/java/org/xbib/elasticsearch/jdbc/strategy/standard/StandardContext.java
@@ -36,7 +36,6 @@
 import java.io.FileWriter;
 import java.io.IOException;
 import java.io.Writer;
-import java.util.Collections;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Locale;
@@ -170,6 +169,7 @@ public void execute() throws Exception {
     }
 
     @Override
+    @SuppressWarnings("unchecked")
     public void beforeFetch() throws Exception {
         logger.debug("before fetch");
         Sink sink = createSink();
@@ -188,7 +188,7 @@ public void fetch() throws Exception {
             getSource().fetch();
         } catch (Throwable e) {
             setThrowable(e);
-            logger.error(e.getMessage(), e);
+            logger.error("at fetch: " + e.getMessage(), e);
         }
     }
 
@@ -200,13 +200,13 @@ public void afterFetch() throws Exception {
             getSource().afterFetch();
         } catch (Throwable e) {
             setThrowable(e);
-            logger.error(e.getMessage(), e);
+            logger.error("after fetch: " + e.getMessage(), e);
         }
         try {
             getSink().afterFetch();
         } catch (Throwable e) {
             setThrowable(e);
-            logger.error(e.getMessage(), e);
+            logger.error("after fetch: " + e.getMessage(), e);
         }
     }
 
@@ -220,14 +220,14 @@ public void shutdown() {
             try {
                 source.shutdown();
             } catch (Exception e) {
-                logger.error(e.getMessage(), e);
+                logger.error("source shutdown: " + e.getMessage(), e);
             }
         }
         if (sink != null) {
             try {
                 sink.shutdown();
             } catch (Exception e) {
-                logger.error(e.getMessage(), e);
+                logger.error("sink shutdown: " + e.getMessage(), e);
             }
         }
         logger.info("shutdown completed");
@@ -293,22 +293,10 @@ protected S createSource() {
     protected Sink createSink() throws IOException {
         Sink sink = StrategyLoader.newSink(strategy());
         logger.info("found sink class {}", sink);
-        String index = settings.get("index", "jdbc");
-        String type = settings.get("type", "jdbc");
-        sink.setIndex(index).setType(type);
-        Map<String,Object> map = settings.getAsStructuredMap();
-        if (map.containsKey("index_settings")) {
-            Settings loadedSettings = settings.getAsSettings("index_settings");
-            sink.setIndexSettings(loadedSettings);
-        }
-        if (map.containsKey("type_mapping")) {
-            XContentBuilder builder = jsonBuilder()
-                    .map(settings.getAsSettings("type_mapping").getAsStructuredMap());
-            sink.setTypeMapping(Collections.singletonMap(type, builder.string()));
-        }
         return sink;
     }
 
+    @SuppressWarnings("unchecked")
     protected void prepareContext(S source, Sink sink) throws IOException {
         Map<String, Object> params = settings.getAsStructuredMap();
         List<SQLCommand> sql = SQLCommand.parse(params);