{"record":{"id":"e674ca320df202f8","repo":"apache/iceberg","slug":"unknown-catalog-type-s-must-be-hive-hadoop","errorCode":null,"errorMessage":"Unknown catalog-type: %s (Must be 'hive', 'hadoop' or 'rest')","messagePattern":"Unknown catalog-type: (.+?) \\(Must be 'hive', 'hadoop' or 'rest'\\)","errorType":"exception","errorClass":"java.lang.UnsupportedOperationException","httpStatus":null,"severity":"error","filePath":"flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalogFactory.java","lineNumber":119,"sourceCode":"    String catalogType = properties.getOrDefault(ICEBERG_CATALOG_TYPE, ICEBERG_CATALOG_TYPE_HIVE);\n    switch (catalogType.toLowerCase(Locale.ROOT)) {\n      case ICEBERG_CATALOG_TYPE_HIVE:\n        // The values of properties 'uri', 'warehouse', 'hive-conf-dir' are allowed to be null, in\n        // that case it will\n        // fallback to parse those values from hadoop configuration which is loaded from classpath.\n        String hiveConfDir = properties.get(HIVE_CONF_DIR);\n        String hadoopConfDir = properties.get(HADOOP_CONF_DIR);\n        Configuration newHadoopConf = mergeHiveConf(hadoopConf, hiveConfDir, hadoopConfDir);\n        return CatalogLoader.hive(name, newHadoopConf, properties);\n\n      case ICEBERG_CATALOG_TYPE_HADOOP:\n        return CatalogLoader.hadoop(name, hadoopConf, properties);\n\n      case ICEBERG_CATALOG_TYPE_REST:\n        return CatalogLoader.rest(name, hadoopConf, properties);\n\n      default:\n        throw new UnsupportedOperationException(\n            \"Unknown catalog-type: \" + catalogType + \" (Must be 'hive', 'hadoop' or 'rest')\");\n    }\n  }\n\n  @Override\n  public Map<String, String> requiredContext() {\n    Map<String, String> context = Maps.newHashMap();\n    context.put(TYPE, \"iceberg\");\n    context.put(PROPERTY_VERSION, \"1\");\n    return context;\n  }\n\n  @Override\n  public List<String> supportedProperties() {\n    return ImmutableList.of(\"*\");\n  }\n\n  @Override","sourceCodeStart":101,"sourceCodeEnd":137,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalogFactory.java#L101-L137","documentation":"FlinkCatalogFactory only supports catalog types 'hive', 'hadoop' and 'rest' via the type.catalog property. When the configured catalog type does not match any known switch case, createCatalogLoader throws UnsupportedOperationException with this message. It indicates a misconfigured catalog factory property.","triggerScenarios":"Setting catalog property 'type.catalog' (or 'catalog-type' depending on version) to a value other than hive/hadoop/rest — e.g. a typo like 'Hadoop', 'jdbc' in a Flink SQL client where JDBC isn't wired, or an empty/unset value mapping to a wrong branch.","commonSituations":"Typo in Flink SQL client YAML/DDL WITH options; using a catalog type available in other Iceberg integrations but not this Flink factory; case mismatch since the lookup is case-sensitive.","solutions":["Set 'type.catalog' to exactly one of: hive, hadoop, rest (lowercase).","If you need a JDBC or other catalog type, use the appropriate Iceberg Flink version/build that supports it or construct a CatalogLoader programmatically instead of via FlinkCatalogFactory.","Double-check you are passing the property in the WITH (...) options of the Flink CREATE CATALOG statement or factory config map."],"exampleFix":"// before\nCREATE CATALOG iceberg WITH ('type'='iceberg', 'type.catalog'='Hadoop');\n// after\nCREATE CATALOG iceberg WITH ('type'='iceberg', 'type.catalog'='hadoop');","handlingStrategy":"validation","validationCode":"String type = options.get(\"type.catalog\");\nif (!Set.of(\"hive\", \"hadoop\", \"rest\").contains(type)) {\n  throw new IllegalArgumentException(\"type.catalog must be hive, hadoop or rest, got: \" + type);\n}","typeGuard":"null","tryCatchPattern":"try {\n  catalogLoader = factory.createCatalogLoader(name, hadoopConf, options);\n} catch (UnsupportedOperationException e) {\n  LOG.error(\"Invalid type.catalog option: {}\", e.getMessage());\n  throw e;\n}","preventionTips":["Keep type.catalog lowercase and limited to hive|hadoop|rest.","Validate catalog options in your deployment tooling before submitting Flink jobs.","Keep a config schema/linter for Flink CREATE CATALOG WITH clauses."],"tags":["flink","configuration","catalog"],"backgroundTag":"invalid-enum-value","analyzedSha":"86d9c8fc543e7c56c9f624eb725f76c9baff9570","analyzedAt":"2026-09-12T00:46:39.097Z","contentChangedAt":"2026-09-12T00:46:39.097Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}