apache/beam · error · IllegalArgumentException

Input has tags but expected input tags

Error message

Input has tags ${actualInputTags} but expected input tags ${inputTags}

What it means

YamlTransform.expand validates that the PCollections fed into the externally-defined transform carry exactly the input tags declared in the YAML spec. The library joins the actual TupleTag ids and the expected set and throws IllegalArgumentException when the two sets differ. This fail-fast check prevents wiring a transform with missing or misnamed inputs.

Solutions

1. Align the input PCollectionRowTuple tags with the input tags declared in the YAML transform spec (extra or missing tags both fail).
2. Rename the upstream transform's output TupleTag ids so they match the YAML-declared input tags.
3. Update the YAML spec's input tags to the tags your pipeline actually produces, if the spec is wrong.
4. Print both tag sets (the message lists them) and diff to find the mismatched tag.

Example fix

// before
PCollectionRowTuple.of("in", pc).apply(yamlTransform);
// after (YAML expects tag "input")
PCollectionRowTuple.of("input", pc).apply(yamlTransform);

Set<String> declared = new HashSet<>(yamlSpecInputs);
Set<String> actual = inputTuple.expand().keySet().stream().map(TupleTag::getId).collect(Collectors.toSet());
if (!declared.equals(actual)) throw new IllegalStateException("Input tag mismatch: " + actual + " vs " + declared);

boolean hasExpectedTags(PCollectionRowTuple t, Set<String> expected) {
  return t.expand().keySet().stream().map(TupleTag::getId).collect(Collectors.toSet()).equals(expected);
}

try {
  result = pipeline.apply(yamlTransform);
} catch (IllegalArgumentException e) {
  if (e.getMessage().contains("Input has tags")) { /* fix tag wiring */ }
  throw e;
}

When it happens

Trigger: Calling YamlTransform.expand (or a wrapper) with a PCollectionRowTuple whose input tag ids do not exactly match the input tags expected by the YAML transform definition — extra tags, missing tags, or renamed tags all trip the check.

Common situations: YAML schema declares inputs like 'input' but the pipeline supplies tags 'input' and 'side_input'; renaming a PCollection's output tag; composing YAML transforms programmatically where tag ids are auto-generated and don't match the spec.

Understand the failure class

Background: "Must be a positive integer", "Invalid value", "Unsupported": the invalid-argument-value error family, when a library rejects the value you pass — this error's family across 35 libraries.

Related errors

• Output has tags but expected output tags
• Batch size is too large! It should be smaller or equal than
• Both clientCertPath and clientCertKeyPath must be specified…
• ChangeStreamName can't be empty
• Column delimiter should be set if headers are present.

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

Appendix: source

Thrown at sdks/java/extensions/yaml/src/main/java/org/apache/beam/sdk/extensions/yaml/YamlTransform.java:146

   * Indicates that this YamlTransform expects multiple, named outputs.
   *
   * @param outputTags the set of expected output tags to this transform
   * @return a PTransform like this but with a {@link PCollectionRowTuple} output type.
   */
  public YamlTransform<InputT, PCollectionRowTuple> withMultipleOutputs(String... outputTags) {
    return new YamlTransform<InputT, PCollectionRowTuple>(
        yamlDefinition, inputTags, ImmutableSet.copyOf(outputTags));
  }

  @Override
  public OutputT expand(InputT input) {
    if (inputTags != null) {
      Set<String> actualInputTags =
          input.expand().keySet().stream()
              .map(TupleTag::getId)
              .collect(Collectors.toCollection(HashSet::new));
      if (!inputTags.equals(actualInputTags)) {
        throw new IllegalArgumentException(
            "Input has tags "
                + Joiner.on(", ").join(actualInputTags)
                + " but expected input tags "
                + Joiner.on(", ").join(inputTags));
      }
    }

    // There is no generic apply...
    POutput output;
    @SuppressWarnings("rawtypes")
    PTransform externalTransform =
        PythonExternalTransform.from("apache_beam.yaml.yaml_transform.YamlTransform")
            .withArgs(yamlDefinition)
            .withExtraPackages(ImmutableList.of("jinja2", "pyyaml", "virtualenv-clone"));
    if (input instanceof PBegin) {
      output = ((PBegin) input).apply(externalTransform);
    } else if (input instanceof PCollection) {
      output = ((PCollection<?>) input).apply(externalTransform);

View on GitHub (pinned to 12126d8942)