Developing Applications On Managed Service For Apache Flink

Resource handler returned message: "Please check the role provided or validity of S3 location
you provided. We are unable to get the specified fileKey: <key> in the specified bucket: <bucket>"

# 1. Deploy infrastructure
aws cloudformation deploy --template-file cfn-infra.yaml --stack-name my-app-infra ...

# 2. Build and upload JAR
mvn clean package -q
aws s3 cp target/my-app.jar s3://${JAR_BUCKET}/${JAR_KEY}

# 3. Deploy application (JAR now exists in S3)
aws cloudformation deploy --template-file cfn-app.yaml --stack-name my-app ...

// InfraStack: buckets, streams, IAM, logs
// AppStack: MSF application (depends on InfraStack)
// Deploy script uploads JAR between the two stack deployments

import * as s3deploy from 'aws-cdk-lib/aws-s3-deployment';

// Upload JAR to S3 as part of the CDK deployment
const jarDeployment = new s3deploy.BucketDeployment(this, 'JarDeployment', {
  sources: [s3deploy.Source.asset('./target')],
  destinationBucket: jarBucket,
  destinationKeyPrefix: 'jars/',
});

// Ensure MSF application is created AFTER the JAR is uploaded
flinkApp.node.addDependency(jarDeployment);

resource "aws_s3_object" "app_jar" {
  bucket = aws_s3_bucket.jar_bucket.id
  key    = "jars/my-app.jar"
  source = "target/my-app.jar"
  etag   = filemd5("target/my-app.jar")
}

resource "aws_kinesisanalyticsv2_application" "flink_app" {
  name                   = "my-flink-app"
  runtime_environment    = "FLINK-2_2"  # Default for new apps. Use FLINK-1_20 only when migrating an existing 1.20 app and state compatibility forbids the upgrade.
  service_execution_role = aws_iam_role.flink_role.arn

  application_configuration {
    application_code_configuration {
      code_content {
        s3_content_location {
          bucket_arn = aws_s3_bucket.jar_bucket.arn
          file_key   = aws_s3_object.app_jar.key
        }
      }
      code_content_type = "ZIPFILE"
    }
    # ... other configuration
  }

  depends_on = [aws_s3_object.app_jar]
}

# CloudWatch Logs (required for application logging)
# Scope to the application's log group, not log-group:* — that grants logs
# permissions across every group in the account.
- Effect: Allow
  Action:
    - logs:PutLogEvents
    - logs:DescribeLogStreams
  Resource:
    - !Sub "arn:aws:logs:${AWS::Region}:${AWS::AccountId}:log-group:/aws/kinesis-analytics/${ApplicationName}"
    - !Sub "arn:aws:logs:${AWS::Region}:${AWS::AccountId}:log-group:/aws/kinesis-analytics/${ApplicationName}:log-stream:*"

# DescribeLogGroups does not support resource-level permissions
- Effect: Allow
  Action:
    - logs:DescribeLogGroups
  Resource: "*"

# S3 JAR bucket (required to read the application JAR), and KMS permissions if encrypted by a CMK
- Effect: Allow
  Action:
    - s3:GetObject
    - s3:GetObjectVersion
  Resource:
    - !Sub "${JarBucket.Arn}/*"

- Effect: Allow
  Action:
    - kinesis:DescribeStream
    - kinesis:GetShardIterator
    - kinesis:GetRecords
    - kinesis:ListShards
    - kinesis:DescribeStreamSummary
    - kinesis:DescribeStreamConsumer
    - kinesis:SubscribeToShard        # Required for EFO
    - kinesis:RegisterStreamConsumer   # Required for EFO
    - kinesis:DeregisterStreamConsumer # Required for EFO
  Resource:
    - !GetAtt KinesisStream.Arn
    - !Sub "${KinesisStream.Arn}/consumer/*"  # Required for EFO

- Effect: Allow
  Action:
    - kinesis:PutRecord
    - kinesis:PutRecords
    - kinesis:DescribeStream
    - kinesis:DescribeStreamSummary
  Resource:
    - !GetAtt OutputKinesisStream.Arn

- Effect: Allow
  Action:
    - s3:PutObject
    - s3:GetObject
    - s3:ListBucket
    - s3:DeleteObject
    - s3:GetBucketLocation
    - s3:AbortMultipartUpload
    - s3:ListMultipartUploadParts
  Resource:
    - !GetAtt OutputBucket.Arn
    - !Sub "${OutputBucket.Arn}/*"

# VPC access for MSK. The describe* and *NetworkInterface* actions don't
# accept ARN-scoped resources, so the resource has to be "*" — but you can
# (and should) constrain them with condition keys to the specific VPC/region.
# Example: ec2:Vpc on the network-interface actions, aws:RequestedRegion on
# the describe actions. See guideline 10 (condition keys) below.
- Effect: Allow
  Action:
    - ec2:DescribeVpcs
    - ec2:DescribeSubnets
    - ec2:DescribeSecurityGroups
    - ec2:DescribeDhcpOptions
    - ec2:CreateNetworkInterface
    - ec2:CreateNetworkInterfacePermission
    - ec2:DescribeNetworkInterfaces
    - ec2:DeleteNetworkInterface
  Resource: "*"
  Condition:
    StringEquals:
      aws:RequestedRegion: !Ref AWS::Region
    # For CreateNetworkInterface / CreateNetworkInterfacePermission you can
    # additionally constrain to the application's VPC:
    # ArnEquals:
    #   ec2:Vpc: !Sub "arn:aws:ec2:${AWS::Region}:${AWS::AccountId}:vpc/${VpcId}"

FlinkApplication:
  Type: AWS::KinesisAnalyticsV2::Application
  Properties:
    ApplicationName: !Ref ApplicationName
    RuntimeEnvironment: !Ref FlinkRuntimeEnvironment  # FLINK-2_2 (default for new apps). FLINK-1_20 only for in-place upgrades of existing 1.20 apps.
    ServiceExecutionRole: !GetAtt FlinkRole.Arn
    ApplicationConfiguration:
      # JAR location — JAR must exist before this resource is created
      ApplicationCodeConfiguration:
        CodeContent:
          S3ContentLocation:
            BucketARN: !GetAtt JarBucket.Arn
            FileKey: !Ref JarS3Key
        CodeContentType: ZIPFILE
      # Runtime properties — equivalent to MSF console "Runtime properties"
      EnvironmentProperties:
        PropertyGroups:
          - PropertyGroupId: "kinesis.source"
            PropertyMap:
              stream.arn: !GetAtt InputStream.Arn
              aws.region: !Ref AWS::Region
          - PropertyGroupId: "s3.sink"
            PropertyMap:
              bucket.name: !Ref OutputBucket
              path.prefix: "output/"
      # Parallelism and scaling
      FlinkApplicationConfiguration:
        ParallelismConfiguration:
          ConfigurationType: CUSTOM
          Parallelism: !Ref Parallelism          # Total parallelism (= KPU count × par/KPU)
          ParallelismPerKPU: !Ref ParallelismPerKPU
          AutoScalingEnabled: true
        CheckpointConfiguration:
          ConfigurationType: CUSTOM
          CheckpointingEnabled: true
          CheckpointInterval: 60000
          MinPauseBetweenCheckpoints: 5000
        MonitoringConfiguration:
          ConfigurationType: CUSTOM
          LogLevel: INFO
          MetricsLevel: APPLICATION

# 1. Upload new JAR
aws s3 cp target/my-app.jar s3://${JAR_BUCKET}/${JAR_KEY}

# 2. Get current application version
CURRENT_VERSION=$(aws kinesisanalyticsv2 describe-application \
  --application-name my-app \
  --query 'ApplicationDetail.ApplicationVersionId' \
  --output text)

# 3. Update application code reference
aws kinesisanalyticsv2 update-application \
  --application-name my-app \
  --current-application-version-id ${CURRENT_VERSION} \
  --application-configuration-update '{
    "ApplicationCodeConfigurationUpdate": {
      "CodeContentUpdate": {
        "S3ContentLocationUpdate": {
          "BucketARNUpdate": "arn:aws:s3:::'"${JAR_BUCKET}"'",
          "FileKeyUpdate": "'"${JAR_KEY}"'"
        }
      },
      "CodeContentTypeUpdate": "ZIPFILE"
    }
  }'

# 4. Pick up the new code based on current application state.
#    From RUNNING, update-application auto-restarts the app (UPDATING → RUNNING,
#    typically 10–30s downtime depending on state size) — no explicit stop/start needed.
#    From READY (stopped), update-application stays in READY — start the app explicitly
#    to pick up the new code.
#    See application-lifecycle.md for the full state-transition table.
STATUS=$(aws kinesisanalyticsv2 describe-application \
  --application-name my-app \
  --query 'ApplicationDetail.ApplicationStatus' --output text)

if [ "$STATUS" = "READY" ]; then
  aws kinesisanalyticsv2 start-application --application-name my-app \
    --run-configuration '{
      "ApplicationRestoreConfiguration": {
        "ApplicationRestoreType": "RESTORE_FROM_LATEST_SNAPSHOT"
      }
    }'
fi
# Otherwise (RUNNING/UPDATING), poll describe-application until ApplicationStatus
# returns to RUNNING to confirm the auto-restart completed.