[FLUME-3364] TailDir BackOff Method always blocks the thread

• Type: Bug
• Status: Open
• Priority: Critical
• Resolution: Unresolved
• Affects Version/s: 1.9.0
• Component/s: Sinks+Sources

Environment

in mac system.  The backoff method will block the thread with the data has been read. 

• Assignee: Unassigned
• Reporter: wang qun
• Created: 26/Apr/20 13:44
• Updated: 07/May/20 04:59

Description

The backoff method in taildir source is invalid because of a code bug.

This part should not set default status as false

//代码占位符
public Status process() {
  Status status = Status.BACKOFF;
  try {
    existingInodes.clear();
    existingInodes.addAll(reader.updateTailFiles());
    for (long inode : existingInodes) {
      TailFile tf = reader.getTailFiles().get(inode);
      if (tf.needTail()) {
        boolean hasMoreLines = tailFileProcess(tf, true);
        if (hasMoreLines) {
          status = Status.READY;
        }
      }
    }
    closeTailFiles();
  } catch (Throwable t) {
    logger.error("Unable to tail files", t);
    sourceCounter.incrementEventReadFail();
    status = Status.BACKOFF;
  }
  return status;
}

 

//代码占位符
private boolean tailFileProcess(TailFile tf, boolean backoffWithoutNL)
    throws IOException, InterruptedException {
  long batchCount = 0;
  while (true) {
    reader.setCurrentFile(tf);
    List<Event> events = reader.readEvents(batchSize, backoffWithoutNL);
    if (events.isEmpty()) {
      return false;
    }
    sourceCounter.addToEventReceivedCount(events.size());
    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");
      sourceCounter.incrementChannelWriteFail();
      TimeUnit.MILLISECONDS.sleep(retryInterval);
      retryInterval = retryInterval << 1;
      retryInterval = Math.min(retryInterval, maxRetryInterval);
      continue;
    }
    retryInterval = 1000;
    sourceCounter.addToEventAcceptedCount(events.size());
    sourceCounter.incrementAppendBatchAcceptedCount();
    if (events.size() < batchSize) {
      logger.debug("The events taken from " + tf.getPath() + " is less than " + batchSize);
      return false;
    }
    if (++batchCount >= maxBatchCount) {
      logger.debug("The batches read from the same file is larger than " + maxBatchCount );
      return true;
    }
  }
}