Quickstart

from confluent_kafka import Consumer

conf = {
    "bootstrap.servers": "<bootstrap-server>",
    "group.id": "<group-id-prefix>my-consumer-group",
    "security.protocol": "SASL_SSL",
    "sasl.mechanism": "SCRAM-SHA-512",
    "sasl.username": "<username>",
    "sasl.password": "<password>",
    "auto.offset.reset": "latest",
}

consumer = Consumer(conf)
consumer.subscribe(["beam.<config_id>.usdc-logs"])

while True:
    msg = consumer.poll(1.0)
    if msg is None:
        continue
    print(msg.value().decode("utf-8"))

import { Kafka } from "kafkajs";

const kafka = new Kafka({
  brokers: ["<bootstrap-server>"],
  ssl: true,
  sasl: {
    mechanism: "scram-sha-512",
    username: "<username>",
    password: "<password>",
  },
});

const consumer = kafka.consumer({ groupId: "<group-id-prefix>my-consumer-group" });

await consumer.connect();
await consumer.subscribe({ topic: "beam.<config_id>.usdc-logs" });

await consumer.run({
  eachMessage: async ({ message }) => {
    console.log(message.value?.toString());
  },
});