{"record":{"id":"8b8d139e28a28bc8","repo":"apache/iceberg","slug":"unrecognized-write-distribution-mode-mode","errorCode":null,"errorMessage":"Unrecognized ${WRITE_DISTRIBUTION_MODE}: ${mode}","messagePattern":"Unrecognized (.+?): (.+?)","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"flink/v1.20/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/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java#L965-L1001","documentation":"IcebergSink.distributeDataStream switches on the DistributionMode resolved from write config; the default branch throws RuntimeException 'Unrecognized WRITE_DISTRIBUTION_MODE: <mode>' for any value other than NONE, HASH, or RANGE. Like the FlinkSink equivalent, it is a defensive guard against an unhandled enum constant, typically caused by classpath/version skew.","triggerScenarios":"write.distribution-mode (FlinkWriteOptions.DISTRIBUTION_MODE / FlinkWriteConf) resolving to a DistributionMode constant not covered by the switch — practically only via mixed Iceberg versions or a future enum value run against this older sink code.","commonSituations":"iceberg-flink-runtime jar on the cluster differs from the one the job was compiled against; shaded dependencies bundling a second Iceberg; copying config from a newer Iceberg release introducing a new mode name.","solutions":["Set the mode explicitly to a supported value: none, hash, or range (via FlinkWriteOptions.DISTRIBUTION_MODE or write options)","Ensure one consistent iceberg-flink-runtime version across job jar and cluster classpath","Check the sink's log line 'Write distribution mode is ...' to confirm what value was parsed and from which config source","Upgrade or downgrade the runtime jar so the parsed enum matches the sink's supported set"],"exampleFix":"// before\nFlinkWriteOptions.DISTRIBUTION_MODE.key() -> \"adaptive\"  // newer mode, old sink\n// after\nMap<String, String> opts = Map.of(\"write-distribution-mode\", \"hash\");\nIcebergSink.forRowData(input).writeOptions(opts).tableLoader(loader).append();","handlingStrategy":"validation","validationCode":"DistributionMode mode = flinkWriteConf.distributionMode();\nif (mode != DistributionMode.NONE && mode != DistributionMode.HASH && mode != DistributionMode.RANGE) {\n  throw new IllegalArgumentException(\"unsupported distribution mode: \" + mode);\n}","typeGuard":null,"tryCatchPattern":"try {\n  sink.append();\n} catch (RuntimeException e) {\n  if (String.valueOf(e.getMessage()).contains(\"Unrecognized\" + \" \" + FlinkWriteOptions.DISTRIBUTION_MODE.key())) {\n    LOG.error(\"bad distribution mode resolved; check write options and jar versions\");\n  }\n  throw e;\n}","preventionTips":["Set DISTRIBUTION_MODE write option explicitly to a known value","Keep one iceberg-flink-runtime version across cluster and job","After upgrades, confirm the log line 'Write distribution mode is ...' shows a supported mode","Avoid copying write.distribution-mode values from newer Iceberg docs into older deployments"],"tags":["flink","config","enum","iceberg"],"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"}