我使用的是flink1.1.2，并在maven中添加了elesticsearch依赖项，如下所示

<dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-elasticsearch2_2.10</artifactId>
            <version>1.2.0</version>
 </dependency>

我的程序包含以下代码，从kafka读取数据并插入到ElasticSearch

public class ReadFromKafka {

    public static void main(String[] args) throws Exception {
        // create execution environment
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "localhost:9092");
        properties.setProperty("zookeeper.connect", "localhost:2181");
        properties.setProperty("group.id", "test");

        DataStream<JoinedStreamEvent> message = env.addSource(new FlinkKafkaConsumer09<JoinedStreamEvent>("test",
                new JoinSchema(), properties));

        System.out.println("reading form kafka ");

        message.print();

        Map<String, String> config = new HashMap<>();
        config.put("bulk.flush.max.actions", "1");   // flush inserts after every event
        config.put("cluster.name", "elasticsearch_amar"); // default cluster name

        List<InetSocketAddress> transports = new ArrayList<>();
// set default connection details
        transports.add(new InetSocketAddress(InetAddress.getByName("127.0.0.1"), 9300));

       message.addSink(new ElasticsearchSink<>(config,transports,new ElasticInserter()));

        env.execute();

    } //main

    public static class ElasticInserter implements ElasticsearchSinkFunction<JoinedStreamEvent>{

        @Override
        public void process(JoinedStreamEvent record, RuntimeContext runtimeContext, RequestIndexer requestIndexer) {

            Map<String, Integer> json = new HashMap<>();

            json.put("Time", record.getPatient_id());
            json.put("heart Rate ", record.getHeartRate());
            json.put("resp rete", record.getRespirationRate());

            IndexRequest rqst = Requests.indexRequest()
                    .index("nyc-places")           // index name
                    .type("popular-locations")     // mapping name
                    .source(json);

            requestIndexer.add(rqst);

        } //process

    } //ElasticInserter

} //ReadFromKafka

我已经安装了elesticsearch使用 homebrew 然后开始使用 elesticsearch 命令如下所示

然而，当我开始我的程序我得到以下错误