FLUME IllegalStateException: begin() called when transaction is OPEN
<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>