[ https://issues.apache.org/jira/browse/STORM-845?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14639782#comment-14639782 ]
ASF GitHub Bot commented on STORM-845: -------------------------------------- Github user HeartSaVioR commented on a diff in the pull request: https://github.com/apache/storm/pull/573#discussion_r35388927 --- Diff: external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/bolt/EsPercolateBolt.java --- @@ -0,0 +1,81 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.storm.elasticsearch.bolt; + +import backtype.storm.task.OutputCollector; +import backtype.storm.task.TopologyContext; +import backtype.storm.topology.OutputFieldsDeclarer; +import backtype.storm.tuple.Fields; +import backtype.storm.tuple.Tuple; +import backtype.storm.tuple.Values; +import org.apache.storm.elasticsearch.common.EsConfig; +import org.elasticsearch.action.percolate.PercolateResponse; +import org.elasticsearch.action.percolate.PercolateSourceBuilder; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Map; + +public class EsPercolateBolt extends AbstractEsBolt { + private static final Logger LOG = LoggerFactory.getLogger(EsPercolateBolt.class); + + /** + * EsPercolateBolt constructor + * @param esConfig Elasticsearch configuration containing node addresses and cluster name {@link EsConfig} + */ + public EsPercolateBolt(EsConfig esConfig) { + super(esConfig); + } + + @Override + public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) { + super.prepare(map, topologyContext, outputCollector); + } + + /** + * Executes percolate request for given tuple. + * @param tuple should contain string values of 3 declared fields: "source", "index", "type" + */ + @Override + public void execute(Tuple tuple) { + try { + String source = tuple.getStringByField("source"); + String index = tuple.getStringByField("index"); + String type = tuple.getStringByField("type"); + + PercolateResponse response = client.preparePercolate().setIndices(index).setDocumentType(type) + .setPercolateDoc(PercolateSourceBuilder.docBuilder().setDoc(source)).execute().actionGet(); + if (response.getCount() > 0) { + for (PercolateResponse.Match match : response) { + String id = match.getId().toString(); + collector.emit(new Values(id)); + } + } + collector.ack(tuple); + } catch (Exception e) { + e.printStackTrace(); --- End diff -- Please use LOG instead of printing to stderr. (Line 71) > Storm ElasticSearch connector > ----------------------------- > > Key: STORM-845 > URL: https://issues.apache.org/jira/browse/STORM-845 > Project: Apache Storm > Issue Type: New Feature > Reporter: Adrian Seungjin Lee > Assignee: Adrian Seungjin Lee > > It would be nice to provide storm driver for elasticsearch, just like it does > for hive, redis and so on. -- This message was sent by Atlassian JIRA (v6.3.4#6332)