Managed Flink VPC Connectivity for MCP Servers — MSK, RDS, Private Subnet Sources

VPC configuration requirements

When you enable VPC configuration on a Managed Flink application, you specify a VPC ID, one or more subnet IDs, and one or more security group IDs. Managed Flink provisions ENIs in the specified subnets and attaches the specified security groups to those ENIs. Every network connection the Flink application makes (to Kinesis, MSK, S3, the Managed Flink control plane) originates from one of these ENIs.

Security group rules for common data sources

The Flink application's security group needs outbound rules allowing traffic to each data source. The data source's security group needs an inbound rule allowing traffic from the Flink application's security group (not the IP range — use security group ID references for dynamic ENI IPs):

// Terraform example — security group rules for Managed Flink accessing MSK and RDS

// 1. Outbound from Flink SG to MSK SG
resource "aws_security_group_rule" "flink_to_msk_tls" {
  type                     = "egress"
  from_port                = 9094  // MSK TLS
  to_port                  = 9094
  protocol                 = "tcp"
  source_security_group_id = aws_security_group.msk.id
  security_group_id        = aws_security_group.flink_app.id
  description              = "Managed Flink → MSK TLS brokers"
}

resource "aws_security_group_rule" "flink_to_msk_iam" {
  type                     = "egress"
  from_port                = 9098  // MSK IAM auth
  to_port                  = 9098
  protocol                 = "tcp"
  source_security_group_id = aws_security_group.msk.id
  security_group_id        = aws_security_group.flink_app.id
  description              = "Managed Flink → MSK IAM authentication port"
}

// 2. Inbound to MSK SG from Flink SG
resource "aws_security_group_rule" "msk_from_flink_tls" {
  type                     = "ingress"
  from_port                = 9094
  to_port                  = 9094
  protocol                 = "tcp"
  source_security_group_id = aws_security_group.flink_app.id
  security_group_id        = aws_security_group.msk.id
  description              = "MSK accepts TLS connections from Managed Flink"
}

// 3. Outbound from Flink SG to RDS (PostgreSQL)
resource "aws_security_group_rule" "flink_to_rds" {
  type                     = "egress"
  from_port                = 5432
  to_port                  = 5432
  protocol                 = "tcp"
  source_security_group_id = aws_security_group.rds.id
  security_group_id        = aws_security_group.flink_app.id
  description              = "Managed Flink → RDS PostgreSQL (for JDBC source/sink)"
}

// 4. Allow Flink to reach VPC endpoints (HTTPS 443 for Kinesis, S3 control plane)
resource "aws_security_group_rule" "flink_to_vpc_endpoints" {
  type      = "egress"
  from_port = 443
  to_port   = 443
  protocol  = "tcp"
  cidr_blocks = [var.vpc_cidr]  // VPC endpoints are local to the VPC CIDR
  security_group_id = aws_security_group.flink_app.id
  description = "Managed Flink → VPC endpoints (Kinesis, S3, Managed Flink control plane)"
}

VPC endpoints for private subnet Managed Flink

When Managed Flink runs in a private subnet without a NAT gateway, it cannot reach AWS service endpoints over the public internet. You must create VPC endpoints for each AWS service the application needs. The endpoint type matters for Kinesis vs S3:

// Terraform: S3 Gateway endpoint (free — no interface endpoint needed)
resource "aws_vpc_endpoint" "s3" {
  vpc_id            = var.vpc_id
  service_name      = "com.amazonaws.${var.region}.s3"
  vpc_endpoint_type = "Gateway"
  route_table_ids   = var.private_route_table_ids  // Must include the Flink subnet route tables
}

// Terraform: Kinesis Data Streams interface endpoint
resource "aws_vpc_endpoint" "kinesis_streams" {
  vpc_id              = var.vpc_id
  service_name        = "com.amazonaws.${var.region}.kinesis-streams"
  vpc_endpoint_type   = "Interface"
  subnet_ids          = var.private_subnet_ids  // Same subnets as Managed Flink
  security_group_ids  = [aws_security_group.vpc_endpoints.id]
  private_dns_enabled = true  // Required — Flink uses the default Kinesis DNS name

  tags = { Name = "kinesis-streams-endpoint" }
}

The private_dns_enabled = true setting on the Kinesis interface endpoint is critical. It overrides the public DNS resolution for kinesis.{region}.amazonaws.com with a private IP pointing to the interface endpoint ENI. Without this, Flink resolves the Kinesis endpoint to the public IP and traffic exits the VPC (failing in a private subnet with no NAT, or incurring NAT data processing charges if a NAT gateway exists).

MSK connectivity: TLS, SASL/SCRAM, and IAM authentication

When Managed Flink connects to MSK as a Kafka source or sink, the authentication method must match what the MSK cluster is configured to accept. MSK supports three client authentication mechanisms, and the Flink Kafka connector properties differ for each:

// Option 1: MSK with TLS (mutual TLS — MSK provisioned)
Properties kafkaProps = new Properties();
kafkaProps.setProperty("bootstrap.servers", "b-1.msk-cluster.example.kafka.us-east-1.amazonaws.com:9094");
kafkaProps.setProperty("security.protocol", "SSL");
kafkaProps.setProperty("ssl.truststore.location", "/etc/ssl/certs/kafka.client.truststore.jks");  // Bundled in application JAR

// Option 2: MSK with SASL/SCRAM (username/password from Secrets Manager)
Properties kafkaProps = new Properties();
kafkaProps.setProperty("bootstrap.servers", "b-1.msk-cluster.example.kafka.us-east-1.amazonaws.com:9096");
kafkaProps.setProperty("security.protocol", "SASL_SSL");
kafkaProps.setProperty("sasl.mechanism", "SCRAM-SHA-512");

// Fetch credentials from Secrets Manager at runtime — do not hardcode in application properties
SecretsManagerClient secrets = SecretsManagerClient.builder().region(Region.US_EAST_1).build();
GetSecretValueResponse secret = secrets.getSecretValue(
    GetSecretValueRequest.builder().secretId("alivemcp/msk/kafka-credentials").build()
);
JsonNode creds = objectMapper.readTree(secret.secretString());
String jaasConfig = String.format(
    "org.apache.kafka.common.security.scram.ScramLoginModule required username=\"%s\" password=\"%s\";",
    creds.get("username").asText(),
    creds.get("password").asText()
);
kafkaProps.setProperty("sasl.jaas.config", jaasConfig);

// Option 3: MSK Serverless or MSK with IAM authentication (port 9098)
Properties kafkaProps = new Properties();
kafkaProps.setProperty("bootstrap.servers", "b-1.msk-cluster.example.kafka.us-east-1.amazonaws.com:9098");
kafkaProps.setProperty("security.protocol", "SASL_SSL");
kafkaProps.setProperty("sasl.mechanism", "AWS_MSK_IAM");
kafkaProps.setProperty("sasl.jaas.config", "software.amazon.msk.auth.iam.IAMLoginModule required;");
kafkaProps.setProperty("sasl.client.callback.handler.class",
    "software.amazon.msk.auth.iam.IAMClientCallbackHandler");
// IAM auth uses the Managed Flink application's execution role — no credentials to manage

IAM authentication (Option 3) is the recommended approach for MSK Serverless and for new MSK provisioned clusters. It eliminates credential rotation (the Managed Flink execution role's IAM permissions govern access) and integrates with CloudTrail for audit. The MSK IAM auth library must be included in the application's fat JAR or uploaded as a Managed Flink dependency.

Cross-VPC connectivity via VPC peering or Transit Gateway

AliveMCP's architecture may have Managed Flink in one VPC and MSK in another (e.g., a shared services VPC). VPC peering or Transit Gateway handles cross-VPC connectivity, but three additional requirements apply beyond standard peering configuration:

• Overlapping CIDR blocks prevent peering. Check that the Managed Flink VPC CIDR and the MSK VPC CIDR do not overlap before establishing peering. If they overlap, Transit Gateway with route table isolation (not peering) is the only option.
• Security group rule references only work within the same VPC. In cross-VPC rules, use CIDR ranges (the MSK VPC's subnet CIDRs for the broker subnets) rather than security group ID references — security group ID references across peered VPCs are not supported.
• Private DNS does not automatically cross VPC peering boundaries. If the MSK VPC has private DNS enabled for its VPC endpoints, those DNS names do not resolve from the Managed Flink VPC. Configure Route 53 Resolver forwarding rules to forward MSK-related DNS queries to the MSK VPC's resolver, or use explicit broker IP addresses in the bootstrap.servers property.

Complete failure mode reference