code: public void pollEvent(DFSInotifyEventInputStream eventStream) {
      EventBatch batch;
      LOG.debug("Polling thread starting, with timeout {} ms", mActiveUfsPollTimeoutMs);
      int count = 0;
      long start = System.currentTimeMillis();
  
      long behind = eventStream.getTxidsBehindEstimate();
  
      while (!Thread.currentThread().isInterrupted()) {
        try {
          batch = eventStream.poll(mActiveUfsPollTimeoutMs, TimeUnit.MILLISECONDS);
  
          if (batch != null) {
            long txId = batch.getTxid();
            count++;
            for (Event event : batch.getEvents()) {
              processEvent(event, mUfsUriList, txId);
            }
          }
          long end = System.currentTimeMillis();
          if (end > (start + mActiveUfsSyncEventRateInterval)) {
            long currentlyBehind = eventStream.getTxidsBehindEstimate();
            LOG.info("HDFS generated {} events in {} ms, at a rate of {} rps",
                count + currentlyBehind - behind ,
                end - start,
                String.format("%.2f", (count + currentlyBehind - behind) * 1000.0 / (end - start)));
            LOG.info("processed {} events in {} ms, at a rate of {} rps", count,
                end - start,
                String.format("%.2f", count * 1000.0 / (end - start)));
            LOG.info("Currently TxidsBehindEstimate by {}", currentlyBehind);
            behind = currentlyBehind;
            start = end;
            count = 0;
          }
        } catch (IOException e) {
          LOG.warn("IOException occured during polling inotify {}", e);
          if (e.getCause() instanceof InterruptedException) {
            return;
          }
        } catch (MissingEventsException e) {
          LOG.warn("MissingEventException during polling {}", e);
          mEventMissed = true;
          // need to sync all syncpoints at this point
        } catch (InterruptedException e) {
          LOG.warn("InterruptedException during polling {}", e);
          return;
        }
      }
    }
docs: Fetch and process events.
  @param eventStream event stream