Setting the file. One moment.
references/kinesis-connector-guide.mdMarkdown9 KBView on GitHub
This guide covers Kinesis connector configuration for Amazon Managed Service for Apache Flink applications.
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.
When creating a Kinesis source, use the KinesisStreamsSource builder pattern. This API is the same for both Flink 1.20 and 2.2.
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");.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 sourceFor 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");.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 sinkThe 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 (opens in a new tab)):
| Config Option | Description | Default | Recommendation |
|---|---|---|---|
SHARD_GET_RECORDS_MAX | Max records per GetRecords call | 10,000 (but Kinesis limit is 10,000 records / 10 MB per call; throttling occurs at 1,000 records for some stream configurations) | Lower to 1,000 or less if seeing throttling |
READER_EMPTY_RECORDS_FETCH_INTERVAL | Interval between polling calls when no records are returned | 200ms (5 calls/sec) | Increase to 500ms–1s if sharing shards with other consumers or if the default rate exceeds the per-shard limit |
SHARD_DISCOVERY_INTERVAL | Interval for discovering new shards via ListShards | 10s | Increase if ListShards rate limiting is observed |
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):
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.See kinesis-efo-guide.md for EFO configuration, consumer lifecycle, and the full when-to-use-EFO checklist.
Diagnosing polling throttling:
ReadProvisionedThroughputExceeded on the Kinesis stream — sustained values > 0 indicate throttling.LimitExceededException errors.GetRecords.Latency and GetRecords.Success metrics to correlate throttling with read performance.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 (opens in a new tab) 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.
DataStream API with operator UIDs defined:
flink-connector-kinesis with flink-connector-aws-kinesis-streams v5.0.0+FlinkKinesisConsumer with KinesisStreamsSource builder patternAT_TIMESTAMP set to just before deployment timeallowNonRestoredState = trueTable API/SQL or DataStream without operator UIDs:
SKIP_RESTORE_FROM_SNAPSHOT since Flink cannot map old operator state to new operatorsRESTORE_FROM_LATEST_SNAPSHOT for future restarts| Feature | Legacy FlinkKinesisConsumer | New KinesisStreamsSource |
|---|---|---|
| Interface | SourceFunction (removed in 2.x) | FLIP-27 Source API |
| Stream identifier | Stream name | Stream ARN (cross-region/account support) |
| Watermarks | Implicit defaults | Explicit WatermarkStrategy required |
| Ordering on reshard | Not guaranteed | Guaranteed via parent-child shard lineage |
| Shard assigner | Even shard distribution | Uniform partition-key distribution |
| AWS SDK | v1 | v2 (non-blocking I/O) |
| JAR size | ~60 MB | ~200 KB |
| KCL/KPL dependency | Included | Removed (no built-in KPL de-aggregation) |
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.
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.