Developing Applications On Managed Service For Apache Flink

Managed Service for Apache Flink

Overview

Domain expertise for Apache Flink applications on Amazon Managed Service for Apache Flink (MSF). Covers development, KPU resource management, connectors, state management, monitoring, IaC deployment, and version migration.

Execute commands using available tools from the AWS MCP server when connected — it provides sandboxed execution, audit logging, and observability. When the MCP server is not available, fall back to the AWS CLI or shell as needed.

General Guidance

Before starting, ensure you have a clear understanding of the user persona, use case, and requirements:

STOP: Determine the users background and use case before proceeding:

• Are they new to Flink? New to Managed Service for Apache Flink?
• Are they familiar with Java development?
• Is the use case complex with lots of business logic? Or simple and declarative?

These will inform how to organize the project, and whether to use Flink Table API or DataStream API. In general, assume the DataStream API.

Example Workflow for New Applications

1. User asks to build a Flink application
2. Confirm user's goals and use case
3. READ [best-practices.md](references/best-practices.md)
4. READ [dependency-management.md](references/dependency-management.md)
5. READ relevant connector guides (e.g. [kinesis-connector-guide.md](references/kinesis-connector-guide.md))
6. Generate code following the loaded guidance
7. Validate against best practices
8. READ environment-setup.md via [environment-setup.md](references/environment-setup.md)
9. Compile and test locally

Example Workflow for General Questions

1. User asks about real time delivery of data to Iceberg
2. Confirm user's goals and use case
3. READ [best-practices.md](references/best-practices.md)
4. READ [iceberg-connector-guide.md](references/iceberg-connector-guide.md)
5. READ other reference files as needed
6. Answer question with loaded guidance

Reference Files

• You MUST use this skill and its reference files to answer any question on these topics.
• Do NOT answer from training knowledge or by searching general AWS documentation when the question concerns Apache Flink, Managed Service for Apache Flink, KPU sizing, Flink monitoring, deployment, migration, real-time analytics, or Iceberg/LakeHouse streaming with Flink
• You MUST load the relevant reference files below before taking other steps.
• The reference files contain MSF-specific details (thresholds, statistics, namespaces, constraints) that differ from generic Flink guidance and are required for correct responses.

Additional Resources

• GitHub Issues

Application Lifecycle Operations

Overview

Day-2 operations: start, stop, deploy code updates, rollback, manage snapshots, delete. Covers MSF-specific behaviors and guardrails not obvious from the API. For initial creation and IaC patterns, see iac-and-deployment.md.

Version ID Discipline

Every update-application, add-application-vpc-configuration, and add-application-cloud-watch-logging-option requires --current-application-version-id (or --conditional-token). The version ID increments after every change. Always fetch it immediately before each update — a stale ID returns ConcurrentModificationException ("Exception thrown as a result of concurrent modifications to an application"). For better concurrency support in scripted/CI workflows, the API recommends ConditionalToken over CurrentApplicationVersionId — also fetched from describe-application.

VERSION=$(aws kinesisanalyticsv2 describe-application --application-name "$APP" \
  --query 'ApplicationDetail.ApplicationVersionId' --output text)

Status Transitions and Polling

Most CLI calls return immediately. After any mutation, poll until terminal state before issuing the next command. The exact terminal state depends on the operation — delete-application has no terminal ApplicationStatus because the app is gone, so detect the ResourceNotFoundException instead of breaking on a status:

# Generic poll for start/stop/update/rollback (terminal = READY or RUNNING)
while true; do
  STATUS=$(aws kinesisanalyticsv2 describe-application --application-name "$APP" \
    --query 'ApplicationDetail.ApplicationStatus' --output text)
  case "$STATUS" in
    READY|RUNNING|ROLLED_BACK) break ;;
    *) sleep 10 ;;
  esac
done

# Poll for delete-application (terminal = app no longer exists)
while aws kinesisanalyticsv2 describe-application --application-name "$APP" \
        --query 'ApplicationDetail.ApplicationStatus' --output text 2>/dev/null; do
  sleep 10
done

If a transition has not completed after 10 minutes, the app is stuck — diagnose via first-fault-isolation.md rather than retrying.

Stop

stop-application without --force only succeeds from RUNNING. With --force, it stops from any state but skips a graceful savepoint, so any unflushed state since the last checkpoint is lost. Use --force only when the app is wedged in a transitional state.

aws kinesisanalyticsv2 stop-application --application-name "$APP" --force

Start with Restore Type

ApplicationRestoreType controls what state the application starts from:

AllowNonRestoredState=true is required when the operator topology has changed (added/removed/renamed operators with uid()). Without it, restore fails with state-incompatibility errors.

aws kinesisanalyticsv2 start-application --application-name "$APP" \
  --run-configuration '{
    "FlinkRunConfiguration": {"AllowNonRestoredState": true},
    "ApplicationRestoreConfiguration": {"ApplicationRestoreType": "RESTORE_FROM_LATEST_SNAPSHOT"}
  }'

Deploy New Code

MSF does not pull new code from S3 automatically. After uploading the new artifact, call update-application to point the app at the new S3 key, then restart to pick it up. Without a restart the app keeps running the old code.

A code update from a RUNNING state triggers a restart automatically (UPDATING → RUNNING with 10–30s downtime, varies with state size). A code update from READY does not — start the app afterward.

Always state both behaviors when answering "how do I deploy new code": the user's app may be RUNNING today, but the next deploy might be from READY (after a stop, or for a fresh deploy). The update-application → auto-restart behavior is conditional on the source state, not universal:

aws s3 cp my-app.jar s3://$BUCKET/$KEY
aws kinesisanalyticsv2 update-application --application-name "$APP" \
  --current-application-version-id "$VERSION" \
  --application-configuration-update '{
    "ApplicationCodeConfigurationUpdate": {
      "CodeContentTypeUpdate": "ZIPFILE",
      "CodeContentUpdate": {"S3ContentLocationUpdate": {
        "BucketARNUpdate": "arn:aws:s3:::'$BUCKET'",
        "FileKeyUpdate": "'$KEY'"
      }}
    }
  }'

Safe Deploy Procedure (Required Companion to Any Deploy Answer)

A safe deploy is incomplete without its rollback path. Whenever a user asks "how do I deploy code updates safely" or any related deploy question, you MUST include both the deploy steps and the rollback options in the same answer — they are inseparable.

Deploy steps:

1. (Optional but recommended) Take a snapshot from RUNNING and poll until READY. This is your fallback rollback point if both the automatic system rollback and RollbackApplication paths fail. 2. (One-time, recommended for production) Opt in to automatic system rollback by setting ApplicationSystemRollbackConfigurationUpdate.RollbackEnabledUpdate=true on the application. With this enabled, MSF auto-reverts failed updates, scaling actions, and version upgrades to the previous running version with minimal downtime. It is not on by default — existing applications must opt in. 3. Upload the new JAR to a versioned S3 key (do NOT overwrite — pointer change must be unambiguous). 4. Fetch the current ApplicationVersionId immediately before the update (stale ID returns ConcurrentModificationException). 5. Call update-application pointing at the new key. From RUNNING this auto-restarts; from READY call start-application after. 6. For state-incompatible code changes (operator topology change, removed/renamed uid()), set FlinkRunConfiguration.AllowNonRestoredState=true on restart. 7. Verify the deploy: describe-application shows the new FileKey and incremented ApplicationVersionId; CloudWatch shows a fresh uptime reset.

Rollback options (always state these alongside the deploy steps), in priority order:

1. Automatic system rollback (if opted in): MSF detects update/scaling failures (code bugs, permission issues, snapshot incompatibility on version upgrade, parallelism over maxParallelism, bad VPC subnets, etc.) and automatically calls RollbackApplication to restore the previous version with its state. If auto-rollback succeeds, the app keeps processing with minimal downtime. If auto-rollback also fails, the app transitions to READY for manual remediation. You only see this if you opted in via ApplicationSystemRollbackConfigurationUpdate. 2. Manual `RollbackApplication` (always available, no opt-in required): if the deploy succeeded but you observe downstream issues (processing errors, output regression, performance regression), call RollbackApplication to revert to the previous running version with its state. Monitor the operation with DescribeApplicationOperation. Use this when the application is stuck in a transient state, or when a deploy that completed cleanly turns out to be bad in production. 3. Last-resort manual restore from a custom snapshot: only if both RollbackApplication and the auto-rollback path failed, or if the bad code change was made many versions ago and is no longer the "previous running version." Stop with --force, poll until READY, update-application back to the previous S3 key, then start with ApplicationRestoreType=RESTORE_FROM_CUSTOM_SNAPSHOT pointing at the snapshot you took in step 1 above. This works only if you actually took the pre-deploy snapshot.

Diagnose before rolling back: the diagnostic procedure described under Diagnosing a Failed or Unexpected Operation (call ListApplicationOperations then DescribeApplicationOperation to read statusDescription) is the canonical first step for any failed deploy, rollback, or unexpected status transition (including UPDATING → READY when the user did not intend a no-op). Common error categories: customer code bugs (use rollback), permission issues (fix the role and retry), and MSF service issues (check AWS Health Dashboard).

Rollback

Diagnosing a Failed or Unexpected Operation

When a deploy goes wrong, or when an UpdateApplication call transitions to UPDATING and back without producing the expected new state, always start by collecting diagnostic context before initiating any recovery action or assuming success. Do not assume the operation succeeded just because the application returned to READY or RUNNING — UpdateApplication can transition through UPDATING and back even when the underlying operation failed, the new version was rolled back, or the change was rejected. Run the diagnostic flow first:

1. ListApplicationOperations — chronological history of all UpdateApplication, Maintenance, RollbackApplication, and other operations. Find the operation ID for the unexpected transition. 2. DescribeApplicationOperation on that operation ID — read OperationStatus (do not trust the application status alone) and especially statusDescription, which contains the actual failure reason. This is the single most informative diagnostic field MSF surfaces. 3. CloudWatch Logs for the application — read runtime errors that appear after the operation summary. Operation-level failures (IAM, parallelism limits, VPC) show up in statusDescription; runtime errors after a successful operation (e.g., the new code crashes on startup) show up in CloudWatch Logs.

Common failure categories surfaced this way: insufficient permissions, incompatible customer code, snapshot incompatibility on a Flink version upgrade, parallelism above maxParallelism, VPC misconfiguration, and MSF service issues (check AWS Health Dashboard). Each often points directly at the fix and may make rollback unnecessary.

Only after statusDescription is read should you decide whether to retry the operation, roll back, or fix the underlying issue. Retrying or rolling back blindly hides the root cause and tends to repeat the failure.

Rollback Paths

MSF has three rollback paths — choose the highest one available:

1. Automatic system rollback (opt-in via ApplicationSystemRollbackConfigurationUpdate.RollbackEnabledUpdate=true). Auto-reverts failed UpdateApplication, autoscaling, or version-upgrade operations to the previous running version. Triggers when the service detects code bugs, permission issues, snapshot incompatibility on Flink version upgrade, parallelism above maxParallelism, or VPC misconfiguration that fails Flink job startup. Important: this only fires when the operation itself fails. A deploy that completes successfully but produces wrong output downstream is not a failed operation from MSF's perspective — auto-rollback will not engage. Use the manual RollbackApplication API (path 2) for that case. If auto-rollback also fails, the app moves to READY.

2. Manual `RollbackApplication` API (always available, no opt-in). Reverts to the previous running version with its state. Use when:

• The deploy succeeded but the new version has downstream issues you only see in production (auto-rollback does not cover this case).
• The application is stuck in a transient state (e.g., long UPDATING).
• Auto-rollback was not enabled.

   aws kinesisanalyticsv2 rollback-application --application-name "$APP" \
     --current-application-version-id "$VERSION"
   aws kinesisanalyticsv2 describe-application-operation \
     --application-name "$APP" --operation-id "$OPERATION_ID"

3. Manual restore from a custom snapshot (last resort). Use only when both 1 and 2 are unavailable or have failed — for example, when the bad code change is older than the previous running version that RollbackApplication would target, or when both rollback paths returned errors. 1. stop-application --force and poll until READY. 2. update-application to point at a known-good previous S3 key. 3. start-application with ApplicationRestoreType=RESTORE_FROM_CUSTOM_SNAPSHOT and SnapshotName=<pre-deploy snapshot>.

This requires a pre-deploy snapshot. Take one before any code update so this fallback is available.

Operation visibility for any failed deploy or rollback: use ListApplicationOperations (chronological history of all UpdateApplication, Maintenance, RollbackApplication, and other operations) and DescribeApplicationOperation for the per-operation failure reason. Common error categories: customer code bugs (use rollback), permission issues (fix the role and retry), and MSF service issues (check AWS Health Dashboard).

Runtime Properties Update

Application code reads runtime properties via KinesisAnalyticsRuntime.getApplicationProperties(). Update them via EnvironmentPropertyUpdates — the application picks up the new values on next restart. They are organized by PropertyGroupId, which the application code uses to look up its property map.

aws kinesisanalyticsv2 update-application --application-name "$APP" \
  --current-application-version-id "$VERSION" \
  --application-configuration-update '{
    "EnvironmentPropertyUpdates": {"PropertyGroups": [{
      "PropertyGroupId": "FlinkApplicationProperties",
      "PropertyMap": {"input.stream": "new-stream"}
    }]}
  }'

Snapshots

Snapshot vs Checkpoint

Create

Snapshots can only be created from RUNNING. Creation is asynchronous — poll until READY.

NAME="snapshot-$(date +%Y%m%d-%H%M%S)"
aws kinesisanalyticsv2 create-application-snapshot \
  --application-name "$APP" --snapshot-name "$NAME"

# Poll until READY (timeout matters; large state can take 10+ min)
while true; do
  STATUS=$(aws kinesisanalyticsv2 list-application-snapshots --application-name "$APP" \
    --query 'SnapshotSummaries[?SnapshotName==`'"$NAME"'`].SnapshotStatus' --output text)
  [ "$STATUS" = "READY" ] && break
  [ "$STATUS" = "FAILED" ] && { echo "Snapshot failed"; exit 1; }
  sleep 5
done

Delete

delete-application-snapshot requires the exact SnapshotCreationTimestamp from list-application-snapshots. Cannot delete a snapshot in CREATING state — wait for READY or FAILED first.

Stuck CREATING

If a snapshot stays in CREATING for >10 minutes, the cause is usually:

• Backpressure slowing state serialization (check backPressuredTimeMsPerSecond)
• S3 permissions missing on execution role
• VPC NAT gateway down (no network path to S3)
• State too large for the snapshot timeout

Do not stop the application while a snapshot is CREATING — that risks state inconsistency. Wait or contact AWS Support.

Retention

Snapshots are billed at $0.023/GB-month and not auto-pruned. Implement retention:

A streaming app with checkpoint-sized snapshots and no retention will accumulate cost over months. Iterate list-application-snapshots, filter by SnapshotCreationTimestamp older than threshold, delete with the exact timestamp.

Delete Application

delete-application is irreversible and deletes all associated snapshots along with the application. There is no flag, grace period, or "soft delete" that preserves snapshots — they go with the app. The MSF console will warn you, but a CLI/SDK call will not.

Before calling `delete-application`, two requirements always apply:

1. The application MUST be in `READY` (stopped) state. A RUNNING application cannot be deleted; call stop-application first and poll until ApplicationStatus=READY. 2. The --create-timestamp argument MUST exactly match the value of ApplicationDetail.CreateTimestamp returned by describe-application. This is a guard against accidentally deleting a re-created same-named app and there is no way to bypass it.

To preserve state across a deletion, you must do one of these before calling delete-application — there is no way to recover snapshots after the fact:

• (a) Create a new application from the snapshot first. Use the existing snapshot as the basis for a new application via CreateApplication with the appropriate ApplicationConfiguration and run the new app from that snapshot. Only delete the original after the new app is verified.
• (b) Copy the underlying S3 checkpoint/snapshot data out-of-band. MSF stores snapshot state in S3 paths derived from your application; you can copy the relevant S3 prefixes to a bucket you control and reconstruct state later via RESTORE_FROM_CUSTOM_SNAPSHOT or by reading with the State Processor API. This is the fallback when option (a) isn't practical.

# 1. Confirm the app is READY (stop first if it isn't)
STATUS=$(aws kinesisanalyticsv2 describe-application --application-name "$APP" \
  --query 'ApplicationDetail.ApplicationStatus' --output text)

# 2. Pull the exact CreateTimestamp — pass this verbatim to delete-application
TIMESTAMP=$(aws kinesisanalyticsv2 describe-application --application-name "$APP" \
  --query 'ApplicationDetail.CreateTimestamp' --output text)

# 3. (Optional) Preserve state via option (a) or (b) above before deleting

# 4. Delete (irreversible — all snapshots gone)
aws kinesisanalyticsv2 delete-application --application-name "$APP" \
  --create-timestamp "$TIMESTAMP"

Pre-Mutation Checklist

Before any stop, code update, scale operation, or deletion:

1. Snapshot the application (RUNNING + poll until READY) 2. Confirm no snapshots are in CREATING (deletion / stop blocks them) 3. Verify the operation is reversible — if not, confirm with the user

Common Mistakes

References

• MSF API: UpdateApplication
• MSF API: RunConfiguration
• MSF Snapshots

Managed Service for Apache Flink Development Best Practices

Overview

This guide provides Managed Service for Apache Flink-optimized development patterns, anti-patterns, and best practices for building robust, performant, and secure Flink applications on Amazon Managed Service for Apache Flink. For existing applications, use the current user's Flink version. For new applications, assume Flink 2.2 and ask if the user has a preference.

Code examples in this guide use Flink 2.2 APIs by default, which are also compatible with Flink 1.20 unless noted otherwise. See flink-2x-migration.md for the complete migration reference.

Development Patterns

Best Practices for Managed Service for Apache Flink

Application Design:

• Design for KPU-based automatic scaling with service-level parallelism configuration
• Use appropriate parallelism levels as suggestions (Managed Service for Apache Flink service-level settings take precedence)
• Implement proper backpressure handling for Managed Service for Apache Flink's automatic scaling algorithms
• Design stateful operations with Managed Service for Apache Flink-managed checkpoint intervals in mind

Resource Management:

• Configure application for KPU-based resource allocation (1 vCPU, 4GB per KPU)
• Let Managed Service for Apache Flink manage checkpoint intervals and retention through service-level configuration
• Monitor resource utilization patterns through CloudWatch metrics
• Implement proper error handling that works with Managed Service for Apache Flink's automatic recovery

Monitoring and Alerting:

• Leverage integrated CloudWatch dashboards and metrics
• Configure Managed Service for Apache Flink-specific alarms for KPU utilization and throughput
• Monitor key performance metrics through Managed Service for Apache Flink console and CloudWatch
• Implement application health checks that integrate with Managed Service for Apache Flink monitoring

Managed Service for Apache Flink-Optimized Application Structure

Best Practice: Clean Application Architecture

public class MSFStreamingApp {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // NO checkpoint configuration in code - managed by MSF service
        DataStream<Event> events = env
            .fromSource(createKinesisSource(), WatermarkStrategy.forMonotonousTimestamps(), "kinesis-source")
            .uid("kinesis-source-uid");
        
        DataStream<ProcessedEvent> processed = events
            .keyBy(Event::getKey)
            .process(new EventProcessor())
            .name("event-processor")
            .uid("event-processor-uid");
        
        processed.sinkTo(createS3Sink())
            .name("s3-sink")
            .uid("s3-sink-uid");
        
        env.execute("MSF Streaming Application");
    }
}

fromSource()/sinkTo() are the recommended APIs for both Flink 1.20 and 2.2. The legacy addSource()/addSink() APIs are deprecated in 1.20 and removed in 2.x. See environment-setup.md for docker-compose.yml setup.

Anti-Pattern: Monolithic Processing

// AVOID: Single large operator doing everything
events.map(event -> {
    // Complex transformation logic
    // Multiple business rules
    // Data enrichment
    // Validation
    // Formatting
    return processedEvent;
}); // Hard to debug, scale, and maintain

For state management best practices (efficient state usage, TTL, state types, Managed Service for Apache Flink state management), see state-management.md.

For serialization best practices (performance hierarchy, POJO, Tuple, Avro, Protobuf, Kryo avoidance, state serialization, anti-patterns), see serialization-guide.md.

Performance Best Practices

KPU-Based Resource Configuration

Best Practice: Managed Service for Apache Flink KPU-Optimized Applications

// Application code should be KPU-agnostic
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// DO NOT set parallelism in application code for Managed Service for Apache Flink deployment
// Managed Service for Apache Flink manages parallelism through KPU configuration

// For local Docker development only:
if (isLocalDevelopment()) {
    int localParallelism = Math.max(1, Runtime.getRuntime().availableProcessors() - 1);
    env.setParallelism(localParallelism);
}

// Set operator-specific parallelism only when business logic or infrastructure requires it
// E.g. you may set parallelism on a Kafka source operator to be equal to number of partitions, if the overall app parallelism is higher than the number of partitions (lower parallelism for source, but enable high parallelism for processing operators with keyBy or similar operators that spread load)
dataStream
    .keyBy(Event::getPartitionKey)
    .process(new HeavyProcessor())
    // Only set if this operator specifically needs different parallelism
    .setParallelism(5);

Error Handling and Recovery in Managed Service for Apache Flink

Best Practice: Use Side Outputs for dead letter queues for bad data handling and dependency failures

public class RobustProcessor extends ProcessFunction<Event, ProcessedEvent> {
    public static final OutputTag<Event> DEAD_LETTER_TAG =
            new OutputTag<Event>("dead-letter") {};
    private transient Counter errorCounter;
    
    @Override
    public void open(OpenContext openContext) throws Exception {
        errorCounter = getRuntimeContext().getMetricGroup().counter("processing_errors");
    }
    
    @Override
    public void processElement(Event event, Context ctx, Collector<ProcessedEvent> out) {
        try {
            // Validate input
            if (!isValidEvent(event)) {
                LOG.warn("Invalid event received: {}", event);
                ctx.output(DEAD_LETTER_TAG, event);
                return;
            }
            
            ProcessedEvent result = processEvent(event);
            out.collect(result);
            
        } catch (TransientException e) {
            // Let Managed Service for Apache Flink handle transient errors through restart strategy
            LOG.warn("Transient error processing event {}, Managed Service for Apache Flink will retry", event.getId(), e);
            throw e; // Managed Service for Apache Flink restart strategy handles this
            
        } catch (Exception e) {
            // Handle permanent errors gracefully
            LOG.error("Permanent error processing event {}", event.getId(), e);
            errorCounter.inc();
            ctx.output(DEAD_LETTER_TAG, event);
            // Don't throw - continue processing other events
        }
    }
}

Configuration Optimization

Critical Principle: Managed Service for Apache Flink applications must clearly separate local development configuration from Managed Service for Apache Flink service-level configuration. Managed Service for Apache Flink manages all advance Flink runtime parameters (i.e. FLINK_PROPERTIES configs such as state.backend and restart-strategy) and should not be a consideration for developers outside of local development.

• Local Docker Configuration: Used only for Kiro-based development with Docker containers
• Managed Service for Apache Flink Service Configuration: Managed through Managed Service for Apache Flink console and service APIs, not in application code
• Application Code: Should be environment-agnostic and avoid hardcoded infrastructure settings
• Advanced Configs: Configurations for the Flink runtime are managed by Managed Service for Apache Flink and should not be a consideration for developers outside of local development - some configurations can be updated via AWS Support Case requests (such as state.backend for RocksDB vs. HashMap) but have significant considerations to weigh for application health and stability and in general should be managed by Managed Service for Apache Flink

Environment-Specific Configuration Management

Best Practice: Clean Configuration Separation

import com.amazonaws.services.kinesisanalytics.runtime.KinesisAnalyticsRuntime;

public class FlinkStreamingJob {
    private static final String LOCAL_PROPS = "flink-application-properties-dev.json";
    
    private static boolean isLocal(StreamExecutionEnvironment env) {
        String runtime = System.getenv("RUNTIME_ENVIRONMENT");
        return env instanceof LocalStreamEnvironment || "local".equalsIgnoreCase(runtime);
    }

    private static Map<String, Properties> loadApplicationProperties(StreamExecutionEnvironment env) throws IOException {
        if (isLocal(env)) {
            InputStream input = FlinkStreamingJob.class.getClassLoader().getResourceAsStream(LOCAL_PROPS);
            if (input == null) throw new IOException("Unable to find " + LOCAL_PROPS);
            java.nio.file.Path tempFile = java.nio.file.Files.createTempFile("flink-app-props", ".json");
            java.nio.file.Files.copy(input, tempFile, java.nio.file.StandardCopyOption.REPLACE_EXISTING);
            input.close();
            tempFile.toFile().deleteOnExit();
            return KinesisAnalyticsRuntime.getApplicationProperties(tempFile.toString());
        } else {
            return KinesisAnalyticsRuntime.getApplicationProperties();
        }
    }

    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        if (isLocal(env)) {
            env.enableCheckpointing(10_000);
            env.setParallelism(3);
        }
        final Map<String, Properties> applicationProperties = loadApplicationProperties(env);
    }
}

In Managed Service for Apache Flink, application properties are configured at the Application level with property groups (same structure as the local JSON file).

Anti-Patterns to Avoid

Deployment Anti-Patterns

1. Single-Stack IaC Without Pre-Uploaded JAR The MSF application resource validates that the JAR exists in S3 at creation time. A single CloudFormation stack (or equivalent) that creates the S3 bucket and the MSF application together will fail because the JAR hasn't been uploaded yet. Always use a two-phase deployment: deploy infrastructure first, upload the JAR, then deploy the application resource. See IaC and Deployment Guide for patterns.

Configuration Anti-Patterns

1. Hardcoded Infrastructure Configuration in Application Code 2. App-Level Savepoint Management

Performance Anti-Patterns

1. KPU-Unaware Parallelism Configuration

   // AVOID: Fixed parallelism that doesn't align with KPU model
   env.setParallelism(7); // Doesn't align with KPU scaling
   dataStream.setParallelism(13); // Arbitrary parallelism, only set when operator requires custom parallelism

2. Excessive Rebalancing

   // AVOID: Unnecessary rebalance operations
   stream.rebalance().map(...).rebalance().filter(...);
   // Breaks Managed Service for Apache Flink's automatic load balancing

3. Blocking Operations in Processing Functions

   // AVOID: Synchronous external calls that block KPU resources, use Async functions instead
   public void processElement(Event event, Context ctx, Collector<Result> out) {
       Result result = externalService.blockingCall(event); // Blocks KPU
       out.collect(result);
   }

4. Large State Objects Without TTL

   // AVOID: Unbounded state growth
   private transient ListState<Event> allEvents; // Can exhaust KPU memory
   private transient MapState<String, LargeObject> cache; // No TTL configured

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.

Checkpoint Impact on Resources

Overview

This guide covers how checkpointing affects Managed Service for Apache Flink application resources, including checkpoint size and memory consumption, frequency vs CPU/network trade-offs, checkpoint duration exceeding interval, and OOM/GC diagnostic steps.

For KPU sizing, operator parallelism tuning, and MSF configuration overrides, see resource-optimization.md.

Checkpoint Size and KPU Memory Consumption

During a checkpoint, Flink snapshots operator state and uploads it to S3 (Managed Service for Apache Flink-managed bucket). This process consumes memory and CPU on each TaskManager:

• RocksDB state backend (default on Managed Service for Apache Flink): For incremental checkpoints, Flink uploads only new SST files created since the last checkpoint to S3. The async phase (uploading to S3) does not block record processing but does consume network bandwidth and some memory for upload buffers. The sync phase (snapshotting RocksDB) briefly blocks the subtask.
• Larger state per TaskManager means more data to read and upload, increasing memory and network pressure during the checkpoint window. Skewed state distribution can cause individual TaskManagers to become bottlenecks even when aggregate resources are sufficient.
• There is no official AWS or Flink guidance specifying a fixed percentage of KPU memory to reserve for checkpoint overhead. The actual impact depends on state size, checkpoint type (incremental vs full), and upload concurrency. Monitor lastCheckpointDuration and heapMemoryUtilization during checkpoint windows to assess whether checkpoint overhead is causing memory pressure.

Checkpoint Frequency vs CPU and Network Bandwidth

Checkpoint frequency is configured via the Managed Service for Apache Flink CheckpointInterval setting (default: 60000ms / 60 seconds). The MinPauseBetweenCheckpoints (default: 5000ms) prevents continuous checkpointing when a checkpoint takes longer than the interval. You must set ConfigurationType to CUSTOM to modify these values. More frequent checkpoints:

• Increase CPU usage: each checkpoint triggers RocksDB file reads and S3 uploads across all TaskManagers. With incremental checkpoints (Managed Service for Apache Flink default), the CPU impact is proportional to state changes, not total state size.
• Increase network bandwidth: checkpoint data flows from TaskManagers to S3. With large state and frequent checkpoints, this can compete with data processing traffic.
• Reduce recovery time: more frequent checkpoints mean less data to replay from sources after a failure.

Trade-off guidance per Flink docs: When checkpoints frequently take longer than the base interval, the system ends up constantly taking checkpoints, tying up resources and reducing operator progress. Use MinPauseBetweenCheckpoints to prevent this. The Flink documentation does not prescribe specific interval ranges for state sizes — tune based on observed lastCheckpointDuration relative to your interval, and ensure checkpoints complete well within the interval with room to spare.

Checkpoint Duration Exceeding Interval

Symptoms:

• lastCheckpointDuration in CloudWatch approaches or exceeds the configured checkpoint interval
• numberOfInProgressCheckpoints stays > 0 for extended periods
• Increasing backPressuredTimeMsPerSecond during checkpoint windows
• millisBehindLatest (Kinesis) or consumer lag (Kafka) grows during checkpoints

Consequences:

• By default, the next checkpoint is triggered immediately once the ongoing one completes. With MinPauseBetweenCheckpoints (default 5s on Managed Service for Apache Flink), there's a minimum gap, but the system can still end up constantly checkpointing.
• For aligned checkpoints (the default mode), checkpoint barriers can cause channels to block while waiting for alignment, contributing to backpressure. Unaligned checkpoints (available from Flink 1.15+, requestable via AWS support for Managed Service for Apache Flink) avoid this alignment delay but have other trade-offs.
• In extreme cases, checkpoint timeouts trigger checkpoint failures, and repeated failures can cause application restarts.

Remediation:

1. Verify incremental checkpoints are active (they are enabled by default on Managed Service for Apache Flink). If for some reason they were overridden, re-enable them — incremental checkpoints only upload state changes since the last checkpoint, dramatically reducing upload size for large state. 2. Increase checkpoint interval via the UpdateApplication API with ConfigurationType: CUSTOM to give more time for completion. Also consider increasing MinPauseBetweenCheckpoints. 3. Add KPUs to spread state across more TaskManagers, reducing per-TaskManager checkpoint size. 4. Reduce state size: add or tighten TTL on keyed state, reduce key cardinality, or use more compact serialization (POJO over Kryo). 5. Request RocksDB tuning overrides via AWS support if compaction or read amplification is the bottleneck. 6. Consider buffer debloating — request enablement via AWS support case. This can help applications with backpressure-related checkpoint issues.

OOM and GC Diagnostic Steps

If the application throws OutOfMemoryError or shows sustained high GC activity:

1. Check `heapMemoryUtilization` in CloudWatch. If sustained > 80%, the application needs investigation and likely a scale-up (see monitoring-and-metrics.md for graduated thresholds: healthy ≤ 75%, scale-up signal > 80% sustained, critical alarm > 90%). 2. Check `lastCheckpointSize` and `lastCheckpointDuration`. Large checkpoints consume significant heap during snapshot creation. 3. Review state TTL configuration. Missing or overly long TTL causes state to accumulate indefinitely. 4. Check for Kryo serialization fallbacks. Kryo uses more memory than POJO serialization. Look for log messages: "Class ... cannot be used as a POJO type". 5. Review operator state usage. Use Flink Web UI to check state size per operator. Identify operators with disproportionately large state.

If the root cause is legitimate memory pressure after optimization:

• Request a JVM heap size increase via AWS support
• Consider increasing KPU count to spread state across more TaskManagers
• Request RocksDB block cache increase if state reads are the bottleneck (high cache miss rate)

Flink Java Dependency Management

This document provides the complete Maven dependency configuration for Flink projects on Amazon Managed Service for Apache Flink. For new applications, default to Flink 2.2. For existing applications, use the user's current Flink version.

For Kinesis-specific API usage and code examples, see kinesis-connector-guide.md.

Version-Specific Dependency Mapping

Not all connector versions are published for every Flink version. Use the correct combination from this table:

Flink 2.2 additional notes:

• flink-statebackend-forst is available as an alternative to flink-statebackend-rocksdb for disaggregated state management
• Logging uses log4j-slf4j-impl (same as 1.20; older Flink versions used slf4j-log4j12)
• See flink-2x-migration.md for API changes that affect application code (e.g., open(OpenContext), Duration instead of Time, removed SourceFunction/SinkFunction)

Example pom.xml for a Flink project

The example below defaults to Flink 2.2 properties (recommended for new applications). To use Flink 1.20 for existing applications, replace the properties block with the Flink 1.20 values from the table above.

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xmlns="http://maven.apache.org/POM/4.0.0"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <!-- REPLACE: Use your application's package name -->
    <groupId>com.example</groupId>
    <!-- REPLACE: Use your application's artifact name -->
    <artifactId>my-flink-app</artifactId>
    <!-- Increment version as new changes are released -->
    <version>1.0</version>
    <packaging>jar</packaging>

    <!-- ============================================================
         Flink 2.2 properties (default for new applications)
         For Flink 1.20, replace with:
           target.java.version  = 11
           flink.version        = 1.20.3
           kafka.version        = 3.4.0-1.20
           kinesis-streams.version = 5.1.0-1.20
           maven.compiler.plugin.version = 3.8.1
         ============================================================ -->
    <properties>
        <!-- Build configs -->
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <buildDirectory>${project.basedir}/target</buildDirectory>
        <jar.finalName>${project.name}-${project.version}</jar.finalName>
        <target.java.version>17</target.java.version>
        <maven.compiler.source>${target.java.version}</maven.compiler.source>
        <maven.compiler.target>${target.java.version}</maven.compiler.target>

        <!-- Dependency versions -->
        <flink.version>2.2.0</flink.version>
        <kda.runtime.version>1.2.0</kda.runtime.version>
        <log4j.version>2.23.1</log4j.version>
        <!-- Not all kafka / flink version combinations are published on Maven -->
        <kafka.version>4.0.1-2.0</kafka.version>
        <msk-iam-auth.version>2.3.5</msk-iam-auth.version>
        <!-- Not all Kinesis / flink version combinations are published on Maven -->
        <kinesis-streams.version>6.0.0-2.0</kinesis-streams.version>
        <maven.compiler.plugin.version>3.11.0</maven.compiler.plugin.version>
    </properties>

    <!-- Java SDK for AWS required in general -->
    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>com.amazonaws</groupId>
                <artifactId>aws-java-sdk-bom</artifactId>
                <version>1.12.677</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>

    <dependencies>
        <!-- Apache Flink dependencies required for most projects -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-runtime-web</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-base</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>

        <!-- Flink Table Runtime — required when application code references
             internal Table Runtime types (InternalTypeInfo, RowDataSerializer,
             internal RowData converters). Commonly needed by Iceberg sinks
             that emit/consume RowData on a side output. Provided by MSF at
             runtime.

             Note: bridge != runtime. flink-table-api-java-bridge gives you
             the public Table API/SQL surface (DDL, TableEnvironment,
             DataStream<->Table conversion). flink-table-runtime gives you
             the internal data-structure types referenced by sinks and
             custom serializers. -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-runtime</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>

        <!-- Managed Service for Apache Flink Runtime required for all projects -->
        <dependency>
            <groupId>com.amazonaws</groupId>
            <artifactId>aws-kinesisanalytics-runtime</artifactId>
            <version>${kda.runtime.version}</version>
            <scope>provided</scope>
        </dependency>

        <!-- Kafka Connector required for Kafka source/sink projects -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-kafka</artifactId>
            <version>${kafka.version}</version>
        </dependency>
        <!-- MSK IAM Auth required for Kafka projects using MSK IAM -->
        <dependency>
            <groupId>software.amazon.msk</groupId>
            <artifactId>aws-msk-iam-auth</artifactId>
            <version>${msk-iam-auth.version}</version>
        </dependency>

        <!-- Kinesis connector for Kinesis Streams source/sink projects -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-aws-kinesis-streams</artifactId>
            <version>${kinesis-streams.version}</version>
        </dependency>


        <!-- RocksDB State Backend required for all projects -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-statebackend-rocksdb</artifactId>
            <version>${flink.version}</version>
        </dependency>

        <!-- JSON Processing required for JSON projects -->
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
            <version>2.15.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-json</artifactId>
            <version>${flink.version}</version>
        </dependency>


        <!-- Logging - generally required -->
        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-slf4j-impl</artifactId>
            <version>${log4j.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-api</artifactId>
            <version>${log4j.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-core</artifactId>
            <version>${log4j.version}</version>
        </dependency>
    </dependencies>


    <!-- Profile for local testing - includes provided dependencies -->
    <profiles>
        <profile>
            <id>local</id>
            <dependencies>
                <dependency>
                    <groupId>org.apache.flink</groupId>
                    <artifactId>flink-clients</artifactId>
                    <version>${flink.version}</version>
                    <scope>compile</scope>
                </dependency>
                <dependency>
                    <groupId>org.apache.flink</groupId>
                    <artifactId>flink-streaming-java</artifactId>
                    <version>${flink.version}</version>
                    <scope>compile</scope>
                </dependency>
                <dependency>
                    <groupId>org.apache.flink</groupId>
                    <artifactId>flink-runtime-web</artifactId>
                    <version>${flink.version}</version>
                    <scope>compile</scope>
                </dependency>
                <dependency>
                    <groupId>org.apache.flink</groupId>
                    <artifactId>flink-connector-base</artifactId>
                    <version>${flink.version}</version>
                    <scope>compile</scope>
                </dependency>

                <!-- Managed Service for Apache Flink Runtime -->
                <dependency>
                    <groupId>com.amazonaws</groupId>
                    <artifactId>aws-kinesisanalytics-runtime</artifactId>
                    <version>${kda.runtime.version}</version>
                    <scope>compile</scope>
                </dependency>
            </dependencies>
        </profile>
    </profiles>

    <!-- Most of the below is boilerplate for managing shaded JARs -->
    <build>
        <directory>${buildDirectory}</directory>
        <finalName>${jar.finalName}</finalName>
        <plugins>
            <!-- Java Compiler -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>${maven.compiler.plugin.version}</version>
                <configuration>
                    <source>${target.java.version}</source>
                    <target>${target.java.version}</target>
                </configuration>
            </plugin>
            <!-- Maven Shade Plugin -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>3.2.1</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <artifactSet>
                                <excludes>
                                    <exclude>org.apache.flink:force-shading</exclude>
                                    <exclude>com.google.code.findbugs:jsr305</exclude>
                                </excludes>
                            </artifactSet>
                            <filters>
                                <filter>
                                    <artifact>*:*</artifact>
                                    <excludes>
                                        <exclude>META-INF/*.SF</exclude>
                                        <exclude>META-INF/*.DSA</exclude>
                                        <exclude>META-INF/*.RSA</exclude>
                                    </excludes>
                                </filter>
                            </filters>
                            <transformers>
                                <transformer
                                        implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
                                <transformer
                                        implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                    <!-- REPLACE: Use your application's fully qualified main class -->
                                    <mainClass>com.example.MyFlinkJob</mainClass>
                                </transformer>
                            </transformers>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

Managed Service for Apache Flink Development Environment Setup

Overview

This guide provides setup instructions for Apache Flink development targeting Amazon Managed Service for Apache Flink deployment. The setup ensures compatibility with Managed Service for Apache Flink requirements while enabling efficient local development and testing workflows using Docker containers that mirror the Managed Service for Apache Flink environment. For new applications, default to Flink 2.2. For existing applications, use the user's current Flink version.

Prerequisites

Before starting, ensure you have:

• Administrative access to your development machine
• Stable internet connection for downloading dependencies
• At least 16GB RAM and 30GB free disk space (for Docker containers)
• Docker Desktop installed and running
• Java 11 (for Flink 1.20) or Java 17 (for Flink 2.2) installed
• Basic familiarity with command-line operations and Docker concepts

Core Development Environment

1. Docker Desktop Installation

Docker is required for running Flink, Kafka, LocalStack, and other infrastructure locally for testing. Ensure Docker is installed.

Verification:

docker --version
docker-compose --version
docker run hello-world

2. Java Development Kit (JDK) for Container Development

While Flink runs in containers, JDK is still needed for compilation and IDE support. Flink 1.20 requires Java 11. Flink 2.2 requires Java 17 (Java 21 also supported).

Verification:

java -version
javac -version
echo $JAVA_HOME

Docker-Based Flink Development Environment

Docker Compose Configuration

Create a comprehensive Docker Compose setup that mirrors the Managed Service for Apache Flink environment. Adjust the Flink image tag to match your target version:

• Flink 1.20: flink:1.20-java11
• Flink 2.2: flink:2.2-java17

Create `docker-compose.yml` in your project root:

version: '3.8'

# Set FLINK_IMAGE_TAG in your environment or .env file:
#   Flink 1.20: FLINK_IMAGE_TAG=1.20-java11
#   Flink 2.2:  FLINK_IMAGE_TAG=2.2-java17

services:
  # Flink JobManager
  jobmanager:
    image: flink:${FLINK_IMAGE_TAG:-2.2-java17}
    hostname: jobmanager
    container_name: flink-jobmanager
    ports:
      - "8081:8081"
    command: jobmanager
    environment:
      - |
        FLINK_PROPERTIES=
        jobmanager.rpc.address: jobmanager
        jobmanager.memory.process.size: 1600m
        jobmanager.execution.failover-strategy: region
      - RUNTIME_ENVIRONMENT=local
    volumes:
      - ./flink-jobs:/opt/flink/jobs
      - flink-checkpoints:/tmp/flink-checkpoints
      - flink-savepoints:/tmp/flink-savepoints
    networks:
      - flink-network

  # Flink TaskManager
  taskmanager:
    image: flink:${FLINK_IMAGE_TAG:-2.2-java17}
    depends_on:
      - jobmanager
    command: taskmanager
    scale: 2
    environment:
      - |
        FLINK_PROPERTIES=
        jobmanager.rpc.address: jobmanager
        taskmanager.numberOfTaskSlots: 2
        parallelism.default: 2        
        taskmanager.memory.process.size: 1728m
        taskmanager.memory.managed.fraction: 0.4
        state.backend: rocksdb
        state.checkpoints.dir: file:///tmp/flink-checkpoints
        state.savepoints.dir: file:///tmp/flink-savepoints
        execution.checkpointing.interval: 60000
        execution.checkpointing.mode: EXACTLY_ONCE
        restart-strategy: exponential-delay
        restart-strategy.exponential-delay.initial-backoff: 10s
        restart-strategy.exponential-delay.max-backoff: 2min
        restart-strategy.exponential-delay.backoff-multiplier: 2.0
      - RUNTIME_ENVIRONMENT=local
    volumes:
      - ./flink-jobs:/opt/flink/jobs
      - flink-checkpoints:/tmp/flink-checkpoints
      - flink-savepoints:/tmp/flink-savepoints
    networks:
      - flink-network

  # Kafka (for streaming data sources)
  zookeeper:
    image: confluentinc/cp-zookeeper:7.4.0
    hostname: zookeeper
    container_name: zookeeper
    ports:
      - "2181:2181"
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    networks:
      - flink-network

  kafka:
    image: confluentinc/cp-kafka:7.4.0
    hostname: kafka
    container_name: kafka
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
      - "9101:9101"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_JMX_PORT: 9101
      KAFKA_JMX_HOSTNAME: localhost
    networks:
      - flink-network

  # LocalStack (for local AWS service emulation: Kinesis, S3, DynamoDB, etc.)
  localstack:
    image: localstack/localstack:3.5.0
    container_name: localstack
    ports:
      - "4566:4566"
    environment:
      SERVICES: kinesis,s3,dynamodb,cloudwatch
      DEFAULT_REGION: us-east-1
    networks:
      - flink-network

volumes:
  flink-checkpoints:
  flink-savepoints:

networks:
  flink-network:
    driver: bridge

Starting the Development Environment

1. Create Project Structure:

# Create Managed Service for Apache Flink project directory
mkdir my-msf-app
cd my-msf-app

# Create required directories
mkdir -p flink-jobs src/main/java src/test/java

# Copy docker-compose.yml to project root

2. Start Docker Environment:

# Start all services
docker-compose up -d

# Verify services are running
docker-compose ps

# Check taskmanager logs for submitted Flink jobs 
docker-compose logs taskmanager

# Check Flink Web UI
open http://localhost:8081

# Check Kafka
docker-compose exec kafka kafka-topics --bootstrap-server localhost:9092 --list

Build Tools and Dependencies

Maven Project Template for Managed Service for Apache Flink

Build the JAR in local mode to include Managed Service for Apache Flink provided dependencies:

# For Flink 1.20 (Java 11):
mvn clean package -Plocal -q

# For Flink 2.2 (Java 17):
# Set JAVA_HOME to your Java 17 installation path before building
export JAVA_HOME=<path-to-java-17>
mvn clean package -Plocal -q

cp target/my-flink-app-1.0.jar flink-jobs/

docker exec flink-jobmanager flink run /opt/flink/jobs/my-flink-app-1.0.jar

See dependency-management.md for examples of dependencies and pom.xml requried for Managed Service for Apache Flink.

Managed Service for Apache Flink-Specific Tools

AWS CLI Managed Service for Apache Flink Commands:

# List Managed Service for Apache Flink applications
aws kinesisanalyticsv2 list-applications

# Describe Managed Service for Apache Flink application
aws kinesisanalyticsv2 describe-application --application-name my-app

# Create savepoint
aws kinesisanalyticsv2 create-application-snapshot --application-name my-app --snapshot-name my-snapshot

Foundation Operations: Quotas, Service Selection, and Architecture

Overview

Cross-cutting operational knowledge: MSF service quotas, ENI capacity planning for VPC apps, MSF vs EMR Flink decision, and source/sink selection. Use when planning capacity, choosing the right streaming service, or designing a new pipeline.

CLI and CloudWatch Identifiers

The CLI/SDK is the only identifier that uses the v2 suffix. Service Quotas, IAM actions, the CloudWatch namespace, and the trust principal all use the legacy kinesisanalytics name. Treating the v2 form as the "default" and applying it everywhere is the single most common source of permission failures, empty metric results, missing service-quota lookups, and trust policy errors.

What Goes In the Execution Role (and What Does NOT)

The MSF execution role is assumed by the MSF service to access your data plane resources on behalf of the application. It is not used by the application code itself, and it does not call MSF's own control plane. The principle of least privilege follows from that:

When generating an execution role, default to omitting kinesisanalytics:* and cloudwatch:PutMetricData unless the user explicitly says they emit custom metrics. The CloudWatch metrics you see in the AWS console come from the service, not from this role.

Service Quotas

Check current values:

aws service-quotas list-aws-default-service-quotas \
  --service-code kinesisanalytics --region "$REGION"

The Service Quotas service-code is kinesisanalytics (legacy name), not kinesisanalyticsv2. The KPUs-per-application quota code is L-3A88E041 (Apache Flink Kinesis Processing Units (KPUs)).

Request increases via request-service-quota-increase with the quota code from the list output. Increases for KPU and snapshot quotas typically approve within a business day.

ENI Capacity Planning (VPC Apps)

MSF creates one ENI per allocated KPU, in each subnet, for VPC-enabled applications. Subnet sizing and the regional ENI quota both matter.

ENIs_per_subnet = KPUs
required_subnet_IPs = ENIs_per_subnet + 20% headroom for scaling and rolling restarts

Example: 16 KPU app 16 ENIs per subnet. Allow 20 available IPs per subnet to account for 20% buffer. A /28 subnet (11 usable IPs) is too small; use /27 (27 usable) or larger.

Regional ENI quota (vpc service code, Network interfaces per region) defaults to 5,000. Each VPC-enabled MSF KPU consumes one. Multiple large MSF apps in the same region can pressure this quota — check before deploying. Other services in the VPC (Lambda, ECS, RDS) also consume ENIs from the same quota.

Subnets should span at least 2 AZs for fault tolerance.

When VPC Is Required

VPC apps without a NAT gateway lose access to public AWS endpoints (CloudWatch, S3 if not using gateway endpoint, Kinesis). Symptoms: silent failure of metric publishing, S3 checkpoint failures, deserialization errors trying to call schema registry.

MSF vs EMR Flink

Choose MSF when: zero infrastructure management, single-job-per-app is acceptable, parallelism fits within KPU limits, fast iteration matters.

Choose EMR Flink when: custom Flink connectors not in MSF, multiple Flink jobs sharing infrastructure, specific Flink version or patch control, parallelism exceeds 256, fine-grained CPU/memory ratios needed.

Cost crossover: EMR is typically cheaper at large scale (10+ KPU equivalent, 24/7) due to EC2 commitment savings, but the operational overhead (cluster patching, scaling, monitoring) typically erases the savings unless EMR expertise already exists.

Source Selection: KDS vs MSK vs S3

KDS pitfalls: shard count drives parallelism; shard count is hard to change post-creation. Cost scales linearly with shard count, not throughput.

MSK pitfalls: broker provisioning takes hours; cross-AZ replication doubles network cost; SASL/IAM auth requires careful security group setup.

S3 source pitfalls: no event-time ordering across files unless designed in; file enumeration is the bottleneck for large prefix counts.

For source-side EFO and Kinesis polling tradeoffs, see kinesis-efo-guide.md. For MSK setup, see the iac-and-deployment.md deployment patterns.

Sink Selection

S3 small files anti-pattern: Streaming writers commit on every checkpoint, creating one file per checkpoint per partition. With 60s checkpoints and 8 partitions, a job writes 11,520 files/day. Use Iceberg with compaction, or batch via Firehose, or increase checkpoint interval. See iceberg-tuning-and-operations.md.

RDS/Aurora sink anti-pattern: Per-record JDBC writes are 10–100× slower than batched writes. Use the JDBC sink with batching enabled. RDS connection pool limits (typically 5,000) cap effective parallelism.

Architecture Patterns

Co-location and Cross-AZ Cost

For private connections (MSK, RDS, OpenSearch in VPC), MSF KPUs must be in the same VPC or peered VPC as the source/sink. Cross-AZ data transfer between MSF and MSK is always billed at $0.01/GB each direction — at 100 MB/s sustained that is $26K/year. Place MSF subnets in the same AZs as MSK brokers and use rack-aware producers/consumers where supported.

Common Mistakes

References

• MSF Quotas
• MSF VPC Configuration
• MSF Pricing
• Cross-AZ Data Transfer Pricing