apache/beam · error · RuntimeException

" + e.getMessage() + "

Error message

" + e.getMessage() + "

What it means

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.

Solutions

1. Ensure the record class used for metadata (MqttRecord) is annotated/registerable: check @DefaultSchema / SchemaCoder registration on the classpath.
2. Register a schema coder manually in pipeline options setup: pipeline.getSchemaRegistry().registerSchemaProvider(MqttRecord.class, ...).
3. If you don't need metadata, drop .withMetadata() so the plain ByteArrayCoder path is used.
4. Unify the Beam version across all sdks/java dependencies to avoid registry incompatibilities.

Example fix

// before
pipeline.apply(MqttIO.<byte[]>read().withConnectorConfiguration(conf).withMetadata());
// after (no schema needed)
pipeline.apply(MqttIO.<byte[]>read().withConnectorConfiguration(conf));

// in pipeline setup, before running
try {
  pipeline.getSchemaRegistry().getSchemaCoder(MqttRecord.class);
} catch (NoSuchSchemaException e) {
  throw new IllegalStateException("MqttRecord not schema-registered; avoid withMetadata() or register a coder", e);
}

try {
  coder = (Coder<T>) input.getPipeline().getSchemaRegistry().getSchemaCoder(MqttRecord.class);
} catch (NoSuchSchemaException e) {
  throw new RuntimeException("No schema coder for MqttRecord; check Beam version/annotations", e);
}

When it happens

Trigger: 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.

Common situations: Using withMetadata() with a Beam version where schema inference for MqttRecord fails; classpath shading/proguard stripping annotations; mixing Beam versions across dependencies.

Understand the failure class

Background: Schema validation failed / invalid input schema: payload rejected because its shape doesn't match the expected schema — this error's family across 28 libraries.

Related errors

• ApproximateUnique.PerKey requires its input to use KvCoder
• array len mismatch. decoding
• AvroCoder for GenericRecord requires a schema
• cannot encode a null BitSet
• cannot encode a null byte[]

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/8d661527f554e95f. Report an issue: GitHub.

Appendix: source

Thrown at sdks/java/io/mqtt/src/main/java/org/apache/beam/sdk/io/mqtt/MqttIO.java:406

     * max read time is not null, the {@link Read} will provide a bounded {@link PCollection}.
     */
    public Read<T> withMaxReadTime(Duration maxReadTime) {
      return builder().setMaxReadTime(maxReadTime).build();
    }

    @Override
    @SuppressWarnings("unchecked")
    public PCollection<T> expand(PBegin input) {
      checkArgument(connectionConfiguration() != null, "connectionConfiguration can not be null");
      checkArgument(connectionConfiguration().getTopic() != null, "topic can not be null");

      Coder<T> coder;
      if (withMetadata()) {
        try {
          coder =
              (Coder<T>) input.getPipeline().getSchemaRegistry().getSchemaCoder(MqttRecord.class);
        } catch (NoSuchSchemaException e) {
          throw new RuntimeException(e.getMessage());
        }
      } else {
        coder = (Coder<T>) ByteArrayCoder.of();
      }

      org.apache.beam.sdk.io.Read.Unbounded<T> unbounded =
          org.apache.beam.sdk.io.Read.from(
              new UnboundedMqttSource<>(this.builder().setCoder(coder).build()));

      PTransform<PBegin, PCollection<T>> transform = unbounded;

      if (maxNumRecords() < Long.MAX_VALUE || maxReadTime() != null) {
        transform = unbounded.withMaxReadTime(maxReadTime()).withMaxNumRecords(maxNumRecords());
      }

      return input.getPipeline().apply(transform);
    }

View on GitHub (pinned to 12126d8942)