[FLUME-2772] SpoolDirectorySource drop event while the channel is full, with no any exception

• Type: Bug
• Status: Open
• Priority: Major
• Resolution: Unresolved
• Affects Version/s: 1.6.0
• Component/s: Sinks+Sources
• Assignee: Unassigned
• Reporter: Ming-Lung Weng
• Created: 26/Aug/15 08:58
• Updated: 19/Sep/15 17:52

Description

I found SpoolDirectorySource drop Event when channel(memory channel i used) is full

at org.apache.flume.source.SpoolDirectorySource.run()

Bar.java

        while (!Thread.interrupted()) {
          List<Event> events = reader.readEvents(batchSize);
          if (events.isEmpty()) {
            break;
          }
          sourceCounter.addToEventReceivedCount(events.size());
          sourceCounter.incrementAppendBatchReceivedCount();

          try {
            getChannelProcessor().processEventBatch(events);
            reader.commit();
          } catch (ChannelException ex) {
            logger.warn("The channel is full, and cannot write data now. The " +
              "source will try again after " + String.valueOf(backoffInterval) +
              " milliseconds");
            hitChannelException = true;
            if (backoff) {
              TimeUnit.MILLISECONDS.sleep(backoffInterval);
              backoffInterval = backoffInterval << 1;
              backoffInterval = backoffInterval >= maxBackoff ? maxBackoff :
                                backoffInterval;
            }
            continue;
          }
          backoffInterval = 250;
          sourceCounter.addToEventAcceptedCount(events.size());
          sourceCounter.incrementAppendBatchAcceptedCount();
        }

SpoolDirectorySource sleep a backoff time and re-calculate backoff time,
and continue to read next event DO NOTHING with the event that not put into channel

There is no exception but only logger.warn that sound not important

This is a bug to me, but I'm not sure is this a bug

I use a simple while loop to fix this, just like

Bar.java

            boolean eventPushed = false;
            while(!eventPushed) {
                try {
                  getChannelProcessor().processEventBatch(events);
                  reader.commit();
                  eventPushed = true;
                } catch (ChannelException ex) {
                  logger.warn("The channel is full, and cannot write data now. The " +
                    "source will try again after " + String.valueOf(backoffInterval) +
                    " milliseconds");
                  hitChannelException = true;
                  if (backoff) {
                    TimeUnit.MILLISECONDS.sleep(backoffInterval);
                    backoffInterval = backoffInterval << 1;
                    backoffInterval = backoffInterval >= maxBackoff ? maxBackoff :
                                      backoffInterval;
                  }
                }
            
            }