[FLUME-3032] taildir source sleeps frequently.

• Type: Bug
• Status: Open
• Priority: Major
• Resolution: Unresolved
• Affects Version/s: 1.7.0
• Component/s: Sinks+Sources
• Labels: newbie

Environment

CentOS Linux release 7.2.1511 (Core)
java version "1.7.0_80"

• Flags: Patch
• Assignee: Unassigned
• Reporter: Liu Tianhao
• Created: 25/Nov/16 02:48
• Updated: 26/Dec/16 02:57

Description

Test configuration.
source - taildir
interceptor - The custom interceptor drops some events
channel - anyone
sink - none

I found that taildir source sleeps frequently.

The tailFileProcess() function in TaildirSource.java break the loop by (events.size() < batchSize), but interceptor may change events.size().

I think the events.size() should be used before interceptor processing.
Avoid unnecessary sleep.

TaildirSource.java

    private void tailFileProcess(TailFile tf, boolean backoffWithoutNL)
            throws IOException, InterruptedException {
        long receivedSize = 0;
        while (true) {
            reader.setCurrentFile(tf);
            List<Event> events = reader.readEvents(batchSize, backoffWithoutNL);
            if (events.isEmpty()) {
                break;
            }
            receivedSize = events.size();
            sourceCounter.addToEventReceivedCount(receivedSize);
            sourceCounter.incrementAppendBatchReceivedCount();
            try {
                getChannelProcessor().processEventBatch(events);
                reader.commit();
            } catch (ChannelException ex) {
                logger.warn("The channel is full or unexpected failure. "
                        + "The source will try again after " + retryInterval
                        + " ms");
                TimeUnit.MILLISECONDS.sleep(retryInterval);
                retryInterval = retryInterval << 1;
                retryInterval = Math.min(retryInterval, maxRetryInterval);
                continue;
            }
            retryInterval = 1000;
            sourceCounter.addToEventAcceptedCount(events.size());
            sourceCounter.incrementAppendBatchAcceptedCount();
            if (receivedSize < batchSize) {
                break;
            }
        }
    }