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.

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)),
);