{"record":{"id":"0bc99051ba39af51","repo":"apache/iceberg","slug":"unknown-row-kind-row-getrowkind","errorCode":null,"errorMessage":"Unknown row kind: <row.getRowKind()>","messagePattern":"Unknown row kind: <row\\.getRowKind\\(\\)>","errorType":"exception","errorClass":"UnsupportedOperationException","httpStatus":null,"severity":"error","filePath":"flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/BaseDeltaTaskWriter.java","lineNumber":109,"sourceCode":"        break;\n\n      case UPDATE_BEFORE:\n        if (upsert) {\n          break; // UPDATE_BEFORE is not necessary for UPSERT, we do nothing to prevent delete one\n          // row twice\n        }\n        writer.delete(row);\n        break;\n      case DELETE:\n        if (upsert) {\n          writer.deleteKey(keyProjection.wrap(row));\n        } else {\n          writer.delete(row);\n        }\n        break;\n\n      default:\n        throw new UnsupportedOperationException(\"Unknown row kind: \" + row.getRowKind());\n    }\n  }\n\n  protected class RowDataDeltaWriter extends BaseEqualityDeltaWriter {\n    RowDataDeltaWriter(PartitionKey partition, PartitioningDVWriter<RowData> dvFileWriter) {\n      super(partition, schema, deleteSchema, DeleteGranularity.FILE, dvFileWriter);\n    }\n\n    @Override\n    protected StructLike asStructLike(RowData data) {\n      return wrapper.wrap(data);\n    }\n\n    @Override\n    protected StructLike asStructLikeKey(RowData data) {\n      return keyWrapper.wrap(data);\n    }\n  }","sourceCodeStart":91,"sourceCodeEnd":127,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/BaseDeltaTaskWriter.java#L91-L127","documentation":"BaseDeltaTaskWriter.write dispatches RowData by RowKind: INSERT goes to the upsert/insert writer, DELETE (and UPDATE_BEFORE in some paths) to the delete writer. Any other RowKind (only UPDATE_AFTER remains) reaches the default arm and throws UnsupportedOperationException. Flink change streams must be normalized to INSERT/DELETE before writing to Iceberg.","triggerScenarios":"Writing a RowData with RowKind.UPDATE_AFTER (or any kind other than INSERT/DELETE/UPDATE_BEFORE handled upstream) into an Iceberg delta sink that expects a delete/insert-only changelog.","commonSituations":"Feeding an upsert changelog stream (from CDC sources like Debezium/Kafka with toRetractStream semantics kept as update kinds) directly into the Iceberg sink instead of retracting to INSERT/DELETE; custom pipeline emitting UPDATE_AFTER rows.","solutions":["Convert the changelog stream to retract semantics with toRetractStream so only INSERT and DELETE kinds reach the sink.","For CDC sources, drop UPDATE_BEFORE and map UPDATE_AFTER to INSERT (e.g. via a map that sets RowKind.INSERT).","Verify upstream operators are not emitting UPDATE_AFTER into the Iceberg sink."],"exampleFix":"// before\nstream.map(row -> row); // RowData with RowKind.UPDATE_AFTER reaches sink\n// after\nDataStream<RowData> normalized = stream.map(row -> {\n  if (row.getRowKind() == RowKind.UPDATE_AFTER) { row.setRowKind(RowKind.INSERT); }\n  return row;\n});","handlingStrategy":"validation","validationCode":"if (row.getRowKind() == RowKind.UPDATE_AFTER || row.getRowKind() == RowKind.UPDATE_BEFORE) { throw new IllegalArgumentException(\"Normalize changelog before Iceberg sink: \" + row.getRowKind()); }","typeGuard":"boolean isSinkCompatible(RowData row) { RowKind k = row.getRowKind(); return k == RowKind.INSERT || k == RowKind.DELETE; }","tryCatchPattern":"try { writer.write(row); } catch (UnsupportedOperationException e) { if (e.getMessage().startsWith(\"Unknown row kind\")) { LOG.error(\"Changelog not normalized; use toRetractStream\", e); throw e; } }","preventionTips":["Use statementset/toRetractStream so only INSERT and DELETE rows reach the sink","For CDC, collapse UPDATE_BEFORE/UPDATE_AFTER pairs before the sink","Add a validation operator before the Iceberg sink in custom pipelines"],"tags":["flink","cdc","rowkind","sink"],"backgroundTag":"unsupported-operation","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"}