{"record":{"id":"ea1d2f45a6b5ca02","repo":"apache/iceberg","slug":"tablename-unrecognized-write-distribution-mo","errorCode":null,"errorMessage":"${tableName}: Unrecognized ${WRITE_DISTRIBUTION_MODE}: ${mode}","messagePattern":"(.+?): Unrecognized (.+?): (.+?)","errorType":"exception","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/HashKeyGenerator.java","lineNumber":194,"sourceCode":"        if (equalityFields.isEmpty()) {\n          LOG.warn(\n              \"{}: Fallback to use 'none' distribution mode, because there are no equality fields set \"\n                  + \"and {}='range' is not supported yet in flink\",\n              tableName,\n              WRITE_DISTRIBUTION_MODE);\n          return tableKeySelector(tableName, writeParallelism, maxWriteParallelism);\n        } else {\n          LOG.info(\n              \"{}: Distribute rows by equality fields, because there are equality fields set \"\n                  + \"and {}='range' is not supported yet in flink\",\n              tableName,\n              WRITE_DISTRIBUTION_MODE);\n          return equalityFieldKeySelector(\n              tableName, schema, equalityFields, writeParallelism, maxWriteParallelism);\n        }\n\n      default:\n        throw new IllegalArgumentException(\n            tableName + \": Unrecognized \" + WRITE_DISTRIBUTION_MODE + \": \" + mode);\n    }\n  }\n\n  private static KeySelector<RowData, Integer> equalityFieldKeySelector(\n      String tableName,\n      Schema schema,\n      Set<String> equalityFields,\n      int writeParallelism,\n      int maxWriteParallelism) {\n    return new TargetLimitedKeySelector(\n        new EqualityFieldKeySelector(\n            schema,\n            FlinkSchemaUtil.convert(schema),\n            DynamicSinkUtil.getEqualityFieldIds(equalityFields, schema)),\n        tableName,\n        writeParallelism,\n        maxWriteParallelism);","sourceCodeStart":176,"sourceCodeEnd":212,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/HashKeyGenerator.java#L176-L212","documentation":"The dynamic Flink sink builds a key selector based on the table's write.distribution-mode property. When the mode string read from table properties is not one of the recognized values (none/hash), getKeySelector throws IllegalArgumentException naming the table and the offending mode. This guards against misconfigured or future/unsupported distribution modes reaching the sink's hashing path.","triggerScenarios":"A table's WRITE_DISTRIBUTION_MODE property contains a value other than none or hash (e.g. 'range' or a typo like 'hsh') when the dynamic sink's HashKeyGenerator.getKeySelector creates the key selector for that table.","commonSituations":"Setting write.distribution-mode=range in table properties while using the Flink dynamic sink which only supports none/hash for equality-field keying; manual property edits with typos; writing to a table whose properties were created by a newer engine version with an unrecognized mode value.","solutions":["Change the table property write.distribution-mode to a supported value (none or hash): ALTER TABLE ... SET TBLPROPERTIES ('write.distribution-mode'='hash')","If range distribution is intended, use the regular (non-dynamic) Iceberg Flink sink that supports RANGE mode, or pre-partition the data yourself","Check for typos in the mode value; values are matched literally after normalization, e.g. use 'hash' not 'HASHED'","Verify which table is named in the message and inspect its properties via the catalog before restarting the job"],"exampleFix":"// before (table properties)\n'write.distribution-mode'='range'\n// after\n'write.distribution-mode'='hash'","handlingStrategy":"validation","validationCode":"String mode = table.properties().getOrDefault(TableProperties.WRITE_DISTRIBUTION_MODE_DEFAULT, TableProperties.WRITE_DISTRIBUTION_MODE_NONE);\nif (!mode.equals(TableProperties.WRITE_DISTRIBUTION_MODE_NONE) && !mode.equals(TableProperties.WRITE_DISTRIBUTION_MODE_HASH)) {\n  throw new IllegalArgumentException(\"Unsupported write.distribution-mode for dynamic sink: \" + mode);\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Set write.distribution-mode only to none or hash for tables consumed by the Flink dynamic sink","Keep table property values lowercase and unmodified from engine defaults","Validate table properties in a pre-flight check before submitting the Flink job"],"tags":["flink","iceberg","config","distribution-mode"],"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"}