Streaming to Upstash (Kafka)

​

What’s Kafka?

Apache Kafka is a distributed event store and stream-processing platform. It lets you store streams of records in a fault-tolerant way and process streams of records as they occur. Kafka is commonly used for building real-time streaming data pipelines and applications.

Upstash is a serverless Kafka service that makes it easy to get started with Kafka without having to manage servers or understand complex configurations. With Upstash, you can create a Kafka cluster in seconds and start streaming data immediately.

Sequin makes it easy to stream data from APIs like Salesforce, Jira, and Stripe into Kafka topics. This enables a variety of real-time use cases, such as:

• Triggering notifications when a new high-priority Jira ticket is created
• Updating an internal database whenever a Salesforce contact is updated
• Calculating real-time metrics on Stripe payments

To stream data to Upstash Kafka via Sequin, you will:

• Create an Upstash Kafka cluster
• Connect Sequin to an API source
• Create a Sequin Kafka consumer that streams your Upstash Kafka topic

Sequin will load all historical data from the API the topic, then will keep your Kafka topic up-to-date by sending new records and updates as they occur. This means your Kafka topics will always have a complete copy of your API data.

​

Create an Upstash Kafka cluster

Sign up for an Upstash account at https://console.upstash.com/signup.

Once logged in, navigate to the Kafka section and click “Create cluster”. Give your cluster a name and select a region (i.e. us-east-1).

​

Create a Kafka topic to receive data

Next, create a Kafka topic to receive data from Sequin. From your cluster overview page, click “Create topic”. Give the topic a name (e.g. sequin-records) and set the number of partitions (1 is fine to start).

Choose “Compact” as the cleanup policy. This will ensure that Kafka retains the latest record for each key. This means your Kafka cluster will contain a complete view of the latest data in your upstream APIs.

Click “Create” to create the topic.

​

Create a user for Sequin

After your cluster is provisioned, create a user that you’ll use later for Sequin:

1. Click the “Credentials” tab.
2. Click “New Credentials.”
3. Give your user a name (i.e. sequin).
4. For topic, enter the topic you created in the previous step (i.e. sequin-records).
5. Give the user permissions to “Produce Only.”
6. Click create.

Note the credentials for the next step.

​

Connect Sequin to an API source

You can connect Sequin to an upstream API using either the Sequin console or API.

Here’s a curl request that connects Sequin to Stripe:

curl --request POST \
  --url https://api.sequin.io/v1/syncs \
  --header 'Authorization: Bearer {your-api-token}' \
  --header 'Content-Type: application/json' \
  --data '{
    "provider": "stripe",
    "collection_ids": ["stripe:customer", "stripe:charge", "stripe:payout"],
    "credential": {
      "kind": "secret_key",
      "secret_key": "sk_live_51I..."
    }
  }'

This request instructs Sequin to sync Stripe customers, charges, and payouts using the provided API key.

​

Create a Sequin kafka consumer

Next, you’ll create a webhook consumer to stream data from Stripe to your Upstash Kafka topic. Consumers are how you stream data from Sequin’s syncs to destinations.

You can create consumers in Sequin’s console or API. Here’s an example curl request to create a Kafka consumer:

curl --request POST \
  --url https://api.sequin.io/v1/kafka_consumers \
  --header 'Authorization: Bearer {your-api-token}' \
  --header 'Content-Type: application/json' \
  --data '{
    "name": "Upstash - Stripe",
    "kafka": {
      "brokers": ["wise-friend-91jas1-us1-kafka.upstash.io:9092"],
      "topic": "sequin-records",
      "key_prefix": "sequin",
      "credential": {
        "kind": "SASL_SSL",
        "mechanism": "SCRAM-SHA-256",
        "username": "ufnEJNH/Z2LSTc7/TPqozRYTSOE7ludR7e",
        "password": "my-password"
      }
    },
    "sync_id": "8ff58ef0-d376-4ae8-b2e2-9f0206aa65b8"
  }'

Use the topic and credential you provisioned for Sequin earlier. key_prefix is optional and can be used to prefix the Kafka record key (defaults to "sequin").

This configuration will send each Stripe record to the Kafka topic sequin-records.

At this point, your Stripe data should be flowing into Kafka! Let’s verify by consuming from the topic.

​

Consume from the Kafka topic

There are many ways to consume from a Kafka topic. For this example, we’ll use the Upstash Kafka REST API.

To consume the latest record from your topic, make a GET request to the following URL:

curl --request GET \
  --url 'https://{rest-api-id}.rest.upstash.io/consume/{topic-name}' \
  --header 'Authorization: Basic {base64-encoded-username-password}'

You should see a response like this:

{
  "records": [
    {
      "topic": "sequin-data",
      "partition": 0,
      "offset": 1,
      "timestamp": 1683749961535,
      "key": "sequin.default.8ff58ef0-d376-4ae8-b2e2-9f0206aa65b8.hubspot.stripe:charge.ch_1N5w6wI5GiWQwPxYVZQ5DvIR",
      "value": "{\"collection_id\":\"stripe:charge\",\"sync_id\":\"63b8f7d2-c71a-4ad4-8e2e-e20877d812ac\",\"data\":{\"id\":\"ch_1N5w6wI5GiWQwPxYVZQ5DvIR\",\"object\":\"charge\",\"amount\":2000,\"amount_captured\":2000,\"amount_refunded\":0, ... }"
    }
  ]
}

The value field contains the stringified JSON record from Sequin.

Now, any time a Stripe customer, charge, or payout is created or updated, Sequin will send the record to your Kafka topic in real-time!

​

Where to next?

By streaming your API data to Kafka, you’ve opened up a world of possibilities for real-time applications.

If you weren’t following along, head over to Upstash to sign up for an account and start streaming.

Check out the Upstash Kafka docs to learn more about their platform, and the Sequin docs for more on streaming API data.