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

Kinesis Connector Guide

This guide covers Kinesis connector configuration for Amazon Managed Service for Apache Flink applications.

Maven Dependencies

CRITICAL: Use the correct Kinesis connector dependency from the official Apache Flink project:

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-aws-kinesis-streams</artifactId>
    <version>${kinesis-streams.version}</version>
</dependency>

Where kinesis-streams.version should match your Flink version (e.g., 5.1.0-1.20 for Flink 1.20, 6.0.0-2.0 for Flink 2.2). See dependency-management.md for the full version mapping.

Kinesis Source Configuration

When creating a Kinesis source, use the KinesisStreamsSource builder pattern. This API is the same for both Flink 1.20 and 2.2.

Correct API Pattern

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

Map<String, Properties> applicationProperties = loadApplicationProperties(env);
Properties inputConfig = applicationProperties.get("input.kinesis.config");
Map<String, String> configMap = new HashMap<>();
inputConfig.forEach((k, v) -> configMap.put(k.toString(), v.toString()));
Configuration sourceConfig = Configuration.fromMap(configMap);
String inputStreamArn = inputConfig.getProperty("stream.arn");

KinesisStreamsSource<Event> source = KinesisStreamsSource.<Event>builder()
    .setStreamArn(inputStreamArn)
    .setDeserializationSchema(new EventDeserializationSchema())
    .setSourceConfig(sourceConfig)
    .build();

DataStream<Event> events = env
        .fromSource(source, 
                    WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                            .withTimestampAssigner((event, timestamp) -> event.getTimestamp())
                            .withIdleness(Duration.ofSeconds(10)),  // CRITICAL for low-throughput streams
                    "kinesis-source",
                    org.apache.flink.api.common.typeinfo.TypeInformation.of(Event.class))
        .name("kinesis-source")
        .uid("kinesis-source-uid")
        .filter(event -> event != null)
        .name("filter-null-events")
        .uid("filter-null-events-uid");

Key Configuration Methods

• .setStreamArn(String) - Set the stream ARN (preferred over stream name)
• .setDeserializationSchema(DeserializationSchema<T>) - Set how to deserialize records
• .setSourceConfig(Configuration) - Set Kinesis client configuration
• .build() - Build the source

Kinesis Sink Configuration

For Kinesis sinks, use the KinesisStreamsSink builder:

Properties outputConfig = applicationProperties.get("output.kinesis.config");
String outputStreamArn = outputConfig.getProperty("stream.arn");

KinesisStreamsSink<Event> sink = KinesisStreamsSink.<Event>builder()
    .setStreamArn(outputStreamArn)
    .setSerializationSchema(new EventSerializationSchema())
    .setPartitionKeyGenerator(event -> String.valueOf(event.hashCode()))
    .setKinesisClientProperties(outputConfig)
    .build();

events.sinkTo(sink)
        .name("kinesis-sink")
        .uid("kinesis-sink-uid");

Key Sink Methods

• .setStreamArn(String) - Set the destination stream ARN
• .setSerializationSchema(SerializationSchema<T>) - Set how to serialize records
• .setPartitionKeyGenerator(PartitionKeyGenerator<T>) - Set partition key logic
• .build() - Build the sink

Polling Configuration and Throttling

The Kinesis GetRecords API has a hard limit of 5 calls per second per shard, shared across all consumers reading from that shard. The Flink Kinesis connector's default polling behavior can be aggressive and lead to ReadProvisionedThroughputExceeded or LimitExceededException errors, especially when multiple consumers share a stream or when polling intervals are too short.

Key polling configuration options (see KinesisSourceConfigOptions Javadoc):

Tuning polling to avoid throttling:

Configuration sourceConfig = new Configuration();

// Reduce max records per GetRecords call to stay within Kinesis limits
sourceConfig.set(KinesisSourceConfigOptions.SHARD_GET_RECORDS_MAX, 1000);

// Increase polling interval to reduce GetRecords call rate
// Default is 200ms (5 calls/sec) — increase if sharing shards with other consumers
sourceConfig.set(KinesisSourceConfigOptions.READER_EMPTY_RECORDS_FETCH_INTERVAL, Duration.ofMillis(500));

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

When polling tuning is sufficient (vs switching to EFO):

• Single consumer reading from the stream: tune polling — EFO is unnecessary cost since there is no shared-quota contention.
• 2+ consumers and still seeing ReadProvisionedThroughputExceeded after raising READER_EMPTY_RECORDS_FETCH_INTERVAL: switch to EFO. Tuning polling intervals across multiple consumers is fragile; EFO eliminates the shared quota entirely.
• Polling interval increase introduces unacceptable latency: switch to EFO (HTTP/2 push has no polling delay).
• You expect to scale to more consumers later: prefer EFO upfront.

See kinesis-efo-guide.md for EFO configuration, consumer lifecycle, and the full when-to-use-EFO checklist.

Diagnosing polling throttling:

• Check CloudWatch metric ReadProvisionedThroughputExceeded on the Kinesis stream — sustained values > 0 indicate throttling.
• Check Managed Service for Apache Flink CloudWatch logs for LimitExceededException errors.
• Monitor GetRecords.Latency and GetRecords.Success metrics to correlate throttling with read performance.

Migrating from Legacy Kinesis Consumer to KinesisStreamsSource

CRITICAL for Flink 2.x upgrades: KDS connector versions below 5.0 have state that is incompatible with the Flink 2.2 connector (v6.0.0-2.0). You must migrate to connector v5.0+ on Flink 1.x before upgrading to Flink 2.x. See the AWS blog post on the Kinesis source connector for full details.

The legacy FlinkKinesisConsumer uses the removed SourceFunction interface and will not work with Flink 2.x. The KinesisStreamsSource uses the FLIP-27 Source API.

Migration Paths

DataStream API with operator UIDs defined:

1. Update dependencies: replace flink-connector-kinesis with flink-connector-aws-kinesis-streams v5.0.0+
2. Replace FlinkKinesisConsumer with KinesisStreamsSource builder pattern
3. Change the UID of the source operator to a new string (this selectively resets source state while preserving all other operator state)
4. Configure starting position with AT_TIMESTAMP set to just before deployment time
5. Deploy with allowNonRestoredState = true

Table API/SQL or DataStream without operator UIDs:

1. Update dependencies and code as above
2. Deploy with SKIP_RESTORE_FROM_SNAPSHOT since Flink cannot map old operator state to new operators
3. After the application is running, switch back to RESTORE_FROM_LATEST_SNAPSHOT for future restarts

Key Differences from Legacy Connector

State Compatibility Warning

The saved state from FlinkKinesisConsumer is not compatible with KinesisStreamsSource. You cannot restore source position from a snapshot taken with the legacy connector. Plan for either selective state reset (with UIDs) or full state reset (without UIDs) as described above.

Authentication

In Managed Service for Apache Flink, authentication to Kinesis is handled automatically via the application's IAM execution role. No explicit credentials configuration is needed in the code.