title: Flink 1.3.1, Elasticsearch 2.4.6, RuntimeException: Elasticsearch client is not connected to any Elasticsearch nodes
body: <p>Trying to progress on Flink training / exersizes. Had following <a href="https://stackoverflow.com/questions/45555420/flink-1-3-1-elasticsearch-5-5-1-elasticsearchsinkfunction-fails-with-java-lang">issue</a>.<br>
  Re-installed Elasticsearch, now having version: <strong>2.4.6</strong>.<br>
  Elasticsearch seems to work. I'm able to access it via <code>http://localhost:9200</code>.<br>
  Created <code>index</code>, and <code>type mapping</code> as per exercise instructions using <code>cURL</code> commands.<br>
  Installed Kibana: <strong>4.6.5</strong>, and Kibana is able to connect to Elasticsearch and resolve the <code>index</code> I created.  </p>
  
  <p>Now, I try to execute <code>Flink</code> job, which reads data from Kafka topic, and attempts to populate the data into Elasticsearch, and, it fails to create a <code>Sink</code> with following error:  </p>
  
  <pre><code>Caused by: java.lang.RuntimeException: Elasticsearch client is not connected to any Elasticsearch nodes!
      at org.apache.flink.streaming.connectors.elasticsearch2.Elasticsearch2ApiCallBridge.createClient(Elasticsearch2ApiCallBridge.java:72)
      at org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkBase.open(ElasticsearchSinkBase.java:272)
      at org.apache.flink.api.common.functions.util.FunctionUtils.openFunction(FunctionUtils.java:36)
      at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.open(AbstractUdfStreamOperator.java:111)
      at org.apache.flink.streaming.runtime.tasks.StreamTask.openAllOperators(StreamTask.java:376)
      at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:253)
      at org.apache.flink.runtime.taskmanager.Task.run(Task.java:702)
      at java.lang.Thread.run(Thread.java:748)
  </code></pre>
  
  <p>Again, here is my <code>built.sbt</code> setting:</p>
  
  <pre><code>libraryDependencies ++= Seq(
    "org.apache.flink" %% "flink-scala" % "1.3.1" % "provided",
    "org.apache.flink" %% "flink-streaming-scala" % "1.3.1" % "provided",
    "org.apache.flink" %% "flink-clients" % "1.3.1" % "provided",
    "joda-time" % "joda-time" % "2.9.9",
    "com.google.guava" % "guava" % "22.0",
    "com.typesafe" % "config" % "1.3.0",
    "org.apache.flink" % "flink-connector-kafka-0.10_2.10" % "1.3.1",
    "org.apache.flink" % "flink-connector-elasticsearch2_2.10" % "1.3.1"
  )
  </code></pre>
  
  <p>Imports:</p>
  
  <pre><code>import org.apache.flink.streaming.connectors.elasticsearch.{ElasticsearchSinkFunction, RequestIndexer}
  import org.apache.flink.streaming.connectors.elasticsearch2.ElasticsearchSink
  </code></pre>
  
  <p>And function which builds, and returns <code>Sink</code>.<br>
  Passed parameters <code>("localhost", 9300, "elasticsearch")</code>. Tried passing <code>"127.0.0.1"</code> for host, had the same error.</p>
  
  <pre><code>  private def getSink(host: String, port: Int, cluster: String) = {
  
      val config = Map(
        // This instructs the sink to emit after every element, otherwise they would be buffered
        "bulk.flush.max.actions" -&gt; "1",
        // default cluster name
        "cluster.name" -&gt; cluster
      )
  
      val jConfig: java.util.Map[String, String] = new java.util.HashMap()
      jConfig.putAll(config.asJava)
  
      println(s"jConfig: ${jConfig.toString()}")
  
      val transports = List(new InetSocketAddress(InetAddress.getByName(host), port))
      val jTransports = new util.ArrayList(transports.asJava)
  
      println(s"jTransports: ${jTransports.toString()}")
  
      val esSink = new ElasticsearchSink(jConfig, jTransports,
        new MyElasticsearchInserter("nyc-idx", "popular-locations"))
  
      esSink
    }
  </code></pre>
  
  <p>Here is additional information from <code>Elasticsearch</code> log to indicate it's status:  </p>
  
  <pre><code>...
  ...
  [2017-08-08 20:12:24,203][INFO ][node                     ] [Beautiful Dreamer] stopping ...
  [2017-08-08 20:12:27,063][INFO ][node                     ] [Beautiful Dreamer] stopped
  [2017-08-08 20:12:27,066][INFO ][node                     ] [Beautiful Dreamer] closing ...
  [2017-08-08 20:12:28,104][INFO ][node                     ] [Beautiful Dreamer] closed
  [2017-08-08 20:21:58,212][INFO ][node                     ] [Aries] version[2.4.6], pid[1502], build[5376dca/2017-07-18T12:17:44Z]
  [2017-08-08 20:21:58,264][INFO ][node                     ] [Aries] initializing ...
  [2017-08-08 20:22:00,569][INFO ][plugins                  ] [Aries] modules [reindex, lang-expression, lang-groovy], plugins [], sites []
  [2017-08-08 20:22:00,857][INFO ][env                      ] [Aries] using [1] data paths, mounts [[/ (rootfs)]], net usable_space [1.6gb], net total_space [45gb], spins? [unknown], types [rootfs]
  [2017-08-08 20:22:00,857][INFO ][env                      ] [Aries] heap size [1007.3mb], compressed ordinary object pointers [true]
  [2017-08-08 20:22:07,759][INFO ][node                     ] [Aries] initialized
  [2017-08-08 20:22:07,759][INFO ][node                     ] [Aries] starting ...
  [2017-08-08 20:22:08,179][INFO ][transport                ] [Aries] publish_address {127.0.0.1:9300}, bound_addresses {[::1]:9300}, {127.0.0.1:9300}
  [2017-08-08 20:22:08,187][INFO ][discovery                ] [Aries] elasticsearch/_BFBmx-XR5aHFhbN9nvX2g
  [2017-08-08 20:22:11,796][INFO ][cluster.service          ] [Aries] new_master {Aries}{_BFBmx-XR5aHFhbN9nvX2g}{127.0.0.1}{127.0.0.1:9300}, reason: zen-disco-join(elected_as_master, [0] joins received)
  [2017-08-08 20:22:12,415][INFO ][http                     ] [Aries] publish_address {127.0.0.1:9200}, bound_addresses {[::1]:9200}, {127.0.0.1:9200}
  [2017-08-08 20:22:12,415][INFO ][node                     ] [Aries] started
  [2017-08-08 20:22:12,654][INFO ][gateway                  ] [Aries] recovered [2] indices into cluster_state
  [2017-08-08 20:22:14,526][INFO ][cluster.routing.allocation] [Aries] Cluster health status changed from [RED] to [YELLOW] (reason: [shards started [[nyc-places][1], [nyc-places][0]] ...]). ..
  ...
  ... 
  </code></pre>
  
  <p>Going back to software versions used, it might not match exactly to the versions used by the training site.<br>
  Mine are Flink: <strong>1.3.1</strong>, and Elasticsearch: <strong>2.4.6</strong>.</p>