{"record":{"id":"d7438d88f3541276","repo":"apache/beam","slug":"spark-receiver-was-not-initialized","errorCode":null,"errorMessage":"Spark Receiver was not initialized","messagePattern":"Spark Receiver was not initialized","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/sparkreceiver/3/src/main/java/org/apache/beam/sdk/io/sparkreceiver/ReadFromSparkReceiverWithOffsetDoFn.java","lineNumber":248,"sourceCode":"            } else if (data instanceof ArrayBuffer) {\n              final ArrayBuffer<V> arrayBuffer = (ArrayBuffer<V>) data;\n              final Iterator<V> iterator = arrayBuffer.iterator();\n              while (iterator.hasNext()) {\n                V record = iterator.next();\n                recordsQueue.offer(record);\n              }\n            } else {\n              V record = (V) data;\n              recordsQueue.offer(record);\n            }\n            return null;\n          };\n\n      try {\n        new WrappedSupervisor(sparkReceiver, new SparkConf(), storeFn);\n      } catch (Exception e) {\n        LOG.error(\"Can not init Spark Receiver!\", e);\n        throw new IllegalStateException(\"Spark Receiver was not initialized\");\n      }\n      LOG.debug(\"Starting receiver\");\n      ((HasOffset) sparkReceiver).setStartOffset(startOffset);\n      sparkReceiver.supervisor().startReceiver();\n      LOG.debug(\"Receiver started\");\n    }\n\n    @Override\n    public void stop() {\n      if (sparkReceiver != null) {\n        sparkReceiver.stop(\"SparkReceiver is stopped.\");\n      }\n      LOG.info(\"Clear records queue: {} records\", recordsQueue.size());\n      recordsQueue.clear();\n    }\n  }\n\n  @ProcessElement","sourceCodeStart":230,"sourceCodeEnd":266,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/sparkreceiver/3/src/main/java/org/apache/beam/sdk/io/sparkreceiver/ReadFromSparkReceiverWithOffsetDoFn.java#L230-L266","documentation":"ReadFromSparkReceiverWithOffsetDoFn.start() instantiates a WrappedSupervisor to drive a Spark Receiver within Beam. If supervisor construction throws, the DoFn logs the cause and rethrows this IllegalStateException, because the receiver cannot stream records without a running supervisor.","triggerScenarios":"new WrappedSupervisor(sparkReceiver, new SparkConf(), storeFn) throws — e.g. invalid SparkConf, missing Spark dependencies at runtime, receiver constructor failures, or storeFn wiring errors.","commonSituations":"Missing spark-core dependency in the worker classpath; incompatible Spark Receiver implementation requiring SparkConf settings not provided; receiver builder returning an instance whose constructor fails on the worker environment.","solutions":["Check the logged \"Can not init Spark Receiver!\" stack trace for the root cause","Ensure spark-core and the receiver's dependencies are bundled in the pipeline jar / available on workers","Provide required SparkConf settings via the receiver's builder options","Verify the SparkReceiverBuilder returns a valid Receiver compatible with sparkreceiver/3 connector"],"exampleFix":"// before\nSparkReceiverIO.readFromSparkReceiver().withSparkReceiverBuilder(() -> new MyReceiver()) // receiver needs config\n// after\nSparkReceiverIO.readFromSparkReceiver().withSparkReceiverBuilder(() -> new MyReceiver(sparkConfParams));","handlingStrategy":"try-catch","validationCode":"// preflight: receiver must build and Spark deps present\nReceiver<?> r = sparkReceiverBuilder.build(); // throws early if broken","typeGuard":null,"tryCatchPattern":"try {\n  readFromSparkReceiver();\n} catch (IllegalStateException e) {\n  if (e.getMessage().contains(\"Spark Receiver was not initialized\")) {\n    LOG.error(\"Check SparkConf and spark-core on worker classpath\", e);\n  }\n  throw e;\n}","preventionTips":["Bundle spark-core and receiver dependencies with the pipeline","Supply required SparkConf settings via builder options","Test receiver construction on a worker-like environment","Keep sparkreceiver/3 connector and Spark versions compatible"],"tags":["java","beam","spark","receiver","initialization"],"backgroundTag":"module-init-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"}