Flume-ng空事件的自定义接收器

Question

我正在尝试为flume-ng编写一个自定义接收器。我查看了现有的接收器和文档并对其进行了编码。然而，应该接收事件的'process()‘方法总是以null结束。我正在执行Event event = channel.take()；但该事件为空。我在日志中看到，此方法被重复调用，因为事件仍在通道中。

谁能给我指个方向？

Answer 1

这是一个过程函数的框架，如果您无法获得回滚事件，请将状态更改为BACKOFF。如果没有，则提交并将status设置为READY。无论如何，您总是关闭事务。

    Status status = null;
    Channel channel = getChannel();
    Transaction transaction = channel.getTransaction();
    transaction.begin();
    try {
        Event event = channel.take();

        if (event != null && validEvent(event.getBody()) >= 0) {
           # make some printing
        }
        transaction.commit();
        status = Status.READY;
    } catch (Throwable ex) {
        transaction.rollback();
        status = Status.BACKOFF;
        logger.error("Failed to deliver event. Exception follows.", ex);
        throw new EventDeliveryException("Failed to deliver event: " + ex);
    } finally {
        transaction.close();
    }
    return status;

我确信这会起作用:)。

Answer 2

这是设计好的。接收器运行器将使用null事件轮询接收器，因此可以确保接收器处于活动状态，并准备好接受未来的事件。当您接收到null事件时，请确保返回Status.BACKOFF，并且接收器处理器将在重试之前等待一段时间。