ThriftSource.java 文件源码

@Override
public Status appendBatch(List<ThriftFlumeEvent> events) throws TException {
  sourceCounter.incrementAppendBatchReceivedCount();
  sourceCounter.addToEventReceivedCount(events.size());

  List<Event> flumeEvents = Lists.newArrayList();
  for (ThriftFlumeEvent event : events) {
    flumeEvents.add(EventBuilder.withBody(event.getBody(), event.getHeaders()));
  }

  try {
    getChannelProcessor().processEventBatch(flumeEvents);
  } catch (ChannelException ex) {
    logger.warn("Thrift source %s could not append events to the channel.", getName());
    return Status.FAILED;
  }

  sourceCounter.incrementAppendBatchAcceptedCount();
  sourceCounter.addToEventAcceptedCount(events.size());
  return Status.OK;
}