{"record":{"id":"8d661527f554e95f","repo":"apache/beam","slug":"e-getmessage-mqttio","errorCode":null,"errorMessage":"\" + e.getMessage() + \"","messagePattern":"\" \\+ e\\.getMessage\\(\\) \\+ \"","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/mqtt/src/main/java/org/apache/beam/sdk/io/mqtt/MqttIO.java","lineNumber":406,"sourceCode":"     * max read time is not null, the {@link Read} will provide a bounded {@link PCollection}.\n     */\n    public Read<T> withMaxReadTime(Duration maxReadTime) {\n      return builder().setMaxReadTime(maxReadTime).build();\n    }\n\n    @Override\n    @SuppressWarnings(\"unchecked\")\n    public PCollection<T> expand(PBegin input) {\n      checkArgument(connectionConfiguration() != null, \"connectionConfiguration can not be null\");\n      checkArgument(connectionConfiguration().getTopic() != null, \"topic can not be null\");\n\n      Coder<T> coder;\n      if (withMetadata()) {\n        try {\n          coder =\n              (Coder<T>) input.getPipeline().getSchemaRegistry().getSchemaCoder(MqttRecord.class);\n        } catch (NoSuchSchemaException e) {\n          throw new RuntimeException(e.getMessage());\n        }\n      } else {\n        coder = (Coder<T>) ByteArrayCoder.of();\n      }\n\n      org.apache.beam.sdk.io.Read.Unbounded<T> unbounded =\n          org.apache.beam.sdk.io.Read.from(\n              new UnboundedMqttSource<>(this.builder().setCoder(coder).build()));\n\n      PTransform<PBegin, PCollection<T>> transform = unbounded;\n\n      if (maxNumRecords() < Long.MAX_VALUE || maxReadTime() != null) {\n        transform = unbounded.withMaxReadTime(maxReadTime()).withMaxNumRecords(maxNumRecords());\n      }\n\n      return input.getPipeline().apply(transform);\n    }\n","sourceCodeStart":388,"sourceCodeEnd":424,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/mqtt/src/main/java/org/apache/beam/sdk/io/mqtt/MqttIO.java#L388-L424","documentation":"In MqttIO.Read's expand, when withMetadata() is enabled the connector fetches a schema coder for MqttRecord from the pipeline's SchemaRegistry. If the MqttRecord class is not registered (NoSuchSchemaException), it rethrows as a plain RuntimeException carrying only e.getMessage(), losing the stack trace. It indicates Beam could not derive a schema for the metadata record type.","triggerScenarios":"Calling MqttIO.read() with .withMetadata() on a Beam version/Java class where MqttRecord has no registered schema — e.g. missing @DefaultSchema annotation resolution failure, or pipeline setup where SchemaRegistry cannot infer a schema for MqttRecord.","commonSituations":"Using withMetadata() with a Beam version where schema inference for MqttRecord fails; classpath shading/proguard stripping annotations; mixing Beam versions across dependencies.","solutions":["Ensure the record class used for metadata (MqttRecord) is annotated/registerable: check @DefaultSchema / SchemaCoder registration on the classpath.","Register a schema coder manually in pipeline options setup: pipeline.getSchemaRegistry().registerSchemaProvider(MqttRecord.class, ...).","If you don't need metadata, drop .withMetadata() so the plain ByteArrayCoder path is used.","Unify the Beam version across all sdks/java dependencies to avoid registry incompatibilities."],"exampleFix":"// before\npipeline.apply(MqttIO.<byte[]>read().withConnectorConfiguration(conf).withMetadata());\n// after (no schema needed)\npipeline.apply(MqttIO.<byte[]>read().withConnectorConfiguration(conf));","handlingStrategy":"try-catch","validationCode":"// in pipeline setup, before running\ntry {\n  pipeline.getSchemaRegistry().getSchemaCoder(MqttRecord.class);\n} catch (NoSuchSchemaException e) {\n  throw new IllegalStateException(\"MqttRecord not schema-registered; avoid withMetadata() or register a coder\", e);\n}","typeGuard":null,"tryCatchPattern":"try {\n  coder = (Coder<T>) input.getPipeline().getSchemaRegistry().getSchemaCoder(MqttRecord.class);\n} catch (NoSuchSchemaException e) {\n  throw new RuntimeException(\"No schema coder for MqttRecord; check Beam version/annotations\", e);\n}","preventionTips":["Only call withMetadata() on Beam versions where MqttRecord schema inference is supported.","Keep all org.apache.beam artifacts at the same version.","Disable shading/proguard on schema-annotated classes."],"tags":["mqtt","apache-beam","schema-registry","coder"],"backgroundTag":"schema-validation-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"}