pentaho/pentaho-kettle · error · KettleException

FixedTimeStreamWindow.SubtransFailed

Error message

FixedTimeStreamWindow.SubtransFailed

What it means

FixedTimeStreamWindow's failOnError throws this KettleException when a batch of rows sent to the sub-transformation comes back with a non-zero error count. It signals that the fixed-time window sub-transformation failed for the buffered input, aborting the streaming pipeline.

Solutions

1. Inspect the sub-transformation's log for the actual failing step and root error.
2. Fix the failing logic in the sub-transformation (e.g. add error handling steps / safe type conversions).
3. Add 'error handling' on the failing sub-step so bad rows are routed instead of counted as errors.
4. Check external dependencies used by the sub-transformation (databases, services) are reachable.
5. Set the abort/row count settings in the sub-transformation so a single bad row doesn't fail the whole window.

Example fix

// before: sub-trans aborts on any bad row
// Text File Input with 'Error handling: Abort' -> NrErrors > 0
// after: put the failing step's error handling to use
step.setLevelOfLogginError... // in the sub-trans, add an error stream
// and route bad rows to a 'bad rows' file instead of failing

// smoke-test the sub-transformation with a sample row before wiring it into the stream
Result r = subTransMeta.executePreview(sampleRows);
if (r.getNrErrors() > 0) throw new IllegalStateException("Sub-trans fails on sample data");

try {
    runStreamingPipeline(trans);
} catch (KettleException e) {
    if (e.getMessage().contains("SubtransFailed")) {
        log.error("Sub-transformation errors; see sub-trans log for the failing step");
    }
    throw e;
}

When it happens

Trigger: sendBufferToSubtrans executes the sub-transformation for a buffer; the returned Result has getNrErrors() > 0 (any step in the sub-transformation logged an error), so failOnError throws SubtransFailed.

Common situations: Sub-transformation step fails on specific data (type conversion, lookup miss, target DB down); the sub-transformation aborts due to a bad row-limit setting; transformation log level hides which inner step actually failed.

Related errors

• Append.Exception.InvalidLayoutDetected
• BaseStreamStepMeta.CheckResult.ResultStepMissing
• ColumnExists.Log.ErrorInStep
• Errors encountered (first 10): ...
• GetXMLData.Log.UnableApplyXPath (localized message)

AI-assisted analysis of pentaho/pentaho-kettle@f3058517a1 (2026-09-13). Data as JSON: /api/errors/1bf8cbdcb0a13f8d. Report an issue: GitHub.

Appendix: source

Thrown at engine/src/main/java/org/pentaho/di/trans/streaming/common/FixedTimeStreamWindow.java:127

      .parallel( parallelism, rxBatchCount )
      .runOn( sharedStreamingBatchPoolSize > 0 ? Schedulers.from( sharedStreamingBatchPool ) : Schedulers.io(),
        rxBatchCount )
      .filter( list -> !list.isEmpty() )
      .map( this.bufferFilter ) // apply any filtering for data that should no longer be processed
      .filter( list -> !list.isEmpty() ) // ensure at least one record is left before sending to subtrans
      .map( this::sendBufferToSubtrans )
      .filter( Optional::isPresent )
      .map( Optional::get )
      .sequential()
      .doOnNext( this::failOnError )
      .doOnNext( postProcessor )
      .map( Map.Entry::getValue )
      .blockingIterable();
  }

  private void failOnError( Map.Entry<List<I>, Result> pair ) throws KettleException {
    if ( pair.getValue().getNrErrors() > 0 ) {
      throw new KettleException( BaseMessages.getString( PKG, "FixedTimeStreamWindow.SubtransFailed"  ) );
    }
  }

  private Optional<Map.Entry<List<I>, Result>> sendBufferToSubtrans( List<I> input ) throws KettleException {
    final List<RowMetaAndData> rows = input.stream()
      .map( row -> row.toArray( new Object[ 0 ] ) )
      .map( objects -> new RowMetaAndData( rowMeta, objects ) )
      .collect( Collectors.toList() );
    Optional<Result> optionalRes = subtransExecutor.execute( rows );
    return optionalRes.map( result -> new AbstractMap.SimpleImmutableEntry<>( input, result ) );
  }

}

View on GitHub (pinned to f3058517a1)