Package com.norconex.crawler.core.ledger

Class CrawlerEntryLedger

java.lang.Object
com.norconex.crawler.core.ledger.CrawlerEntryLedger
public final class CrawlerEntryLedger extends Object

Tracks document state and any other meta information required for document processing. Acts as a facade over a few processing state-specific data stores. Includes persisting of information necessary for incremental crawls and resumes.

The processing of a document has the following stages:

  • Queued: References extracted from documents are first queued for processing.
  • Processing: A reference is currently being processed.
  • Processed: A reference has been processed. If the same reference is encountered again during the same run, it will be ignored.

Once a crawl completes, processed references become "cached" on the next run and are used to establish deltas and save processing.

  • Constructor Details

    • CrawlerEntryLedger

      public CrawlerEntryLedger()

  • Method Details

    • setQueuedListener

      public void setQueuedListener(Consumer<CrawlerEntry> listener)
      Registers a listener that is called each time a new reference is successfully added to the queue. Use this to decouple application-level event publishing from the ledger.
      Parameters:
      listener - callback invoked with the queued CrawlerEntry

    • ensureCurrentLedgerAliasExists

      public void ensureCurrentLedgerAliasExists()
      Ensures the current ledger alias exists in session cache. Sets default (LEDGER_A) if not present. This should be called by the coordinator before any ledger rotation to establish the initial state for all cluster nodes.

    • init

      public void init(CrawlerSession session)

    • requeueQueuedEntries

      public int requeueQueuedEntries()
      Re-queues entries that were in QUEUED state from a previous run. This is needed when the persistent queue fails to restore items (e.g., due to partition ownership changes across restarts). Called by CrawlerEntryLedgerBootstrapper during RUN LEVEL initialization.
      Returns:
      the number of entries re-queued

    • updateEntry

      public Optional<CrawlerEntry> updateEntry(CrawlerEntry entry)
      Updates an entry in the ledger, maintaining the status counters.
      Parameters:
      entry - the entry to update
      Returns:
      the previous entry if it existed

    • removeEntry

      public Optional<CrawlerEntry> removeEntry(String reference)
      Removes an entry from the ledger, updating the status counters.
      Parameters:
      reference - the reference to remove
      Returns:
      the removed entry if it existed

    • exists

      public boolean exists(String ref)
      Whether a reference exists in the current ledger.
      Parameters:
      ref - document reference
      Returns:
      true if existing

    • getProcessingStatus

      public ProcessingStatus getProcessingStatus(String ref)
      Gets a reference processing status. If a document does not exist, the processing status will be ProcessingStatus.UNTRACKED.
      Parameters:
      ref - document reference
      Returns:
      the processing status

    • getEntry

      public Optional<CrawlerEntry> getEntry(String ref)
      Gets the current ledger entry for the given reference, if it exists.
      Parameters:
      ref - document reference
      Returns:
      the current crawl entry, or empty if not found

    • isQueueEmpty

      public boolean isQueueEmpty()

    • getQueueCount

      public long getQueueCount()

    • getQueuedEntryCount

      public long getQueuedEntryCount()
      Returns the number of entries tracked as ProcessingStatus.QUEUED in the current ledger.

      Unlike the physical queue size, this value comes from the authoritative ledger state and avoids expensive queue-store size operations.

      Returns:
      queued entry count from the ledger

    • isQueuedEntryEmpty

      public boolean isQueuedEntryEmpty()
      Returns whether the current ledger has any entries still tracked in ProcessingStatus.QUEUED state.
      Returns:
      true if no queued entries remain in the ledger

    • clearQueue

      public void clearQueue()

    • clearQueuedEntriesInLedger

      public void clearQueuedEntriesInLedger()
      Clears QUEUED entries from the active ledger without issuing a physical distributed queue clear.

    • forEachQueued

      public void forEachQueued(Consumer<CrawlerEntry> c)

    • queue

      public void queue(@NonNull @NonNull CrawlerEntry crawlEntry)

    • requeueEntry

      public boolean requeueEntry(@NonNull @NonNull String reference)

    • requeueEntry

      public boolean requeueEntry(@NonNull @NonNull CrawlerEntry crawlEntry)

    • nextQueuedBatch

      public List<CrawlerEntry> nextQueuedBatch(int batchSize)

    • requeueProcessingEntries

      public int requeueProcessingEntries()
      Re-queues entries that were in PROCESSING state from a previous run. stopped. These entries were pulled from the queue but not completed, so they need to be queued again on resume to avoid losing them. Updates their status back to QUEUED.
      Returns:
      the number of entries re-queued

    • getProcessingCount

      public long getProcessingCount()

    • isProcessingEmpty

      public boolean isProcessingEmpty()

    • forEachProcessing

      public void forEachProcessing(Consumer<CrawlerEntry> c)

    • getProcessedCount

      public long getProcessedCount()

    • isProcessedEmpty

      public boolean isProcessedEmpty()

    • forEachProcessed

      public void forEachProcessed(Consumer<CrawlerEntry> c)

    • getBaselineCount

      public long getBaselineCount()

    • forEachBaseline

      public void forEachBaseline(Consumer<CrawlerEntry> c)

    • getBaselineEntry

      public Optional<CrawlerEntry> getBaselineEntry(String id)

    • getEntriesByStatus

      public Iterator<CrawlerEntry> getEntriesByStatus(ProcessingStatus status)
      Gets all entries matching the given processing status.
      Parameters:
      status - the processing status to match
      Returns:
      matching entries

    • countByStatus

      public long countByStatus(ProcessingStatus status)
      Counts entries with the given processing status. Uses an efficient O(1) counter instead of executing a query.
      Parameters:
      status - the processing status to count
      Returns:
      count of entries with the given status

    • deleteByStatus

      public void deleteByStatus(ProcessingStatus status)
      Deletes all entries with the given processing status. Also updates the status counter accordingly.
      Parameters:
      status - the processing status of entries to delete

    • archiveCurrentLedger

      public void archiveCurrentLedger()
      Archives the current ledger by making it the baseline and resetting the current ledger to empty. Must only be called by the coordinator node. In a multi-node cluster, calling this on a non-coordinator node is a no-op (logged as a warning).

    • isMaxDocsProcessedReached

      public boolean isMaxDocsProcessedReached()