{"record":{"id":"ab85057c3c1c6162","repo":"apache/beam","slug":"unknown-watermark-type-type-supported-types-are","errorCode":null,"errorMessage":"Unknown watermark type: ${type}. Supported types are ProcessingTime, LogAppendTime, CreateTime.","messagePattern":"Unknown watermark type: (.+?)\\. Supported types are ProcessingTime, LogAppendTime, CreateTime\\.","errorType":"validation","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/kafka/KafkaTableProvider.java","lineNumber":143,"sourceCode":"      String type = properties.get(\"watermark.type\").asText().toUpperCase();\n\n      switch (type) {\n        case \"PROCESSINGTIME\":\n          timestampPolicyFactory = TimestampPolicyFactory.withProcessingTime();\n          break;\n        case \"LOGAPPENDTIME\":\n          timestampPolicyFactory = TimestampPolicyFactory.withLogAppendTime();\n          break;\n        case \"CREATETIME\":\n          Duration delay = Duration.ZERO;\n          if (properties.has(\"watermark.delay\")) {\n            String delayStr = properties.get(\"watermark.delay\").asText();\n            delay = PeriodFormat.getDefault().parsePeriod(delayStr).toStandardDuration();\n          }\n          timestampPolicyFactory = TimestampPolicyFactory.withCreateTime(delay);\n          break;\n        default:\n          throw new IllegalArgumentException(\n              \"Unknown watermark type: \"\n                  + type\n                  + \". Supported types are ProcessingTime, LogAppendTime, CreateTime.\");\n      }\n    }\n\n    BeamKafkaTable kafkaTable = null;\n    if (Schemas.isNestedSchema(schema)) {\n      Optional<PayloadSerializer> serializer =\n          payloadFormat.map(\n              format ->\n                  PayloadSerializers.getSerializer(\n                      format,\n                      checkArgumentNotNull(schema.getField(PAYLOAD_FIELD).getType().getRowSchema()),\n                      TableUtils.convertNode2Map(properties)));\n      kafkaTable =\n          new NestedPayloadKafkaTable(\n              schema, bootstrapServers, topics, serializer, timestampPolicyFactory);","sourceCodeStart":125,"sourceCodeEnd":161,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/kafka/KafkaTableProvider.java#L125-L161","documentation":"Thrown by KafkaTableProvider.buildBeamSqlTable when the DDL table property 'watermark.type' (mapped from the TBLPROPERTIES 'type') is not one of ProcessingTime, LogAppendTime, or CreateTime. The switch over the watermark type string has a default branch that rejects any unrecognized value at table-creation time.","triggerScenarios":"Creating a Beam SQL Kafka table whose 'watermark.type' property is misspelled (e.g. 'procesingtime'), lowercase ('processingtime'), or an entirely unsupported policy string; the parser hits the default branch and throws IllegalArgumentException.","commonSituations":"Typo in CREATE EXTERNAL TABLE TBLPROPERTIES; copying a watermark type from another connector (e.g. Flink) that supports extra values; case-sensitivity mistakes; upgrading Beam and using a value documented elsewhere but not implemented in this provider.","solutions":["Set 'watermark.type' to exactly ProcessingTime, LogAppendTime, or CreateTime (match capitalization).","If no watermark behavior is needed, omit the 'watermark.type' property so the default policy is used.","Check the Kafka SQL provider docs for the exact supported enum values for your Beam version."],"exampleFix":"// before\nCREATE EXTERNAL TABLE kafka_table (...) TBLPROPERTIES {'type':'processingtime'}\n// after\nCREATE EXTERNAL TABLE kafka_table (...) TBLPROPERTIES {'type':'ProcessingTime'}","handlingStrategy":"validation","validationCode":"Set<String> allowed = Set.of(\"ProcessingTime\", \"LogAppendTime\", \"CreateTime\");\nif (watermarkType != null && !allowed.contains(watermarkType)) {\n  throw new IllegalArgumentException(\"watermark.type must be one of \" + allowed);\n}","typeGuard":null,"tryCatchPattern":"try { table = tableProvider.buildBeamSqlTable(table); } catch (IllegalArgumentException e) { /* fall back to default watermark policy */ }","preventionTips":["Copy the exact enum-capitalized names from the docs","Omit the property when the default policy suffices","Add a startup DDL validation test"],"tags":["java","apache-beam","kafka","sql","configuration"],"backgroundTag":"invalid-enum-value","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"}