if (readConfiguration.getInstance() .getPropertyMap() .containsKey(word.replace("#", ""))) { String companyName = readConfiguration .getPropertyMap().get( word.replace("#", "")); This logic is been implemented and getting the desired output. Thanks. ... View more

The PutElasticsearch processor uses the Transport API for Elasticsearch, not the HTTP API. This means your port should be 9300 not 9200. The "Identifier Attribute" property is the name of a flow file attribute that contains a unique identifer for the document. If you don't have an identifier you want to use, you can put "uuid", this will use the flow file's UUID as the identifier for the Elasticsearch document. If you do have an identifier for the document, put its value (using UpdateAttribute, EvaluateJsonPath, etc.) into a flow file attribute, and put that attribute's name in the "Identifier Attribute" property. Note you don't use Expression language here, so if your attribute's name is "doc_id", you put "doc_id" in the Identifier Attribute property, not "${doc_id}" ... View more

You can kill a storm topology as shown below. Use set_wait_secs to set some buffer time so messages already in topology are completely processed before topology is killed. Its equivalent to -w option using storm kill CLI command. Map conf = Utils.readStormConfig(); Client client = NimbusClient.getConfiguredClient(conf).getClient(); KillOptions killOpts = new KillOptions(); killOpts.set_wait_secs(waitSeconds); // time to wait before killing client.killTopologyWithOpts(topology_name, killOpts); //provide topology name I'm not sure if there is any direct way to achieve what you want without 1) changing this value for every run or 2) set this to a very high value (like 10 min) so that its guarantees that all messages are processed before killing topology. Please keep in mind that the main use case of storm is to do continuous computation on data by having your topologies running for ever. ... View more