title: Flume1.5.2 throws Unable to deliver event with com.frontier45.flume.sink.elasticsearch2.ElasticSearchSink
body: <p>We are using Apache Flume for indexing data into Elasticsearch.
  Flume throws the exception upon start</p>
  
  <pre><code>17 May 2016 17:57:30,121 ERROR [SinkRunner-PollingRunner-DefaultSinkProcessor] (com.frontier45.flume.sink.elasticsearch2.ElasticSearchSink.process:201)  - Failed to commit transaction. Transaction rolled back.
  java.lang.NullPointerException
          at com.frontier45.flume.sink.elasticsearch2.ElasticSearchSink.process(ElasticSearchSink.java:169)
          at org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:68)
          at org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:147)
          at java.lang.Thread.run(Thread.java:745)
  17 May 2016 17:57:30,124 ERROR [SinkRunner-PollingRunner-DefaultSinkProcessor] (org.apache.flume.SinkRunner$PollingRunner.run:160)  - Unable to deliver event. Exception follows.
  java.lang.NullPointerException
          at com.frontier45.flume.sink.elasticsearch2.ElasticSearchSink.process(ElasticSearchSink.java:169)
          at org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:68)
          at org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:147)
          at java.lang.Thread.run(Thread.java:745)
  </code></pre>
  
  <hr>
  
  <p>Flume conf file settings</p>
  
  <pre><code>a1.sources = r1
  a1.sinks = k1
  a1.channels = c1
  
  # Describe/configure the source
  a1.sources.r1.type=exec
  a1.sources.r1.command=tail -f /tmp/storm/logs/perf.log
  
  #interceptors
  a1.sources.r1.interceptors = i1
  a1.sources.r1.interceptors.i1.type = REGEX_EXTRACTOR 
  #the following regexp parse a line like 
  #2015.06.29 08:59:51,650 INFO perf-logger 76 pool-13-thread-1 Reducer-38:97177163
  a1.sources.r1.interceptors.i1.regex = (\\d{4}-\\d{2}-\\d{2}\\s\\d{2}:\\d{2}:\\d{2},\\d{3})\\s*([A-Z]{4})\\s*([a-zA-Z-.]*)\\s(\\d+)\\s([a-zA-Z-.0.9]*)\\s([a-zA-Z-.0.9]*):(\\d*)
  a1.sources.r1.interceptors.i1.serializers = s1 s2 s3 s4 s5 s6 s7
  a1.sources.r1.interceptors.i1.serializers.s1.name = timestamp
  a1.sources.r1.interceptors.i1.serializers.s1.type = org.apache.flume.interceptor.RegexExtractorInterceptorMillisSerializer
  a1.sources.r1.interceptors.i1.serializers.s1.pattern = yyyy-MM-dd HH:mm:ss,SSS
  a1.sources.r1.interceptors.i1.serializers.s2.name = level
  a1.sources.r1.interceptors.i1.serializers.s3.name = logger
  a1.sources.r1.interceptors.i1.serializers.s4.name = line
  a1.sources.r1.interceptors.i1.serializers.s5.name = thread
  a1.sources.r1.interceptors.i1.serializers.s6.name = component
  a1.sources.r1.interceptors.i1.serializers.s7.name = value
  
  # Describe the sink
  a1.sinks.k1.type = com.frontier45.flume.sink.elasticsearch2.ElasticSearchSink 
  a1.sinks.k1.hostNames = localhost:9300
  a1.sinks.k1.indexName = storm_logs  
  a1.sinks.k1.indexType = storm_logs
  a1.sinks.k1.clusterName = log_cluster
  a1.sinks.k1.batchSize = 500
  a1.sinks.k1.ttl = 5d
  a1.sinks.k1.serializer = com.frontier45.flume.sink.elasticsearch2.ElasticSearchDynamicSerializer 
  a1.sinks.k1.indexNameBuilder = com.frontier45.flume.sink.elasticsearch2.TimeBasedIndexNameBuilder 
  
  # Use a channel which buffers events in memory
  a1.channels.c1.type = memory
  a1.channels.c1.memoryCapacity=100
  #a1.channels.c1.byteCapacity=5242880
  a1.channels.c1.checkpointDir = /tmp/elasticsearch/log
  a1.channels.c1.dataDirs = /tmp/elasticsearch/data
  a1.channels.c1.use-fast-replay = false
  
  # Bind the source and sink to the channel
  a1.sources.r1.channels = c1
  a1.sinks.k1.channel = c1
  </code></pre>
  
  <hr>
  
  <p>Using Flume1.5.0
  Elasticsearch2.3.0</p>
  
  <p>Anyone faced this errror? please share the resolution.</p>