apache/beam · error · IllegalStateException

Unknown ChangelogScanTask type

Error message

Unknown ChangelogScanTask type: {}

What it means

SerializableChangelogTask.from serializes a ChangelogScanTask into a portable form by branching on the task's concrete type. If getType(task) yields a Type the switch does not handle (a ChangelogScanTask subtype outside the four supported ones), it throws IllegalStateException. This guards against new Iceberg changelog task types appearing without a serialization path.

Solutions

1. Ensure only Iceberg versions producing the four supported task types are used
2. Extend both getType() and from() to handle the new task subtype
3. Validate task types before serialization and fail with a clear pipeline-level error
4. Pin/align the Iceberg runtime version with the one Beam's CDC was built against

Example fix

// before
SerializableChangelogTask t = SerializableChangelogTask.from(unknownTask, specs, true);
// after
if (!SerializableChangelogTask.isSupported(unknownTask)) {
  throw new IllegalArgumentException("Unsupported task type: " + unknownTask.getClass());
}
SerializableChangelogTask t = SerializableChangelogTask.from(unknownTask, specs, true);

if (!(task instanceof AddedRowsScanTask)
    && !(task instanceof DeletedRowsScanTask)
    && !(task instanceof DeletedDataFileScanTask)) {
  throw new IllegalArgumentException("Unsupported changelog task type: " + task.getClass());
}

static boolean isSerializableTask(ChangelogScanTask task) {
  return task instanceof AddedRowsScanTask
      || task instanceof DeletedRowsScanTask
      || task instanceof DeletedDataFileScanTask;
}

try {
  SerializableChangelogTask t = SerializableChangelogTask.from(task, specs, includeMetrics);
} catch (IllegalStateException e) {
  throw new UnsupportedOperationException("Iceberg runtime produced an unsupported changelog task; check version alignment", e);
}

When it happens

Trigger: Calling SerializableChangelogTask.from(task) with a ChangelogScanTask that is not an AddedRowsScanTask, DeletedRowsScanTask, DeletedDataFileScanTask, or the type mapped by getType — e.g. a new task subtype introduced by a newer Iceberg incremental scan.

Common situations: Iceberg library upgrade adding a new ChangelogScanTask subtype; a custom ChangelogScanTask implementation passed into the Beam CDC pipeline; mismatch between getType's classification and the serialization switch after partial code changes.

Understand the failure class

Background: Invalid enum value errors: "Unknown type", "Invalid scope", "must be one of" — when a string is not on the library's allowed list — this error's family across 23 libraries.

Related errors

• Unknown delete file content
• Cannot serialize DataFile: its partition spec id
• Cannot serialize DeleteFile: its partition spec id
• Equality field is not a top-level column of schema
• Expected at least one overlapping task in bidirectional list

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

Appendix: source

Thrown at sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/SerializableChangelogTask.java:191

                  toSerializableDeletes(addedRowsTask.deletes(), specs, includeMetrics));
    } else if (task instanceof DeletedRowsScanTask) {
      DeletedRowsScanTask deletedRowsTask = (DeletedRowsScanTask) task;
      builder =
          builder
              .setType(Type.DELETED_ROWS)
              .setAddedDeletes(
                  toSerializableDeletes(deletedRowsTask.addedDeletes(), specs, includeMetrics))
              .setExistingDeletes(
                  toSerializableDeletes(deletedRowsTask.existingDeletes(), specs, includeMetrics));
    } else if (task instanceof DeletedDataFileScanTask) {
      DeletedDataFileScanTask deletedFileTask = (DeletedDataFileScanTask) task;
      builder =
          builder
              .setType(Type.DELETED_FILE)
              .setExistingDeletes(
                  toSerializableDeletes(deletedFileTask.existingDeletes(), specs, includeMetrics));
    } else {
      throw new IllegalStateException("Unknown ChangelogScanTask type: " + task.getClass());
    }
    return builder.build();
  }

  static Type getType(ChangelogScanTask task) {
    if (task instanceof AddedRowsScanTask) {
      return Type.ADDED_ROWS;
    } else if (task instanceof DeletedRowsScanTask) {
      return Type.DELETED_ROWS;
    } else if (task instanceof DeletedDataFileScanTask) {
      return Type.DELETED_FILE;
    } else {
      throw new IllegalStateException("Unknown ChangelogScanTask type: " + task.getClass());
    }
  }

  static long getTotalLength(List<ChangelogScanTask> tasks) {
    return tasks.stream().mapToLong(SerializableChangelogTask::getLength).sum();

View on GitHub (pinned to 12126d8942)