{"record":{"id":"7c6dc462b8464e08","repo":"apache/iceberg","slug":"writersink-is-used-only-for-writing-committing-is","errorCode":null,"errorMessage":"WriterSink is used only for writing; committing is handled by the main sink","messagePattern":"WriterSink is used only for writing; committing is handled by the main sink","errorType":"exception","errorClass":"UnsupportedOperationException","httpStatus":null,"severity":"error","filePath":"flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java","lineNumber":221,"sourceCode":"      this.cacheMaximumSize = cacheMaximumSize;\n    }\n\n    @SuppressWarnings(\"deprecation\")\n    @Override\n    public SinkWriter<DynamicRecordInternal> createWriter(InitContext context) throws IOException {\n      return new DynamicWriter(\n          catalogLoader.loadCatalog(),\n          writeProperties,\n          flinkConfig,\n          cacheMaximumSize,\n          new DynamicWriterMetrics(context.metricGroup()),\n          context.getSubtaskId(),\n          context.getAttemptNumber());\n    }\n\n    @Override\n    public Committer<DynamicWriteResult> createCommitter(CommitterInitContext context) {\n      throw new UnsupportedOperationException(\n          \"WriterSink is used only for writing; committing is handled by the main sink\");\n    }\n\n    @Override\n    public SimpleVersionedSerializer<DynamicWriteResult> getCommittableSerializer() {\n      return new DynamicWriteResultSerializer();\n    }\n  }\n\n  public static class Builder<T> {\n    private DataStream<T> input;\n    private DynamicRecordGenerator<T> generator;\n    private CatalogLoader catalogLoader;\n    private String uidPrefix = null;\n    private final Map<String, String> writeOptions = Maps.newHashMap();\n    private final Map<String, String> snapshotSummary = Maps.newHashMap();\n    private ReadableConfig readableConfig = new Configuration();\n    private TableCreator tableCreator = TableCreator.DEFAULT;","sourceCodeStart":203,"sourceCodeEnd":239,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java#L203-L239","documentation":"In the dynamic sink's split writer/follower topology, the WriterSink branch only produces write results; committing is the responsibility of the main sink operator. Its FlinkSinkBuilder/createCommitter hook therefore unconditionally throws UnsupportedOperationException if a committer is requested. This is a programming-model guard: constructing a committer on the writer branch is never valid.","triggerScenarios":"Calling createCommitter() on the WriterSink's write-branch operator, e.g. by configuring the sink so the framework tries to instantiate a committer for the writing subtasks, or custom code invoking the builder's committer hook on this branch.","commonSituations":"Misconfiguring the DynamicIcebergSink topology so the writer subtasks are also asked to commit; upgrading Flink sink APIs where committer placement changed and generic glue code calls createCommitter on all operators.","solutions":["Do not attach a committer to the WriterSink branch; let the main sink operator perform commits","Check sink topology setup code so committer construction is only invoked for the main sink","If using sink v2 APIs, ensure the write-branch uses only the writer/(writer sink) and the committable flow is wired to the dedicated committer operator"],"exampleFix":null,"handlingStrategy":"validation","validationCode":"// Configure the sink so committer creation only happens for the main sink operator,\n// never for the WriterSink write branch.","typeGuard":null,"tryCatchPattern":"try {\n  committer = writerSink.createCommitter(ctx);\n} catch (UnsupportedOperationException e) {\n  // wrong branch: use the main sink's committer instead\n}","preventionTips":["Wire commitables to the dedicated main sink committer operator","Do not reuse generic sink-builder code that instantiates committers on every operator","Review sink topology after Flink sink API upgrades"],"tags":["flink","sink-topology","committer"],"backgroundTag":"unsupported-operation","analyzedSha":"86d9c8fc543e7c56c9f624eb725f76c9baff9570","analyzedAt":"2026-09-12T00:46:39.097Z","contentChangedAt":"2026-09-12T00:46:39.097Z","schemaVersion":2},"datasetVersion":"2026-09-23T08:17:48.524Z"}