[REST] Fixing connection leak in the InitializationUtils.

Mismanagement of a RestRepository connection would occur if no connection was passed to the check, the index auto create was disabled, and the index already exists upon check.
Fixes elastic#883

Author: jbaiera

mr/src/main/java/org/elasticsearch/hadoop/mr/EsOutputFormat.java

@@ -256,11 +256,8 @@ private void init(Configuration cfg) throws IOException {
         Settings settings = HadoopSettingsManager.loadFrom(cfg);
         Assert.hasText(settings.getResourceWrite(), String.format("No resource ['%s'] (index/query/location) specified", ES_RESOURCE));
 
-        // lazy-init
-        RestRepository client = null;
-
         InitializationUtils.checkIdForOperation(settings);
-        InitializationUtils.checkIndexExistence(settings, client);
+        InitializationUtils.checkIndexExistence(settings);
 
         if (HadoopCfgUtils.getReduceTasks(cfg) != null) {
             if (HadoopCfgUtils.getSpeculativeReduce(cfg)) {

mr/src/main/java/org/elasticsearch/hadoop/rest/InitializationUtils.java

@@ -253,23 +253,31 @@ public static EsMajorVersion discoverEsVersion(Settings settings, Log log) {
     }
 
     public static void checkIndexExistence(RestRepository client) {
-        checkIndexExistence(client.getSettings(), client);
+        if (!client.getSettings().getIndexAutoCreate()) {
+            doCheckIndexExistence(client.getSettings(), client);
+        }
     }
 
-    public static void checkIndexExistence(Settings settings, RestRepository client) {
-        // check index existence
+    public static void checkIndexExistence(Settings settings) {
+        // Only open a connection and check if autocreate is disabled
         if (!settings.getIndexAutoCreate()) {
-            if (client == null) {
-                client = new RestRepository(settings);
-            }
-            if (!client.indexExists(false)) {
-                client.close();
-                throw new EsHadoopIllegalArgumentException(String.format("Target index [%s] does not exist and auto-creation is disabled [setting '%s' is '%s']",
-                        settings.getResourceWrite(), ConfigurationOptions.ES_INDEX_AUTO_CREATE, settings.getIndexAutoCreate()));
+            RestRepository repository = new RestRepository(settings);
+            try {
+                doCheckIndexExistence(settings, repository);
+            } finally {
+                repository.close();
             }
         }
     }
 
+    private static void doCheckIndexExistence(Settings settings, RestRepository client) {
+        // check index existence
+        if (!client.indexExists(false)) {
+            throw new EsHadoopIllegalArgumentException(String.format("Target index [%s] does not exist and auto-creation is disabled [setting '%s' is '%s']",
+                    settings.getResourceWrite(), ConfigurationOptions.ES_INDEX_AUTO_CREATE, settings.getIndexAutoCreate()));
+        }
+    }
+
     public static boolean setFieldExtractorIfNotSet(Settings settings, Class<? extends FieldExtractor> clazz, Log log) {
         if (!StringUtils.hasText(settings.getMappingIdExtractorClassName())) {
             Log logger = (log != null ? log : LogFactory.getLog(clazz));

spark/core/main/scala/org/elasticsearch/spark/rdd/EsSpark.scala

@@ -96,8 +96,8 @@ object EsSpark {
     val config = new PropertiesSettings().load(sparkCfg.save())
     config.merge(cfg.asJava)
 
-    InitializationUtils.checkIdForOperation(config);
-    InitializationUtils.checkIndexExistence(config, null);
+    InitializationUtils.checkIdForOperation(config)
+    InitializationUtils.checkIndexExistence(config)
 
     rdd.sparkContext.runJob(rdd, new EsRDDWriter(config.save(), hasMeta).write _)
   }

spark/sql-13/src/main/scala/org/elasticsearch/spark/sql/EsSparkSQL.scala

@@ -71,8 +71,8 @@ object EsSparkSQL {
     val esCfg = new PropertiesSettings().load(sparkCfg.save())
     esCfg.merge(cfg.asJava)
 
-    InitializationUtils.checkIdForOperation(esCfg);
-    InitializationUtils.checkIndexExistence(esCfg, null);
+    InitializationUtils.checkIdForOperation(esCfg)
+    InitializationUtils.checkIndexExistence(esCfg)
 
     sparkCtx.runJob(srdd.rdd, new EsDataFrameWriter(srdd.schema, esCfg.save()).write _)
   }

spark/sql-20/src/main/scala/org/elasticsearch/spark/sql/EsSparkSQL.scala

@@ -88,8 +88,8 @@ object EsSparkSQL {
     val esCfg = new PropertiesSettings().load(sparkCfg.save())
     esCfg.merge(cfg.asJava)
 
-    InitializationUtils.checkIdForOperation(esCfg);
-    InitializationUtils.checkIndexExistence(esCfg, null);
+    InitializationUtils.checkIdForOperation(esCfg)
+    InitializationUtils.checkIndexExistence(esCfg)
 
     sparkCtx.runJob(srdd.toDF().rdd, new EsDataFrameWriter(srdd.schema, esCfg.save()).write _)
   }