{"record":{"id":"f62c82a4953b5b9a","repo":"apache/beam","slug":"unexpected-beam-type","errorCode":null,"errorMessage":"Unexpected Beam type: {}","messagePattern":"Unexpected Beam type: (.+?)","errorType":"exception","errorClass":"UnsupportedOperationException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/ApplyWatermarkColumn.java","lineNumber":89,"sourceCode":"      case INT64:\n        return Instant.ofEpochMilli(timeUnit.toMillis((Long) value));\n      case DATETIME:\n        return (Instant) value;\n      case LOGICAL_TYPE:\n        String logicalType =\n            Preconditions.checkStateNotNull(field.getType().getLogicalType()).getIdentifier();\n        if (logicalType.equals(SqlTypes.DATETIME.getIdentifier())) {\n          return Instant.ofEpochMilli(\n              MICROSECONDS.toMillis(DateTimeUtil.microsFromTimestamp((LocalDateTime) value)));\n        } else if (logicalType.equals(SqlTypes.TIMESTAMP.getIdentifier())\n            || logicalType.equals(org.apache.beam.sdk.schemas.logicaltypes.Timestamp.IDENTIFIER)) {\n          return Instant.ofEpochMilli(\n              MICROSECONDS.toMillis(DateTimeUtil.microsFromInstant((java.time.Instant) value)));\n        } else {\n          throw new UnsupportedOperationException(\"Unexpected logical type: \" + logicalType);\n        }\n      default:\n        throw new UnsupportedOperationException(\"Unexpected Beam type: \" + field.getType());\n    }\n  }\n\n  @Override\n  public Duration getAllowedTimestampSkew() {\n    // Generous skew to cover backfill of historical data and late-arriving CDC patterns.\n    return Duration.standardDays(365);\n  }\n}\n","sourceCodeStart":71,"sourceCodeEnd":99,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/ApplyWatermarkColumn.java#L71-L99","documentation":"ApplyWatermarkColumn.getInstant switches on the field's Beam primitive type; only a few primitive types are supported for extracting an Instant (via their logical types). If the field's Beam type falls outside all supported cases, the default branch throws UnsupportedOperationException with the unexpected Beam type.","triggerScenarios":"A watermark field typed as a non-timestamp Beam primitive (e.g. STRING, INT64, BOOLEAN, or a nested row/array type) is supplied to ApplyWatermarkColumn, reaching the default branch of the type switch.","commonSituations":"Misconfigured watermark column name pointing at a string-encoded timestamp, schema drift changing the field type after a connector upgrade, or selecting the wrong column in the CDC record schema.","solutions":["Point ApplyWatermarkColumn at a field whose Beam type is one of the supported timestamp variants (logical SqlTypes.TIMESTAMP / beam Timestamp).","Coerce the column in an upstream transform (e.g. parse string timestamps into a proper logical timestamp field).","Update ApplyWatermarkColumn to support the missing Beam type and rebuild.","Verify the schema of records at runtime and align the configured watermark field with it."],"exampleFix":"// before\n.apply(\"watermark\", ApplyWatermarkColumn.create(\"event_ts\")); // event_ts is STRING\n// after\n.apply(\"parse-ts\", MapElements.into(SqlTypes.TIMESTAMP).via(s -> Instant.parse(s)))\n.apply(\"watermark\", ApplyWatermarkColumn.create(\"event_ts\"));","handlingStrategy":"type-guard","validationCode":"if (!isSupportedTimestamp(schema.getField(col).getType())) {\n  throw new IllegalArgumentException(\"Watermark column is not a timestamp-typed field: \" + col);\n}","typeGuard":"static boolean isTimestampField(Schema.Field f) {\n  FieldType t = f.getType();\n  return t.getTypeName().isNumericType() == false && t.getLogicalType() != null &&\n      (SqlTypes.TIMESTAMP.getIdentifier().equals(t.getLogicalType().getIdentifier()) ||\n       org.apache.beam.sdk.schemas.logicaltypes.Timestamp.IDENTIFIER.equals(t.getLogicalType().getIdentifier()));\n}","tryCatchPattern":"try {\n  records.apply(ApplyWatermarkColumn.create(col));\n} catch (UnsupportedOperationException e) {\n  if (e.getMessage().startsWith(\"Unexpected Beam type\")) {\n    throw new IllegalArgumentException(\"Watermark field must be timestamp-typed, got: \" + e.getMessage());\n  }\n  throw e;\n}","preventionTips":["Verify the configured watermark column name points at a timestamp field, not a string.","Parse string timestamps into proper logical timestamp fields upstream.","Pin schema contracts between producer and pipeline to avoid silent type drift."],"tags":["java","beam","iceberg","cdc","type-mismatch"],"backgroundTag":"type-mismatch","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}