This is the multi-page printable view of this section. Click here to print.

Return to the regular view of this page.

Runner

Component that manages a Lucille Run end-to-end.

1: Runner Orchestration

The Runner is the command-line entry point for launching a Lucille run. When invoked, it reads the configuration file, validates all component configurations, generates a unique runId, launches the configured components, waits for all work to complete, and prints a run summary.

What a Run Is

A Lucille Run is a sequence of Connectors executed one after the other. Each Connector feeds a specific Pipeline. A run can include multiple Connectors feeding multiple Pipelines, all sharing the same Indexer.

Connectors run strictly in sequence: the next Connector does not start until all documents from the previous Connector have been fully processed and indexed. This ordering guarantee is enforced automatically by the Publisher’s accounting system.

Run Lifecycle

For each Connector in the configured sequence, the Runner:

Validates the full configuration (fails fast on any misconfiguration).
Starts a WorkerPool (N Worker threads based on worker.threads).
Starts an Indexer thread.
Creates a PublisherImpl and launches the Connector in a ConnectorThread.
Blocks on publisher.waitForCompletion() until all work is done.
Logs the run summary and moves to the next Connector (or exits).

Starting a Run

Local mode (default):

java \
  -Dconfig.file=/path/to/config.conf \
  -cp 'lucille-core/target/lucille.jar:lucille-core/target/lib/*' \
  com.kmwllc.lucille.core.Runner

Distributed mode:

java \
  -Dconfig.file=/path/to/config.conf \
  -cp 'lucille-core/target/lucille.jar:lucille-core/target/lib/*' \
  com.kmwllc.lucille.core.Runner \
  -distributed

External mode (single JVM, Kafka messaging):

java \
  -Dconfig.file=/path/to/config.conf \
  -cp 'lucille-core/target/lucille.jar:lucille-core/target/lib/*' \
  com.kmwllc.lucille.core.Runner \
  -external

Config validation only (no run):

Validates every Connector, Stage, and Indexer spec and prints all errors. Exits without executing anything.

java \
  -Dconfig.file=/path/to/config.conf \
  -cp 'lucille-core/target/lucille.jar:lucille-core/target/lib/*' \
  com.kmwllc.lucille.core.Runner \
  -validate

Render effective config (no run):

Prints the fully resolved configuration after HOCON substitutions (environment variables, include directives, etc.). Useful for debugging config variable expansion.

java \
  -Dconfig.file=/path/to/config.conf \
  -cp 'lucille-core/target/lucille.jar:lucille-core/target/lib/*' \
  com.kmwllc.lucille.core.Runner \
  -render

Run Configuration

runner {
  # Log detailed stage-by-stage metrics at end of run (default: INFO)
  metricsLoggingLevel: "INFO"

  # Connector timeout in milliseconds (default: 86400000 = 24 hours; set <= 0 to disable)
  connectorTimeout: 86400000
}

Run ID

The Runner generates a UUID runId for each run. The run ID is:

Stamped on every Document by the Publisher (run_id field).
Used as part of the Kafka event topic name in distributed mode.
Included in log MDC so all log lines during a run include the run ID for filtering.

Run Summary

At the end of every run, the Runner logs a structured summary:

RUN SUMMARY: Success. 1/1 connectors complete. All published docs succeeded.
connector1: complete. 200000 docs succeeded. 0 docs failed. 0 docs dropped. Time: 416.47 secs.
Run took 417.46 secs.

A connector that failed entirely is distinguished from one that completed with individual document failures. Connectors after a failed one are listed as skipped.

Graceful Shutdown

The Runner handles SIGINT (Ctrl+C) and SIGTERM. On signal receipt:

The Connector stops publishing.
Workers drain remaining documents.
The Indexer flushes its current batch.
A partial run summary is logged.

RunType

Lucille supports four run types, selected via command-line flags:

RunType	Flag(s)	Description
`LOCAL`	(none)	Single JVM, in-memory queues. Default.
`EXTERNAL`	`-external`	Single JVM, Kafka messaging.
`DISTRIBUTED`	`-distributed`	Separate JVMs per component, Kafka messaging.
`TEST`	(API only)	Single JVM, in-memory, search backend bypassed, messages captured.

Practical Guide

For deployment instructions — starting runs in each mode, command-line flags, and operational considerations — see Deployment.

For runner and other top-level configuration parameters, see Writing a Config.

1 - Runner Orchestration

How Runner.run() coordinates the full lifecycle — validation, connector loop, signal handling, and reporting.

Overview

The Runner is Lucille’s top-level orchestrator. It coordinates the full lifecycle of a run: validating configuration, instantiating components, executing connectors sequentially, and reporting results. All methods are static — the Runner is never instantiated.

A “run” is a sequential execution of one or more Connectors. Each Connector’s work must complete before the next begins. If any Connector fails, the run aborts.

How `Runner.run()` Coordinates the Full Lifecycle

The main execution flow:

public static RunResult run(Config config, RunType type, String runId) throws Exception {
    if (runId == null) {
        runId = Runner.generateRunId();  // UUID
    }
    MDC.put(RUNID_FIELD, runId);

    // 1. Validate FIRST
    Map<String, List<Exception>> validationErrors = runInValidationMode(config);
    if (!validationErrors.isEmpty()) {
        return new RunResult(false, ...);
    }

    // 2. Create connectors
    List<Connector> connectors = Connector.fromConfig(config);

    // 3. Execute each connector sequentially
    for (Connector connector : connectors) {
        // Create messenger factories based on RunType
        // Run connector with components
        ConnectorResult result = runConnectorWithComponents(...);
        if (!result.getStatus()) {
            return new RunResult(false, ...);  // Abort on failure
        }
    }

    return new RunResult(true, ...);
}

The Validation Step

Before any work begins, the Runner validates the entire configuration:

Map<String, List<Exception>> validationErrors = runInValidationMode(config);

This validates:

Pipelines — each pipeline’s stages are instantiated to check for config errors
Connectors — each connector’s config is checked for required/optional properties
Indexer — the indexer config block is validated
Other parents — publisher, runner, kafka, and other top-level config blocks

Validation is fail-all, not fail-fast. All errors are collected and reported together. If any validation errors exist, the run returns immediately with a failure result.

The Connector Loop

For each connector, the Runner:

Creates messenger factories appropriate for the RunType
Calls runConnectorWithComponents() which:
- Starts a WorkerPool (if local mode)
- Creates and starts an Indexer thread (if local mode)
- Creates a Publisher
- Calls runConnector() which:
  - Calls connector.preExecute(runId)
  - Launches a ConnectorThread that calls connector.execute(publisher) then publisher.flush()
  - Calls publisher.waitForCompletion(connectorThread, timeout)
  - Calls connector.postExecute(runId) (only if publishing succeeded)
Stops WorkerPool and Indexer in the finally block

private static ConnectorResult runConnectorWithComponents(...) {
    try {
        if (startWorkerAndIndexer && connector.getPipelineName() != null) {
            workerPool = new WorkerPool(config, pipelineName, localRunId, workerMessengerFactory, metricsPrefix);
            workerPool.start();

            IndexerMessenger indexerMessenger = indexerMessengerFactory.create();
            indexer = IndexerFactory.fromConfig(config, indexerMessenger, bypassIndexer, metricsPrefix, localRunId);
            indexerThread = new Thread(indexer);
            indexerThread.start();
        }

        publisher = new PublisherImpl(config, publisherMessenger, runId, ...);
        return runConnector(config, runId, connector, publisher);
    } finally {
        if (workerPool != null) { workerPool.stop(); workerPool.join(3000); }
        if (indexerThread != null) { indexer.terminate(); indexerThread.join(3000); }
    }
}

How RunType Affects Component Startup

public enum RunType {
    LOCAL,             // Workers + Indexer as threads; in-memory queues
    TEST,              // Same as LOCAL but bypass indexer backend; record message history
    EXTERNAL,          // Workers + Indexer as threads; Kafka for messaging
    DISTRIBUTED        // No local Workers/Indexer; Kafka messaging; assume external processes
}

RunType	Start Worker/Indexer?	Bypass Indexer Backend?	Messenger Type
`LOCAL`	Yes	No	`LocalMessenger`
`TEST`	Yes	Yes	`TestMessenger`
`EXTERNAL`	Yes	No	`KafkaWorkerMessenger` / `KafkaIndexerMessenger`
`DISTRIBUTED`	No	No	`KafkaWorkerMessenger` / `KafkaIndexerMessenger`

The key decision:

boolean startWorkerAndIndexer = !type.equals(RunType.DISTRIBUTED);
boolean bypassSolr = type.equals(RunType.TEST);

The MessengerFactory Pattern

The Runner uses factory interfaces to decouple component creation from the RunType decision:

if (RunType.TEST.equals(type)) {
    TestMessenger messenger = new TestMessenger();
    history.put(connector.getName(), messenger);
    workerMessengerFactory = WorkerMessengerFactory.getConstantFactory(messenger);
    indexerMessengerFactory = IndexerMessengerFactory.getConstantFactory(messenger);
    publisherMessengerFactory = PublisherMessengerFactory.getConstantFactory(messenger);
} else if (RunType.LOCAL.equals(type)) {
    LocalMessenger messenger = new LocalMessenger(config);
    workerMessengerFactory = WorkerMessengerFactory.getConstantFactory(messenger);
    // ...
} else {
    workerMessengerFactory = WorkerMessengerFactory.getKafkaFactory(config, connector.getPipelineName());
    // ...
}

For LOCAL/TEST modes, a single messenger instance is shared (via getConstantFactory). For Kafka modes, each factory call creates a new messenger with its own Kafka consumer/producer.

Signal Handling for Clean Shutdown

When running via main(), the Runner registers an INT signal handler:

state = new RunnerState();

Signal.handle(new Signal("INT"), signal -> {
    if (state != null) {
        state.close();  // Close connector, publisher, workerPool, indexer
    }
    SystemHelper.exit(0);
});

RunnerState holds references to the currently-active components. When a connector starts, the state is populated:

if (state != null) {
    state.set(publisher, connector, workerPool, indexer, indexerThread);
}

When the connector finishes, the state is cleared. The RunnerState.close() method attempts orderly shutdown of each component, logging errors but not throwing.

ConnectorResult and RunResult Reporting

ConnectorResult captures per-connector outcomes:

Status (success/failure)
Error message (if failed)
Duration in seconds
Document counts from the Publisher (succeeded, failed)

RunResult aggregates across all connectors:

Overall status
List of all ConnectorResults
Summary message (e.g., “2/3 connectors complete. Some docs failed.”)
For TEST mode: a Map<String, TestMessenger> containing message history per connector

The Connector Timeout Mechanism

Each connector has a configurable timeout (default: 24 hours):

final int connectorTimeout = config.hasPath("runner.connectorTimeout") ?
    config.getInt("runner.connectorTimeout") : DEFAULT_CONNECTOR_TIMEOUT;
pubResult = publisher.waitForCompletion(connectorThread, connectorTimeout);

Inside waitForCompletion, the timeout is checked on each poll iteration:

if (timeout > 0 && ChronoUnit.MILLIS.between(start, Instant.now()) > timeout) {
    return new PublisherResult(false, "Connector timeout.");
}

This prevents a stuck connector from blocking the entire run indefinitely.

Sequential Connector Composition

Connectors execute strictly sequentially. The next connector only starts after the previous one fully completes (all documents indexed or failed):

for (Connector connector : connectors) {
    ConnectorResult result = runConnectorWithComponents(...);
    if (!result.getStatus()) {
        log.error("Aborting run because " + connector.getName() + " failed.");
        return new RunResult(false, ...);
    }
}

Each connector gets its own WorkerPool, Indexer, and Publisher. Components from one connector are fully stopped before the next connector’s components are created.

The `main()` Method and CLI Options

Option distributedOpt = Option.builder("distributed").hasArg(false)
    .desc("Uses Kafka for inter-component communication and doesn't execute pipelines locally.")
    .build();

Option external = Option.builder("external").hasArg(false)
    .desc("Modified local mode where workers and indexers are separate threads within the JVM communicating "
        + "through kafka")
    .build();

OptionGroup distributedType = new OptionGroup().addOption(distributedOpt).addOption(external);

Options cliOptions = new Options()
    .addOptionGroup(distributedType)
    .addOption(Option.builder("validate").hasArg(false)
        .desc("Validate the configuration and exit").build())
    .addOption(Option.builder("render").hasArg(false)
        .desc("Print out the configuration file with substitutions applied and exit").build());

Flag	Effect
(none)	`RunType.LOCAL` — full local execution with in-memory queues
`-distributed`	`RunType.DISTRIBUTED` — only run connectors, assume external workers/indexers
`-external`	`RunType.EXTERNAL` — local workers/indexers but communicate via Kafka
`-validate`	Validate config and exit (no execution)
`-render`	Print resolved config as JSON and exit

The -validate and -render flags can be combined. -distributed and -external cannot as they are mutually exclusive, supplying both is rejected.

Anything the parser doesn’t recognize (unknown flags or leftover positional arguments) cause the Runner to log the usage text and exit 1 rather than starting a run.

Thread Model (Local Mode)

For each connector in a local run, there are 4+ threads:

Main thread — runs waitForCompletion(), polling for events
ConnectorThread — calls connector.execute(publisher), publishes documents
Worker thread(s) — poll documents, run pipeline, emit results (configurable count)
Indexer thread — polls processed documents, sends to search engine

Plus a WorkerPool watcher thread that monitors worker health and logs statistics.