{"record":{"id":"721cb4d6d2787f21","repo":"apache/beam","slug":"pipeline-failed-for-unknown-reason","errorCode":null,"errorMessage":"Pipeline failed for unknown reason","messagePattern":"Pipeline failed for unknown reason","errorType":"exception","errorClass":"java.lang.RuntimeException","httpStatus":null,"severity":"error","filePath":"sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamEnumerableConverter.java","lineNumber":195,"sourceCode":"      PipelineOptions options,\n      BeamRelNode node,\n      DoFn<Row, Void> doFn,\n      Queue<Row> values,\n      int limitCount) {\n    options.as(DirectOptions.class).setBlockOnRun(false);\n    Pipeline pipeline = Pipeline.create(options);\n    PCollection<Row> resultCollection = BeamSqlRelUtils.toPCollection(pipeline, node);\n    resultCollection.apply(ParDo.of(doFn));\n\n    PipelineResult result = pipeline.run();\n\n    State state;\n    while (true) {\n      // Check pipeline state in every second\n      state = result.waitUntilFinish(Duration.standardSeconds(1));\n      if (state != null && state.isTerminal()) {\n        if (PipelineResult.State.FAILED.equals(state)) {\n          throw new RuntimeException(\"Pipeline failed for unknown reason\");\n        }\n        break;\n      }\n\n      try {\n        if (values.size() >= limitCount) {\n          result.cancel();\n          break;\n        }\n      } catch (IOException e) {\n        LOG.warn(\"{}\", e.toString());\n        break;\n      }\n    }\n\n    return result;\n  }\n","sourceCodeStart":177,"sourceCodeEnd":213,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamEnumerableConverter.java#L177-L213","documentation":"BeamEnumerableConverter.limitRun executes a Beam SQL pipeline (built from the BeamRelNode) and polls the pipeline result state every second until it finishes, while collecting rows up to a LIMIT. If the pipeline reaches a terminal FAILED state, the original underlying exception is discarded and this generic RuntimeException(\"Pipeline failed for unknown reason\") is thrown instead, hiding the real cause from the caller.","triggerScenarios":"Calling LIMIT-style SQL execution (BeamSqlStringUtils/BeamSqlEnv query paths that route through limitCollect -> limitRun) on a pipeline whose job fails during execution — e.g. a bad source/sink configuration, an exception inside a user DoFn, quota/resource errors on the runner, or invalid input data causing a worker crash.","commonSituations":"Developers running Beam SQL interactively via calcite CLI or sqlline, or unit-testing BeamSql queries with a DirectRunner/SparkRunner/FlinkRunner, where the pipeline aborts (missing module, bad test table config, serialization errors of the BeamRelNode or UDF classes) and the real stack trace is swallowed.","solutions":["Run the pipeline with DEBUG logging for org.apache.beam to find the root-cause exception printed before this error","Inspect the runner's job logs (Flink/Spark/Dataflow UI or DirectRunner console) for the original failure","Call the query without LIMIT (collectRows path) or with a plain PipelineResult to surface the underlying exception","Verify input table configs (beamSql table types, connection params) and that all UDFs/POJOs are Serializable","Pin/upgrade Beam and runner versions to compatible releases to rule out known converter bugs"],"exampleFix":"// before (library code loses the cause)\nif (PipelineResult.State.FAILED.equals(state)) {\n  throw new RuntimeException(\"Pipeline failed for unknown reason\");\n}\n// after\nif (PipelineResult.State.FAILED.equals(state)) {\n  Throwable cause = result instanceof CompleteFutureResult\n      ? null : null; // or capture error callback\n  throw new RuntimeException(\"Pipeline failed for unknown reason\", cause);\n}","handlingStrategy":"try-catch","validationCode":"// verify pipeline completes and inspect state before relying on the limited rows\nPipelineResult result = pipeline.run();\nPipelineResult.State state = result.waitUntilFinish();\nif (state != PipelineResult.State.DONE) {\n  throw new IllegalStateException(\"Beam SQL pipeline did not finish: \" + state);\n}","typeGuard":"boolean pipelineSucceeded(PipelineResult r) {\n  return r != null && PipelineResult.State.DONE.equals(r.waitUntilFinish());\n}","tryCatchPattern":"try {\n  Enumerable<Object> rows = limitRun(options, node, limit);\n} catch (RuntimeException e) {\n  if (\"Pipeline failed for unknown reason\".equals(e.getMessage())) {\n    // consult runner logs for the suppressed root cause; wrap or retry\n  } else throw e;\n}","preventionTips":["Always check pipeline state / runner logs immediately when this message appears — the real cause is logged separately","Prefer running queries without LIMIT first to validate the pipeline works end-to-end","Test SQL queries with DirectRunner in CI before deploying to a cluster runner","Keep Beam SDK and runner versions aligned","Make all UDFs and referenced POJOs Serializable"],"tags":["java","beam","apache-beam-sql","pipeline-execution","root-cause-suppressed"],"backgroundTag":"pipeline-execution-failed","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}