What causes "Target field {} was added twice to input {}" in apache/flink?

Audit the @ForwardedFieldsFirst / @ForwardedFieldsSecond annotations: each output field index must appear as a target exactly once per input.

apache/flink · error · InvalidSemanticAnnotationException

Target field {} was added twice to input {}

Error message

Target field {} was added twice to input {}

What it means

Thrown by DualInputSemanticProperties.addForwardedField() when the same target field position is already mapped from a (different) source field in the same input. Flink's two-input semantic model requires each output field to be forwarded from at most one source field per input; a duplicate target means the annotation is internally contradictory and the optimizer cannot trust it.

Source

Thrown at flink-core/src/main/java/org/apache/flink/api/common/operators/DualInputSemanticProperties.java:128

     *
     * @param input the input of the source field
     * @param sourceField the position in the source record
     * @param targetField the position in the destination record
     */
    public void addForwardedField(int input, int sourceField, int targetField) {

        Map<Integer, FieldSet> fieldMapping;

        if (input != 0 && input != 1) {
            throw new IndexOutOfBoundsException();
        } else if (input == 0) {
            fieldMapping = this.fieldMapping1;
        } else {
            fieldMapping = this.fieldMapping2;
        }

        if (isTargetFieldPresent(targetField, fieldMapping)) {
            throw new InvalidSemanticAnnotationException(
                    "Target field " + targetField + " was added twice to input " + input);
        }

        FieldSet targetFields = fieldMapping.get(sourceField);
        if (targetFields != null) {
            fieldMapping.put(sourceField, targetFields.addField(targetField));
        } else {
            fieldMapping.put(sourceField, new FieldSet(targetField));
        }
    }

    private boolean isTargetFieldPresent(int targetField, Map<Integer, FieldSet> fieldMapping) {

        for (FieldSet targetFields : fieldMapping.values()) {
            if (targetFields.contains(targetField)) {
                return true;
            }
        }

View on GitHub (pinned to 2f3c205e92)

Solutions

Audit the @ForwardedFieldsFirst / @ForwardedFieldsSecond annotations: each output field index must appear as a target exactly once per input.
If the function genuinely produces one field from two inputs, drop the forwarded-field annotation for that target and use @ReadFields or no annotation instead.
Check the exception's input number (0 or 1) and targetField value to localise which annotation is duplicated.

Example fix

// before
@ForwardedFieldsFirst("0->0")
@ForwardedFieldsSecond("0->0") // both claim output field 0
public class MyJoin extends RichJoinFunction<Tuple2<String,Integer>, Tuple2<String,Integer>, Tuple2<String,Integer>> { ... }
// after — only one source legitimately forwards to output field 0
@ForwardedFieldsFirst("0->0;1->1")
public class MyJoin extends RichJoinFunction<...> { ... }

Defensive patterns

Strategy: validation

Validate before calling

// before annotating, ensure each output field appears as a target at most once per input
Set<Integer> targetsForInput0 = new HashSet<>();
for (ForwardedFieldSpec s : specsFirstInput) {
    if (!targetsForInput0.add(s.target)) throw new IllegalStateException("duplicate target " + s.target);
}

Try / catch

try {
    props.addForwardedField(input, src, tgt);
} catch (InvalidSemanticAnnotationException e) {
    // annotations are static metadata; fix the annotation rather than catch at runtime
    throw new IllegalStateException("Bad @ForwardedFields annotation: " + e.getMessage(), e);
}

Prevention

Each output field index must be a forwarded target of at most one source field per input.
Validate forwarded-field annotations in a unit test that constructs the SemanticProperties.
When copying single-input annotations to a two-input operator, re-check the second-input targets.

When it happens

Trigger: Annotating a CoGroup/Join/FlatJoin function with @ForwardedFieldsSecond or @ForwardedFieldsFirst where two distinct source fields both claim to land in the same output position; manually calling addForwardedField(input, srcA, tgtX) then addForwardedField(input, srcB, tgtX).

Common situations: Writing @ForwardedFields annotations on two-input operators by copying from a single-input pattern without realising the second-input annotation also targets output fields; miscounting output field positions when the function emits a Tuple whose field index is reused.

Related errors

AI-assisted analysis of apache/flink@2f3c205e92 (2026-08-14). Data as JSON: /api/errors/7f77f4354d356aed. Report an issue: GitHub.