Upgrade path from 0.8.2.Final to 0.9.0.Final

[gilesrapkinb]

Hello,

What is the upgrade path for 0.9.0 for iceberg tables that have been created using version 0.8.2? We are having some issues with the handling of date/time types when trying to upgrade from 0.8.2 to 0.9.0.

From the release notes I can see that handling of date/time fields has changed, as have some of the default settings.

• Improved Data Type Handling:
  • Enhanced data type handling now supports datetime, timestamp, and date formats. (When time.precision.mode is not dynamic) The connector accurately converts these data types to their Iceberg equivalents.
    ...
    NOTE: Please be aware that these modifications may affect backward compatibility. Thorough testing is strongly recommended before deployment.

In our test environment on startup of 0.9.0, without changing any of the new default settings we do see some compatibility issues with tables that were previously created by 0.8.2. It appears to be trying to change the field type in the iceberg tables:

java.lang.IllegalArgumentException: Cannot change column type: last_active: long -> timestamp
java.lang.IllegalArgumentException: Cannot change column type: deleted_at: string -> timestamptz

Is there any way to configure 0.9.0 to be backward compatible with tables written by older versions? We have some quite large tables (1 billion+ rows), and having to backfill them again is not a viable option for us.

[ismailsimsek]

@gilesrapkinb version 0.9 has better data type handling, with temporal types.

which mode are you running upsert, append? could you share full stack trace to see full detail? and your config (after removing sensitive values)

Regarding adding backward compatibility, I propose the following:

• Introduce a configuration variable that allows users to enable or disable it.
• Utilize this configuration setting to determine the schema, vaue conversion logic (for example in the code below).

debezium-server-iceberg/debezium-server-iceberg-sink/src/main/java/io/debezium/server/iceberg/SchemaConverter.java

Lines 236 to 245 in 14686ad

	case "string":
	return switch (fieldTypeName) {
	case "io.debezium.data.Uuid" -> Types.UUIDType.get();
	case "io.debezium.time.IsoDate" -> Types.DateType.get();
	case "io.debezium.time.IsoTimestamp" -> Types.TimestampType.withoutZone();
	case "io.debezium.time.ZonedTimestamp" -> Types.TimestampType.withZone();
	// NOTE: Time type is disable for the moment, it's not supported by spark
	// case "io.debezium.time.IsoTime" -> Types.TimeType.get();
	// case "io.debezium.time.ZonedTime" -> Types.TimeType.get();
	default -> Types.StringType.get();

[gilesrapkinb]

@ismailsimsek we're running in append mode.

Here's the stack trace from one of the error logs:

{
  "timestamp": "2025-05-16T15:10:22.168266663Z",
  "sequence": 769,
  "loggerClassName": "org.slf4j.impl.Slf4jLogger",
  "loggerName": "io.debezium.embedded.async.AsyncEmbeddedEngine",
  "level": "ERROR",
  "message": "Engine has failed with ",
  "threadName": "pool-8-thread-1",
  "threadId": 43,
  "mdc": {},
  "ndc": "",
  "hostName": "debezium-server-core-54ff6664bc-f5l8q",
  "processName": "/opt/java/openjdk/bin/java",
  "processId": 1,
  "exception": {
    "refId": 1,
    "exceptionType": "java.util.concurrent.ExecutionException",
    "message": "java.lang.IllegalArgumentException: Cannot change column type: deleted_at: string -> timestamptz",
    "frames": [
      {
        "class": "java.util.concurrent.FutureTask",
        "method": "report"
      },
      {
        "class": "java.util.concurrent.FutureTask",
        "method": "get"
      },
      {
        "class": "io.debezium.embedded.async.AsyncEmbeddedEngine",
        "method": "runTasksPolling",
        "line": 496
      },
      {
        "class": "io.debezium.embedded.async.AsyncEmbeddedEngine",
        "method": "run",
        "line": 220
      },
      {
        "class": "io.debezium.server.DebeziumServer",
        "method": "lambda$start$1",
        "line": 182
      },
      {
        "class": "java.util.concurrent.ThreadPoolExecutor",
        "method": "runWorker"
      },
      {
        "class": "java.util.concurrent.ThreadPoolExecutor$Worker",
        "method": "run"
      },
      {
        "class": "java.lang.Thread",
        "method": "run"
      }
    ],
    "causedBy": {
      "exception": {
        "refId": 2,
        "exceptionType": "java.lang.IllegalArgumentException",
        "message": "Cannot change column type: deleted_at: string -> timestamptz",
        "frames": [
          {
            "class": "org.apache.iceberg.relocated.com.google.common.base.Preconditions",
            "method": "checkArgument",
            "line": 464
          },
          {
            "class": "org.apache.iceberg.SchemaUpdate",
            "method": "updateColumn",
            "line": 287
          },
          {
            "class": "org.apache.iceberg.schema.UnionByNameVisitor",
            "method": "updateColumn",
            "line": 175
          },
          {
            "class": "org.apache.iceberg.schema.UnionByNameVisitor",
            "method": "lambda$struct$0",
            "line": 99
          },
          {
            "class": "java.util.stream.Streams$RangeIntSpliterator",
            "method": "forEachRemaining"
          },
          {
            "class": "java.util.stream.IntPipeline$Head",
            "method": "forEach"
          },
          {
            "class": "org.apache.iceberg.schema.UnionByNameVisitor",
            "method": "struct",
            "line": 88
          },
          {
            "class": "org.apache.iceberg.schema.UnionByNameVisitor",
            "method": "struct",
            "line": 34
          },
          {
            "class": "org.apache.iceberg.schema.SchemaWithPartnerVisitor",
            "method": "visit",
            "line": 68
          },
          {
            "class": "org.apache.iceberg.schema.SchemaWithPartnerVisitor",
            "method": "visit",
            "line": 45
          },
          {
            "class": "org.apache.iceberg.schema.UnionByNameVisitor",
            "method": "visit",
            "line": 71
          },
          {
            "class": "org.apache.iceberg.SchemaUpdate",
            "method": "unionByNameWith",
            "line": 373
          },
          {
            "class": "io.debezium.server.iceberg.tableoperator.IcebergTableOperator",
            "method": "applyFieldAddition",
            "line": 120
          },
          {
            "class": "io.debezium.server.iceberg.tableoperator.IcebergTableOperator",
            "method": "addToTable",
            "line": 161
          },
          {
            "class": "io.debezium.server.iceberg.IcebergChangeConsumer",
            "method": "handleBatch",
            "line": 122
          },
          {
            "class": "io.debezium.embedded.async.ParallelSmtAndConvertBatchProcessor",
            "method": "processRecords",
            "line": 56
          },
          {
            "class": "io.debezium.embedded.async.AsyncEmbeddedEngine$PollRecords",
            "method": "doCall",
            "line": 1203
          },
          {
            "class": "io.debezium.embedded.async.AsyncEmbeddedEngine$PollRecords",
            "method": "doCall",
            "line": 1183
          },
          {
            "class": "io.debezium.embedded.async.RetryingCallable",
            "method": "call",
            "line": 47
          },
          {
            "class": "java.util.concurrent.FutureTask",
            "method": "run"
          },
          {
            "class": "java.util.concurrent.Executors$RunnableAdapter",
            "method": "call"
          },
          {
            "class": "java.util.concurrent.FutureTask",
            "method": "run"
          },
          {
            "class": "java.util.concurrent.ThreadPoolExecutor",
            "method": "runWorker"
          },
          {
            "class": "java.util.concurrent.ThreadPoolExecutor$Worker",
            "method": "run"
          },
          {
            "class": "java.lang.Thread",
            "method": "run"
          }
        ]
      }
    }
  }
}

..and our debezium config

debezium.format.key = json
debezium.format.key.schemas.enable = true
debezium.format.value = json
debezium.format.value.schemas.enable = true
debezium.sink.batch.batch-size-wait = MaxBatchSizeWait
debezium.sink.batch.batch-size-wait.max-wait-ms = 30000
debezium.sink.batch.batch-size-wait.wait-interval-ms = 5000
debezium.sink.iceberg.allow-field-addition = true
debezium.sink.iceberg.catalog-impl = org.apache.iceberg.aws.glue.GlueCatalog
debezium.sink.iceberg.catalog-name = REDACTED
debezium.sink.iceberg.create-identifier-fields = true
debezium.sink.iceberg.io-impl = org.apache.iceberg.aws.s3.S3FileIO
debezium.sink.iceberg.s3.multipart.num-threads = 5
debezium.sink.iceberg.s3.path-style-access = true
debezium.sink.iceberg.s3.sse.key = REDACTED
debezium.sink.iceberg.s3.sse.type = kms
debezium.sink.iceberg.table-namespace = REDACTED
debezium.sink.iceberg.table.prefix = core
debezium.sink.iceberg.upsert = false
debezium.sink.iceberg.upsert-dedup-column = __source_ts_ms
debezium.sink.iceberg.warehouse = REDACTED
debezium.sink.iceberg.write.format.default = parquet
debezium.sink.type = iceberg
debezium.source.connector.class = io.debezium.connector.mysql.MySqlConnector
debezium.source.database.hostname = REDACTED
debezium.source.database.include.list = REDACTED
debezium.source.database.password = REDACTED
debezium.source.database.port = 3306
debezium.source.database.server.id = 10198
debezium.source.database.server.name = iceberg
debezium.source.database.user = REDACTED
debezium.source.include.schema.changes = false
debezium.source.incremental.snapshot.allow.schema.changes = true
debezium.source.incremental.snapshot.chunk.size = 10240
debezium.source.max.batch.size = 10240
debezium.source.max.queue.size = 40960
debezium.source.offset.flush.interval.ms = 0
debezium.source.offset.storage = org.apache.kafka.connect.storage.FileOffsetBackingStore
debezium.source.offset.storage.file.filename = /data/offset.dat
debezium.source.record.processing.shutdown.timeout.ms = 70000
debezium.source.schema.history.internal = io.debezium.storage.file.history.FileSchemaHistory
debezium.source.schema.history.internal.file.filename = /data/schemahistory.dat
debezium.source.schema.history.internal.skip.unparseable.ddl = true
debezium.source.signal.data.collection = REDACTED
debezium.source.snapshot.locking.mode = none
debezium.source.snapshot.mode = no_data
debezium.source.task.management.timeout.ms = 280000
debezium.source.topic.prefix = core
debezium.table.exclude.list = REDACTED
debezium.transforms = unwrap
debezium.transforms.unwrap.add.fields = op,table,source.ts_ms,db,ts_ms
debezium.transforms.unwrap.delete.handling.mode = rewrite
debezium.transforms.unwrap.delete.tombstone.handling.mode = rewrite
debezium.transforms.unwrap.drop.tombstones = true
debezium.transforms.unwrap.type = io.debezium.transforms.ExtractNewRecordState
quarkus.log.console.json = true
quarkus.log.level = INFO

[ismailsimsek]

one idea:
We could create variable (example debezium.sink.iceberg.schema-handling-version) to handle backward compatibility.

0.8 could be version 1
0.9 could be version 2
1.0 could be version 3

We could then use this version number to conditionally enable relevant code sections.

[ismailsimsek]

@gilesrapkinb The recommended migration process is documented here: https://memiiso.github.io/debezium-server-iceberg/migration/

If you run into any issues or have questions while going through the documentation, please don't hesitate to reach out.

there is also full test demonstrating it here

[github-actions]

This issue has been automatically marked as stale because it has been open for 180 days with no activity. It will be closed in next 14 days if no further activity occurs. To permanently prevent this issue from being considered stale, add the label 'not-stale', but commenting on the issue is preferred when possible.

[github-actions]

This issue has been closed because it has not received any activity in the last 14 days since being marked as 'stale'