apache/beam · error · IllegalStateException

Could not get value Coder

Error message

Could not get value Coder

What it means

In CdapIO.expand for the read path, the connector attempts to obtain a Coder for the value type from the pipeline's CoderRegistry. If the registry cannot provide a coder (CannotProvideCoderException), an IllegalStateException wrapping the cause is thrown.

Solutions

1. Register a coder for the value type in the pipeline's CoderRegistry.
2. Make the value type a standard codable type (e.g. String, Long, byte[], TableRow) or implement a Coder<V> and supply it via setCoder after the read.
3. Implement Serializable/`@DefaultSchema` annotations so Beam can infer a coder automatically.

Example fix

// before
CdapIO.<String, MyPojo>read()...
// after
pipeline.getCoderRegistry().registerCoderForClass(MyPojo.class, new MyPojoCoder());
CdapIO.<String, MyPojo>read()...

try {
  pipeline.getCoderRegistry().getCoder(valueClass);
} catch (CannotProvideCoderException e) {
  pipeline.getCoderRegistry().registerCoderForClass(valueClass, new MyValueCoder());
}

try { pipeline.run().waitUntilFinish(); } catch (IllegalStateException e) { if (e.getMessage().equals("Could not get value Coder")) { /* register coder */ } throw e; }

When it happens

Trigger: Using CdapIO.read() with a value type V that has no registered/default coder (custom class without codable structure) in the Beam pipeline.

Common situations: Reading records into a custom POJO/class that Beam cannot infer a coder for; missing registration of a custom Coder in the CoderRegistry.

Understand the failure class

Background: "missing required config value" errors: why libraries refuse to start when a configuration key is empty, unset, or blank — this error's family across 48 libraries.

Related errors

• ApproximateUnique.PerKey requires its input to use KvCoder
• AvroCoder for GenericRecord requires a schema
• Can not call prepareRun
• 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/27554e6b6ada13a2. Report an issue: GitHub.

Appendix: source

Thrown at sdks/java/io/cdap/src/main/java/org/apache/beam/sdk/io/cdap/CdapIO.java:356

        Long pullFrequencySec = getPullFrequencySec();
        if (pullFrequencySec != null) {
          reader = reader.withPullFrequencySec(pullFrequencySec);
        }
        Long startPollTimeoutSec = getStartPollTimeoutSec();
        if (startPollTimeoutSec != null) {
          reader = reader.withStartPollTimeoutSec(startPollTimeoutSec);
        }
        Long startOffset = getStartOffset();
        if (startOffset != null) {
          reader = reader.withStartOffset(startOffset);
        }
        try {
          Coder<V> coder = input.getPipeline().getCoderRegistry().getCoder(valueClass);
          PCollection<V> values = input.apply(reader).setCoder(coder);
          SerializableFunction<V, KV<K, V>> fn = input1 -> KV.of(null, input1);
          return values.apply(MapElements.into(new TypeDescriptor<KV<K, V>>() {}).via(fn));
        } catch (CannotProvideCoderException e) {
          throw new IllegalStateException("Could not get value Coder", e);
        }
      } else {
        cdapPlugin.withHadoopConfiguration(keyClass, valueClass).prepareRun();
        Configuration hConf = cdapPlugin.getHadoopConfiguration();
        HadoopFormatIO.Read<K, V> readFromHadoop =
            HadoopFormatIO.<K, V>read().withConfiguration(hConf);
        return input.apply(readFromHadoop);
      }
    }
  }

  /** A {@link PTransform} to write to CDAP sink. */
  @AutoValue
  @AutoValue.CopyAnnotations
  public abstract static class Write<K, V> extends PTransform<PCollection<KV<K, V>>, PDone> {

    abstract @Nullable PluginConfig getPluginConfig();

View on GitHub (pinned to 12126d8942)