basic-lib/src/main/java/com/streamsets/pipeline/stage/origin/logtail/FileTailSource.java
@Override
  public String produce(String lastSourceOffset, int maxBatchSize, BatchMaker batchMaker) throws StageException {
    int recordCounter = 0;
    long startTime = System.currentTimeMillis();
    maxBatchSize = Math.min(conf.batchSize, maxBatchSize);

    // deserializing offsets of all directories
    Map<String, String> offsetMap = deserializeOffsetMap(lastSourceOffset);

    boolean offsetSet = false;
    while (!offsetSet) {
      try {
        multiDirReader.setOffsets(offsetMap);
        offsetSet = true;
      } catch (IOException ex) {
        LOG.warn("Error while creating reading previous offset: {}", ex.toString(), ex);
        multiDirReader.purge();
      }
    }

    while (recordCounter < maxBatchSize && !isTimeout(startTime)) {
      LiveFileChunk chunk = multiDirReader.next(getRemainingWaitTime(startTime));

      if (chunk != null) {
        String tag = chunk.getTag();
        tag = (tag != null && tag.isEmpty()) ? null : tag;
        String liveFileStr = chunk.getFile().serialize();

        List<FileLine> lines = chunk.getLines();
        int truncatedLine = chunk.isTruncated() ? lines.size()-1 : -1;

        for (int i = 0; i < lines.size(); i++) {
          FileLine line = lines.get(i);
          String sourceId = liveFileStr + "::" + line.getFileOffset();
          try (DataParser parser = parserFactory.getParser(sourceId, line.getText())) {
            if(i == truncatedLine) {
              //set truncated
              parser.setTruncated();
            }
            Record record = parser.parse();
            if (record != null) {
              if (tag != null) {
                record.getHeader().setAttribute("tag", tag);
              }
              record.getHeader().setAttribute(HeaderAttributeConstants.FILE, chunk.getFile().getPath().toString());
              record.getHeader().setAttribute(HeaderAttributeConstants.FILE_NAME, chunk.getFile().getPath().getFileName().toString());
              record.getHeader().setAttribute(HeaderAttributeConstants.OFFSET, String.valueOf(line.getFileOffset()));
              record.getHeader().setAttribute(
                HeaderAttributeConstants.LAST_MODIFIED_TIME,
                String.valueOf(Files.getLastModifiedTime(chunk.getFile().getPath()).toMillis())
              );
              batchMaker.addRecord(record, outputLane);
              recordCounter++;
            }
          } catch (IOException | DataParserException ex) {
            errorRecordHandler.onError(Errors.TAIL_12, sourceId, ex.toString(), ex);
          }
        }
      }
    }

    boolean metadataGenerationFailure = false;
    Date now = new Date(startTime);
    for (FileEvent event : multiDirReader.getEvents()) {
      try {
        LiveFile file = event.getFile().refresh();
        Record metadataRecord = getContext().createRecord("");
        Map<String, Field> map = new HashMap<>();
        map.put("fileName", Field.create(file.getPath().toString()));
        map.put("inode", Field.create(file.getINode()));
        map.put("time", Field.createDate(now));
        map.put("event", Field.create((event.getAction().name())));
        metadataRecord.set(Field.create(map));
        batchMaker.addRecord(metadataRecord, metadataLane);

        // We're also sending the same information on event lane
        String eventRecordSourceId =
            Utils.format("event:{}:{}:{}", event.getAction().name(), 1, file.getPath().toString());
        EventRecord eventRecord = getContext().createEventRecord(event.getAction().name(), 1, eventRecordSourceId);
        eventRecord.set(Field.create(map));
        getContext().toEvent(eventRecord);
      } catch (IOException ex) {
        LOG.warn("Error while creating metadata records: {}", ex.toString(), ex);
        metadataGenerationFailure = true;
      }
    }
    if (metadataGenerationFailure) {
      multiDirReader.purge();
    }

    boolean offsetExtracted = false;
    while (!offsetExtracted) {
      try {
        offsetMap = multiDirReader.getOffsets();
        offsetExtracted = true;
      } catch (IOException ex) {
        LOG.warn("Error while creating creating new offset: {}", ex.toString(), ex);
        multiDirReader.purge();
      }
    }

    //Calculate Offset lag Metric.
    calculateOffsetLagMetric(offsetMap);

    //Calculate Pending Files Metric
    calculatePendingFilesMetric();

    // serializing offsets of all directories
    return serializeOffsetMap(offsetMap);
  }
@Override
  public String produce(String lastSourceOffset, int maxBatchSize, BatchMaker batchMaker) throws StageException {
    int recordCounter = 0;
    long startTime = System.currentTimeMillis();
    maxBatchSize = Math.min(conf.batchSize, maxBatchSize);

    // deserializing offsets of all directories
    Map<String, String> offsetMap = deserializeOffsetMap(lastSourceOffset);

    boolean offsetSet = false;
    while (!offsetSet) {
      try {
        multiDirReader.setOffsets(offsetMap);
        offsetSet = true;
      } catch (IOException ex) {
        LOG.warn("Error while creating reading previous offset: {}", ex.toString(), ex);
        multiDirReader.purge();
      }
    }

    while (recordCounter < maxBatchSize && !isTimeout(startTime)) {
      LiveFileChunk chunk = multiDirReader.next(getRemainingWaitTime(startTime));

      if (chunk != null) {
        String tag = chunk.getTag();
        tag = (tag != null && tag.isEmpty()) ? null : tag;
        String liveFileStr = chunk.getFile().serialize();

        List<FileLine> lines = chunk.getLines();
        int truncatedLine = chunk.isTruncated() ? lines.size()-1 : -1;

        for (int i = 0; i < lines.size(); i++) {
          FileLine line = lines.get(i);
          String sourceId = liveFileStr + "::" + line.getFileOffset();
          try (DataParser parser = parserFactory.getParser(sourceId, line.getText())) {
            if(i == truncatedLine) {
              //set truncated
              parser.setTruncated();
            }
            Record record = parser.parse();
            if (record != null) {
              if (tag != null) {
                record.getHeader().setAttribute("tag", tag);
              }
              record.getHeader().setAttribute(HeaderAttributeConstants.FILE, chunk.getFile().getPath().toString());
              record.getHeader().setAttribute(HeaderAttributeConstants.FILE_NAME, chunk.getFile().getPath().getFileName().toString());
              record.getHeader().setAttribute(HeaderAttributeConstants.OFFSET, String.valueOf(line.getFileOffset()));
              record.getHeader().setAttribute(
                HeaderAttributeConstants.LAST_MODIFIED_TIME,
                String.valueOf(Files.getLastModifiedTime(chunk.getFile().getPath()).toMillis())
              );
              batchMaker.addRecord(record, outputLane);
              recordCounter++;
            }
          } catch (IOException | DataParserException ex) {
            errorRecordHandler.onError(Errors.TAIL_12, sourceId, ex.toString(), ex);
          }
        }
      }
    }

    boolean metadataGenerationFailure = false;
    Date now = new Date(startTime);
    for (FileEvent event : multiDirReader.getEvents()) {
      try {
        LiveFile file = event.getFile().refresh();
        Record metadataRecord = getContext().createRecord("");
        Map<String, Field> map = new HashMap<>();
        map.put("fileName", Field.create(file.getPath().toString()));
        map.put("inode", Field.create(file.getINode()));
        map.put("time", Field.createDate(now));
        map.put("event", Field.create((event.getAction().name())));
        metadataRecord.set(Field.create(map));
        batchMaker.addRecord(metadataRecord, metadataLane);

        // We're also sending the same information on event lane
        String eventRecordSourceId =
            Utils.format("event:{}:{}:{}", event.getAction().name(), 1, file.getPath().toString());
        EventRecord eventRecord = getContext().createEventRecord(event.getAction().name(), 1, eventRecordSourceId);
        eventRecord.set(Field.create(map));
        getContext().toEvent(eventRecord);
      } catch (IOException ex) {
        LOG.warn("Error while creating metadata records: {}", ex.toString(), ex);
        metadataGenerationFailure = true;
      }
    }
    if (metadataGenerationFailure) {
      multiDirReader.purge();
    }

    boolean offsetExtracted = false;
    while (!offsetExtracted) {
      try {
        offsetMap = multiDirReader.getOffsets();
        offsetExtracted = true;
      } catch (IOException ex) {
        LOG.warn("Error while creating creating new offset: {}", ex.toString(), ex);
        multiDirReader.purge();
      }
    }

    //Calculate Offset lag Metric.
    calculateOffsetLagMetric(offsetMap);

    //Calculate Pending Files Metric
    calculatePendingFilesMetric();

    // serializing offsets of all directories
    return serializeOffsetMap(offsetMap);
  }
/*
When we start with a file (empty or not) the file offset is zero.
If the file is a rolled file, the file will be EOF immediately triggering a close of the reader and setting the
offset to Long.MAX_VALUE (this happens in the MultiDirectoryReader class). This is the signal that in the next
read a directory scan should be triggered to get the next rolled file or the live file if we were scanning the last
rolled file.
If the file you are starting is the live file, we don't get an EOF as we expect data to be appended. We just return
null chunks while there is no data. If the file is rolled we'll detect that and then do what is described in the
previous paragraph.

When offset for  file is "" it means we never processed things in the directory, at that point we start from the
first file (according to the defined order) in the directory, or if a 'first file' as been set in the configuration,
we start from that file.

We encode in lastSourceOffset the current file and offset from all directories in JSON.