remove delta writers

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/IcebergSinkConfig.java

@@ -71,9 +71,6 @@ public class IcebergSinkConfig extends AbstractConfig {
   private static final String TABLES_DEFAULT_COMMIT_BRANCH = "iceberg.tables.default-commit-branch";
   private static final String TABLES_DEFAULT_ID_COLUMNS = "iceberg.tables.default-id-columns";
   private static final String TABLES_DEFAULT_PARTITION_BY = "iceberg.tables.default-partition-by";
-  private static final String TABLES_CDC_FIELD_PROP = "iceberg.tables.cdc-field";
-  private static final String TABLES_UPSERT_MODE_ENABLED_PROP =
-      "iceberg.tables.upsert-mode-enabled";
   private static final String TABLES_AUTO_CREATE_ENABLED_PROP =
       "iceberg.tables.auto-create-enabled";
   private static final String TABLES_EVOLVE_SCHEMA_ENABLED_PROP =
@@ -152,18 +149,6 @@ private static ConfigDef newConfigDef() {
         null,
         Importance.MEDIUM,
         "Default partition spec to use when creating tables, comma-separated");
-    configDef.define(
-        TABLES_CDC_FIELD_PROP,
-        ConfigDef.Type.STRING,
-        null,
-        Importance.MEDIUM,
-        "Source record field that identifies the type of operation (insert, update, or delete)");
-    configDef.define(
-        TABLES_UPSERT_MODE_ENABLED_PROP,
-        ConfigDef.Type.BOOLEAN,
-        false,
-        Importance.MEDIUM,
-        "Set to true to treat all appends as upserts, false otherwise");
     configDef.define(
         TABLES_AUTO_CREATE_ENABLED_PROP,
         ConfigDef.Type.BOOLEAN,
@@ -422,14 +407,6 @@ public String hadoopConfDir() {
     return getString(HADDOP_CONF_DIR_PROP);
   }
 
-  public String tablesCdcField() {
-    return getString(TABLES_CDC_FIELD_PROP);
-  }
-
-  public boolean upsertModeEnabled() {
-    return getBoolean(TABLES_UPSERT_MODE_ENABLED_PROP);
-  }
-
   public boolean autoCreateEnabled() {
     return getBoolean(TABLES_AUTO_CREATE_ENABLED_PROP);
   }

kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/IcebergWriter.java

@@ -60,13 +60,7 @@ public void write(SinkRecord record) {
       // ignore tombstones...
       if (record.value() != null) {
         Record row = convertToRow(record);
-        String cdcField = config.tablesCdcField();
-        if (cdcField == null) {
-          writer.write(row);
-        } else {
-          Operation op = extractCdcOperation(record.value(), cdcField);
-          writer.write(new RecordWrapper(row, op));
-        }
+        writer.write(row);
       }
     } catch (Exception e) {
       throw new DataException(
@@ -99,28 +93,6 @@ private Record convertToRow(SinkRecord record) {
     return row;
   }
 
-  private Operation extractCdcOperation(Object recordValue, String cdcField) {
-    Object opValue = Utilities.extractFromRecordValue(recordValue, cdcField);
-
-    if (opValue == null) {
-      return Operation.INSERT;
-    }
-
-    String opStr = opValue.toString().trim().toUpperCase();
-    if (opStr.isEmpty()) {
-      return Operation.INSERT;
-    }
-
-    switch (opStr.charAt(0)) {
-      case 'U':
-        return Operation.UPDATE;
-      case 'D':
-        return Operation.DELETE;
-      default:
-        return Operation.INSERT;
-    }
-  }
-
   private void flush() {
     WriteResult writeResult;
     try {