apache/beam · error · IllegalArgumentException

Expected the engine to not be null

Error message

Expected the engine to not be null

What it means

processElement() requires a Delta Kernel Engine instance to construct the scan state, parquet handler, and read data. The engine field should have been initialized in setup() from the serialized Hadoop configuration; if it is still null when elements arrive, the DoFn is in an invalid lifecycle state and throws IllegalArgumentException.

Solutions

1. Ensure delta-kernel-defaults (and its Engine implementation) plus Hadoop dependencies are on the worker classpath
2. Check worker logs for an earlier failure/exception in setup() that left engine uninitialized
3. Verify the DoFn is only used through DeltaIO.ReadRows so setup()/startBundle lifecycle hooks run correctly
4. Upgrade/align Beam Delta IO and io.delta kernel versions to a compatible set

Example fix

// before
// engine null at process time because setup never ran
DoFnProcessContext ctx = ...; doFn.processElement(ctx);
// after
doFn.setup(); // initializes engine before processing
DoFnProcessContext ctx = ...; doFn.processElement(ctx);

// Ensure dependencies ship the kernel Engine impl on the worker classpath
// gradle: implementation 'io.delta:delta-kernel-defaults:<version>'
// and let DeltaIO construct DeltaSourceDoFn itself rather than reusing it manually

try {
  rows = input.apply(deltaIO);
} catch (IllegalArgumentException e) {
  if (e.getMessage().contains("Expected the engine to not be null")) {
    // check worker logs for setup() failure; fix classpath/dependencies and rerun
  } else { throw e; }
}

When it happens

Trigger: The DoFn's setup() failed to initialize the engine (e.g. missing Hadoop configuration, classpath issue loading engine classes), or the DoFn is invoked in a context where setup was skipped.

Common situations: Missing/incorrect delta-kernel or Hadoop dependencies on the worker classpath so engine construction silently fails or is skipped; custom pipelines reusing DeltaSourceDoFn without running setup; serialization/deserialization problems losing the engine field.

Related errors

• Buffer counter not initialized for UUID:
• Cannot access FinishBundleContext outside of @FinishBundle…
• Cannot access StartBundleContext outside of @StartBundle…
• Cannot set both endVersion and endTimestamp.
• Cannot set both startVersion and startTimestamp.

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/d2346582f2833ab7. Report an issue: GitHub.

Appendix: source

Thrown at sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaSourceDoFn.java:139

  public void setUp() {
    engine = DefaultEngine.create(getConfiguration());
  }

  @ProcessElement
  public ProcessContinuation processElement(
      @Element DeltaReadTask task,
      RestrictionTracker<OffsetRange, Long> tracker,
      OutputReceiver<Row> out)
      throws Exception {

    SerializableRow scanStateRow = task.getScanStateRow();
    StructType physicalSchema = ScanStateRow.getPhysicalDataReadSchema(scanStateRow);
    StructType logicalSchema = ScanStateRow.getLogicalSchema(scanStateRow);
    Schema beamSchema = DeltaIO.ReadRows.convertToBeamSchema(logicalSchema);

    Engine currentEngine = engine;
    if (currentEngine == null) {
      throw new IllegalArgumentException("Expected the engine to not be null");
    }

    // `BeamParquetHandler` takes a reference to the `RestrictionTracker` so that it
    // can perform `getFrom`, `getTo`, `tryClaim` requests to return the correct set
    // of row groups that map to the current restriction.
    BeamParquetHandler parquetHandler =
        new BeamParquetHandler(getConfiguration(), currentEngine.getParquetHandler(), tracker);
    BeamEngine beamEngine = new BeamEngine(currentEngine, parquetHandler);

    long currentStartRgIndex = 0L;

    // We have to go through files in the `DeltaReadTask` in order so that the
    // `RestrictionTracker`
    // can correctly handle the range of the current split.
    List<SerializableRow> scanFileRows = task.getScanFileRows();
    List<List<Long>> rowGroupSizesPerFile = task.getRowGroupSizesPerFile();
    for (int i = 0; i < scanFileRows.size(); i++) {
      if (currentStartRgIndex >= tracker.currentRestriction().getTo()) {

View on GitHub (pinned to 12126d8942)