Kafka Subscriber

5 minute read

Streams Kafka subscriber allows clients to route events published in a Streams topic to an external Kafka topic.

Create a Kafka subscription

You can create a kafka subscription by making an HTTP post request on the following endpoint:

POST /streams/subscribers/kafka/api/v1/topics/{topicID}/subscriptions

The request body must contain a JSON kafka subscription configuration. For example:

{
    "subscriptionMode": "snapshot-only",
    "bootstrapServers": "my-kafka-cluster:9092",
    "topic": "kafka-topic-to-publish-to",
    "partition": 0,
    "recordKey": "optional-custom-record-key",
    "authorization": {
        "type": "sasl_ssl",
        "sasl": {
          "mechanism": "plain",
          "username": "myusername",
          "password": "mypassword"
        }
    }
}

After the kafka subscription is successfully created, Streams start sending records to your kafka cluster.

Security configuration with SASL and SSL

The Kafka subscriber supports connection with external kafka broker configured with Secure sockets layer (SSL) and Simple authentication and security layer (SASL).

The following table describes the security configuration that you can setup when creating a kafka subscription:

Configure a custom SSL root certificate authority

With SSL enabled, Streams needs to trust the root certificate authority that signed your kafka broker certificates. If you are not using one of the generic SSL certificates providers (for example, Digicert, Let’s Encrypt, and so on), but a custom root certificate authority instead, you must add it to Streams:

Ensure that the root certificate authority you have is in PEM format. You can list and extract the root CA from a truststore with the following command:

keytool -list -v -keystore <KAFKA TRUSTORE IN JKS>
...
Your keystore contains 1 entry

Alias name: caroot
Creation date: Apr 22, 2022
Entry type: trustedCertEntry
...

keytool -exportcert -rfc -keystore <KAFKA TRUSTORE IN JKS> -alias <ALIAS NAME OF ROOT CA> -file kafka.truststore.pem

In this example, the alias name of the root CA is caroot.

For more information on how to add the custom certificate, see Add self-signed TLS certificates, or ask your operator to perform this task if you are not operating Streams yourself.

Create status codes

The following are HTTP status codes that can be returned when trying to create a kafka subscription:

Besides those specific errors, you can also receive any of the generic Streams API errors.

Stop a kafka subscription

To stop sending records to your kafka cluster, delete the corresponding kafka subscription with the following request:

DELETE /streams/subscribers/kafka/api/v1/subscriptions/{subscriptionId}

Delete status codes

The following are HTTP status codes that can be returned when deleting the kafka subscription:

Besides those specific errors, you can also receive any of the generic Streams API errors.

Getting a kafka subscription

To get an existing subscription, use the following GET request:

GET /streams/subscribers/kafka/api/v1/subscriptions/{subscriptionId}

Get status codes

List of HTTP status codes that can be returned when trying to get a kafka subscription:

Besides those specific errors, you can also receive any of the generic Streams API errors.

Testing a Kafka subscription

You can test a Kafka subscription by making an HTTP Post request on the following endpoint:

POST /streams/subscribers/kafka/api/v1/subscriptions/{subscriptionId}/test

The request body can contain any JSON object and it will be sent as-is to the identified subscription.

Test status codes

The following HTTP status codes can be returned while testing a Kafka subscription:

Besides those specific errors, you can also receive any of the generic Streams API errors.

Get kafka subscriptions for a topic

Use the following HTTP Get request on your topic to get existing subscriptions:

GET /streams/subscribers/kafka/api/v1/topics/{topicId}/subscriptions

For more information on how pagination and sorting work, see Pagination.

The field names allowed for sorting are:

• subscriptionMode
• kafkaBootstrapServers
• kafkaTopic
• kafkaPartition
• kafkaRecordKey

Kafka record

As soon as the publisher starts to publish data, the kafka subscribers start to receive the messages, and they send a record with custom headers and a payload.

Record headers

The following table describes the custom headers, which are added to the messages.

Record payload

The payload of the record contains your actual data, for example, events and snapshots, and subscription errors.