# Managed Service for Apache Flink

Source: /aws/services/kinesisanalyticsv2/

:::note
This service was formerly known as 'Kinesis Data Analytics for Apache Flink'.
:::

## Introduction

[Apache Flink](https://flink.apache.org/) is a framework for building applications that process and analyze streaming data.
[Managed Service for Apache Flink (MSF)](https://docs.aws.amazon.com/managed-flink/latest/java/what-is.html) is an AWS service that provides the underlying infrastructure and a hosted Apache Flink cluster that can run Apache Flink applications.

LocalStack lets you to run Flink applications locally and implements several [AWS-compatible API operations](#api-coverage).

A separate Apache Flink cluster is started in [application mode](https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/deployment/overview/#application-mode) for every Managed Flink application created.
Flink cluster deployment on LocalStack consists of two separate containers for [JobManager](https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/concepts/flink-architecture/#jobmanager) and [TaskManager](https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/concepts/flink-architecture/#taskmanagers).

## Deployment Considerations

The default container runtime for MSF on LocalStack is Docker.
LocalStack creates Flink JobManager and TaskManager containers on the Docker network specified by `MAIN_DOCKER_NETWORK` (defaults to `bridge`).

**Running LocalStack inside a Kubernetes cluster with the Docker executor is not supported.**
Mounting the host Docker socket (`/var/run/docker.sock`) into a LocalStack pod is not sufficient — the Docker executor cannot create Flink containers in this topology and applications will remain stuck in `STARTING` indefinitely.

If you are running LocalStack outside of Kubernetes (for example, with Docker Compose or the `lstk` CLI), no additional configuration is required and the Docker executor is used automatically.

## Getting Started

This guide builds a demo Flink application and deploys it to LocalStack.
The application generates synthetic records, processes them and sends the output to an S3 bucket.

Start the LocalStack container using your preferred method.

### Build Application Code

Begin by cloning the AWS sample repository.
We will use the [S3 Sink](https://github.com/localstack-samples/amazon-managed-service-for-apache-flink-examples/tree/main/java/S3Sink) application in this example.

```bash
git clone https://github.com/localstack-samples/amazon-managed-service-for-apache-flink-examples.git
cd java/S3Sink
```

Next, use [Maven](https://maven.apache.org/) to compile and package the Flink application into a jar.

```bash
mvn package
```

The Flink application jar file will be placed in the `./target/flink-kds-s3.jar` directory.

### Upload Application Code

MSF requires that all application code resides in S3.

Create an S3 bucket and upload the compiled Flink application jar.

```bash
lstk aws s3api create-bucket --bucket flink-bucket
lstk aws s3api put-object --bucket flink-bucket --key job.jar --body ./target/flink-kds-s3.jar
```

### Output Sink

As mentioned earlier, this Flink application writes the output to an S3 bucket.

Create the S3 bucket that will serve as the sink.

```bash
lstk aws s3api create-bucket --bucket sink-bucket
```

### Permissions

MSF requires a service execution role which allows it to connect to other services.
Without the proper permissions policy and role, this example application will not be able to connect to S3 sink bucket to output the result.

Create an IAM role for the running MSF application to assume.

```json showshowLineNumbers
# role.json
{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Principal": {"Service": "kinesisanalytics.amazonaws.com"},
            "Action": "sts:AssumeRole"
        }
    ]
}
```

```bash
lstk aws iam create-role --role-name msaf-role --assume-role-policy-document file://role.json
```

Next create add a permissions policy to this role that permits read and write access to S3.

```json showshowLineNumbers
# policy.json
{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": ["s3:GetObject", "s3:GetObjectVersion", "s3:PutObject"],
            "Resource": "*"
        }
    ]
}
```

```bash
lstk aws iam put-role-policy --role-name msaf-role --policy-name msaf-policy --policy-document file://policy.json
```

Now, when the running MSF application assumes this role, it will have the necessary permissions to write to the S3 sink.

### Deploy Application

With all prerequisite resources in place, the Flink application can now be created and started.

```bash showshowLineNumbers
lstk aws kinesisanalyticsv2 create-application \
    --application-name msaf-app \
    --runtime-environment FLINK-1_20 \
    --application-mode STREAMING \
    --service-execution-role arn:aws:iam::000000000000:role/msaf-role \
    --application-configuration '{
        "ApplicationCodeConfiguration": {
            "CodeContent": {
                "S3ContentLocation": {
                    "BucketARN": "arn:aws:s3:::flink-bucket",
                    "FileKey": "job.jar"
                }
            },
            "CodeContentType": "ZIPFILE"
        },
        "EnvironmentProperties": {
            "PropertyGroups": [{
                "PropertyGroupId": "bucket", "PropertyMap": {"name": "sink-bucket"}
            }]
        }
    }'

lstk aws kinesisanalyticsv2 start-application --application-name msaf-app
```

Once the Flink cluster is up and running, the application will stream the results to the sink S3 bucket.
You can verify this with:

```bash
lstk aws s3api list-objects --bucket sink-bucket
```

## CloudWatch Logging

LocalStack MSF supports [CloudWatch Logs integration](https://docs.aws.amazon.com/managed-flink/latest/java/cloudwatch-logs.html) to help monitor the Flink cluster for application events or configuration problems.
The logging option can be added at the time of creating the Flink application using the [CreateApplication](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_CreateApplication.html) operation.
Logging options can also be managed at a later point using the [AddApplicationCloudWatchLoggingOption](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_AddApplicationCloudWatchLoggingOption.html) and [DeleteApplicationCloudWatchLoggingOption](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_DeleteApplicationCloudWatchLoggingOption.html) operations.

There are following prerequisites for CloudWatch Logs integration:
- You must create the application's log group and log stream.
  Flink will not create it for you.
- You must add the permissions your application needs to write to the log stream to the service execution role.
  Generally the following IAM actions are sufficient: `logs:DescribeLogGroups`, `logs:DescribeLogStreams` and `logs:PutLogEvents`

To add a logging option:

```bash showshowLineNumbers
lstk aws kinesisanalyticsv2 add-application-cloud-watch-logging-option \
    --application-name msaf-app \
    --cloud-watch-logging-option '{"LogStreamARN": "arn:aws:logs:us-east-1:000000000000:log-group:msaf-log-group:log-stream:msaf-log-stream"}'
```

```bash title="Output"
{
    "ApplicationARN": "arn:aws:kinesisanalytics:us-east-1:000000000000:application/msaf-app",
    "ApplicationVersionId": 2,
    "CloudWatchLoggingOptionDescriptions": [
        {
            "CloudWatchLoggingOptionId": "1.1",
            "LogStreamARN": "arn:aws:logs:us-east-1:000000000000:log-group:msaf-log-group:log-stream:msaf-log-stream"
        }
    ]
}
```

:::note
Enabling CloudWatch Logs integration has a significant performance hit.
:::

Configured logging options can be retrieved using [DescribeApplication](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_DescribeApplication.html):

```bash
lstk aws kinesisanalyticsv2 describe-application --application-name msaf-app | jq .ApplicationDetail.CloudWatchLoggingOptionDescriptions
```

```bash title="Output"
[
  {
    "CloudWatchLoggingOptionId": "1.1",
    "LogStreamARN": "arn:aws:logs:us-east-1:000000000000:log-group:msaf-log-group:log-stream:msaf-log-stream"
  }
]
```

:::note
Logs events are reported to CloudWatch every 10 seconds.
:::

Log events can be retrieved from CloudWatch Logs using the appropriate operation.
To retrieve all events:

```bash
lstk aws logs get-log-events --log-group-name msaf-log-group --log-stream-name msaf-log-stream
```

LocalStack reports both Flink application and Flink framework logs to CloudWatch.
However, certain extended information such as stack traces may be missing.
You may obtain this information by execing into the Flink Docker container created by LocalStack and inspecting `/opt/flink/log`.

## Resource Tagging

You can manage [resource tags](https://docs.aws.amazon.com/managed-flink/latest/java/how-tagging.html) using [TagResource](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_TagResource.html), [UntagResource](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_UntagResource.html) and [ListTagsForResource](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_ListTagsForResource.html).
Tags can also be specified when creating the Flink application using the [CreateApplication](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_CreateApplication.html) operation.

```bash
lstk aws kinesisanalyticsv2 tag-resource \
    --resource-arn arn:aws:kinesisanalytics:us-east-1:000000000000:application/msaf-app \
    --tags Key=country,Value=SE

lstk aws kinesisanalyticsv2 list-tags-for-resource \
    --resource-arn arn:aws:kinesisanalytics:us-east-1:000000000000:application/msaf-app
```

```bash title="Output"
{
    "Tags": [
        {
            "Key": "country",
            "Value": "SE"
        }
    ]
}
```

You can also untag the resource:

```bash
lstk aws kinesisanalyticsv2 untag-resource \
    --resource-arn arn:aws:kinesisanalytics:us-east-1:000000000000:application/msaf-app \
    --tag-keys country
```

## Supported Flink Versions

| Flink version | Supported by LocalStack | Supported by Apache |
|:---:|:---:|:---:|
| 1.20.0 | yes | yes |
| 1.19.1 | yes | yes |
| 1.18.1 | yes | yes |
| 1.15.2 | yes | no |

## Troubleshooting

### Application stuck in `STARTING`

If `describe-application` returns `STARTING` indefinitely and no Flink containers appear, the most likely cause is that LocalStack cannot reach the Docker daemon or Kubernetes API to create the Flink JobManager and TaskManager.

When this occurs, LocalStack logs an internal error after `CLUSTER_READY_WAIT_TIMEOUT` elapses:

```
Exception: Error submitting job: Flink cluster <id> is not running
```

The application does not automatically transition to `FAILED` — it remains in `STARTING`.
This means IaC tools that use state waiters (such as the Terraform AWS provider or Crossplane) will block until their own timeout expires.

**Common causes and fixes:**

- **Running LocalStack inside Kubernetes with the Docker executor** — the Docker executor is not supported in this topology.
  See [Deployment Considerations](#deployment-considerations) for details.
- **Wrong or missing Docker network** — if LocalStack is running in a non-default Docker network, set `MAIN_DOCKER_NETWORK` to the name of that network so Flink containers are attached to the correct network.
- **Docker socket not accessible** — confirm that the Docker socket is mounted and functional.
  You can verify by listing containers from inside the LocalStack container: `curl --unix-socket /var/run/docker.sock http://localhost/containers/json`.

## Limitations

- Application versions are not maintained
- Only S3 zipfile code is supported
- Values of 20,000 ms for `execution.checkpointing.interval` and 5,000 ms for `execution.checkpointing.min-pause` are used for checkpointing.
  They can not be overridden.
- In-place [version upgrades](https://docs.aws.amazon.com/managed-flink/latest/java/how-in-place-version-upgrades.html) and [roll-backs](https://docs.aws.amazon.com/managed-flink/latest/java/how-system-rollbacks.html) are not supported
- [Snapshot/savepoint management](https://docs.aws.amazon.com/managed-flink/latest/java/how-snapshots.html) is not implemented
- CloudTrail integration and CloudWatch metrics is not implemented.
  The application logging level defaults to `INFO` and can not be overridden.
- Parallelism is limited to the default value of 1, with one TaskManager that has one [Task Slot](https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/concepts/flink-architecture/#task-slots-and-resources) allocated.
  [Parallelism configuration](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_FlinkApplicationConfiguration.html#APIReference-Type-FlinkApplicationConfiguration-ParallelismConfiguration) provided on Flink application creation or update is ignored.
- When a Flink cluster fails to start, the application remains in `STARTING` rather than transitioning to `FAILED`.
  Check LocalStack logs and see [Troubleshooting](#troubleshooting) for guidance.
- The Docker executor is not supported when LocalStack runs as a pod inside a Kubernetes cluster.

## API Coverage


### Managed Service for Apache Flink API coverage

Source service: `kinesisanalyticsv2`. 13 of 33 tracked operations are implemented.

Service documentation: /aws/services/kinesisanalyticsv2/
License availability: available starting with the Ultimate plan. See /aws/licensing/ for current plan details.

| Operation | Status |
| --- | --- |
| AddApplicationCloudWatchLoggingOption | Implemented |
| AddApplicationInput | Not implemented |
| AddApplicationInputProcessingConfiguration | Not implemented |
| AddApplicationOutput | Not implemented |
| AddApplicationReferenceDataSource | Not implemented |
| AddApplicationVpcConfiguration | Not implemented |
| CreateApplication | Implemented |
| CreateApplicationPresignedUrl | Not implemented |
| CreateApplicationSnapshot | Not implemented |
| DeleteApplication | Implemented |
| DeleteApplicationCloudWatchLoggingOption | Implemented |
| DeleteApplicationInputProcessingConfiguration | Not implemented |
| DeleteApplicationOutput | Not implemented |
| DeleteApplicationReferenceDataSource | Not implemented |
| DeleteApplicationSnapshot | Not implemented |
| DeleteApplicationVpcConfiguration | Not implemented |
| DescribeApplication | Implemented |
| DescribeApplicationOperation | Not implemented |
| DescribeApplicationSnapshot | Not implemented |
| DescribeApplicationVersion | Not implemented |
| DiscoverInputSchema | Not implemented |
| ListApplicationOperations | Not implemented |
| ListApplicationSnapshots | Not implemented |
| ListApplicationVersions | Not implemented |
| ListApplications | Implemented |
| ListTagsForResource | Implemented |
| RollbackApplication | Not implemented |
| StartApplication | Implemented |
| StopApplication | Implemented |
| TagResource | Implemented |
| UntagResource | Implemented |
| UpdateApplication | Implemented |
| UpdateApplicationMaintenanceConfiguration | Implemented |