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时，您只需返回它。但是，正确的方法是提交或回滚。事务应该经历三个阶段：begin、commit或rollback、finally close。
基本通道语义：

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;
  }

当currenttransaction为null或其状态为close时，通道将创建一个新的，否则返回旧的。这种例外情况不会立即发生。第一次执行process方法时，得到一个新的事务，但是事件为null，只返回并最终关闭，close方法因为它的实现而不起作用。所以第二次执行process方法时，没有得到一个新的事务，它是旧的事务。下面的代码是关于事务如何实现的。
基本交易语义：

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();
  }

创建时，状态是新的。
开始时，状态必须是新的，然后状态变为开放的。
提交或回滚时，状态必须是open，然后状态变为complete。
关闭时，状态必须完整，然后状态变为关闭。
所以当你以正确的方式执行close方法时，下次你会得到一个新的事务，否则状态为new的旧事务就不能执行transaction.begin（），它需要一个新的事务。