@Nayan Gondaliya joined #flink-cdc. They’re also new to Apache Flink.

Hi everyone I’m currently testing Apache Flink CDC

3.7.0-SNAPSHOT

using the Pipeline connector with: • Apache Flink

1.20

• Source: PostgreSQL • Sink: Apache Iceberg

1.9.1

on Amazon Web Services I’m seeing a few issues and wanted to check whether these are known limitations / known bugs, or if fixes are already planned / on the roadmap. ------- New tables are not discovered after restart from savepoint/checkpoint The pipeline discovers new tables correctly during normal runtime. However, after stopping the job and restarting from a savepoint/checkpoint, newly added matching tables are not discovered anymore. Question: Is table discovery state intentionally frozen/restored from the savepoint, or is this a known issue? ------- Schema reconciliation fails after restart if schema changed while job was stopped Scenario: 1. Stop job 2. Apply schema changes to source table 3. Restart job from savepoint The job fails during startup / schema reconciliation instead of evolving the schema and continuing. Question: Is this expected behavior currently, or should schema evolution handle this case? ------- Column rename behavior with Iceberg sink When a column is renamed in PostgreSQL: • Iceberg receives it as a new column (which I understand may be expected in lenient/evolve modes) • old data remains under the old field / does not populate the new renamed field as expected. I observed two scenarios: Scenario 1 (Pipeline running continuously) 1. Column is renamed in PostgreSQL 2. New column is added to sink (Iceberg) with schema change enabled 3. Old data remains under old column, new column receives only new data 4. Old field is removed manually from sink table 5. Subsequent DML operations do not populate the new field; it remains

NULL

Scenario 2 (Pipeline restarted after step 3) 1. Column is renamed in PostgreSQL 2. New column is added to sink (Iceberg) with schema change enabled 3. Old data remains under old column, new column receives only new data 4. Old field is removed manually from sink table 5. After restarting the pipeline, subsequent DML operations populate the new field correctly Questions: 1. Is rename currently treated only as add-column semantics for Iceberg sink? 2. Is there an internal field mapping cache/state that only refreshes after restart? 3. Is proper rename mapping / seamless evolution planned? ------- Would appreciate any guidance on whether these are: • known/open issues • expected current limitations • already fixed in newer snapshots • planned for future releases Thanks

Hi everyone I’m using Apache Flink CDC 3.4.0 (Flink 1.2) with the PostgreSQL source connector to replicate data from PostgreSQL into Apache Iceberg on AWS. I’m trying to use Debezium signaling to trigger ad-hoc/incremental snapshots of specific tables. I configured the source with:

--  java --
'debezium.signal.enabled.channels' = 'source',
'debezium.signal.data.collection' = 'public.debezium_signal'

and created the signal table in Postgres:

--- signal table ---

CREATE TABLE public.debezium_signal (
  id VARCHAR(42) PRIMARY KEY,
  type VARCHAR(32) NOT NULL,
  data VARCHAR(2048)
);

Then I insert signals like:

--- trigger adhoc snapshot --- 
INSERT INTO public.debezium_signal (id, type, data)
VALUES (
  '1',
  'execute-snapshot',
  '{"data-collections": ["public.my_table"], "type": "incremental"}'
);

My understanding is that Flink CDC embeds Debezium and passes

debezium.*

properties through, but Flink CDC 3.x also has its own incremental snapshot framework (split enumerator / chunk snapshotting / checkpoint-aware recovery). So I’m unsure whether Flink CDC 3.4.0 actually supports Debezium signaling semantics for triggering snapshots, or whether these configs are simply ignored / only partially honored. Has anyone successfully used Debezium signaling with Flink CDC PostgreSQL source in 3.4.x? Specifically: 1. Does Flink CDC read/process the signal table? 2. Can

execute-snapshot

trigger Flink CDC incremental snapshots? 3. Are there recommended alternatives for ad-hoc table re-snapshotting without restarting the job? Any guidance would be greatly appreciated.

@Toby C joined #flink-cdc. They’re also new to Apache Flink.

@Alkiad joined #flink-cdc. They’re also new to Apache Flink.

Hi good people, I'm developing a flink CDC data project for a production environment. However, I'm having trouble with the layout and configuration of directories to meet production standards. I'm looking for a GitHub repository that can solve this problem, set up a proper flink service for production, and ideally provide step-by-step instructions. If anyone knows of such a resource, please help me. I would greatly appreciate it. Thank you.

@Rishabh Kedia joined #flink-cdc. They’re also new to Apache Flink.

@邓钒 joined #flink-cdc. They’re also new to Apache Flink.

@Junbo Wang joined #flink-cdc. They’re also new to Apache Flink.

@Chengbing Liu joined #flink-cdc. They’re also new to Apache Flink.

Hey guys, I have raised one pr https://github.com/apache/flink-cdc/pull/4181, it had gotten initial comment, I have made some changes, The PR has gone a bit stale since then, would love if someone can review it again thanks!

Hi Flink CDC community, I opened a PR for FLINK-39797 to add a Fluss pipeline sink option

auto.create.table.enabled

. PR: https://github.com/apache/flink-cdc/pull/4420 JIRA: https://issues.apache.org/jira/browse/FLINK-39797 The option defaults to

true

to preserve the current behavior. When set to

false

, the Fluss sink will fail if the target table does not already exist, instead of creating it automatically. This is useful for users who want to manage Fluss table creation explicitly. Would appreciate your review when you have time. Thanks!

Hi Flink CDC dev, I have a question about a design choice of Iceberg CDC pipeline connector. From what I see flink-cdc has its own implementation of iceberg writer and not using flink iceberg connector implementation. May I ask what was the context and motivation behind this design choice? From what i see, the latest version of flink-iceberg has many features that can benefit CDC integration such as: • https://iceberg.apache.org/docs/1.10.2/flink-writes/ dynamic sink: multi tables write and schema evolution • https://iceberg.apache.org/docs/1.10.2/flink-maintenance/ inline table maintenance I feel like this can make the flink-cdc integration falling behind with flink iceberg new features, for example the table maintenance in flink-cdc iceberg is using an old api of table maintenance and flink-iceberg has an implementation with better performance. Do you think it makes sense to reuse the flink iceberg connector with flink-cdc? If so, I'd happy to make a testing POC for this

Hi Flink CDC de, I am using Flink 2.2 and Flink CDC 3.6.0 (Postgres). I am trying to take advantage of the FLIP-510 optimization to drop the

ChangelogNormalizer

from my Job Graph to save state memory. My Postgres table is configured with

REPLICA IDENTITY DEFAULT

, so deletes only contain the Primary Key. My pipeline is a pure 1:1 Upsert pipeline with no SQL transformations. Does the

flink-connector-postgres-cdc

currently support and propagate the new

keyOnlyDeletes(true)

flag (from FLIP-510) to the Flink planner?

Good morning, team! I'm using the Flink CDC Pipeline connector with MySQL as the source and Iceberg as the sink. I have a question regarding Iceberg equality delete files. Is there a recommended approach for managing them in production? I have

sink.compaction.enabled

enabled, but it doesn't seem to rewrite or eliminate these files. I also occasionally see a cryptic "Array index out of range" error that disappears when I disable compaction. Is it better to run a separate Flink job that periodically calls the Iceberg table maintenance APIs, or any other approach? Open to suggestions, thanks! Thanks in advance!

Hi everyone, I see that the convertCDCTypeToIcebergType doesn't handle

ARRAY

data types. Because of this 1. MySQL to Iceberg pipeline is not able to handle

SET

datatypes in MySQL. a. It mentions here that

SET

MySQL datatype is converted to

ARRAY<STRING>

Flink SQL data type. This should be mapped to a Iceberg datatype which is not currently handled. 2. Postgres to Iceberg pipeline is not able to handle

text[]

datatypes in Postgres. i. Similarly

text[]

Postgres datatype is converted to

ARRAY<STRING>

Flink SQL data type. This should be mapped to a Iceberg datatype which is not currently handled.

Caused by: org.apache.flink.cdc.common.exceptions.SchemaEvolveException: java.lang.IllegalArgumentException: Unsupported cdc type in iceberg: ARRAY<STRING>
	at org.apache.flink.cdc.connectors.iceberg.sink.IcebergMetadataApplier.applyCreateTable(IcebergMetadataApplier.java:238) ~[flink-cdc-pipeline-connector-iceberg-3.6.0.jar:3.6.0]
	at org.apache.flink.cdc.connectors.iceberg.sink.IcebergMetadataApplier.lambda$applySchemaChange$2(IcebergMetadataApplier.java:130) ~[flink-cdc-pipeline-connector-iceberg-3.6.0.jar:3.6.0]
	at org.apache.flink.cdc.common.event.visitor.SchemaChangeEventVisitor.visit(SchemaChangeEventVisitor.java:59) ~[flink-cdc-dist-3.6.0.jar:3.6.0]
	at org.apache.flink.cdc.connectors.iceberg.sink.IcebergMetadataApplier.applySchemaChange(IcebergMetadataApplier.java:119) ~[flink-cdc-pipeline-connector-iceberg-3.6.0.jar:3.6.0]
	at org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator.applyAndUpdateEvolvedSchemaChange(SchemaCoordinator.java:440) ~[flink-cdc-dist-3.6.0.jar:3.6.0]
	at org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator.applySchemaChange(SchemaCoordinator.java:409) ~[flink-cdc-dist-3.6.0.jar:3.6.0]
	at org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator.lambda$startSchemaChangesEvolve$1(SchemaCoordinator.java:250) ~[flink-cdc-dist-3.6.0.jar:3.6.0]
	... 5 more
Caused by: java.lang.IllegalArgumentException: Unsupported cdc type in iceberg: ARRAY<STRING>
	at org.apache.flink.cdc.connectors.iceberg.sink.utils.IcebergTypeUtils.convertCDCTypeToIcebergType(IcebergTypeUtils.java:164) ~[flink-cdc-pipeline-connector-iceberg-3.6.0.jar:3.6.0]
	at org.apache.flink.cdc.connectors.iceberg.sink.utils.IcebergTypeUtils.convertCdcColumnToIcebergField(IcebergTypeUtils.java:59) ~[flink-cdc-pipeline-connector-iceberg-3.6.0.jar:3.6.0]
	at org.apache.flink.cdc.connectors.iceberg.sink.IcebergMetadataApplier.applyCreateTable(IcebergMetadataApplier.java:171) ~[flink-cdc-pipeline-connector-iceberg-3.6.0.jar:3.6.0]
	at org.apache.flink.cdc.connectors.iceberg.sink.IcebergMetadataApplier.lambda$applySchemaChange$2(IcebergMetadataApplier.java:130) ~[flink-cdc-pipeline-connector-iceberg-3.6.0.jar:3.6.0]
	at org.apache.flink.cdc.common.event.visitor.SchemaChangeEventVisitor.visit(SchemaChangeEventVisitor.java:59) ~[flink-cdc-dist-3.6.0.jar:3.6.0]
	at org.apache.flink.cdc.connectors.iceberg.sink.IcebergMetadataApplier.applySchemaChange(IcebergMetadataApplier.java:119) ~[flink-cdc-pipeline-connector-iceberg-3.6.0.jar:3.6.0]
	at org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator.applyAndUpdateEvolvedSchemaChange(SchemaCoordinator.java:440) ~[flink-cdc-dist-3.6.0.jar:3.6.0]
	at org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator.applySchemaChange(SchemaCoordinator.java:409) ~[flink-cdc-dist-3.6.0.jar:3.6.0]
	at org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator.lambda$startSchemaChangesEvolve$1(SchemaCoordinator.java:250) ~[flink-cdc-dist-3.6.0.jar:3.6.0]
	... 5 more

Anyone observed this before? is this a known issue?

Hi,

2026-07-22 23:00:54,362 INFO  org.apache.flink.cdc.runtime.operators.schema.common.SchemaRegistry [] - Handling evolved schema request: GetEvolvedSchemaRequest{tableId=sbschema.dea_tarmac_flight, schemaVersion=-1}
2026-07-22 23:00:54,366 INFO  org.apache.flink.runtime.executiongraph.ExecutionGraph       [] - Transform:Data -> SchemaOperator -> PrePartition (1/1) (b9d8de1899282a1b56c00c6f3a1f7483_90bea66de1c231edf33913ecd54406c1_0_8) switched from RUNNING to FAILED on postgres-cdc-to-iceberg-flight-taskmanager-1-1 @ 241.140.95.178 (dataPort=42617).
java.io.IOException: Can't get next record for channel InputChannelInfo{gateIdx=0, inputChannelIdx=0}
	at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.emitNext(AbstractStreamTaskNetworkInput.java:106) ~[flink-dist-1.20.0.jar:1.20.0]
	at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65) ~[flink-dist-1.20.0.jar:1.20.0]
	at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:638) ~[flink-dist-1.20.0.jar:1.20.0]
	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:231) ~[flink-dist-1.20.0.jar:1.20.0]
	at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:973) ~[flink-dist-1.20.0.jar:1.20.0]
	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:917) ~[flink-dist-1.20.0.jar:1.20.0]
	at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:970) ~[flink-dist-1.20.0.jar:1.20.0]
	at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:949) ~[flink-dist-1.20.0.jar:1.20.0]
	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:763) ~[flink-dist-1.20.0.jar:1.20.0]
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:575) ~[flink-dist-1.20.0.jar:1.20.0]
	at java.lang.Thread.run(Unknown Source) ~[?:?]
Caused by: java.io.IOException: Serializer consumed more bytes than the record had. This indicates broken serialization. If you are using custom serialization types (Value or Writable), check their serialization methods. If you are using a Kryo-serialized type, check the corresponding Kryo serializer.
	at org.apache.flink.runtime.io.network.api.serialization.NonSpanningWrapper.readInto(NonSpanningWrapper.java:339) ~[flink-dist-1.20.0.jar:1.20.0]
	at org.apache.flink.runtime.io.network.api.serialization.SpillingAdaptiveSpanningRecordDeserializer.readNonSpanningRecord(SpillingAdaptiveSpanningRecordDeserializer.java:128) ~[flink-dist-1.20.0.jar:1.20.0]
	at org.apache.flink.runtime.io.network.api.serialization.SpillingAdaptiveSpanningRecordDeserializer.readNextRecord(SpillingAdaptiveSpanningRecordDeserializer.java:103) ~[flink-dist-1.20.0.jar:1.20.0]
	at org.apache.flink.runtime.io.network.api.serialization.SpillingAdaptiveSpanningRecordDeserializer.getNextRecord(SpillingAdaptiveSpanningRecordDeserializer.java:93) ~[flink-dist-1.20.0.jar:1.20.0]
	at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.emitNext(AbstractStreamTaskNetworkInput.java:103) ~[flink-dist-1.20.0.jar:1.20.0]
	... 10 more
Caused by: java.lang.IndexOutOfBoundsException
	at org.apache.flink.core.memory.MemorySegment.get(MemorySegment.java:389) ~[flink-dist-1.20.0.jar:1.20.0]
	at org.apache.flink.runtime.io.network.api.serialization.NonSpanningWrapper.readByte(NonSpanningWrapper.java:112) ~[flink-dist-1.20.0.jar:1.20.0]
	at org.apache.flink.runtime.io.network.api.serialization.NonSpanningWrapper.readBoolean(NonSpanningWrapper.java:107) ~[flink-dist-1.20.0.jar:1.20.0]
	at org.apache.flink.cdc.runtime.serializer.MapSerializer.deserialize(MapSerializer.java:152) ~[flink-cdc-dist-3.6.0.jar:3.6.0]
	at org.apache.flink.cdc.runtime.serializer.MapSerializer.deserialize(MapSerializer.java:43) ~[flink-cdc-dist-3.6.0.jar:3.6.0]
	at org.apache.flink.cdc.runtime.serializer.NullableSerializerWrapper.deserialize(NullableSerializerWrapper.java:91) ~[flink-cdc-dist-3.6.0.jar:3.6.0]
	at org.apache.flink.cdc.runtime.serializer.event.DataChangeEventSerializer.deserialize(DataChangeEventSerializer.java:95) ~[flink-cdc-dist-3.6.0.jar:3.6.0]
	at org.apache.flink.cdc.runtime.serializer.event.DataChangeEventSerializer.deserialize(DataChangeEventSerializer.java:40) ~[flink-cdc-dist-3.6.0.jar:3.6.0]
	at org.apache.flink.cdc.runtime.serializer.event.EventSerializer.deserialize(EventSerializer.java:123) ~[flink-cdc-dist-3.6.0.jar:3.6.0]
	at org.apache.flink.cdc.runtime.serializer.event.EventSerializer.deserialize(EventSerializer.java:39) ~[flink-cdc-dist-3.6.0.jar:3.6.0]
	at org.apache.flink.streaming.runtime.streamrecord.StreamElementSerializer.deserialize(StreamElementSerializer.java:206) ~[flink-dist-1.20.0.jar:1.20.0]
	at org.apache.flink.streaming.runtime.streamrecord.StreamElementSerializer.deserialize(StreamElementSerializer.java:44) ~[flink-dist-1.20.0.jar:1.20.0]
	at org.apache.flink.runtime.plugable.NonReusingDeserializationDelegate.read(NonReusingDeserializationDelegate.java:53) ~[flink-dist-1.20.0.jar:1.20.0]
	at org.apache.flink.runtime.io.network.api.serialization.NonSpanningWrapper.readInto(NonSpanningWrapper.java:337) ~[flink-dist-1.20.0.jar:1.20.0]
	at org.apache.flink.runtime.io.network.api.serialization.SpillingAdaptiveSpanningRecordDeserializer.readNonSpanningRecord(SpillingAdaptiveSpanningRecordDeserializer.java:128) ~[flink-dist-1.20.0.jar:1.20.0]
	at org.apache.flink.runtime.io.network.api.serialization.SpillingAdaptiveSpanningRecordDeserializer.readNextRecord(SpillingAdaptiveSpanningRecordDeserializer.java:103) ~[flink-dist-1.20.0.jar:1.20.0]
	at org.apache.flink.runtime.io.network.api.serialization.SpillingAdaptiveSpanningRecordDeserializer.getNextRecord(SpillingAdaptiveSpanningRecordDeserializer.java:93) ~[flink-dist-1.20.0.jar:1.20.0]
	at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.emitNext(AbstractStreamTaskNetworkInput.java:103) ~[flink-dist-1.20.0.jar:1.20.0]
	... 10 more

I am seeing with error when using Flink Postgres CDC. Anyone observed this issue before?

Hello, Anyone working on flinkcdc with oracle? What's your strategy to recover if the lag builds up during incremental phase? Ofcourse tuning log mining configs is the first approach. Do you have any other strategies? Appreciate your suggestions. Thanks

Hi 👋 We are using Flink CDC MySQL 3.6 and facing serialization issues caused by:

io.debezium.DebeziumException: Failed to deserialize data of EventHeaderV4

. We have several MySQL consumers working perfectly with Flink CDC, but one specific consumer refuses to work. We've tried various troubleshooting steps, including the official Flink recommended configurations for this issue, but haven't had any success. Has anyone run into this or found a fix?

Hi all, running into an issue with table discovery. Setup: • Flink CDC: 3.6.0 • Flink: 1.20 • Pipeline: MySQL CDC source to StarRocks sink (YAML pipeline) Issue: On job start, the source correctly discovers and captures all tables that already exist and match my

tables:

pattern. However, if a new table is created in MySQL after the job has started (while it's actively streaming), it's never picked up automatically, even though it matches the same pattern and would've been captured had it existed at startup. Confirmed workaround: restarting the job fresh does pick up the new table correctly. So the discovery logic itself works fine; it's specifically the "table created while streaming, without a restart" case that doesn't. What I need: a way for the newly created table to be picked up automatically without restarting the job. I tried enabling

scan.newly-added-table.enabled

along with several

debezium.*

schema history passthrough options, but still couldn't get it to work without a restart:

source:
  type: mysql
  hostname: mysql
  port: 3306
  username: root
  password: rootpassword
  tables: cdc_demo.users,cdc_demo.users_new
  server-id: 5400-5404
  scan.startup.mode: initial
  jdbc.properties.allowPublicKeyRetrieval: "true"
  jdbc.properties.useSSL: "false"
  # Use flat dot notation for Debezium properties:
  debezium.database.history.store.only.captured.tables.ddl: "false"
  debezium.database.history.store.only.captured.databases.ddl: "false"
  debezium.schema.history.internal.store.only.captured.tables.ddl: "false"
  debezium.include.schema.changes: "true"
  scan.newly-added-table.enabled: "true"

sink:
  type: starrocks
  jdbc-url: jdbc:<mysql://starrocks-fe:9030>
  load-url: starrocks-fe:8030
  username: root
  password: ""
  sink.buffer-flush.interval-ms: 1000

pipeline:
  name: MySQL_to_StarRocks_CDC
  parallelism: 1
  schema.change.behavior: evolve

Question: Is live (no restart) discovery of newly created tables supported today in 3.6.0, and if so, what am I missing? If it's genuinely not supported, I'd like to raise a feature/bug ticket for it. Is this the right place to flag that first, or should I go straight to JIRA? Original config for reference (MySQL to StarRocks):

source:
  type: mysql
  tables: cdc_demo.users,cdc_demo.users_new
  scan.startup.mode: initial
  server-id: 5400-5404

Thanks in advance!

Hey team! Ran into something interesting with checkpoint intervals on the CDC → Iceberg pipeline and wanted to get your take. Our cluster is configured to checkpoint every 30s in

docker-compose

. For this particular pipeline, I wanted a shorter interval, so I added:

execution.checkpointing.interval: 3s

under the

pipeline:

block in the pipeline YAML, assuming the job-level setting would override the cluster-level configuration as it normally does in Flink. However, that didn't happen—the job continued checkpointing every 30s regardless of the value in the YAML. After digging into it, it looks like this isn't really a case of the cluster configuration taking precedence. The Flink CDC pipeline YAML doesn't currently support

execution.checkpointing.interval

as a pipeline option, so the setting is simply ignored. The only way I was able to override it was by passing it through the CLI during submission:

-D execution.checkpointing.interval=3s

While looking into this, I also came across FLINK-34834, which proposes adding a

flink-config:

block to the pipeline YAML to allow pipeline-level overrides of

flink-conf.yaml

. That's essentially what I was trying to achieve here. For reference, our setup: • Flink 2.2.1 (Java 21) • Flink CDC 3.6.0 • MySQL Pipeline Connector 3.6.0-2 • MySQL 8.0 / Driver 8.4.0 • Iceberg 1.11.0 • Nessie 0.108.1 • Trino 482 Has anyone else run into this, or is there a recommended way to configure job-specific Flink properties directly through the CDC pipeline YAML?

Hey team! somebody know how to fix this error with Mysqlconnector? stackoverflow.com/questions/68482482/error-while-trying-to-run-connector-class-io-debezium-connector-mysql-mysqlconn

Hey all, I’m running into an issue. We’re using

flink-cdc-pipeline-connector-iceberg (3.6.0-2.2)

to stream MySQL → Iceberg on Glue. When the partition key column is updated, we’re getting duplicate rows instead of the row being moved to the new partition. Repro:

CREATE TABLE t (
  id BIGINT,
  val VARCHAR,
  region VARCHAR
) PARTITIONED BY identity(region);

INSERT id=1, region=north;

UPDATE id=1 SET region=east;

After the update,

id=1

appears twice: one row with

region=north

(old/stale) and one with

region=east

(new). From what I understand, this happens because the equality delete only uses the primary key and doesn’t include the partition column, so the delete doesn’t remove the row from the old partition. Has anyone solved this for partitioned tables? Is scheduled compaction the only practical workaround, or is there a write-side fix? Also, does

flink-cdc-pipeline-connector-iceberg

expose any configuration to set the equality fields? Any pointers would be appreciated.

Hello team anyone using oracle stream api based flink cdc pipeline ? I would like to understand your experience in production with it. Your response will be appreciated. Thanks .

Hi team, I have a question about the Flink CDC → Iceberg sink. Is there a way to periodically roll or flush Iceberg data files based on file size or record count, independent of the checkpoint interval? I want to increase the checkpoint interval to 2–3 minutes, but I don’t want all the data to remain buffered in the TaskManager until the checkpoint completes. Ideally, I’d like to have multiple files written within a single checkpoint based on size or record count. I tried the following properties, but they didn’t seem to have any effect:

sink:
  type: iceberg
  table.properties.write.target-file-size-bytes: "4194304"
  table.properties.write.parquet.row-group-size-bytes: "4194304"
  table.properties.write.parquet.row-group-check-max-record-count: "10000"

A large batch arriving within one checkpoint still gets written as a single file, with the file size depending on how much data arrived during that checkpoint. Does the Flink CDC pipeline Iceberg connector have any parameter to periodically flush or roll writer files by size or record count, independent of the checkpoint interval? Or is the checkpoint interval the only way to control the write frequency?

Hi everyone, Has anyone benchmarked memory usage for a MySQL → Flink CDC → Iceberg pipeline at scale? I’m trying to establish a baseline for estimating Flink TaskManager memory requirements per table, considering factors such as: • Number of columns and their data types • Peak CDC throughput (records/sec) I’m curious how others in production or large-scale deployments approach this. Do you use any rule of thumb or estimation model based on these metrics, or is load testing the only reliable way to determine memory requirements? If anyone has done similar benchmarking or has shared benchmark results, it would be great to hear about your approach and findings. Thanks!

Hi guys Excuse me, I don't understand this post👇. Could you help me? For example, how was it possible for the "drop columns" operation to occur without us noticing? @Leonard Xu @kunni

Hi everyone, I'm planning to introduce Flink MySQL CDC 3.6 into our company’s data platform for a MySQL → Kafka pipeline using the DataStream API. Our DB team runs pt-online-schema-change with

sql_log_bin=0

config for performance reasons, particularly to avoid generating binlog events for the table copy. This setting also applies to the shadow table’s DDL and the final

RENAME TABLE

. Application connections continue writing DML to the binlog normally. As a result, the connector does not receive the schema changes or the table swap. After the rename, it continues decoding row events using the old schema. In our PoC, adding a column in the middle of a table caused subsequent values to be mapped to incorrect columns. The job remained running without reporting a deserialization error, so this appears to present a risk of silent data corruption. We understand that scan.parse.online.schema.changes.enabled=true supports pt-osc by interpreting shadow-table schema changes and the subsequent swap. However, those DDL events are absent from the binlog in our environment. Is there someone who struggled with the same problem and could advise on the following? 1. Is there a supported way to detect and safely handle this scenario without logging the entire pt-osc copy operation? (I’m already aware of the

scan.parse.online.schema.changes.enabled

option, but my understanding is that it cannot handle this case because

sql_log_bin=0

prevents the shadow table’s DDL and the final

RENAME TABLE

from being written to the binlog. Please correct me if I’m mistaken.) 2. Can the connector use

TABLE_MAP_EVENT

metadata, potentially with

binlog_row_metadata=FULL

, to detect schema mismatches and refresh the schema, or at least fail rather than emit incorrectly mapped records? 3. If external notifications provide the DDL statement and binlog positions captured immediately before and after the rename, is there a supported mechanism to reconcile schema history and replay the affected events safely? These positions bracket the rename but do not identify its exact boundary. We would prefer to avoid maintaining a custom connector fork or synchronously blocking DB schema changes on a CDC acknowledgment. Any recommended configuration, operational approach, or related issue would be appreciated.

Hi @ksg980614, with

sql_log_bin=0

, pt-osc will not write any schema change related info into the binlog. Downstream CDC (and other binlog subscribers) may still receive data changes but interpret them using the old schema, causing errors or incorrect column mappings. I believe we need another approach to inform CDC that the schema has been changed, since it's not available from binlog. Sadly the only approach to recover from this situation is to re-snapshotting related tables.

Hi everyone, I ran into a checkpoint recovery issue with the Fluss pipeline source: if a referenced partition has been dropped or expired, the source may fail to resume consumption even though the table and other partitions still exist. I opened FLINK-41313 to track the issue and submitted a fix in PR #4608. Local validation passed on Flink 1.20.3 / Java 11 and Flink 2.2.0 / Java 17, including checkpoint recovery ITs after partition deletion with both

earliest

and

full

startup modes. Could a maintainer please approve the CI workflow run and help review the PR? Thanks!