Struct IndexerExecutor

pub struct IndexerExecutor<P> { /* private fields */ }

The Executor of the main ingestion pipeline process.

This struct orchestrates the execution of multiple worker pools, handling checkpoint distribution, progress tracking, and shutdown. It utilizes ProgressStore for persisting checkpoint progress and provides metrics for monitoring the indexing process.

Example

use async_trait::async_trait;
use iota_data_ingestion_core::{
    DataIngestionMetrics, FileProgressStore, IndexerExecutor, IngestionError, ReaderOptions,
    Worker, WorkerPool,
};
use iota_types::full_checkpoint_content::CheckpointData;
use prometheus::Registry;
use tokio_util::sync::CancellationToken;
use std::{path::PathBuf, sync::Arc};

struct CustomWorker;

#[async_trait]
impl Worker for CustomWorker {
    type Message = ();
    type Error = IngestionError;

    async fn process_checkpoint(
        &self,
        checkpoint: Arc<CheckpointData>,
    ) -> Result<Self::Message, Self::Error> {
        // custom processing logic.
        println!(
            "Processing Local checkpoint: {}",
            checkpoint.checkpoint_summary.to_string()
        );
        Ok(())
    }
}

#[tokio::main]
async fn main() {
    let concurrency = 5;
    let progress_store = FileProgressStore::new("progress.json").await.unwrap();
    let mut executor = IndexerExecutor::new(
        progress_store,
        1, // number of registered WorkerPools.
        DataIngestionMetrics::new(&Registry::new()),
        CancellationToken::new(),
    );
    // register a worker pool with 5 workers to process checkpoints in parallel
    let worker_pool = WorkerPool::new(CustomWorker, "local_reader".to_string(), concurrency, Default::default());
    // register the worker pool to the executor.
    executor.register(worker_pool).await.unwrap();
    // run the ingestion pipeline.
    executor
        .run(
            PathBuf::from("./chk".to_string()), // path to a local directory where checkpoints are stored.
            None,
            vec![],                   // optional remote store access options.
            ReaderOptions::default(), // remote_read_batch_size.
        )
        .await
        .unwrap();
}

Implementations

impl<P: ProgressStore> IndexerExecutor<P>

pub fn new( progress_store: P, number_of_jobs: usize, metrics: DataIngestionMetrics, token: CancellationToken, ) -> Self

pub async fn register<W: Worker + 'static>( &mut self, pool: WorkerPool<W>, ) -> IngestionResult<()>

Registers new worker pool in executor.

pub async fn update_watermark( &mut self, task_name: String, watermark: CheckpointSequenceNumber, ) -> IngestionResult<()>

pub async fn read_watermark( &mut self, task_name: String, ) -> IngestionResult<CheckpointSequenceNumber>

pub async fn run( self, path: PathBuf, remote_store_url: Option<String>, remote_store_options: Vec<(String, String)>, reader_options: ReaderOptions, ) -> IngestionResult<HashMap<String, CheckpointSequenceNumber>>

Main executor loop.

Error

Returns an IngestionError::EmptyWorkerPool if no worker pool was registered.

pub async fn run_with_config( self, config: CheckpointReaderConfig, ) -> IngestionResult<HashMap<String, CheckpointSequenceNumber>>

Alternative main executor loop. Uses the new iteration of the CheckpointReader supporting syncing checkppints from hybrid historical store.

Error

Returns an IngestionError::EmptyWorkerPool if no worker pool was registered.