{"record":{"id":"132d0d49cd2f2680","repo":"apache/iceberg","slug":"cannot-create-the-table-with-connector-iceberg-132d0d","errorCode":null,"errorMessage":"Cannot create the table with 'connector'='iceberg' table property in an iceberg catalog, Please create table with 'connector'='iceberg' property in a non-iceberg catalog or create table without 'connector'='iceberg' related properties in an iceberg table.","messagePattern":"Cannot create the table with 'connector'='iceberg' table property in an iceberg catalog, Please create table with 'connector'='iceberg' property in a non-iceberg catalog or create table without 'connector'='iceberg' related properties in an iceberg table\\.","errorType":"exception","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java","lineNumber":420,"sourceCode":"          toIdentifier(new ObjectPath(tablePath.getDatabaseName(), newTableName)));\n    } catch (org.apache.iceberg.exceptions.NoSuchTableException e) {\n      if (!ignoreIfNotExists) {\n        throw new TableNotExistException(getName(), tablePath, e);\n      }\n    } catch (AlreadyExistsException e) {\n      throw new TableAlreadyExistException(getName(), tablePath, e);\n    }\n  }\n\n  @Override\n  public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)\n      throws CatalogException, TableAlreadyExistException {\n    // Creating Iceberg table using connector is allowed only when table is created using LIKE\n    if (Objects.equals(\n            table.getOptions().get(FlinkCreateTableOptions.CONNECTOR_PROPS_KEY),\n            FlinkDynamicTableFactory.FACTORY_IDENTIFIER)\n        && table.getOptions().get(FlinkCreateTableOptions.SRC_CATALOG_PROPS_KEY) == null) {\n      throw new IllegalArgumentException(\n          \"Cannot create the table with 'connector'='iceberg' table property in \"\n              + \"an iceberg catalog, Please create table with 'connector'='iceberg' property in a non-iceberg catalog or \"\n              + \"create table without 'connector'='iceberg' related properties in an iceberg table.\");\n    }\n\n    Preconditions.checkArgument(\n        table instanceof ResolvedCatalogTable,\n        \"Expected a ResolvedCatalogTable but got: %s. \"\n            + \"Iceberg Flink catalog only supports resolved catalog tables \"\n            + \"(Materialized tables and other table kinds are not supported).\",\n        table == null ? \"null\" : table.getClass().getName());\n    createIcebergTable(tablePath, (ResolvedCatalogTable) table, ignoreIfExists);\n  }\n\n  void createIcebergTable(ObjectPath tablePath, ResolvedCatalogTable table, boolean ignoreIfExists)\n      throws CatalogException, TableAlreadyExistException {\n    validateFlinkTable(table);\n","sourceCodeStart":402,"sourceCodeEnd":438,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java#L402-L438","documentation":"createTable rejects a Flink table definition that explicitly sets 'connector'='iceberg' unless it carries the src-catalog property that marks it as derived via CREATE TABLE LIKE. Inside an Iceberg catalog the connector option is redundant and reserved, so a bare connector='iceberg' definition is treated as a misuse (usually copy-paste from a generic Flink catalog setup).","triggerScenarios":"CREATE TABLE ... WITH ('connector'='iceberg') executed in an Iceberg catalog where the options map contains CONNECTOR_PROPS_KEY equal to 'iceberg' and SRC_CATALOG_PROPS_KEY is null (FlinkCatalog.java:416-423).","commonSituations":"Users following generic Flink SQL docs that require 'connector'='iceberg' for non-Iceberg catalogs, then running the same DDL in an Iceberg catalog; DDL generated by tools that always emit the connector option.","solutions":["Remove the 'connector'='iceberg' option from the WITH clause when the table is created in an Iceberg catalog.","If this table was produced by CREATE TABLE LIKE from an Iceberg table, keep the like-origin property (do not strip the src-catalog key).","If the intent is a generic Flink connector table, create it in a non-Iceberg Flink catalog instead."],"exampleFix":"// before\nCREATE TABLE iceberg_catalog.db.t (id BIGINT) WITH ('connector'='iceberg');\n\n// after\nCREATE TABLE iceberg_catalog.db.t (id BIGINT) WITH ();","handlingStrategy":"validation","validationCode":"if (\"iceberg\".equals(options.get(\"connector\"))) {\n  throw new IllegalArgumentException(\"drop 'connector'='iceberg' inside an iceberg catalog\");\n}","typeGuard":null,"tryCatchPattern":"try {\n  catalog.createTable(tablePath, table, ignoreIfExists);\n} catch (IllegalArgumentException e) {\n  // strip connector option and retry\n}","preventionTips":["In Iceberg catalogs, omit 'connector'='iceberg' from WITH clauses.","If using CREATE TABLE LIKE from an Iceberg table, keep the generated like-origin properties.","Lint generated DDL to strip connector options when the target catalog is Iceberg."],"tags":["flink","catalog","create-table","connector","illegal-argument"],"backgroundTag":"invalid-argument-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"}