bryanck commented on code in PR #9641:
URL: https://github.com/apache/iceberg/pull/9641#discussion_r1506868255
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/data/IcebergWriter.java:
##########
@@ -77,8 +78,47 @@ public void write(SinkRecord record) {
}
private Record convertToRow(SinkRecord record) {
- // FIXME: update this when the record converter is added
- return null;
+ if (!config.evolveSchemaEnabled()) {
+ return recordConverter.convert(record.value());
+ }
+
+ SchemaUpdate.Consumer updates = new SchemaUpdate.Consumer();
+ Record row = recordConverter.convert(record.value(), updates);
+
+ if (!updates.empty()) {
+ // complete the current file
+ flush();
+ // apply the schema updates, this will refresh the table
+ SchemaUtils.applySchemaUpdates(table, updates);
+ // initialize a new writer with the new schema
+ initNewWriter();
+ // convert the row again, this time using the new table schema
+ row = recordConverter.convert(record.value(), null);
+ }
+
+ 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;
+ }
Review Comment:
This code has been removed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@iceberg.apache.org
For additional commands, e-mail: issues-h...@iceberg.apache.org
