Managed Flink Connectors for MCP Servers — Kinesis, Kafka, S3, OpenSearch Source and Sink

Kinesis Data Streams source connector

Managed Flink supports two Kinesis source implementations. The newer KinesisStreamsSource (available in Flink 1.15+, which is the minimum version for current Managed Flink) uses the unified Source interface and supports exactly-once with enhanced fan-out. The legacy FlinkKinesisConsumer uses the SourceFunction interface and supports at-least-once by default with optional exactly-once.

import org.apache.flink.connector.kinesis.source.KinesisStreamsSource;
import org.apache.flink.connector.kinesis.source.enumerator.assigner.ShardAssignerFactory;
import org.apache.flink.api.common.serialization.SimpleStringSchema;

// New KinesisStreamsSource (Flink 1.15+) — recommended
Properties kinesisConsumerConfig = new Properties();
kinesisConsumerConfig.put(AWSConfigConstants.AWS_REGION, "us-east-1");

// Enhanced fan-out: dedicated 2 MB/s per shard (vs 2 MB/s shared across all GetRecords consumers)
// Required for high-throughput pipelines or when multiple consumers share the same stream
kinesisConsumerConfig.put(
    ConsumerConfigConstants.RECORD_PUBLISHER_TYPE,
    ConsumerConfigConstants.RecordPublisherType.EFO.toString()
);
kinesisConsumerConfig.put(
    ConsumerConfigConstants.EFO_CONSUMER_NAME,
    "alivemcp-flink-probe-analytics"
);

KinesisStreamsSource<String> kdsSource = KinesisStreamsSource.<String>builder()
    .setStreamArn("arn:aws:kinesis:us-east-1:123456789012:stream/alivemcp-probe-events")
    .setDeserializationSchema(new SimpleStringSchema())
    .setKinesisConsumerConfig(kinesisConsumerConfig)
    .setStartingPosition(StartingPosition.fromTrimHorizon())  // Initial position
    .setShardAssignerFactory(ShardAssignerFactory.uniformShardAssigner())  // Even distribution
    .build();

DataStream<String> probeStream = env
    .fromSource(kdsSource, WatermarkStrategy.noWatermarks(), "Kinesis Probe Source")
    .uid("kinesis-source");

The IAM role attached to the Managed Flink application must include permissions to call kinesis:GetRecords, kinesis:GetShardIterator, kinesis:DescribeStream, kinesis:ListShards. For enhanced fan-out, additionally: kinesis:RegisterStreamConsumer, kinesis:DeregisterStreamConsumer, kinesis:DescribeStreamConsumer, kinesis:SubscribeToShard.

Kinesis Data Streams sink connector

The KinesisStreamsSink writes records to a Kinesis stream using PutRecords in batches. It is at-least-once — on recovery from a checkpoint, records near the checkpoint boundary may be written twice. The partition key strategy determines shard distribution:

import org.apache.flink.connector.kinesis.sink.KinesisStreamsSink;

// KinesisStreamsSink — at-least-once, batches PutRecords calls
KinesisStreamsSink<String> kinesisSink = KinesisStreamsSink.<String>builder()
    .setStreamName("alivemcp-uptime-windows")
    .setSerializationSchema(new SimpleStringSchema())
    // Partition key: use serverSlug to co-locate the same server's windows on the same shard
    // This allows shard-level consumers downstream to process all windows for a server together
    .setPartitionKeyGenerator(element -> {
        // Parse serverSlug from the serialized window JSON
        // Alternatively, use a field in the record if using a typed serializer
        return extractServerSlug(element);  // Returns consistent key for each server
    })
    // Batching configuration — balance latency vs throughput
    .setMaxBatchSize(500)           // Max records per PutRecords call (KDS limit is 500)
    .setMaxInFlightRequests(50)     // Concurrent in-flight PutRecords batches
    .setMaxBufferedRequests(10000)  // Buffer before backpressure (blocks upstream operators)
    .setMaxBatchSizeInBytes(5 * 1024 * 1024)  // 5 MB max per PutRecords call (KDS limit)
    .setMaxTimeInBufferMS(5000)     // Flush buffer after 5 seconds even if not full
    .setFailOnError(false)          // Log and continue on partial PutRecords failure (at-least-once)
    .build();

windowStream
    .map(window -> objectMapper.writeValueAsString(window))
    .sinkTo(kinesisSink)
    .uid("kinesis-output-sink");

If downstream consumers require exactly-once guarantees from the Kinesis output, implement deduplication on the consumer side using the probeId or (serverSlug + windowStart) composite key as the deduplication key. The Kinesis record's SequenceNumber is not stable across retries — two PutRecords calls for the same data produce different sequence numbers. Use a field from the record payload for deduplication, not the sequence number.

S3 FileSink connector — exactly-once Parquet output

The S3 FileSink (replacement for the deprecated StreamingFileSink) provides exactly-once guarantees when checkpointing is enabled. It uses a two-phase file naming protocol: files are written as .part-{taskId}-{checkpointId}.inprogress during a checkpoint boundary and renamed to their final name only when the checkpoint commits. If the application fails before the checkpoint commits, the in-progress file is discarded on recovery — no partial data reaches the committed path.

import org.apache.flink.connector.file.sink.FileSink;
import org.apache.flink.formats.parquet.avro.ParquetAvroWriters;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.OnCheckpointRollingPolicy;
import org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.DateTimeBucketAssigner;

// S3 FileSink with Parquet bulk writer — exactly-once
FileSink<ProbeUptimeWindow> s3Sink = FileSink
    .forBulkFormat(
        new Path("s3://alivemcp-probe-analytics/uptime-windows/"),
        ParquetAvroWriters.forReflectRecord(ProbeUptimeWindow.class)  // Avro-backed Parquet writer
    )
    // Hive-compatible bucket path: year=YYYY/month=MM/day=DD/hour=HH
    // Athena partition projection reads this structure natively
    .withBucketAssigner(new DateTimeBucketAssigner<>("'year='yyyy/'month='MM/'day='dd/'hour='HH"))
    // Roll file on every checkpoint — ensures files are closed and committed predictably
    .withRollingPolicy(OnCheckpointRollingPolicy.build())
    // Compact small files: merge files below 128 MB into a single file per partition per hour
    // Prevents small-file proliferation from low-throughput time windows
    .withOutputFileConfig(
        OutputFileConfig.builder()
            .withPartPrefix("part")   // part-0-0.parquet, part-0-1.parquet, ...
            .withPartSuffix(".parquet")
            .build()
    )
    .build();

windowStream.sinkTo(s3Sink).uid("s3-parquet-sink");

For MCP probe analytics, use OnCheckpointRollingPolicy to roll files on every checkpoint rather than time-based rolling. This ensures files are closed at checkpoint boundaries, which aligns with Flink's exactly-once commit protocol. Time-based rolling (e.g., roll every 10 minutes) can create files that span checkpoint boundaries, violating the exactly-once guarantee — a rolling file straddles the checkpoint boundary and may contain records from both before and after the checkpoint commit point.

Kafka/MSK connector

The Flink Kafka connector (sourced from the Apache Flink Kafka connector library) supports both MSK and self-managed Kafka. The source provides exactly-once semantics; the sink requires explicit transactional configuration for exactly-once output:

import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;

// Kafka source with MSK IAM authentication
KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
    .setBootstrapServers("b-1.msk-cluster.kafka.us-east-1.amazonaws.com:9098")
    .setTopics("alivemcp-msk-probe-events")
    .setGroupId("alivemcp-flink-consumer")
    .setStartingOffsets(OffsetsInitializer.earliest())
    .setValueOnlyDeserializer(new SimpleStringSchema())
    // MSK IAM authentication
    .setProperty("security.protocol", "SASL_SSL")
    .setProperty("sasl.mechanism", "AWS_MSK_IAM")
    .setProperty("sasl.jaas.config", "software.amazon.msk.auth.iam.IAMLoginModule required;")
    .setProperty("sasl.client.callback.handler.class",
        "software.amazon.msk.auth.iam.IAMClientCallbackHandler")
    .build();

// Kafka sink with transactional exactly-once output (Kafka transactions)
// Requires MSK cluster with transaction coordinator (all MSK Provisioned clusters have this)
KafkaSink<String> kafkaSink = KafkaSink.<String>builder()
    .setBootstrapServers("b-1.msk-cluster.kafka.us-east-1.amazonaws.com:9098")
    .setRecordSerializer(
        KafkaRecordSerializationSchema.builder()
            .setTopic("alivemcp-uptime-results")
            .setValueSerializationSchema(new SimpleStringSchema())
            // Partition by serverSlug hash for co-location
            .setPartitioner(new FlinkFixedPartitioner<>())  // Or custom FlinkKafkaPartitioner
            .build()
    )
    // Exactly-once: Flink coordinates Kafka transactions with Flink checkpoints
    // Transaction timeout must be > checkpoint interval + max checkpoint duration
    .setDeliverGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
    .setTransactionalIdPrefix("alivemcp-flink-sink")  // Unique prefix per application
    .setProperty("transaction.timeout.ms", "300000")  // 5 min; must be > checkpoint interval
    // MSK IAM authentication (same as source)
    .setProperty("security.protocol", "SASL_SSL")
    .setProperty("sasl.mechanism", "AWS_MSK_IAM")
    .setProperty("sasl.jaas.config", "software.amazon.msk.auth.iam.IAMLoginModule required;")
    .setProperty("sasl.client.callback.handler.class",
        "software.amazon.msk.auth.iam.IAMClientCallbackHandler")
    .build();

Kafka exactly-once with Flink requires the Kafka producer to use transactional mode. Managed Flink applications using Kafka exactly-once must configure setTransactionalIdPrefix with a unique prefix per application deployment. If two application versions run with the same transactional ID prefix (e.g., during a deployment), the new producer fences the old producer and the old application fails with ProducerFencedException — this is the expected behavior for exactly-once handover during upgrades.

Connector exactly-once support matrix

Complete failure mode reference