{"record":{"id":"46ab9a1cae1e0727","repo":"apache/iceberg","slug":"unrecognized-write-distribution-mode","errorCode":null,"errorMessage":"Unrecognized write.distribution-mode: ","messagePattern":"Unrecognized write\\.distribution-mode: ","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java","lineNumber":721,"sourceCode":"            shuffleStream = shuffleStream.uid(uidPrefix + \"-shuffle\");\n          }\n\n          return shuffleStream\n              .partitionCustom(new RangePartitioner(iSchema, sortOrder), r -> r)\n              .flatMap(\n                  (FlatMapFunction<StatisticsOrRecord, RowData>)\n                      (statisticsOrRecord, out) -> {\n                        if (statisticsOrRecord.hasRecord()) {\n                          out.collect(statisticsOrRecord.record());\n                        }\n                      })\n              // Set the parallelism same as writerParallelism to\n              // promote operator chaining with the downstream writer operator\n              .setParallelism(writerParallelism)\n              .returns(RowData.class);\n\n        default:\n          throw new RuntimeException(\"Unrecognized \" + WRITE_DISTRIBUTION_MODE + \": \" + writeMode);\n      }\n    }\n  }\n\n  /**\n   * Clean up after removing {@link Builder#tableSchema}\n   *\n   * @deprecated since 1.10.0, will be removed in 2.0.0. Use {@link #toFlinkRowType(Schema,\n   *     ResolvedSchema)} instead.\n   */\n  @Deprecated\n  static RowType toFlinkRowType(Schema schema, TableSchema requestedSchema) {\n    if (requestedSchema != null) {\n      // Convert the flink schema to iceberg schema using the table schema as the reference.\n      Schema writeSchema = FlinkSchemaUtil.convert(schema, requestedSchema);\n      TypeUtil.validateWriteSchema(schema, writeSchema, true, true);\n\n      // We use this flink schema to read values from RowData. The flink's TINYINT and SMALLINT will","sourceCodeStart":703,"sourceCodeEnd":739,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java#L703-L739","documentation":"distributeDataStream switches on the table's write.distribution-mode; the default branch throws when the mode string is not none/hash/range. The library only implements those three distribution strategies for the pre-write topology.","triggerScenarios":"Setting table property write.distribution-mode to an unrecognized value (typo like 'hashes', mixed case after custom parsing, or a future/newer value read by an older runtime) before FlinkSink.append().","commonSituations":"Hand-editing table properties; a table created by a newer Iceberg/Spark writer using a mode the Flink runtime doesn't know; configuration copied from another engine with extra values.","solutions":["Set write.distribution-mode to one of none, hash, or range (or remove the property to use the default).","Check the exact value: ALTER TABLE ... SET TBLPROPERTIES ('write.distribution-mode'='hash').","Upgrade the iceberg-flink-runtime if the mode was introduced by a newer version."],"exampleFix":"// before\nproperties.put(\"write.distribution-mode\", \"key-by\");\n// after\nproperties.put(\"write.distribution-mode\", \"hash\");","handlingStrategy":"validation","validationCode":"String mode = table.properties().getOrDefault(\"write.distribution-mode\", \"none\");\nif (!java.util.Set.of(\"none\", \"hash\", \"range\").contains(mode.toLowerCase(java.util.Locale.ROOT))) {\n  throw new IllegalArgumentException(\"unsupported write.distribution-mode: \" + mode);\n}","typeGuard":null,"tryCatchPattern":"try {\n  sink.append();\n} catch (RuntimeException e) {\n  if (e.getMessage().contains(\"write.distribution-mode\")) {\n    // fix the table property then resubmit\n  }\n  throw e;\n}","preventionTips":["Only set distribution-mode to none/hash/range.","Set table properties via SQL/catalog APIs, not manual edits.","Keep engine and Flink runtime versions aligned."],"tags":["flink","config","distribution-mode","invalid-enum"],"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"}