{"record":{"id":"1bf8cbdcb0a13f8d","repo":"pentaho/pentaho-kettle","slug":"fixedtimestreamwindow-subtransfailed","errorCode":null,"errorMessage":"FixedTimeStreamWindow.SubtransFailed","messagePattern":"FixedTimeStreamWindow\\.SubtransFailed","errorType":"exception","errorClass":"KettleException","httpStatus":null,"severity":"error","filePath":"engine/src/main/java/org/pentaho/di/trans/streaming/common/FixedTimeStreamWindow.java","lineNumber":127,"sourceCode":"      .parallel( parallelism, rxBatchCount )\n      .runOn( sharedStreamingBatchPoolSize > 0 ? Schedulers.from( sharedStreamingBatchPool ) : Schedulers.io(),\n        rxBatchCount )\n      .filter( list -> !list.isEmpty() )\n      .map( this.bufferFilter ) // apply any filtering for data that should no longer be processed\n      .filter( list -> !list.isEmpty() ) // ensure at least one record is left before sending to subtrans\n      .map( this::sendBufferToSubtrans )\n      .filter( Optional::isPresent )\n      .map( Optional::get )\n      .sequential()\n      .doOnNext( this::failOnError )\n      .doOnNext( postProcessor )\n      .map( Map.Entry::getValue )\n      .blockingIterable();\n  }\n\n  private void failOnError( Map.Entry<List<I>, Result> pair ) throws KettleException {\n    if ( pair.getValue().getNrErrors() > 0 ) {\n      throw new KettleException( BaseMessages.getString( PKG, \"FixedTimeStreamWindow.SubtransFailed\"  ) );\n    }\n  }\n\n  private Optional<Map.Entry<List<I>, Result>> sendBufferToSubtrans( List<I> input ) throws KettleException {\n    final List<RowMetaAndData> rows = input.stream()\n      .map( row -> row.toArray( new Object[ 0 ] ) )\n      .map( objects -> new RowMetaAndData( rowMeta, objects ) )\n      .collect( Collectors.toList() );\n    Optional<Result> optionalRes = subtransExecutor.execute( rows );\n    return optionalRes.map( result -> new AbstractMap.SimpleImmutableEntry<>( input, result ) );\n  }\n\n}\n","sourceCodeStart":109,"sourceCodeEnd":141,"githubUrl":"https://github.com/pentaho/pentaho-kettle/blob/f3058517a153da500bf4551f46d79b91bf8ec552/engine/src/main/java/org/pentaho/di/trans/streaming/common/FixedTimeStreamWindow.java#L109-L141","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Inspect the sub-transformation's log for the actual failing step and root error.","Fix the failing logic in the sub-transformation (e.g. add error handling steps / safe type conversions).","Add 'error handling' on the failing sub-step so bad rows are routed instead of counted as errors.","Check external dependencies used by the sub-transformation (databases, services) are reachable.","Set the abort/row count settings in the sub-transformation so a single bad row doesn't fail the whole window."],"exampleFix":"// before: sub-trans aborts on any bad row\n// Text File Input with 'Error handling: Abort' -> NrErrors > 0\n// after: put the failing step's error handling to use\nstep.setLevelOfLogginError... // in the sub-trans, add an error stream\n// and route bad rows to a 'bad rows' file instead of failing","handlingStrategy":"try-catch","validationCode":"// smoke-test the sub-transformation with a sample row before wiring it into the stream\nResult r = subTransMeta.executePreview(sampleRows);\nif (r.getNrErrors() > 0) throw new IllegalStateException(\"Sub-trans fails on sample data\");","typeGuard":null,"tryCatchPattern":"try {\n    runStreamingPipeline(trans);\n} catch (KettleException e) {\n    if (e.getMessage().contains(\"SubtransFailed\")) {\n        log.error(\"Sub-transformation errors; see sub-trans log for the failing step\");\n    }\n    throw e;\n}","preventionTips":["Add error handling streams on fallible steps inside the sub-transformation.","Preview/execute the sub-transformation standalone on sample data before production use.","Monitor sub-transformation external dependencies (DBs, services) health."],"tags":["streaming","subtransformation","error-handling","kettle"],"backgroundTag":"subtransformation-failed","analyzedSha":"f3058517a153da500bf4551f46d79b91bf8ec552","analyzedAt":"2026-09-13T14:04:16.340Z","contentChangedAt":"2026-09-13T14:04:16.340Z","schemaVersion":2},"datasetVersion":"2026-09-20T23:17:15.980Z"}