{"record":{"id":"c903b049ca7c8054","repo":"apache/iceberg","slug":"unrecognized-write-distribution-mode-writemo","errorCode":null,"errorMessage":"Unrecognized ${WRITE_DISTRIBUTION_MODE}: ${writeMode}","messagePattern":"Unrecognized (.+?): (.+?)","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java","lineNumber":719,"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":701,"sourceCodeEnd":737,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java#L701-L737","documentation":"FlinkSink.distributeDataStream switches on the resolved DistributionMode for the write; if the mode is not NONE, HASH, or RANGE it throws a RuntimeException naming the WRITE_DISTRIBUTION_MODE property and the offending value. This is a defensive default-branch guard — the enum is parsed by FlinkWriteConf, so this only fires when a DistributionMode value escapes validation.","triggerScenarios":"Builder#distributionMode / write option 'write-distribution-mode' / FlinkWriteOptions.DISTRIBUTION_MODE resolves to a DistributionMode constant outside the three handled cases (no current user-facing value can); effectively only reachable via a newer enum constant running against older sink code, or custom DistributionMode injection.","commonSituations":"Version mismatch: a Flink job or SQL planner compiled against a newer Iceberg that added an enum value is run with an older iceberg-flink-runtime jar on the classpath; or fat-jar shading mixes two Iceberg versions so the parsed enum differs from the one the sink switch handles.","solutions":["Check for duplicate/mixed iceberg-flink-runtime versions on the classpath (mvn dependency:tree, shade config) and pin a single version","Align the Iceberg runtime jar version used by the cluster and the job (e.g. upgrade iceberg-flink-runtime to match your build)","Set write-distribution-mode explicitly to none|hash|range so a known-good value is used","If nothing changed, inspect the log line 'Write distribution mode is ...' to see the resolved mode and trace where it came from"],"exampleFix":"// before\nbuilder.set(\"write-distribution-mode\", \"batch-shuffle\"); // unrecognized\n// after\nbuilder.set(\"write-distribution-mode\", DistributionMode.HASH.modeName()); // \"hash\"","handlingStrategy":"validation","validationCode":"String mode = writeOptions.getOrDefault(\"write-distribution-mode\", \"none\");\nif (!Set.of(\"none\", \"hash\", \"range\").contains(mode.toLowerCase(Locale.ROOT))) {\n  throw new IllegalArgumentException(\"unsupported write-distribution-mode: \" + mode);\n}","typeGuard":null,"tryCatchPattern":"try {\n  sinkBuilder.append();\n} catch (RuntimeException e) {\n  if (e.getMessage() != null && e.getMessage().contains(\"Unrecognized\")) {\n    LOG.error(\"distribution mode rejected: {}\", e.getMessage());\n    // fall back to explicit hash mode\n  }\n  throw e;\n}","preventionTips":["Always set write-distribution-mode explicitly to none|hash|range","Keep iceberg-flink-runtime and compile-time Iceberg versions identical","Audit fat jars for duplicate org.apache.iceberg classes","Log the resolved DistributionMode before append() during upgrades"],"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"}