<a id="cmf-s3-checkpointing"></a>

# How to Set Up Checkpointing to AWS S3 with Confluent Manager for Apache Flink

This guide shows how to set up checkpointing for a Flink application using Confluent Manager for Apache Flink.
This example shows how to set up checkpointing to AWS S3. The steps are mostly
the same for other cloud providers, but this example is specific to AWS S3.

## Overview of steps

Use the following steps to set up checkpointing.

1. Enable the S3 filesystem plugin for Flink. Do this by setting the
   `ENABLE_BUILT_IN_PLUGINS` environment variable using a `spec.podTemplate` format for all
   the pods in the Flink cluster. Make sure to use the correct version in the plugin
   jar name.
2. Set the S3 credentials in the Flink configuration. The following example shows the
   `s3.access-key: your-access-key` and `s3.secret-key: your-secret-key` settings.
3. Enable checkpointing by setting the `state.checkpoints.dir` to the location on the S3 bucket where
   you want the checkpoints to be saved.

## Examples

The following examples show different ways to configure S3 credentials for checkpointing.

#### IMPORTANT
Avoid setting S3 credentials as plain text in `flinkConfiguration` for production
deployments. Use Kubernetes Secrets or IAM Roles instead.

### Plain text credentials

This minimal example sets S3 credentials directly in `flinkConfiguration`. Use this
approach only for local development or testing.

```yaml
apiVersion: cmf.confluent.io/v1
kind: FlinkApplication
metadata:
  name: checkpointing-example
spec:
  image: confluentinc/cp-flink:1.19.2-cp1
  flinkVersion: v1_19
  flinkConfiguration:
    taskmanager.numberOfTaskSlots: '1'
    # Set the S3 credentials
    s3.access-key: your-access-key
    s3.secret-key: your-secret-key
    # Enable checkpointing to S3
    state.checkpoints.dir: s3://your-bucket/flink-checkpoints/
  serviceAccount: flink
  podTemplate:
    spec:
      containers:
        - name: flink-main-container
          env:
            - name: ENABLE_BUILT_IN_PLUGINS
              value: "flink-s3-fs-hadoop-1.19.2-cp1.jar"
  jobManager:
    resource:
      memory: 1024m
      cpu: 1
  taskManager:
    resource:
      memory: 1024m
      cpu: 1
  job:
    jarURI: local:///opt/flink/examples/streaming/StateMachineExample.jar
    state: running
    parallelism: 1
```

### Kubernetes Secrets

Store your AWS credentials in a Kubernetes Secret and inject them as environment
variables using `secretKeyRef`:

```yaml
spec:
  podTemplate:
    spec:
      containers:
        - name: flink-main-container
          env:
            - name: AWS_ACCESS_KEY_ID
              valueFrom:
                secretKeyRef:
                  name: s3-credentials
                  key: access-key
            - name: AWS_SECRET_ACCESS_KEY
              valueFrom:
                secretKeyRef:
                  name: s3-credentials
                  key: secret-key
```

When using this approach, remove the `s3.access-key` and `s3.secret-key`
entries from `flinkConfiguration`. The Flink S3 filesystem plugin picks up
credentials from the standard `AWS_ACCESS_KEY_ID` and `AWS_SECRET_ACCESS_KEY`
environment variables automatically.

### IAM Roles (recommended on AWS EKS)

Use [EKS Pod Identity](https://docs.aws.amazon.com/eks/latest/userguide/pod-identities.html)
or IAM Roles for Service Accounts (IRSA) to associate an IAM role with the
Kubernetes service account used by the Flink pods. This eliminates the need to
manage credentials. With EKS Pod Identity, the AWS SDK credential chain
resolves permissions automatically from the pod’s assigned IAM role.

If you use IRSA, also set the credentials provider in `flinkConfiguration`.
The default credential providers of the Flink S3 filesystem plugin do not
include the web identity token provider that IRSA requires:

```yaml
spec:
  flinkConfiguration:
    s3.aws.credentials.provider: com.amazonaws.auth.WebIdentityTokenCredentialsProvider
```

## Related content

- [Flink Jobs for Confluent Manager for Apache Flink](../jobs/overview.md#af-jobs-cmf)