{"record":{"id":"c6eeaa6c7cc004aa","repo":"apache/iceberg","slug":"unrecognized","errorCode":null,"errorMessage":"Unrecognized : ","messagePattern":"Unrecognized : ","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"flink/v2.1/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.1/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java#L703-L739","documentation":"When building the sink, FlinkSink.distributeDataStream() applies a distribution transformation keyed by the table's write.distribution-mode (NONE/HASH). A writeMode value outside the supported set (e.g. an unrecognized string or a future mode like RANGE in a build lacking support) hits the default branch and throws RuntimeException.","triggerScenarios":"Setting table property write.distribution-mode to an unsupported/typo'd value (e.g. 'range' with a build that doesn't support it, or 'hash ' with stray characters) before constructing FlinkSink with keyed output.","commonSituations":"Hand-editing table properties; copying distribution-mode settings from Spark docs into a Flink table where only none/hash are handled; case/format mistakes like NONE vs none depending on parse path.","solutions":["Set write.distribution-mode to a supported value: none or hash (for this code path).","Correct typos/whitespace in the table property: ALTER TABLE ... SET TBLPROPERTIES ('write.distribution-mode'='hash').","Remove the property entirely to fall back to the default behavior if no specific distribution is needed.","Upgrade Iceberg if a newer distribution mode (e.g. range) support is required for Flink."],"exampleFix":"// before\nALTER TABLE t SET TBLPROPERTIES ('write.distribution-mode'='range');\n// after\nALTER TABLE t SET TBLPROPERTIES ('write.distribution-mode'='hash');","handlingStrategy":"validation","validationCode":"String mode = table.properties().getOrDefault(TableProperties.WRITE_DISTRIBUTION_MODE, TableProperties.WRITE_DISTRIBUTION_MODE_NONE);\nif (!mode.equals(\"none\") && !mode.equals(\"hash\")) {\n  throw new IllegalArgumentException(\"write.distribution-mode must be none or hash for Flink sink, got: \" + mode);\n}","typeGuard":null,"tryCatchPattern":"try { buildSink(); } catch (RuntimeException e) {\n  if (e.getMessage() != null && e.getMessage().contains(WRITE_DISTRIBUTION_MODE)) {\n    LOG.error(\"Fix write.distribution-mode table property\", e);\n  }\n}","preventionTips":["Only set write.distribution-mode to values supported by your Iceberg Flink version (none, hash).","Avoid copying Spark-oriented table properties into Flink-consumed tables.","Read the property back after setting it to catch typos and whitespace."],"tags":["flink","iceberg","config","distribution-mode","sink"],"backgroundTag":"invalid-config-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"}