/developing-applications-on-managed-service-for-apache-flink

Enhanced Fan-Out (EFO) Configuration Guide

Overview

This guide covers Enhanced Fan-Out (EFO) configuration for Kinesis sources in Managed Service for Apache Flink applications, including when to use EFO, source configuration, consumer lifecycle management, parallelism and shard count considerations, auto-scaling interaction, and troubleshooting.

For Kinesis source/sink configuration, authentication, and legacy consumer migration, see kinesis-connector-guide.md.

When to Use EFO vs Standard Polling

Use Enhanced Fan-Out when:

• Multiple consumers read from the same Kinesis stream (EFO gives each consumer dedicated 2 MB/s per shard, vs shared 2 MB/s with polling)
• You need sub-200ms read latency (EFO uses HTTP/2 push, polling has up to 200ms delay per GetRecords call)
• Your application cannot tolerate ReadProvisionedThroughputExceeded throttling from competing consumers
• You are hitting the 5 GetRecords calls per second per shard limit due to multiple consumers or aggressive polling intervals

Standard polling is sufficient when:

• Your Flink application is the only consumer on the stream
• Latency requirements are relaxed (200ms+ acceptable)
• You want to minimize cost (EFO incurs additional per-consumer, per-shard-hour charges)

For polling configuration tuning, the 5 GetRecords calls/sec per shard limit, and diagnosing ReadProvisionedThroughputExceeded, see the "Polling Configuration and Throttling" section in kinesis-connector-guide.md. EFO is the right answer when polling tuning is not enough; the connector guide explains when to choose which.

EFO Source Configuration

Enable EFO by setting READER_TYPE and EFO_CONSUMER_NAME in the source configuration:

import org.apache.flink.connector.kinesis.source.KinesisStreamsSource;
import org.apache.flink.connector.kinesis.source.config.KinesisSourceConfigOptions;
import org.apache.flink.configuration.Configuration;

Configuration sourceConfig = new Configuration();
sourceConfig.set(KinesisSourceConfigOptions.READER_TYPE, KinesisSourceConfigOptions.ReaderType.EFO);
sourceConfig.set(KinesisSourceConfigOptions.EFO_CONSUMER_NAME, "my-flink-efo-consumer");

KinesisStreamsSource<String> source = KinesisStreamsSource.<String>builder()
    .setStreamArn("arn:aws:kinesis:us-east-1:123456789012:stream/my-stream")
    .setDeserializationSchema(new SimpleStringSchema())
    .setSourceConfig(sourceConfig)
    .build();

// Recommended: set parallelism >= shard count for optimal per-shard throughput isolation
env.fromSource(source,
        WatermarkStrategy.<String>forBoundedOutOfOrderness(Duration.ofSeconds(5))
            .withIdleness(Duration.ofSeconds(10)),
        "kinesis-efo-source")
    .setParallelism(shardCount)  // match or exceed shard count
    .uid("kinesis-efo-source-uid");

The consumer name must be unique per stream but can be reused across different streams. Reusing an existing consumer name on the same stream will terminate the previous subscription.

Consumer Lifecycle Management

By default (JOB_MANAGED), the KinesisStreamsSource automatically registers the stream consumer on job start and deregisters on graceful stop. For environments where you want external control:

• JOB_MANAGED (default): Flink registers/deregisters the consumer automatically. Preferred for most Managed Service for Apache Flink applications.
• SELF_MANAGED: You register the consumer externally via AWS CLI (aws kinesis register-stream-consumer) or SDK, then provide the consumer ARN to the job. Use this when multiple jobs share a consumer or when you need to control the consumer lifecycle independently.

Source Parallelism and Shard Count

Source parallelism should ideally match or exceed the Kinesis shard count when using EFO for optimal throughput. If parallelism is less than the shard count, some subtasks handle multiple shards — this still works but reduces the throughput benefit of EFO. If parallelism exceeds the shard count, idle subtasks will block watermark generation unless withIdleness() is configured on the WatermarkStrategy.

EFO Interaction with Managed Service for Apache Flink Auto-Scaling and KPU Allocation

When Managed Service for Apache Flink auto-scaling adjusts KPU count, the total parallelism changes. This affects EFO consumers:

• Scaling up increases parallelism, potentially creating idle subtasks if parallelism exceeds shard count — ensure withIdleness() is set
• Scaling down reduces parallelism, causing subtasks to handle more shards — EFO still provides dedicated throughput per shard but each subtask processes more data
• During scaling events, the EFO consumer subscription is re-established automatically; expect brief transient errors in logs (this is normal)
• Set Managed Service for Apache Flink auto-scaling min KPU to ensure parallelism never drops below shard count for optimal EFO throughput

Troubleshooting EFO

Consumer registration errors (LimitExceededException):

• Kinesis limits concurrent consumer registrations to 5 in CREATING state per account. If you see this during job startup, the consumer will retry automatically. For persistent failures, check if other applications are registering consumers simultaneously.
• Maximum consumers per stream: 20 (Provisioned/On-demand Standard) or 50 (On-demand Advantage). Verify you haven't hit the limit with aws kinesis list-stream-consumers.

Throughput exceptions (SubscribeToShard failures):

• Each EFO consumer gets dedicated 2 MB/s per shard. If you see throughput errors, verify the consumer is properly registered (ACTIVE status) using aws kinesis describe-stream-consumer.
• Transient SubscribeToShard errors are expected — subscriptions last 5 minutes and are automatically re-acquired.

IAM permissions for EFO: The Managed Service for Apache Flink application's IAM execution role needs these additional permissions beyond standard Kinesis read access:

{
    "Effect": "Allow",
    "Action": [
        "kinesis:RegisterStreamConsumer",
        "kinesis:DeregisterStreamConsumer",
        "kinesis:DescribeStreamConsumer",
        "kinesis:SubscribeToShard"
    ],
    "Resource": [
        "arn:aws:kinesis:us-east-1:123456789012:stream/my-stream",
        "arn:aws:kinesis:us-east-1:123456789012:stream/my-stream/consumer/*"
    ]
}

Note the consumer resource ARN (stream/*/consumer/*) — SubscribeToShard and DescribeStreamConsumer require permissions on the consumer resource, not just the stream.

Retry strategy tuning: If DescribeStreamConsumer calls fail during startup (common when the consumer was just registered and is still in CREATING state), tune the dedicated retry strategy:

sourceConfig.set(KinesisSourceConfigOptions.EFO_DESCRIBE_CONSUMER_RETRY_STRATEGY_MAX_ATTEMPTS, 30);
sourceConfig.set(KinesisSourceConfigOptions.EFO_DESCRIBE_CONSUMER_RETRY_STRATEGY_MIN_DELAY, Duration.ofSeconds(2));
sourceConfig.set(KinesisSourceConfigOptions.EFO_DESCRIBE_CONSUMER_RETRY_STRATEGY_MAX_DELAY, Duration.ofSeconds(10));