Amazon Managed Service for Apache Flink

Amazon Managed Service for Apache Flink (kinesisanalyticsv2, formerly Kinesis Data Analytics v2) on fakecloud: full 33-operation control plane for SQL and Flink streaming applications, versions, snapshots, operations, VPC and CloudWatch logging configuration, with persistence.

fakecloud implements Amazon Managed Service for Apache Flink (kinesisanalyticsv2, formerly Kinesis Data Analytics v2), the managed streaming-application service. All 33 operations from the AWS Smithy model ship now, backed by account-partitioned state that persists across restarts in persistent mode. The wire protocol is awsJson1.1 (x-amz-target KinesisAnalytics_20180523.<Op>), signing as kinesisanalytics.

Supported features

  • Applications (CreateApplication, DescribeApplication, UpdateApplication, DeleteApplication, ListApplications). Both application flavors are modeled: SQL applications (RuntimeEnvironment SQL-1_0, with inputs/outputs/reference data sources) and Flink applications (FLINK-1_15 ... FLINK-1_20, with ApplicationCodeConfiguration, EnvironmentProperties, and checkpoint/monitoring/parallelism configuration). Whatever configuration is created or updated is persisted and echoed back on describe. Applications carry arn:aws:kinesisanalytics:<region>:<account>:application/<name> ARNs.
  • Lifecycle (StartApplication, StopApplication, RollbackApplication). For a Flink-flavor application with a JAR in S3, StartApplication spawns a REAL Apache Flink job in a Docker container (see Data plane below) and only reaches RUNNING once Flink reports the job RUNNING; StopApplication cancels the real job and tears the container down. In the degraded / control-plane-only path (no container runtime, or a SQL app), StartApplication moves READY -> STARTING -> RUNNING (settling on the next describe) and StopApplication moves RUNNING -> STOPPING -> READY. Force is supported; RollbackApplication restores the previous version's configuration into a new version.
  • Versioning (DescribeApplicationVersion, ListApplicationVersions). Every configuration-changing operation increments ApplicationVersionId and records the full version history. CurrentApplicationVersionId and ConditionalToken optimistic-concurrency checks are enforced.
  • Snapshots (CreateApplicationSnapshot, DescribeApplicationSnapshot, ListApplicationSnapshots, DeleteApplicationSnapshot) capture the application version and settle CREATING -> READY.
  • Operations (DescribeApplicationOperation, ListApplicationOperations). Each Start/Stop/Update/Rollback records an async operation that settles SUCCESSFUL.
  • CloudWatch logging and VPC configuration (AddApplicationCloudWatchLoggingOption, DeleteApplicationCloudWatchLoggingOption, AddApplicationVpcConfiguration, DeleteApplicationVpcConfiguration) mint ids, bump the version, and are reflected in DescribeApplication.
  • SQL sub-resources (AddApplicationInput, AddApplicationInputProcessingConfiguration, DeleteApplicationInputProcessingConfiguration, AddApplicationOutput, DeleteApplicationOutput, AddApplicationReferenceDataSource, DeleteApplicationReferenceDataSource) assign ids and update the version.
  • Schema discovery (DiscoverInputSchema) infers a record format and columns from the provided input.
  • Presigned dashboard URL (CreateApplicationPresignedUrl) returns the REAL reachable Flink Web Dashboard / REST base URL when a Flink app is running against the live container runtime, and a well-formed synthesized authorized URL otherwise.
  • Maintenance windows (UpdateApplicationMaintenanceConfiguration).
  • Tagging (TagResource, UntagResource, ListTagsForResource), ARN-keyed; CreateApplication honors inline Tags.

100% conformance: all 1,232 generated Smithy probe variants pass.

Data plane

Starting a Flink-flavor application (RuntimeEnvironment FLINK-1_x) whose ApplicationCodeConfiguration points at a JAR in a fakecloud S3 bucket runs a GENUINE Apache Flink job, the same Docker-backed data-plane bar Amazon MQ, MSK, RDS, ElastiCache, and Lambda meet:

  • StartApplication spawns a real flink:1.19 session cluster (a JobManager
    • TaskManager) in a container, publishing the Flink REST port 8081 on a free host port.
  • fakecloud fetches the application's code JAR from its own S3 service, uploads it to the cluster via the Flink REST API (POST /jars/upload), and submits it (POST /jars/{id}/run) with the configured parallelism.
  • The application settles to RUNNING only once the real Flink job reaches RUNNING (polled via GET /jobs/{id}); DescribeApplication reflects the live job status.
  • CreateApplicationPresignedUrl returns the cluster's reachable dashboard/REST URL, so you can drive the live JobManager directly.
  • StopApplication cancels the real job (PATCH /jobs/{id}?mode=cancel) and tears the container down; DeleteApplication reaps it. Persisted RUNNING apps re-attach their container on restart (no RUNNING-with-dead-container).

The backing runtime is used automatically when a container CLI (Docker/Podman) is available. It is skipped (and the pure control-plane state machine used instead) when none is present or when FAKECLOUD_KINESISANALYTICSV2_DISABLE_BACKEND=1 is set; the image is overridable via FAKECLOUD_KINESISANALYTICSV2_IMAGE. The end-to-end Docker-backed job is proven by the flink-runtime CI job, gated on FAKECLOUD_E2E_FLINK=1.

Honest scope: SQL applications do not run for real. A SQL-flavor application (RuntimeEnvironment SQL-1_0) keeps the control-plane state machine (STARTING -> RUNNING in memory) — running arbitrary SQL as a live Flink job needs the Flink SQL gateway, which is out of scope for now. Only Flink-flavor apps with a JAR get the real runtime.

CloudFormation

fakecloud provisions every AWS::KinesisAnalyticsV2::* resource type through CloudFormation:

  • AWS::KinesisAnalyticsV2::Application
  • AWS::KinesisAnalyticsV2::ApplicationOutput
  • AWS::KinesisAnalyticsV2::ApplicationReferenceDataSource
  • AWS::KinesisAnalyticsV2::ApplicationCloudWatchLoggingOption

Each is written through to the kinesisanalyticsv2 service via the same shared record builders the direct API uses, so a CFN-provisioned application (and its outputs / reference data sources / CloudWatch logging options) reads back identically on DescribeApplication and persists across a restart, exactly like its direct-API equivalent.

Ref follows the AWS resource spec: an Application Ref resolves to the application name (its physical id), and each sub-resource Ref resolves to its minted id (the output id, reference id, or CloudWatch-logging-option id). The Application resource exposes no returnable Fn::GetAtt attributes. Creating an application never starts a Flink job (StartApplication is a separate action), so CloudFormation provisioning settles the application to READY in the control plane and never spawns a Flink container.

Terraform

fakecloud exercises the upstream terraform-provider-aws internal/service/kinesisanalyticsv2 acceptance suite in tfacc, split across the kinesisanalyticsv2-flink, kinesisanalyticsv2-sql, and kinesisanalyticsv2-snapshot jobs: SQL + Flink flavor applications (create/describe/update/delete, tags, import, disappears), inputs / outputs / reference data sources, CloudWatch logging options, VPC configurations, StartApplication on create + update, snapshot restore, and the standalone application snapshot resource. The real Flink-job data plane is disabled for tfacc (proven separately by the Docker E2E) so the control plane's STARTING -> RUNNING settle doesn't race the provider's status assertions.

Example

import boto3
flink = boto3.client("kinesisanalyticsv2", endpoint_url="http://localhost:4566")

flink.create_application(
    ApplicationName="clickstream",
    RuntimeEnvironment="FLINK-1_20",
    ServiceExecutionRole="arn:aws:iam::000000000000:role/service-role/kinesis-analytics",
    ApplicationConfiguration={
        "FlinkApplicationConfiguration": {
            "ParallelismConfiguration": {"ConfigurationType": "DEFAULT"}
        },
        "ApplicationCodeConfiguration": {
            "CodeContentType": "ZIPFILE",
            "CodeContent": {
                "S3ContentLocation": {
                    "BucketARN": "arn:aws:s3:::my-code",
                    "FileKey": "app.jar",
                }
            },
        },
    },
)

flink.start_application(ApplicationName="clickstream")
app = flink.describe_application(ApplicationName="clickstream")["ApplicationDetail"]
print(app["ApplicationStatus"], app["ApplicationVersionId"])