title: FLUME IllegalStateException: begin() called when transaction is OPEN
body: <p>I have written custom flume sink, named MySink, whose process method is indicated in the first snippet below. I am getting an IllegalStateException as follows (detailed stack trace is available in the 2nd snippet below):</p>
  
  <blockquote>
    <p>Caused by: java.lang.IllegalStateException: begin() called when
    transaction is OPEN!</p>
  </blockquote>
  
  <p><strong>QUESTION:</strong> I have followed the KafkaSink and similar existing sink implementations in flume code base while writing the process method and I am applying the very same transaction handling logic with those exiting sinks. Could you please tell me what is wrong in my process method here? How can I fix the problem?</p>
  
  <p><strong>PROCESS method (I have marked where the exception is thrown):</strong></p>
  
  <pre><code>@Override
  public Status process() throws EventDeliveryException {
      Status status = Status.READY;
      Channel ch = getChannel();
      Transaction txn = ch.getTransaction();
      Event event = null;
  
      try {
          LOG.info(getName() + " BEFORE txn.begin()");
      //!!!! EXCEPTION IS THROWN in the following LINE !!!!!!
          txn.begin();
          LOG.info(getName() + " AFTER txn.begin()");
          LOG.info(getName() + " BEFORE ch.take()");
          event = ch.take();
          LOG.info(getName() + " AFTER ch.take()");
  
          if (event == null) {
              // No event found, request back-off semantics from the sink runner
              LOG.info(getName() + " - EVENT is null! ");
              return Status.BACKOFF;
          }
  
          Map&lt;String, String&gt; keyValueMapInTheMessage = event.getHeaders();
          if (!keyValueMapInTheMessage.isEmpty()) {
              mDBWriter.insertDataToDB(keyValueMapInTheMessage);
          }
  
          LOG.info(getName() + " - EVENT: " + EventHelper.dumpEvent(event));
          if (txn != null) {
              txn.commit();                
          }
  
      } catch (Exception ex) {
          String errMsg = getName() + " - Failed to publish events. Exception: ";
          LOG.info(errMsg);
          status = Status.BACKOFF;
          if (txn != null) {
              try {
                  txn.rollback();
              } catch (Exception e) {
                  LOG.info(getName() + " - EVENT: " + EventHelper.dumpEvent(event));
                  throw Throwables.propagate(e);
              }
          }
          throw new EventDeliveryException(errMsg, ex);
      } finally {
          if (txn != null) {
              txn.close();
          }
      }
  
      return status;
  }
  </code></pre>
  
  <p><strong>EXCEPTION STACK:</strong></p>
  
  <pre><code>2016-01-22 14:01:15,440 (SinkRunner-PollingRunner-DefaultSinkProcessor) [ERROR - org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:160)]  
  
  Unable to deliver event. Exception follows.
  </code></pre>
  
  <blockquote>
  <pre><code>org.apache.flume.EventDeliveryException: MySink - Failed to publish events.
  Exception:    at com.XYZ.flume.maprdb.MySink.process(MySink.java:116)
  </code></pre>
  </blockquote>
  
  <pre><code>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>
  
  <blockquote>
  <pre><code>Caused by: java.lang.IllegalStateException: begin() called when transaction is OPEN!
  </code></pre>
  </blockquote>
  
  <pre><code>at com.google.common.base.Preconditions.checkState(Preconditions.java:145)
  at org.apache.flume.channel.BasicTransactionSemantics.begin(BasicTransactionSemantics.java:131)
  at com.XYZ.flume.maprdb.MySink.process(MySink.java:82)
  ... 3 more
  </code></pre>