[jira] [Created] (FLINK-38281) Schema changes still caused column size does not match the data size after enabling scan.parse.online.schema.changes.enabled

["TemiChan (Jira)" <ji...@apache.org>]

TemiChan created FLINK-38281:
--------------------------------

             Summary: Schema changes still caused column size does not match the data size after enabling scan.parse.online.schema.changes.enabled
                 Key: FLINK-38281
                 URL: https://issues.apache.org/jira/browse/FLINK-38281
             Project: Flink
          Issue Type: Improvement
          Components: Flink CDC
    Affects Versions: cdc-3.4.0, cdc-3.3.0
            Reporter: TemiChan


During the process of adding a field to the source table in CDC3.3 using pt-osc or gh-ost, if data is inserted into the source table, the error "column size does not match the data size" will be reported by CDC. However, if there is no data change in the source table during the online field addition process, and the data change is made to the source table after the field addition is completed by the tool, no error will be reported.

*1.while true insert data*
{{while}} {{{}true{}}}{{{};{}}}{{{}do{}}} {{mysql --login-path=dba doris_sync -Be }}{{{}"insert into test(ptest21,tcc10)select ptest21,tcc10 from test limit 10"{}}}{{{};done;{}}}

*{{2.add column}}*
{{{}pt-online-schema-change  {-}{{-}}charset=utf8mb4 {-}{{-}}recursion-method=none -{-}no-version-check --skip-check-slave-lag  --user=root --password={-}{}}}{{{}"xx"{}}} {{-host=127.0.0.1-}} {{{}u=chenqimi,P={}}}{{{}3006{}}}{{{},D=doris_sync,t={}}}{{{}test{}}}{{{},A=utf8mb4 --alter {{}}}}{{{}"add column tax_code VARCHAR(30) NULL "{}}} {{-execute}}

*3.flinkcdc  error*
{{java.lang.IllegalStateException: Column size does not match the data size}}
{{    }}{{{}at org.apache.flink.cdc.common.utils.Preconditions.checkState(Preconditions.java:{}}}{{{}161{}}}{{{}){}}}
{{    }}{{{}at org.apache.flink.cdc.connectors.doris.sink.DorisEventSerializer.serializerRecord(DorisEventSerializer.java:{}}}{{{}121{}}}{{{}){}}}
{{    }}{{{}at org.apache.flink.cdc.connectors.doris.sink.DorisEventSerializer.applyDataChangeEvent(DorisEventSerializer.java:{}}}{{{}100{}}}{{{}){}}}
{{    }}{{{}at org.apache.flink.cdc.connectors.doris.sink.DorisEventSerializer.serialize(DorisEventSerializer.java:{}}}{{{}71{}}}{{{}){}}}
{{    }}{{{}at org.apache.flink.cdc.connectors.doris.sink.DorisEventSerializer.serialize(DorisEventSerializer.java:{}}}{{{}49{}}}{{{}){}}}
{{    }}{{{}at org.apache.doris.flink.sink.batch.DorisBatchWriter.write(DorisBatchWriter.java:{}}}{{{}120{}}}{{{}){}}}
{{    }}{{{}at org.apache.flink.streaming.runtime.operators.sink.SinkWriterOperator.processElement(SinkWriterOperator.java:{}}}{{{}160{}}}{{{}){}}}
{{    }}{{{}at org.apache.flink.cdc.runtime.operators.sink.DataSinkWriterOperator.processElement(DataSinkWriterOperator.java:{}}}{{{}178{}}}{{{}){}}}
{{    }}{{{}at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:{}}}{{{}75{}}}{{{}){}}}
{{    }}{{{}at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:{}}}{{{}50{}}}{{{}){}}}
{{    }}{{{}at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:{}}}{{{}29{}}}{{{}){}}}
{{    }}{{{}at org.apache.flink.streaming.api.operators.StreamMap.processElement(StreamMap.java:{}}}{{{}38{}}}{{{}){}}}
{{    }}{{{}at org.apache.flink.streaming.runtime.tasks.OneInputStreamTask$StreamTaskNetworkOutput.emitRecord(OneInputStreamTask.java:{}}}{{{}238{}}}{{{}){}}}
{{    }}{{{}at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.processElement(AbstractStreamTaskNetworkInput.java:{}}}{{{}157{}}}{{{}){}}}
{{    }}{{{}at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.emitNext(AbstractStreamTaskNetworkInput.java:{}}}{{{}114{}}}{{{}){}}}
{{    }}{{{}at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:{}}}{{{}65{}}}{{{}){}}}
{{    }}{{{}at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:{}}}{{{}579{}}}{{{}){}}}
{{    }}{{{}at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:{}}}{{{}231{}}}{{{}){}}}
{{    }}{{{}at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:{}}}{{{}909{}}}{{{}){}}}
{{    }}{{{}at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:{}}}{{{}858{}}}{{{}){}}}
{{    }}{{{}at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:{}}}{{{}958{}}}{{{}){}}}
{{    }}{{{}at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:{}}}{{{}937{}}}{{{}){}}}
{{    }}{{{}at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:{}}}{{{}751{}}}{{{}){}}}
{{    }}{{{}at org.apache.flink.runtime.taskmanager.Task.run(Task.java:{}}}{{{}566{}}}{{{}){}}}
{{    }}{{{}at java.lang.Thread.run(Thread.java:{}}}{{{}748{}}}{{{}){}}}

 



--
This message was sent by Atlassian Jira
(v8.20.10#820010)