code: @Override
      public void onEvent(Object event, long sequence, boolean endOfBatch) throws Exception {
          try {
              if (event == null) {
                  return;
              }
              Runnable runnable = null;
              if (event instanceof Tuple) {
                  if (((TupleExt) event).isBatchTuple()) {
                      List<Object> values = ((Tuple) event).getValues();
                      for (Object value : values) {
                          Pair<MessageId, List<Object>> val = (Pair<MessageId, List<Object>>) value;
                          TupleImplExt tuple = new TupleImplExt(
                                  sysTopologyCtx, val.getSecond(), val.getFirst(), ((TupleImplExt) event));
                          runnable = processTupleEvent(tuple);
                          if (runnable != null) {
                              runnable.run();
                              runnable = null;
                          }
                      }
                  } else {
                      runnable = processTupleEvent((Tuple) event);
                  }
              } else if (event instanceof TimerTrigger.TimerEvent) {
                  processTimerEvent((TimerTrigger.TimerEvent) event);
                  return;
              } else if (event instanceof IAckMsg) {
                  runnable = (Runnable) event;
              } else if (event instanceof Runnable) {
                  runnable = (Runnable) event;
              } else {
  
                  LOG.warn("Receive one unknown event-" + event.toString() + " " + idStr);
                  return;
              }
  
              if (runnable != null)
                  runnable.run();
  
          } catch (Throwable e) {
              if (!taskStatus.isShutdown()) {
                  LOG.info("Unknown exception ", e);
                  reportError.report(e);
              }
          }
      }
docs: Handle acker message
  
  @see com.lmax.disruptor.EventHandler#onEvent(java.lang.Object, long, boolean)