FLUME IllegalStateException:当事务打开时调用begin()

Question

我编写了名为MySink的自定义水槽接收器，其处理方法在下面的第一个代码段中指示。我收到一个非法状态异常，如下所示（下面的第二个代码段中提供了详细的堆栈跟踪）：

原因：java.lang.IllegalStateException： begin（） 调用时 交易已开放！

问：在编写进程方法时，我在 flume 代码库中遵循了 KafkaSink 和类似的现有接收器实现，并且我对这些现有接收器应用了完全相同的事务处理逻辑。你能告诉我我的工艺方法有什么问题吗？如何解决问题？

进程方法（我已经标记了抛出异常的位置）：

@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<String, String> 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;
}

异常堆栈：

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.

org.apache.flume.EventDeliveryException: MySink - Failed to publish events.
Exception:    at com.XYZ.flume.maprdb.MySink.process(MySink.java:116)

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)

Caused by: java.lang.IllegalStateException: begin() called when transaction is OPEN!

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

Answer 1

if (event == null) {
    // No event found, request back-off semantics from the sink runner
    LOG.info(getName() + " - EVENT is null! ");
    return Status.BACKOFF;
 }

当 event 为 null 时，只需返回它。但是，正确的方法是提交或回滚。事务应经历三个阶段：开始、提交或回滚、最后关闭。我们可以看到以下源代码来了解它是如何实现的。

基本通道语义：

  public Transaction getTransaction() {
    if (!initialized) {
      synchronized (this) {
        if (!initialized) {
          initialize();
          initialized = true;
        }
      }
    }
    BasicTransactionSemantics transaction = currentTransaction.get();
    if (transaction == null || transaction.getState().equals(
            BasicTransactionSemantics.State.CLOSED)) {
      transaction = createTransaction();
      currentTransaction.set(transaction);
    }
    return transaction;
  }

当当前事务为空或其状态为关闭时，通道将创建一个新的，否则返回旧的。 此异常不会立即发生。 第一次执行 Process 方法时，你得到一个新的事务，但事件为 null，你只是返回并最终关闭， 关闭方法因其 implement.so 而不起作用 第二次执行流程方法， 您不会得到一个新事务，它是旧事务。以下代码是关于事务如何实现的。

基本事务语义：

  protected BasicTransactionSemantics() {
    state = State.NEW;
    initialThreadId = Thread.currentThread().getId();
  }
  
  public void begin() {
    Preconditions.checkState(Thread.currentThread().getId() == initialThreadId,
        "begin() called from different thread than getTransaction()!");
    Preconditions.checkState(state.equals(State.NEW),
        "begin() called when transaction is " + state + "!");
    try {
      doBegin();
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
      throw new ChannelException(e.toString(), e);
    }
    state = State.OPEN;
  }
  public void commit() {
    Preconditions.checkState(Thread.currentThread().getId() == initialThreadId,
        "commit() called from different thread than getTransaction()!");
    Preconditions.checkState(state.equals(State.OPEN),
        "commit() called when transaction is %s!", state);
    try {
      doCommit();
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
      throw new ChannelException(e.toString(), e);
    }
    state = State.COMPLETED;
  }
 public void rollback() {
    Preconditions.checkState(Thread.currentThread().getId() == initialThreadId,
        "rollback() called from different thread than getTransaction()!");
    Preconditions.checkState(state.equals(State.OPEN),
        "rollback() called when transaction is %s!", state);
    state = State.COMPLETED;
    try {
      doRollback();
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
      throw new ChannelException(e.toString(), e);
    }
  }
  
 public void close() {
    Preconditions.checkState(Thread.currentThread().getId() == initialThreadId,
        "close() called from different thread than getTransaction()!");
    Preconditions.checkState(
            state.equals(State.NEW) || state.equals(State.COMPLETED),
            "close() called when transaction is %s"
            + " - you must either commit or rollback first", state);
    state = State.CLOSED;
    doClose();
  }

创建时，状态为新状态。

开始时，状态

必须是新的，然后状态变为开放。

提交或回滚时，状态必须为打开，然后状态变为完成。

当"关闭

"时，状态必须为"完整"，然后状态变为"关闭"。

所以当你以正确的方式执行 close 方法时，下次你会得到一个新的事务，否则旧的状态一定不是新的，所以你不能执行 transaction.begin（），它需要一个新的。