# ESSpark : Callback function on success and fail

**URL:** <https://discuss.elastic.co/t/esspark-callback-function-on-success-and-fail/61893>\
**Category:** Elasticsearch\
**Tags:** es-hadoop\
**Created:** [September 30, 2016, 4:49am UTC](https://discuss.elastic.co/t/esspark-callback-function-on-success-and-fail/61893 "2016-09-30T04:49:51Z")\
**Posts on this page:** 2\
**Page:** 1

<div class="post-metadata">

**Author:** ![Akshay\_Mhetre](https://avatars.discourse-cdn.com/v4/letter/a/3ec8ea/32.png) [@Akshay\_Mhetre](https://discuss.elastic.co/u/Akshay_Mhetre)\
**Post date:** [September 30, 2016, 4:49am UTC](https://discuss.elastic.co/t/esspark-callback-function-on-success-and-fail/61893/1 "2016-09-30T04:49:51Z")

</div>

Hi,

I am trying to save rdd in ES using :  
EsSpark.saveJsonToEs(rdd, "logstash-{@timestamp:YYYY.MM.dd}/logs"))  
This rdd I am getting from a kafka stream created using kafka direct api. As this api. enforces you to store offset in zookeeper by yourself, I want to save offset whenever the message is successfully sent to ES. However, as this rdd might contain set of messages and there could be chances of failure in between., I need a method such that I can call save offset on each successful message sent.

e.g.

EsSpark.saveJsonToEs(rdd, "logstash-{@timestamp:YYYY.MM.dd}/logs")).onEachSuccessfulMsg(str =\> saveOffsetForMsg(str))

Is there any way with which I can hook to each message sent ?  
or is there any api using which I can save each message in ES instead of rdd.

Thanks,  
Akshay

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\
**Post date:** [July 6, 2017, 1:23pm UTC](https://discuss.elastic.co/t/esspark-callback-function-on-success-and-fail/61893/2 "2017-07-06T13:23:04Z")

</div>