{"record":{"id":"4b926893f39c4b81","repo":"apache/iceberg","slug":"unrecognized-write-distribution-mode-4b9268","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/IcebergSink.java","lineNumber":983,"sourceCode":"    }\n  }\n\n  private DataStream<RowData> distributeDataStream(DataStream<RowData> input) {\n    DistributionMode mode = flinkWriteConf.distributionMode();\n    Schema schema = table.schema();\n    PartitionSpec spec = table.spec();\n    SortOrder sortOrder = table.sortOrder();\n\n    LOG.info(\"Write distribution mode is '{}'\", mode.modeName());\n    switch (mode) {\n      case NONE:\n        return distributeDataStreamByNoneDistributionMode(input, schema);\n      case HASH:\n        return distributeDataStreamByHashDistributionMode(input, schema, spec);\n      case RANGE:\n        return distributeDataStreamByRangeDistributionMode(input, schema, spec, sortOrder);\n      default:\n        throw new RuntimeException(\"Unrecognized \" + WRITE_DISTRIBUTION_MODE + \": \" + mode);\n    }\n  }\n\n  private DataStream<RowData> distributeDataStreamByNoneDistributionMode(\n      DataStream<RowData> input, Schema iSchema) {\n    if (equalityFieldIds.isEmpty()) {\n      return input;\n    } else {\n      LOG.info(\"Distribute rows by equality fields, because there are equality fields set\");\n      return input.keyBy(new EqualityFieldKeySelector(iSchema, flinkRowType, equalityFieldIds));\n    }\n  }\n\n  private DataStream<RowData> distributeDataStreamByHashDistributionMode(\n      DataStream<RowData> input, Schema iSchema, PartitionSpec partitionSpec) {\n    if (equalityFieldIds.isEmpty()) {\n      if (partitionSpec.isUnpartitioned()) {\n        LOG.warn(","sourceCodeStart":965,"sourceCodeEnd":1001,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java#L965-L1001","documentation":"In IcebergSink's write-distribution dispatch, an unrecognized DistributionMode value reaches the default branch and throws this RuntimeException. Only NONE, HASH and RANGE are implemented for the pre-write topology of the new sink API.","triggerScenarios":"A table's write.distribution-mode property parses to a DistributionMode the sink cannot handle — practically, a table property set to an invalid string that produced an unexpected enum mapping, or a newer enum constant read by an older runtime.","commonSituations":"Hand-set table properties (typos or custom values); tables written by newer engines with modes unknown to the deployed iceberg-flink-runtime; copy-pasted configs from Spark jobs.","solutions":["Reset the property: ALTER TABLE t SET TBLPROPERTIES ('write.distribution-mode'='none'|'hash'|'range') or unset it.","Match the iceberg-flink-runtime version to the engine version that created the table.","Confirm equality-field/partition settings so the sink can pick a supported mode automatically."],"exampleFix":"// before\nproperties.put(\"write.distribution-mode\", \"broadcast\");\n// after\nproperties.put(\"write.distribution-mode\", \"hash\");","handlingStrategy":"validation","validationCode":"DistributionMode mode = DistributionMode.fromName(\n    table.properties().getOrDefault(\"write.distribution-mode\", \"none\"));\nif (mode != DistributionMode.NONE && mode != DistributionMode.HASH && mode != DistributionMode.RANGE) {\n  throw new IllegalArgumentException(\"unsupported mode: \" + mode);\n}","typeGuard":null,"tryCatchPattern":"try {\n  sink.append();\n} catch (RuntimeException e) {\n  if (e.getMessage().contains(WRITE_DISTRIBUTION_MODE)) {\n    // reset the table property and resubmit\n  }\n  throw e;\n}","preventionTips":["Use only documented distribution-mode values in table properties.","Align table-creating engine versions with the Flink runtime.","Validate table properties at deploy time with a preflight loadTable()."],"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"}