Drop null Flume events before sending to channel

4
votes

I had read in a Flume book that if in the intercept method of interceptor an event is returned as null, the event will be dropped. Hence i have created a custom interceptor which on basis of a condition returns the event as null like:

public Event intercept(Event event) {
    // TODO Auto-generated method stub
    Event finalEvent = event;
    check = new String(event.getBody(),Charsets.UTF_8);

    if(check.matches("([0-9]-.+?-.+?-[0-9][0-9]+)")){

        try {
            fileWriter.append(new String(event.getBody(),Charsets.UTF_8)+ "\n");
        } catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        finalEvent = null;
    }
    System.out.println("Event is : " + finalEvent);
    return finalEvent;
}

The interceptor emits null event but the file channel still passes it to the HDFS sink as empty. Why doesn't the event get dropped?? I am using Spooling directory as source.

2
hadoopnullinterceptorclouderaflume
Maybe you could do a test? Make your code return null all the time and see if all the events are dropped. If they are, then the issue is likely in the matching logic rather than the flume config. - Andrew Ehrlich
@AndrewEhrlich I have a similar issue. In my case I have modified the event Body in the interceptor based on a regex. In the sink i.e the HDFS I can see the lines that match the regex modified as per applied logic, however the other events which should have been dropped are present unmodified so my regex works fine.I am doing a simple "return null" in the code . Do I need to do something else? - Suvarna Pattayil
Could you give a link to your flume config? Perhaps something is misconfigured there. - Andrew Ehrlich

2 Answers

2
votes

Let's see what happens.you use Spooling directory as source, the source call function processEventBatch(events),inside processEventBatch() :

events = interceptorChain.intercept(events);//use your custom interceptor
...
eventQueue.add(event); // add user event to queue,even the event == null

if Spooling directory source use processEvent(),then your interceptor will works: inside processEvent():

event = interceptorChain.intercept(event);
if (event == null) {
  //null event then return !! intercept works!!
  return;
}

so you should modiry processEventBatch(),and do:

if (event == null){
    //dont add to eventQueue
}
2
votes

In my interceptor class, the method intercept(Event event) contains the logic as to how the data flows(as written in the question) which returns null in case the regex is not satisfied, and in the method intercept(List<Event> events) the null events are excluded which servers the purpose. Following is the intercept(List<Event> events) code:

public List<Event> intercept(List<Event> events) 
    {
          List<Event> interceptedEvents = new ArrayList<Event>(events.size());
          for (Event event : events) 
          {
              // Intercept any event
              Event interceptedEvent = intercept(event);
              if(interceptedEvent!=null)
                  interceptedEvents.add(interceptedEvent);
          }

          return interceptedEvents;
    }