/developing-applications-on-managed-service-for-apache-flink

Flink CDC Connector Guide

This guide covers Change Data Capture (CDC) connector configuration for Apache Flink applications on Amazon Managed Service for Apache Flink. Flink CDC enables reading snapshot and incremental change data from databases without requiring Kafka or Kafka Connect — Debezium runs embedded within the Flink application.

Overview

Flink CDC is a streaming data integration tool built on Apache Flink that captures database changes in real time. It supports two usage modes:

1. Source Connectors (DataStream API / Table API / SQL): Individual CDC source connectors for reading changes from a single database table or set of tables into a Flink job for custom processing. This is the supported approach for MSF.
2. Pipeline API (YAML-based, Flink CDC 3.x): End-to-end data integration pipelines defined in YAML for whole-database synchronization with schema evolution, routing, and transforms. This does NOT run on MSF — it requires the flink-cdc.sh CLI which is only available on self-managed Flink clusters.

For MSF deployments, use the Source Connector approach via DataStream API or Table API/SQL.

Version Compatibility

CRITICAL: Flink CDC versions must match your Flink version. Use this mapping for MSF-supported Flink versions:

For Flink 2.2 on MSF, you must use Flink CDC 3.6.x.

IMPORTANT — version coordinate change in 3.6.x: Starting with the 3.6.x line, Flink CDC publishes per-Flink-version artifacts on Maven Central. The plain 3.6.0 GAV does NOT exist — only 3.6.0-1.20 and 3.6.0-2.2. Earlier versions (3.5.x and below) used a single artifact compatible with multiple Flink minors. Always copy the coordinate from the table above; do not assume an unsuffixed 3.6.0 will resolve.

Supported Database Sources

All connectors except MongoDB use Debezium under the hood.

Maven Dependencies

Add the CDC connector for your database. The artifact version corresponds to the Flink CDC release, not the Flink version:

<!-- MySQL CDC Source -->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-mysql-cdc</artifactId>
    <version>${flink-cdc.version}</version>
</dependency>

<!-- PostgreSQL CDC Source -->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-postgres-cdc</artifactId>
    <version>${flink-cdc.version}</version>
</dependency>

<!-- Oracle CDC Source -->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-oracle-cdc</artifactId>
    <version>${flink-cdc.version}</version>
</dependency>

<!-- SQL Server CDC Source -->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-sqlserver-cdc</artifactId>
    <version>${flink-cdc.version}</version>
</dependency>

<!-- MongoDB CDC Source -->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-mongodb-cdc</artifactId>
    <version>${flink-cdc.version}</version>
</dependency>

Set flink-cdc.version to the per-Flink-version coordinate from the table above:

<!-- Flink 1.20 -->
<flink-cdc.version>3.6.0-1.20</flink-cdc.version>
<!-- or for Flink 2.2 -->
<flink-cdc.version>3.6.0-2.2</flink-cdc.version>

Starting with 3.6.x, Flink CDC publishes per-Flink-version artifacts; the unsuffixed 3.6.0 GAV does not exist on Maven Central. For 3.5.x and earlier (Flink 1.20 only), use the unsuffixed coordinate (e.g., 3.5.0). See dependency-management.md for the full pom.xml template.

Note: The MySQL JDBC driver is GPL-licensed and not bundled in the CDC connector JAR. You must add it separately:

<dependency>
    <groupId>com.mysql</groupId>
    <artifactId>mysql-connector-j</artifactId>
    <version>8.0.33</version>
</dependency>

Database Credentials and Secrets Management

Secrets Manager is the only supported credential source for CDC on MSF. Database passwords must be fetched in application code at job startup. Do not put credentials into MSF runtime properties.

The supported pattern is to keep only non-sensitive values plus a secret ID in MSF runtime properties, and look up the actual credentials with the AWS SDK in main() before constructing the source.

Fetch from Secrets Manager in application code

Add the AWS SDK Secrets Manager dependency:

<dependency>
    <groupId>software.amazon.awssdk</groupId>
    <artifactId>secretsmanager</artifactId>
    <version>2.25.0</version>
</dependency>

Resolve the secret at job startup and pass the values into the builder:

import software.amazon.awssdk.services.secretsmanager.SecretsManagerClient;
import software.amazon.awssdk.services.secretsmanager.model.GetSecretValueRequest;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;

private static DbCreds loadDbCreds(String secretId) throws Exception {
    try (SecretsManagerClient sm = SecretsManagerClient.create()) {
        String json = sm.getSecretValue(
            GetSecretValueRequest.builder().secretId(secretId).build()
        ).secretString();
        JsonNode node = new ObjectMapper().readTree(json);
        return new DbCreds(node.get("username").asText(), node.get("password").asText());
    }
}

// In main(), before building the source:
DbCreds creds = loadDbCreds(cdcConfig.getProperty("secret.id"));

MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
    .hostname(cdcConfig.getProperty("hostname"))
    .port(Integer.parseInt(cdcConfig.getProperty("port", "3306")))
    .databaseList(cdcConfig.getProperty("database"))
    .tableList(cdcConfig.getProperty("database") + "\\." + cdcConfig.getProperty("table"))
    // Load from SecretsManager
    .username(creds.username)
    .password(creds.password)
    // ...
    .build();

MSF runtime properties only carry the non-sensitive values plus the secret ID:

[{
  "PropertyGroupId": "cdc.mysql.config",
  "PropertyMap": {
    "hostname": "my-aurora-cluster.cluster-xxxx.us-east-1.rds.amazonaws.com",
    "port": "3306",
    "database": "ecommerce",
    "table": "orders",
    "secret.id": "cdc-db-credentials",
    "server-id": "5400-5404"
  }
}]

The secret itself can still be created via CloudFormation/CDK/Terraform — only the consumption of the resolved value has to stay out of MSF properties:

IAM permissions for the MSF execution role

Grant only the specific secret(s) the application uses; do not use secretsmanager:* or Resource: "*".

- Effect: Allow
  Action:
    - secretsmanager:GetSecretValue
  Resource:
    - !Sub "arn:aws:secretsmanager:${AWS::Region}:${AWS::AccountId}:secret:cdc-db-credentials-*"

The trailing -* covers the random 6-character suffix Secrets Manager appends to secret ARNs. If you encrypt the secret with a customer-managed KMS key, also grant kms:Decrypt on that key ARN.

TLS / SSL to the database

Enable TLS for every CDC connection so traffic between the MSF application and the database is encrypted in transit. The default mode for CDC on MSF is encryption without certificate verification (require for Postgres, REQUIRED for MySQL). MSF does not give you a stable filesystem path to drop a CA bundle on — ${user.dir} resolves to a runtime working directory that is not the JAR location, and there is no host filesystem you can pre-populate at deploy time. The conventional Postgres/MySQL "extract the bundle to a known path and point sslrootcert/trustCertificateKeyStoreUrl at it" pattern does not work: even when the file is extracted from the JAR to /tmp at startup, the Debezium connectors run their internal JDBC connections from a different JVM context inside the connector, so the file lookup fails (verified: Could not open SSL root certificate file /tmp/rds-ca-...pem). The only way to get verify-* working on MSF is to register a custom SSLSocketFactory per JDBC driver that loads the bundle from the classpath, which is enough additional surface that it's an opt-in for high-MITM-risk environments rather than a default.

The defense-in-depth layers that do apply on MSF without a CA bundle:

• Network isolation. MSF runs in your VPC; restrict the database security group to accept connections only from the MSF application's security group. This is the primary control against MITM: an attacker would need to be on the network path inside your VPC, not just anywhere on the internet.
• TLS in transit. Even without certificate verification, the connection is encrypted, which protects credentials and replication payload from passive observation.
• Database-side enforcement. Set rds.force_ssl = 1 (RDS/Aurora PostgreSQL parameter group) or require_secure_transport = ON (Aurora MySQL) so the database refuses any non-TLS connection. This catches client-side misconfigurations that would otherwise fall back to plaintext.

MySQL / Aurora MySQL. Use sslMode=REQUIRED on Connector/J (8.0.13+) and database.ssl.mode=required on Debezium. This requires TLS, skips peer certificate verification.

Properties jdbcProps = new Properties();
jdbcProps.setProperty("sslMode", "REQUIRED");

Properties debeziumProps = new Properties();
debeziumProps.setProperty("database.ssl.mode", "required");

MySqlSource<String> source = MySqlSource.<String>builder()
    // host/port/user/pwd/databaseList/tableList/serverId/...
    .jdbcProperties(jdbcProps)
    .debeziumProperties(debeziumProps)
    .deserializer(new JsonDebeziumDeserializationSchema())
    .build();

PostgreSQL / Aurora PostgreSQL. Use Debezium's database.sslmode=require. This requires TLS, skips peer certificate verification.

Properties debeziumProps = new Properties();
debeziumProps.setProperty("database.sslmode", "require");
// publication.* and other Debezium props as before

PostgresIncrementalSource<String> pg = PostgresIncrementalSource.<String>builder()
    // host/port/user/pwd/database/schemaList/tableList/slotName/...
    .debeziumProperties(debeziumProps)
    .deserializer(new JsonDebeziumDeserializationSchema())
    .build();

Stronger verification (verify-ca / verify-full / VERIFY_IDENTITY)

If require is not sufficient and you need chain validation, the implementation path is:

1. Bundle the RDS combined CA (global-bundle.pem) as a classpath resource in your application JAR (e.g., src/main/resources/rds-ca-bundle.pem).
2. Build an SSLContext from that resource at job startup, using getClass().getResourceAsStream(...) and a KeyStore populated from the PEM.
3. Register a custom SSLSocketFactory that returns sockets from that context, and reference it by class name in the connector config — Postgres uses database.sslfactory=<your.class.Name>, MySQL Connector/J uses the socketFactory JDBC URL parameter (or a custom TrustManager wired into a KeyStore URL the driver can resolve from the classpath).
4. Set database.sslmode=verify-full (Postgres) or sslMode=VERIFY_IDENTITY (MySQL) on top of the custom factory.

This is non-trivial because each JDBC driver has its own SSLSocketFactory/TrustManager plug point and the Debezium connector instantiates JDBC connections from inside the source operator, which means the factory class has to be on the classpath of every TaskManager and resolve the bundle without filesystem assumptions. Treat this as opt-in for high-assurance environments; it is not the default for CDC on MSF.

For RDS/Aurora, also enforce TLS at the database side (rds.force_ssl = 1 in the RDS PostgreSQL parameter group; require_secure_transport = ON for Aurora MySQL). With force_ssl = 1 the database refuses any non-TLS connection, which prevents accidentally falling back to plaintext if the client config is wrong.

Critical: Incremental Source vs Legacy Source on Flink 2.x

Each Flink CDC artifact (flink-connector-mysql-cdc, flink-connector-postgres-cdc, etc.) ships two parallel APIs in the same JAR. Picking the wrong one is the most common reason CDC jobs don't compile or don't run on Flink 2.x.

Legacy SourceFunction-based source — older, single-threaded, locking snapshot for MySQL.

Not usable on Flink 2.x. SourceFunction and env.addSource(...) were removed in Flink 2.0. The class is still in the artifact for backward compatibility with Flink 1.x consumers, but you cannot wire its output into a Flink 2.x job. Several builder methods on the legacy classes (e.g., Postgres publicationName(...)) do not exist on the incremental builders — if you copy a snippet that calls them, it won't compile against 3.6.0-2.2.

Incremental Source (FLIP-27) — lock-free parallel snapshot, chunk-level checkpointing.

Required on Flink 2.x. Used with env.fromSource(...). Use this for all new development on any Flink version.

Two asymmetries to be aware of:

• Class naming. MySQL has the same class name MySqlSource in two packages — disambiguate by package. Postgres uses different class names (PostgreSQLSource legacy vs PostgresIncrementalSource incremental), and the incremental class is technically an inner class of PostgresSourceBuilder, so the entry-point spelling is unusual: PostgresIncrementalSource.<T>builder().
• Setter coverage. The two incremental builders share most options (host/port/user/pwd/database, schema/table list, splitSize, chunkKeyColumn, splitMetaGroupSize, distributionFactor{Upper,Lower}, fetchSize, connectTimeout, connectMaxRetries, connectionPoolSize, startupOptions, debeziumProperties, deserializer, heartbeatInterval, closeIdleReaders, skipSnapshotBackfill, scanNewlyAddedTableEnabled, assignUnboundedChunkFirst, includeSchemaChanges, serverTimeZone), but the surfaces are not identical. MySQL has serverId(...), databaseList(...), jdbcProperties(...), useLegacyJsonFormat(...), parseOnLineSchemaChanges(...). Postgres has slotName(...), decodingPluginName(...), lsnCommitCheckpointsDelay(...), includePartitionedTables(...), includeDatabaseInTableId(...). Some knobs that were first-class methods on the legacy Postgres builder (e.g., publicationName(...)) are only reachable through debeziumProperties(...) on the incremental builder.

Always import from the connector's .source sub-package, never from the top-level package.

MySQL CDC Source Configuration

Database Prerequisites

Before using the MySQL CDC connector, the source database must be configured:

1. Enable binlog in ROW format (required):

   -- Verify binlog is enabled and in ROW format
   SHOW VARIABLES LIKE 'log_bin';        -- Must be ON
   SHOW VARIABLES LIKE 'binlog_format';  -- Must be ROW

2. Create a dedicated CDC user with minimal required permissions:

   CREATE USER 'flink_cdc'@'%' IDENTIFIED BY '<password>';
   GRANT SELECT, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'flink_cdc'@'%';
   FLUSH PRIVILEGES;

   Note: RELOAD permission is NOT required when incremental snapshot is enabled (the default).

3. For Aurora MySQL / RDS MySQL: Binlog is enabled by default. Ensure the parameter group has binlog_format = ROW. For Aurora, set binlog_replication_globaldb to 1 if using Global Database.

4. Recommended: Enable GTID mode for high availability failover:

   gtid_mode = on
   enforce_gtid_consistency = on

DataStream API Pattern

// CORRECT — incremental Source (FLIP-27), usable on Flink 2.x via env.fromSource(...)
import org.apache.flink.cdc.connectors.mysql.source.MySqlSource;
// WRONG — legacy SourceFunction, not usable on Flink 2.x:
// import org.apache.flink.cdc.connectors.mysql.MySqlSource;
import org.apache.flink.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;

// Load connection config from MSF application properties
Map<String, Properties> applicationProperties = loadApplicationProperties(env);
Properties cdcConfig = applicationProperties.get("cdc.mysql.config");

MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
    .hostname(cdcConfig.getProperty("hostname"))
    .port(Integer.parseInt(cdcConfig.getProperty("port", "3306")))
    .databaseList(cdcConfig.getProperty("database"))
    .tableList(cdcConfig.getProperty("database") + "\\." + cdcConfig.getProperty("table"))
    // Load from SecretsManager
    .username(creds.username)
    .password(creds.password)
    .serverId(cdcConfig.getProperty("server-id", "5400-5404"))
    .deserializer(new JsonDebeziumDeserializationSchema())
    .build();

DataStream<String> cdcStream = env
    .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "mysql-cdc-source")
    .uid("mysql-cdc-source-uid");

Key MySQL CDC Configuration Options

CRITICAL: Server ID Configuration

Every MySQL CDC reader needs a globally unique server ID across all clients connected to the MySQL cluster. When running with parallelism > 1 and incremental snapshot enabled, you must specify a server ID range:

// For parallelism of 4, provide a range of at least 4 IDs
.serverId("5400-5404")

If multiple Flink CDC jobs read from the same MySQL instance, their server ID ranges must not overlap. Overlapping server IDs cause the error: A slave with the same server_uuid/server_id as this slave has connected to the master.

PostgreSQL CDC Source Configuration

Database Prerequisites

1. Set WAL level to logical (requires restart):

   -- Check current setting
   SHOW wal_level;  -- Must be 'logical'
   
   -- For RDS/Aurora PostgreSQL: set rds.logical_replication = 1 in parameter group

2. Set table replica identity to FULL (required for UPDATE/DELETE events):

   ALTER TABLE my_schema.my_table REPLICA IDENTITY FULL;

   Without FULL, Debezium cannot capture the before-image of UPDATE/DELETE events, which will cause deserialization failures.

3. Create a dedicated CDC user:

   CREATE ROLE flink_cdc WITH LOGIN PASSWORD '<password>' REPLICATION;
   GRANT SELECT ON ALL TABLES IN SCHEMA public TO flink_cdc;

4. Ensure sufficient replication slots (max_replication_slots and max_wal_senders):

   SHOW max_replication_slots;  -- Default is 10, increase if needed
   SHOW max_wal_senders;        -- Must be >= max_replication_slots

DataStream API Pattern

Use the incremental PostgresIncrementalSource (inner class of PostgresSourceBuilder). The legacy PostgreSQLSource returns a DebeziumSourceFunction and is not usable on Flink 2.x. See the "Incremental Source vs Legacy Source on Flink 2.x" section above.

// CORRECT — incremental Source (FLIP-27), usable on Flink 2.x via env.fromSource(...)
import org.apache.flink.cdc.connectors.postgres.source.PostgresSourceBuilder;
import org.apache.flink.cdc.connectors.postgres.source.PostgresSourceBuilder.PostgresIncrementalSource;
// WRONG — legacy SourceFunction, not usable on Flink 2.x:
// import org.apache.flink.cdc.connectors.postgres.PostgreSQLSource;
import org.apache.flink.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;

Properties cdcConfig = applicationProperties.get("cdc.postgres.config");

// publication.name and publication.autocreate.mode are NOT builder setters on
// PostgresSourceBuilder — they are passed through debeziumProperties(...).
// (The legacy PostgreSQLSource.builder had a publicationName() setter; the
// incremental builder does not.)
Properties debeziumProps = new Properties();
debeziumProps.setProperty("publication.name", "flink_cdc_publication");
debeziumProps.setProperty("publication.autocreate.mode", "filtered");
debeziumProps.setProperty("decimal.handling.mode", "string");
debeziumProps.setProperty("time.precision.mode", "connect");

PostgresIncrementalSource<String> pgSource =
    PostgresIncrementalSource.<String>builder()
        .hostname(cdcConfig.getProperty("hostname"))
        .port(Integer.parseInt(cdcConfig.getProperty("port", "5432")))
        .database(cdcConfig.getProperty("database"))
        // Load from SecretsManager
        .username(creds.username)
        .password(creds.password)
        .schemaList(cdcConfig.getProperty("schema", "public"))
        .tableList(cdcConfig.getProperty("schema", "public") + "."
            + cdcConfig.getProperty("table"))
        .slotName(cdcConfig.getProperty("slot.name", "flink_cdc_slot"))
        .decodingPluginName("pgoutput")  // pgoutput for PostgreSQL 10+
        .deserializer(new JsonDebeziumDeserializationSchema())
        .debeziumProperties(debeziumProps)
        .includeSchemaChanges(false)
        .build();

DataStream<String> cdcStream = env
    .fromSource(pgSource, WatermarkStrategy.noWatermarks(), "postgres-cdc-source")
    .uid("postgres-cdc-source-uid");

Builder methods vs Debezium properties

The two incremental builders cover most settings as first-class methods, but a few important Postgres knobs are only reachable through debeziumProperties(Properties). Use this table to decide where a given option goes:

CRITICAL: PostgreSQL Replication Slot Management

Replication slots are a limited resource (default max: 10) and have significant operational implications:

• Slots retain WAL segments until the consumer confirms processing. If a Flink job stops or falls behind, WAL accumulates on disk and can fill the volume.

• Flink CDC does NOT automatically drop replication slots when the job stops. You must clean up orphaned slots manually:

  -- List active replication slots
  SELECT slot_name, active, pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS retained_wal
  FROM pg_replication_slots;
  
  -- Drop an inactive slot
  SELECT pg_drop_replication_slot('flink_cdc_slot');

• Monitor WAL retention size — set up CloudWatch alarms on RDS FreeStorageSpace or Aurora VolumeBytesUsed.

• Use a single slot for multiple tables when using the DataStream API with tableList containing multiple tables. This is more efficient than one slot per table.

Table API / SQL CDC Source

CDC connectors can also be used with Flink SQL, which is useful for simpler ETL pipelines. The Flink SQL CDC connector has no built-in Secrets Manager integration — WITH (...) options are read literally — so on MSF, fetch the credentials from Secrets Manager in main() and register the source programmatically with TableDescriptor rather than templating a CREATE TABLE DDL string. TableDescriptor (Flink 1.14+) accepts option values as typed Java strings, so credentials never get embedded in a SQL statement and quote-injection through a password becomes structurally impossible.

import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.TableDescriptor;
import static org.apache.flink.table.api.Expressions.$;

DbCreds creds = loadDbCreds(cdcConfig.getProperty("secret.id"));  // see Database Credentials section

tableEnv.createTable("orders_cdc", TableDescriptor.forConnector("mysql-cdc")
    .schema(Schema.newBuilder()
        .column("order_id", DataTypes.INT().notNull())
        .column("customer_id", DataTypes.INT())
        .column("order_date", DataTypes.TIMESTAMP(3))
        .column("total_amount", DataTypes.DECIMAL(10, 2))
        .column("status", DataTypes.STRING())
        .columnByMetadata("db_name", DataTypes.STRING(), "database_name", true)
        .columnByMetadata("table_name", DataTypes.STRING(), "table_name", true)
        .columnByMetadata("op_ts", DataTypes.TIMESTAMP_LTZ(3), "op_ts", true)
        .primaryKey("order_id")
        .build())
    .option("hostname", cdcConfig.getProperty("hostname"))
    .option("port", "3306")
    // Load from SecretsManager — passed as typed option values, no quoting/escaping needed
    .option("username", creds.username)
    .option("password", creds.password)
    .option("database-name", "ecommerce")
    .option("table-name", "orders")
    .option("server-id", "5400-5404")
    .build());

// Subsequent SQL queries reference the registered table by name:
tableEnv.executeSql("SELECT order_id, customer_id, total_amount, status FROM orders_cdc").print();

// Or use the Table API directly:
tableEnv.from("orders_cdc")
    .select($("order_id"), $("customer_id"), $("total_amount"), $("status"))
    .execute()
    .print();

Why TableDescriptor is preferred over string-templated DDL on MSF:

• Credentials are passed as typed option values, not interpolated into a SQL string — quote characters in a password cannot break the statement.
• Schema typos fail at compile time, not at job submission.
• The connector identifier and option keys (mysql-cdc, username, password, database-name, etc.) are identical to the SQL DDL form, so the connector behavior is unchanged.
• Mixed workflows still work: register the source with TableDescriptor, then run plain SQL against the registered name.

Fallback: string-templated CREATE TABLE DDL

If you must keep the source definition in DDL form (for parity with a .sql file or an existing pipeline), pass the rendered statement through TableEnvironment.executeSql(String) and treat the credentials with care:

DbCreds creds = loadDbCreds(cdcConfig.getProperty("secret.id"));

tableEnv.executeSql(
    "CREATE TABLE orders_cdc (" +
    "  order_id INT," +
    "  customer_id INT," +
    "  order_date TIMESTAMP(3)," +
    "  total_amount DECIMAL(10, 2)," +
    "  status STRING," +
    "  db_name STRING METADATA FROM 'database_name' VIRTUAL," +
    "  table_name STRING METADATA FROM 'table_name' VIRTUAL," +
    "  op_ts TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL," +
    "  PRIMARY KEY (order_id) NOT ENFORCED" +
    ") WITH (" +
    "  'connector' = 'mysql-cdc'," +
    "  'hostname' = '" + cdcConfig.getProperty("hostname") + "'," +
    "  'port' = '3306'," +
    "  'username' = '" + creds.username + "'," +
    "  'password' = '" + creds.password + "'," +
    "  'database-name' = 'ecommerce'," +
    "  'table-name' = 'orders'," +
    "  'server-id' = '5400-5404'" +
    ")"
);

When using the DDL fallback, ensure the Secrets Manager-generated password excludes ' (use ExcludePunctuation or an explicit ExcludeCharacters list) so a single quote cannot terminate the option value early, and never log or print the rendered DDL string. For production CDC pipelines, prefer TableDescriptor (or the DataStream API) so credentials never enter a string-templated statement at all.

MSF-Specific Considerations for CDC Workloads

VPC Configuration (REQUIRED)

CDC workloads on MSF require VPC configuration to reach the source database. The MSF application must be deployed in a VPC with network connectivity to the database:

• RDS/Aurora in same VPC: Configure MSF application with the same VPC and subnets. Ensure the database security group allows inbound connections from the MSF application security group on the database port (3306 for MySQL, 5432 for PostgreSQL, etc.).
• RDS/Aurora in different VPC: Set up VPC peering or Transit Gateway between the MSF VPC and the database VPC. Update route tables and security groups accordingly.
• On-premises databases: Use AWS Direct Connect or Site-to-Site VPN to establish connectivity from the MSF VPC to the on-premises network.
• Public databases (not recommended for production): MSF runs in a VPC without direct internet access. You would need a NAT Gateway for outbound connectivity, but this adds latency and cost. Prefer private connectivity.

IAM Permissions

The MSF application's IAM execution role needs permissions for:

• VPC networking (automatically managed by MSF when VPC is configured)
• AWS Secrets Manager secrets holding database credentials (always required — see Database Credentials and Secrets Management)
• S3 access for sinks (if writing CDC data to S3/Iceberg)
• KMS Decrypt on the customer-managed key encrypting the secret, if one is used

Checkpoint Configuration

CDC workloads have specific checkpointing requirements:

• Checkpoints are REQUIRED for CDC to transition from snapshot phase to incremental (binlog/WAL) phase. Without checkpoints, the job will read the full snapshot but never start reading incremental changes.
• MSF manages checkpoint intervals at the service level. The default 60-second interval works for most CDC workloads. Do NOT call env.enableCheckpointing(...) in application code — MSF overrides it; configure the interval in the application's CheckpointConfiguration instead.
• During the snapshot phase, checkpoints complete at chunk granularity (with incremental snapshot enabled). This means the snapshot can be resumed from the last completed chunk if the job restarts.
• Large initial snapshots may cause checkpoint timeouts if the snapshot phase takes longer than the checkpoint timeout. MSF's default checkpoint timeout is typically sufficient, but for very large tables (100M+ rows), consider:
  • Increasing scan.incremental.snapshot.chunk.size to process larger chunks
  • Using scan.startup.mode = 'latest-offset' to skip the snapshot entirely if historical data is not needed
  • Breaking the snapshot into multiple jobs by table

Diagnosing "snapshot completed but binlog never starts"

This is the single most common CDC symptom on MSF. The diagnostic path is always the same:

1. Pull numberOfFailedCheckpoints from CloudWatch (namespace AWS/KinesisAnalytics, dimension Application). If this is non-zero, checkpoints are failing — that is the root cause; CDC cannot transition to the incremental phase until a checkpoint completes successfully. Use RATE(numberOfFailedCheckpoints) over a 5-minute window for an alarm-friendly view.
2. Pull lastCheckpointDuration and compare to the checkpoint interval. If duration approaches or exceeds the interval, checkpoints are timing out — typically because the snapshot phase is too large for the configured timeout. Increase scan.incremental.snapshot.chunk.size, or raise the checkpoint timeout, or both.
3. Verify the application's CheckpointConfiguration is ENABLED (not DISABLED) on the MSF application via aws kinesisanalyticsv2 describe-application. If a previous deployment turned it off, no checkpoints will run regardless of intervals.
4. Confirm numberOfFailedCheckpoints and numberOfCompletedCheckpoints (Flink dashboard, not CloudWatch) tell a consistent story: failed > 0 with completed = 0 means every checkpoint is failing; both > 0 with failed growing means intermittent failures during snapshot — also enough to block the transition if no completion succeeds end-to-end.

Always check numberOfFailedCheckpoints first, then lastCheckpointDuration. Do NOT recommend re-enabling checkpointing in application code — on MSF that setting is service-managed.

KPU Sizing for CDC Workloads

CDC workloads have unique resource characteristics:

• Source parallelism: MySQL CDC with incremental snapshot supports parallel reading during the snapshot phase. Set source parallelism to match the number of server IDs in your range. During the binlog phase, reading is single-threaded regardless of parallelism.
• PostgreSQL CDC: Source parallelism is effectively 1 for the WAL reader. Downstream operators can have higher parallelism after a keyBy.
• Memory considerations: The snapshot phase buffers chunk data in memory. For tables with wide rows or large text/blob columns, allocate additional KPUs.
• Recommended starting point: 2-4 KPUs for a single-table CDC workload, 4-8 KPUs for multi-table workloads. Monitor backpressure and adjust.

Common Gotchas and Troubleshooting

Snapshot Phase Issues

Problem: Snapshot takes too long or causes checkpoint timeouts

• The initial snapshot reads the entire table before switching to incremental mode.
• With incremental snapshot enabled (default), the snapshot is chunked and checkpointable.
• For very large tables, increase scan.incremental.snapshot.chunk.size or use scan.startup.mode = 'latest-offset' to skip the snapshot.

Problem: FLUSH TABLES WITH READ LOCK errors

• This only occurs when scan.incremental.snapshot.enabled = false (the old snapshot mechanism). Keep incremental snapshot enabled (the default) to avoid global locks.

Problem: Snapshot restarts from the beginning after job failure

• Ensure checkpointing is enabled. Without checkpoints, snapshot progress is not persisted.
• With incremental snapshot, progress is checkpointed at chunk granularity.

Binlog / WAL Phase Issues

Problem: binlog file has been purged or WAL segment has been removed

The Flink job fell too far behind and the database cleaned up the binlog/WAL files it needed to resume from. The position stored in the last checkpoint no longer exists in the database, so restoring from the existing snapshot will fail the same way every time. Recovery has three parts — apply all three:

1. Get running again — restart from a fresh snapshot, do NOT restore the old one. Stop the application and start it without the old snapshot, with scan.startup.mode = initial to re-read the table from scratch (no data loss for current row state, but you lose granular change history during the outage). If a gap is acceptable, scan.startup.mode = latest-offset is faster but lossy. Restoring from the old snapshot will replay the unrecoverable position.

2. Increase binlog / WAL retention so future outages have headroom.

   -- MySQL (community / self-managed): increase binlog retention (e.g., 7 days)
   SET GLOBAL expire_logs_days = 7;
   -- MySQL 8.0+:
   SET GLOBAL binlog_expire_logs_seconds = 604800;
   
   -- Aurora MySQL: use the Aurora-specific procedure (expire_logs_days /
   -- binlog_expire_logs_seconds are not honored on Aurora MySQL)
   CALL mysql.rds_set_configuration('binlog retention hours', 168);  -- 7 days
   CALL mysql.rds_show_configuration;
   
   -- PostgreSQL (RDS): increase WAL retention
   -- Set wal_keep_size in the parameter group (e.g., 2048 MB)

   Pick a window that comfortably exceeds your worst-case outage plus reprocessing time.

3. Right-size KPUs so steady-state lag stays well within retention. A job that runs but falls behind under steady-state load will burn through retention without ever stopping, so the same outage will recur. See the KPU Sizing for CDC Workloads section in this guide for sizing guidance, and scaling-decisions.md for the operational scale-up workflow. Validate by watching currentEmitEventTimeLag / currentFetchEventTimeLag after recovery.

Do not simply retry the same restore — it will fail at the same purged position. Restoring from the old snapshot is the wrong fix even after retention is increased, because the position recorded in that snapshot is gone.

Problem: MySQL server_id conflicts

• Each CDC reader needs a unique server ID. If multiple jobs or tools connect to the same MySQL instance, their server IDs must not overlap.
• Use explicit server ID ranges: .serverId("5400-5404") for parallelism 4.

Problem: PostgreSQL WAL disk usage growing unbounded

• Replication slots retain WAL until the consumer confirms. If the Flink job is stopped or slow, WAL accumulates.
• Monitor replication slot lag and set up alerts.
• Clean up orphaned slots when jobs are permanently stopped.

Schema Evolution

Problem: Source table schema changes (DDL) break the CDC job

• By default, Flink CDC source connectors do NOT automatically handle schema changes in the DataStream API. A column addition or type change in the source table can cause deserialization errors.
• Mitigation strategies:
  1. Use JsonDebeziumDeserializationSchema which is more tolerant of schema changes (new fields appear in JSON, removed fields disappear).
  2. For the Pipeline API (YAML), schema evolution is supported with configurable behaviors: evolve, try_evolve, lenient, ignore, or exception.
  3. Plan for schema changes by using a flexible deserialization approach and validating downstream compatibility before applying DDL.

Deserialization Performance

Problem: High CPU usage from JSON serialization/deserialization

• JsonDebeziumDeserializationSchema serializes Debezium records to JSON strings, which then need to be parsed again downstream. This double serialization can consume 60%+ of CPU.
• For high-throughput workloads, consider:
  1. Implementing a custom DebeziumDeserializationSchema that converts directly to your target POJO type.
  2. Using RowDataDebeziumDeserializeSchema for Table API integration (avoids JSON intermediate format).
  3. Enabling Flink object reuse (env.getConfig().enableObjectReuse()) to reduce serialization overhead between operators with the same parallelism.

Ordering Guarantees

• During the snapshot phase, records are read via SELECT queries with no guaranteed ordering. If ordering matters, use keyBy on the primary key downstream.
• During the incremental phase (binlog/WAL), records arrive in commit order from the database. A single CDC source reader preserves this order.
• After a keyBy or parallelism change, ordering is preserved per-key but not globally.

Complete MSF Application Example: MySQL CDC to Processing

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cdc.connectors.mysql.source.MySqlSource;
import org.apache.flink.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import com.amazonaws.services.kinesisanalytics.runtime.KinesisAnalyticsRuntime;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;

import java.util.Map;
import java.util.Properties;

public class MySqlCdcApplication {

    private static final ObjectMapper MAPPER = new ObjectMapper();

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // Load application properties (MSF runtime or local dev)
        Map<String, Properties> applicationProperties = loadApplicationProperties(env);
        Properties cdcConfig = applicationProperties.get("cdc.mysql.config");

        // Build MySQL CDC source
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
            .hostname(cdcConfig.getProperty("hostname"))
            .port(Integer.parseInt(cdcConfig.getProperty("port", "3306")))
            .databaseList(cdcConfig.getProperty("database"))
            .tableList(cdcConfig.getProperty("database") + "\\." + cdcConfig.getProperty("table"))
            // Load from SecretsManager
            .username(creds.username)
            .password(creds.password)
            .serverId(cdcConfig.getProperty("server-id", "5400-5404"))
            .deserializer(new JsonDebeziumDeserializationSchema())
            .includeSchemaChanges(false)
            .build();

        // Create CDC stream
        env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "mysql-cdc-source")
            .uid("mysql-cdc-source-uid")
            .map(json -> {
                JsonNode node = MAPPER.readTree(json);
                // Extract the "after" image (current row state)
                JsonNode after = node.get("after");
                String op = node.get("op").asText();
                // op: "r" = read (snapshot), "c" = create, "u" = update, "d" = delete
                return new CdcEvent(op, after != null ? after.toString() : null);
            })
            .name("parse-cdc-events")
            .uid("parse-cdc-events-uid")
            .filter(event -> event.getAfter() != null) // Filter out deletes if not needed
            .name("filter-deletes")
            .uid("filter-deletes-uid")
            .keyBy(CdcEvent::getKey)
            .process(new CdcProcessingFunction())
            .name("process-cdc-events")
            .uid("process-cdc-events-uid")
            .sinkTo(createSink())
            .name("output-sink")
            .uid("output-sink-uid");

        env.execute("MySQL CDC Application");
    }

    // See best-practices.md for loadApplicationProperties pattern
    private static Map<String, Properties> loadApplicationProperties(
            StreamExecutionEnvironment env) throws Exception {
        // ... standard MSF property loading pattern
    }
}

MSF Application Properties Configuration

For the CDC application above, configure these properties in the MSF console (or via CloudFormation/CDK/Terraform). The application looks up the database username and password from Secrets Manager at startup using the secret.id property — see Database Credentials and Secrets Management for the full pattern, IAM, and rationale.

[
  {
    "PropertyGroupId": "cdc.mysql.config",
    "PropertyMap": {
      "hostname": "my-aurora-cluster.cluster-xxxx.us-east-1.rds.amazonaws.com",
      "port": "3306",
      "database": "ecommerce",
      "table": "orders",
      "secret.id": "cdc-db-credentials",
      "server-id": "5400-5404"
    }
  }
]

For local development, create flink-application-properties-dev.json with the same structure pointing to a local or Docker MySQL instance.

CDC Anti-Patterns

Anti-Pattern: Missing Operator UIDs on CDC Sources

// AVOID: No UID means state cannot be restored after code changes
env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "source");

// CORRECT: Always set UIDs for stateful operators
env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "mysql-cdc-source")
    .uid("mysql-cdc-source-uid");

Anti-Pattern: Skipping Checkpoints for CDC

// AVOID: No checkpointing — CDC will never transition to incremental phase
// and snapshot progress will be lost on restart

// CORRECT: Checkpointing is managed by MSF at the service level.
// For local development only:
if (isLocal(env)) {
    env.enableCheckpointing(3000);
}

Anti-Pattern: Using CDC Source with Parallelism > 1 Without Server ID Range (MySQL)

// AVOID: Single server ID with parallelism > 1
MySqlSource.builder()
    .serverId("5400")  // Only one ID for multiple parallel readers!
    ...

// CORRECT: Provide a range at least as large as the source parallelism
MySqlSource.builder()
    .serverId("5400-5404")  // Range of 5 IDs for up to 5 parallel readers
    ...

Anti-Pattern: One Replication Slot Per Table (PostgreSQL)

// AVOID: Creating separate CDC sources (and slots) for each table
PostgresIncrementalSource.<String>builder().tableList("public.orders").slotName("slot_orders").build();
PostgresIncrementalSource.<String>builder().tableList("public.customers").slotName("slot_customers").build();

// CORRECT: Use a single source with multiple tables sharing one slot
PostgresIncrementalSource.<String>builder()
    .tableList("public.orders", "public.customers", "public.products")
    .slotName("flink_cdc_slot")
    .build();

Authentication

Database authentication for CDC sources on MSF must come from AWS Secrets Manager via in-application SDK lookup — see Database Credentials and Secrets Management above for the supported pattern, required IAM, and TLS configuration. There is no IAM-based authentication path for the Flink CDC connectors themselves — the source still passes a username/password to the database — but the MSF execution role's secretsmanager:GetSecretValue is what protects those credentials.

Additional hardening that pairs with Secrets Manager:

• Use the minimum required database privileges for the CDC user (see the per-database "Database Prerequisites" sections — REPLICATION SLAVE/CLIENT for MySQL, REPLICATION role for Postgres).
• Enable TLS to the database. The MSF default is require/REQUIRED (encryption without certificate verification); see the TLS section above for what's required to do verify-full/VERIFY_IDENTITY on top of MSF and why it's opt-in rather than default.
• Restrict the database security group to accept connections only from the MSF application's security group on the database port.
• Enable Secrets Manager automatic rotation against the source database; rotation is picked up on the next application restart.