Flume Spool Source 源码过程分析（未运行）

 1 @Override
 2     public void run() {
 3       int backoffInterval = 250;
 4       try {
 5         while (!Thread.interrupted()) {
 6           List<Event> events = reader.readEvents(batchSize);
 7           if (events.isEmpty()) {
 8             break;
 9           }
10           sourceCounter.addToEventReceivedCount(events.size());
11           sourceCounter.incrementAppendBatchReceivedCount();
12 
13           try {
14             getChannelProcessor().processEventBatch(events);
15             reader.commit();
16           } catch (ChannelException ex) {
17             logger.warn("The channel is full, and cannot write data now. The " +
18               "source will try again after " + String.valueOf(backoffInterval) +
19               " milliseconds");
20             hitChannelException = true;
21             if (backoff) {
22               TimeUnit.MILLISECONDS.sleep(backoffInterval);
23               backoffInterval = backoffInterval << 1;
24               backoffInterval = backoffInterval >= maxBackoff ? maxBackoff :
25                                 backoffInterval;
26             }
27             continue;
28           }
29           backoffInterval = 250;
30           sourceCounter.addToEventAcceptedCount(events.size());
31           sourceCounter.incrementAppendBatchAcceptedCount();
32         }
33         logger.info("Spooling Directory Source runner has shutdown.");
34       } catch (Throwable t) {
35         logger.error("FATAL: " + SpoolDirectorySource.this.toString() + ": " +
36             "Uncaught exception in SpoolDirectorySource thread. " +
37             "Restart or reconfigure Flume to continue processing.", t);
38         hasFatalError = true;
39         Throwables.propagate(t);
40       }
41     }

 1  public List<Event> readEvents(int numEvents) throws IOException {
 2     if (!committed) {
 3       if (!currentFile.isPresent()) {
 4         throw new IllegalStateException("File should not roll when " +
 5             "commit is outstanding.");
 6       }
 7       logger.info("Last read was never committed - resetting mark position.");
 8       currentFile.get().getDeserializer().reset();
 9     } else {
10       // Check if new files have arrived since last call
11       if (!currentFile.isPresent()) {
12         currentFile = getNextFile();
13       }
14       // Return empty list if no new files
15       if (!currentFile.isPresent()) {
16         return Collections.emptyList();
17       }
18     }
19 
20     EventDeserializer des = currentFile.get().getDeserializer();
21     List<Event> events = des.readEvents(numEvents);
22 
23     /* It‘s possible that the last read took us just up to a file boundary.
24      * If so, try to roll to the next file, if there is one. */
25     if (events.isEmpty()) {
26       retireCurrentFile();
27       currentFile = getNextFile();
28       if (!currentFile.isPresent()) {
29         return Collections.emptyList();
30       }
31       events = currentFile.get().getDeserializer().readEvents(numEvents);
32     }
33 
34     //写入header值，略47 
48     committed = false;
49     lastFileRead = currentFile;
50     return events;
51   }

 1 String nextPath = file.getPath();
 2       PositionTracker tracker =
 3           DurablePositionTracker.getInstance(metaFile, nextPath);
 4       if (!tracker.getTarget().equals(nextPath)) {
 5         tracker.close();
 6         deleteMetaFile();
 7         tracker = DurablePositionTracker.getInstance(metaFile, nextPath);
 8       }
15       ResettableInputStream in =
16           new ResettableFileInputStream(file, tracker,
17               ResettableFileInputStream.DEFAULT_BUF_SIZE, inputCharset,
18               decodeErrorPolicy);
19       EventDeserializer deserializer = EventDeserializerFactory.getInstance
20           (deserializerType, deserializerContext, in);

 public void commit() throws IOException {
    if (!committed && currentFile.isPresent()) {
      currentFile.get().getDeserializer().mark();
      committed = true;
    }
  }

 @Override
  public void mark() throws IOException {
    ensureOpen();
    in.mark();
  }

 @Override
  public void mark() throws IOException {
    tracker.storePosition(tell());
  }

 @Override
  public synchronized void storePosition(long position) throws IOException {
    metaCache.setOffset(position);
    writer.append(metaCache);
    writer.sync();
    writer.flush();
  }