AWS.Kafka reference

KafkaEventSource

Source: src/AWS/Kafka/ClusterEventSource.ts

Event source connecting topics on an MSK ServerlessCluster to the hosting compute.

The contract is a Binding.Service; the Lambda implementation layer (AWS.Lambda.KafkaEventSource) grants the IAM actions MSK IAM authentication requires, creates an event source mapping on the cluster, and forwards aws:kafka records into the handler’s Stream. Use the consumeKafkaTopic helper rather than calling the service directly.

KafkaEventSource: Consuming Topics

export default MyFunction.make(
  { main: import.meta.url },
  Effect.gen(function* () {
    const cluster = yield* AWS.Kafka.ServerlessCluster("Events", {
      subnetIds,
    });

    // init — registers the event source mapping and the record handler
    yield* AWS.Kafka.consumeKafkaTopic(
      cluster,
      { topics: ["orders"], consumerGroupId: "my-service" },
      (records) =>
        records.pipe(Stream.runForEach((r) => Effect.log(r.value))),
    );

    return {};
  }).pipe(Effect.provide(AWS.Lambda.KafkaEventSource)),
);

ConnectRead

Source: src/AWS/Kafka/Connect.ts

Read-only (consumer) runtime access to an MSK ServerlessCluster’s data plane.

Grants the MSK IAM-auth actions a Kafka consumer needs (kafka-cluster:Connect on the cluster, kafka-cluster:ReadData + kafka-cluster:DescribeTopic on the cluster’s topics, and kafka-cluster:DescribeGroup + kafka-cluster:AlterGroup on its consumer groups), publishes the SASL/IAM bootstrap endpoint as environment variables on the host Function, and resolves a typed ClusterConnectionInfo at runtime.

The Kafka data plane is VPC-only — the host Function must be attached to the cluster’s VPC and allowed ingress on port 9098 by the cluster’s security groups. Provide the implementation with Effect.provide(AWS.Kafka.ConnectReadHttp).

ConnectRead: Connecting to a Cluster

const connect = yield* Kafka.ConnectRead(cluster);
// inside a handler:
const { brokers, authentication } = yield* connect;

CreateTopic

Source: src/AWS/Kafka/CreateTopic.ts

Runtime binding for the CreateTopic operation (IAM action kafka:CreateTopic), scoped to one ServerlessCluster.

Creates a Kafka topic on the bound cluster through the MSK control plane — no Kafka admin client or VPC connectivity required. Provide the implementation with Effect.provide(AWS.Kafka.CreateTopicHttp).

CreateTopic: Managing Topics

const createTopic = yield* Kafka.CreateTopic(cluster);

const topic = yield* createTopic({
  TopicName: "orders",
  PartitionCount: 3,
});
// topic.TopicArn, topic.Status → "CREATING" | "ACTIVE"

DeleteTopic

Source: src/AWS/Kafka/DeleteTopic.ts

Runtime binding for the DeleteTopic operation (IAM action kafka:DeleteTopic), scoped to one ServerlessCluster.

Deletes a Kafka topic on the bound cluster through the MSK control plane. Provide the implementation with Effect.provide(AWS.Kafka.DeleteTopicHttp).

DeleteTopic: Managing Topics

const deleteTopic = yield* Kafka.DeleteTopic(cluster);

yield* deleteTopic({ TopicName: "orders" });

DescribeTopic

Source: src/AWS/Kafka/DescribeTopic.ts

Runtime binding for the DescribeTopic operation (IAM action kafka:DescribeTopic), scoped to one ServerlessCluster.

Reads a topic’s partition count, replication factor, configuration, and status through the MSK control plane. Provide the implementation with Effect.provide(AWS.Kafka.DescribeTopicHttp).

DescribeTopic: Managing Topics

const describeTopic = yield* Kafka.DescribeTopic(cluster);

const topic = yield* describeTopic({ TopicName: "orders" });
// topic.PartitionCount, topic.Configs, topic.Status

DescribeTopicPartitions

Source: src/AWS/Kafka/DescribeTopicPartitions.ts

Runtime binding for the DescribeTopicPartitions operation (IAM action kafka:DescribeTopicPartitions), scoped to one ServerlessCluster.

Reads per-partition detail (leader, replicas, ISR) for a topic through the MSK control plane. Pass NextToken from the previous page to paginate. Provide the implementation with Effect.provide(AWS.Kafka.DescribeTopicPartitionsHttp).

DescribeTopicPartitions: Managing Topics

const describeTopicPartitions = yield* Kafka.DescribeTopicPartitions(cluster);

const page = yield* describeTopicPartitions({ TopicName: "orders" });
// page.Partitions → [{ Partition: 0, Leader: …, Isr: […] }, …]

GetBootstrapBrokers

Source: src/AWS/Kafka/GetBootstrapBrokers.ts

Runtime binding for the GetBootstrapBrokers operation (IAM action kafka:GetBootstrapBrokers), scoped to one ServerlessCluster.

Resolves the cluster’s bootstrap broker connection strings at runtime — for MSK Serverless the SASL/IAM string (BootstrapBrokerStringSaslIam) is the one Kafka clients connect to. Provide the implementation with Effect.provide(AWS.Kafka.GetBootstrapBrokersHttp).

GetBootstrapBrokers: Connecting to a Cluster

const getBootstrapBrokers = yield* Kafka.GetBootstrapBrokers(cluster);

const brokers = yield* getBootstrapBrokers();
// brokers.BootstrapBrokerStringSaslIam → "b-1.….kafka-serverless.…:9098"

ListTopics

Source: src/AWS/Kafka/ListTopics.ts

Runtime binding for the ListTopics operation (IAM action kafka:ListTopics), scoped to one ServerlessCluster.

Lists the topics on the bound cluster through the MSK control plane, optionally filtered by name prefix. Provide the implementation with Effect.provide(AWS.Kafka.ListTopicsHttp).

ListTopics: Managing Topics

const listTopics = yield* Kafka.ListTopics(cluster);

const page = yield* listTopics({ TopicNameFilter: "orders" });
// page.Topics → [{ TopicName: "orders", PartitionCount: 3, … }]

ServerlessCluster

Source: src/AWS/Kafka/ServerlessCluster.ts

An Amazon MSK (Managed Streaming for Apache Kafka) Serverless cluster.

MSK Serverless clusters use IAM authentication exclusively and scale broker capacity automatically — there is no broker count, instance type, or storage to configure. They are reachable only from inside a VPC. Creation takes roughly 5-10 minutes.

The provisioned (broker-count) MSK cluster is a separate, much slower (~20-40 minute) resource and is intentionally not modeled here.

ServerlessCluster: Creating a Serverless Cluster

const cluster = yield* ServerlessCluster("Events", {
  subnetIds: [subnetA.subnetId, subnetB.subnetId],
  securityGroupIds: [kafkaSecurityGroup.securityGroupId],
});

ServerlessCluster: Consuming from a Lambda Function

yield* Kafka.consumeKafkaTopic(cluster, { topics: ["orders"] }, (records) =>
  records.pipe(Stream.runForEach((r) => Effect.log(r.value))),
);

UpdateTopic

Source: src/AWS/Kafka/UpdateTopic.ts

Runtime binding for the UpdateTopic operation (IAM action kafka:UpdateTopic), scoped to one ServerlessCluster.

Updates a topic’s configuration or increases its partition count through the MSK control plane. Provide the implementation with Effect.provide(AWS.Kafka.UpdateTopicHttp).

UpdateTopic: Managing Topics

const updateTopic = yield* Kafka.UpdateTopic(cluster);

yield* updateTopic({ TopicName: "orders", PartitionCount: 6 });