{"record":{"id":"a7618f0396e2f487","repo":"apache/iceberg","slug":"unrecognized-write-distribution-mode-a7618f","errorCode":null,"errorMessage":": Unrecognized write.distribution-mode: ","messagePattern":": Unrecognized write\\.distribution-mode: ","errorType":"exception","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"flink/v2.2/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/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/HashKeyGenerator.java#L176-L212","documentation":"HashKeyGenerator.getKeySelector switches on the table's write.distribution-mode. Only NONE/HASH (and equality-field modes handled by earlier branches) are supported; any other value falls to default and throws IllegalArgumentException with the table name, property name, and offending mode value.","triggerScenarios":"Setting write.distribution-mode to a value other than none/hash (e.g. a typo like 'hase' or an unsupported 'range' in this Flink dynamic sink path) and building a key selector for the writer.","commonSituations":"Table properties copied from Spark (where 'range' is valid) but consumed by the Flink dynamic sink; hand-edited table properties with typos.","solutions":["Set write.distribution-mode to 'none' or 'hash' on the table.","If you need range distribution, pre-partition upstream or drop the property so the sink picks its default.","Fix typos in the property value; values are matched case-sensitively via DistributionMode parsing."],"exampleFix":"// before\ntable.updateProperties().set(TableProperties.WRITE_DISTRIBUTION_MODE, \"range\").commit();\n// after\ntable.updateProperties().set(TableProperties.WRITE_DISTRIBUTION_MODE, \"hash\").commit();","handlingStrategy":"validation","validationCode":"String mode = table.properties().getOrDefault(TableProperties.WRITE_DISTRIBUTION_MODE, TableProperties.WRITE_DISTRIBUTION_MODE_NONE); if (!mode.equalsIgnoreCase(\"none\") && !mode.equalsIgnoreCase(\"hash\")) { throw new IllegalArgumentException(\"Unsupported write.distribution-mode for Flink: \" + mode); }","typeGuard":null,"tryCatchPattern":"try { keySelector = HashKeyGenerator.getKeySelector(...); } catch (IllegalArgumentException e) { /* reset the table property to none/hash and retry */ throw e; }","preventionTips":["Only set write.distribution-mode to none or hash for Flink sinks","Do not copy Spark-only distribution settings (e.g. range) to Flink jobs","Validate table properties at job startup with a preflight check"],"tags":["flink","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-14T21:17:11.552Z"}