Kinesis

The Kinesis source connector pulls data from Amazon Kinesis and persists it to Pulsar topics.

Create the connector

  1. Optional: If you are using the pulsar-admin CLI or Pulsar Admin API, set the following commonly-used environment variables:

    export TENANT="TENANT_NAME"
export TOPIC="OUTPUT_TOPIC_NAME"
export NAMESPACE="NAMESPACE_NAME" # or default
export SOURCE_NAME="SOURCE_CONNECTOR_UNIQUE_NAME"
export PULSAR_TOKEN="TENANT_PULSAR_TOKEN" # API only
export WEB_SERVICE_URL="TENANT_PULSAR_WEB_SERVICE_URL" # API only

    SOURCE_NAME is the name for your new source connector. DataStax recommends a memorable, human-readable name that summarizes the connector’s purpose. For example: kinesis-source-prod-us-east-1.

  2. Create the connector using JSON-formatted connector configuration data. You can pass the configuration directly or with a configuration file.

    pulsar-admin CLI
    ./bin/pulsar-admin sources create \
  --source-type kinesis \
  --name "$SOURCE_NAME" \
  --destination-topic-name "persistent://$TENANT/$NAMESPACE/$TOPIC" \
  --tenant "$TENANT" \
  --source-config-file configs.json
    Pulsar Admin API
    curl -sS --fail -L -X POST "$WEB_SERVICE_URL/admin/v3/sources/$TENANT/$NAMESPACE/$SOURCE_NAME" \
  --header "Authorization: Bearer $PULSAR_TOKEN" \
  --form "sourceConfig=@mynetty-source-config.json;type=application/json"
    Example configuration data structure
    {
  "tenant": "${TENANT}",
  "namespace": "${NAMESPACE}",
  "topicName": "persistent://${TENANT}/${NAMESPACE}/${TOPIC}",
  "name": "${SOURCE_NAME}",
  "producerConfig": {
    "maxPendingMessages": 0,
    "maxPendingMessagesAcrossPartitions": 0,
    "useThreadLocalProducers": true,
    "cryptoConfig": {
      "cryptoKeyReaderClassName": "string",
      "cryptoKeyReaderConfig": {
        "property1": {},
        "property2": {}
      },
      "encryptionKeys": [
        "string"
      ],
      "producerCryptoFailureAction": "FAIL",
      "consumerCryptoFailureAction": "FAIL"
    },
    "batchBuilder": "string"
  },
  "serdeClassName": "string",
  "schemaType": "string",
  "configs": {
    "awsEndpoint": "https://some.endpoint.aws",
    "awsRegion": "us-east-1",
    "awsKinesisStreamName": "my-stream",
    "awsCredentialPluginParam": "{\"accessKey\":\"myKey\",\"secretKey\":\"my-Secret\"}",
    "applicationName": "My test application",
    "checkpointInterval": "30000",
    "backoffTime": "4000",
    "numRetries": "3",
    "receiveQueueSize": 2000,
    "initialPositionInStream": "TRIM_HORIZON",
    "startAtTime": "2020-08-09T19:28:58.000Z"
  },
  "secrets": {},
  "parallelism": 1,
  "processingGuarantees": "ATLEAST_ONCE",
  "resources": {
    "cpu": 0.25,
    "disk": 1000000000,
    "ram": 1000000000
  },
  "archive": "builtin://kinesis",
  "runtimeFlags": "string",
  "customRuntimeOptions": "string",
  "batchSourceConfig": {
    "discoveryTriggererClassName": "string",
    "discoveryTriggererConfig": {
      "property1": {},
      "property2": {}
    }
  },
  "batchBuilder": "string"
}

Edit the connector

To update a connector, pass the new configuration definition to the connector. For example, if you created the connector with a configuration file, you can pass an updated configuration file.

You can include the entire configuration or only the properties that you want to change. Additionally, some properties can be modified with specific arguments, such as --parallelism.

To get the current configuration, see Get source connector configuration data.

pulsar-admin CLI
./bin/pulsar-admin sources update \
  --source-type kinesis \
  --name "$SOURCE_NAME" \
  --destination-topic-name "persistent://$TENANT/$NAMESPACE/$TOPIC" \
  --tenant "$TENANT" \
  --parallelism 2
Pulsar Admin API
curl -sS --fail -L -X PUT "$WEB_SERVICE_URL/admin/v3/sources/$TENANT/$NAMESPACE/$SOURCE_NAME" \
  --header "Authorization: Bearer $PULSAR_TOKEN" \
  --form "sourceConfig=@mynetty-source-config.json;type=application/json"

Manage the connector

See Create and manage connectors for details on how to manage connectors after you create them, including:

  • Get connector status

  • Get existing connectors

  • Get connector configuration details

  • Start connectors

  • Stop connectors

  • Restart connectors

  • Delete connectors

Connector configuration reference

To configure Astra Streaming Pulsar connectors, you use a combination of common Pulsar properties and provider-specific properties.

Because Astra Streaming and the Astra Streaming Pulsar connectors are based on a specific version of the open-source Apache Pulsar project and connectors, some properties and values might be unsupported or ignored by Astra Streaming.

Pulsar source connector properties

Pulsar connectors and functions both use Pulsar functions workers. Therefore, some function configuration properties are also used to configure connectors.

The following table lists Astra Streaming Pulsar source connector configuration properties by JSON field name. For the equivalent command line arguments and more information about these properties, see the following:

Pulsar configuration properties for source connectors
Field name Required Description

archive

Yes

The type of built-in Astra Streaming Pulsar connector that you want to deploy. Formatted as builtin://CONNECTOR, such as builtin://kafka.

batchBuilder

No

The batch construction method: DEFAULT or KEY_BASED.

Default: DEFAULT

batchSourceConfig

No

The batch source configuration key-value as a JSON string.

configs

Yes

A JSON-formatted key-value map containing configuration properties specific to the connector type, including provider-specific authentication and integration settings. For available subproperties, see Kinesis source connector properties (configs).

Default: {} (Empty map, uses defaults if they exist, fails otherwise)

customRuntimeOptions

No

A string that encodes options to configure the Apache Pulsar function runtime.

name

Yes

The name for the connector. It must start with a lowercase letter, and contain only numbers, hyphens (-), and lowercase letters. DataStax recommends a memorable, human-readable name that summarizes the connector’s purpose. For example: kinesis-prod-us-east-1.

namespace

Yes

The namespace in your Pulsar tenant where you want to create the connector.

parallelism

Yes

The number of Pulsar function instances to run.

Default: 1

processingGuarantees

Yes

The messaging delivery semantic to use when writing to topics: ATLEAST_ONCE, ATMOST_ONCE, or EFFECTIVELY_ONCE.

Respect for the processing guarantee depends on the connector implementation.

For more information, see the Pulsar documentation on Function processing guarantees and Processing guarantees in I/O connectors.

Default: ATLEAST_ONCE

producerConfig

No

A JSON string containing a custom producer configuration. By default, most values are null or empty. For example:

  "producerConfig": {
    "maxPendingMessages": null,
    "maxPendingMessagesAcrossPartitions": null,
    "useThreadLocalProducers": false,
    "cryptoConfig": null,
    "batchBuilder": ""
  },

resources

No

A JSON string describing the compute resources to allocate to each Pulsar function instance. For example: {"cpu":0.25,"disk":1000000000,"ram":500000000}.

runtimeFlags

No

A string that encodes flags to pass to the Apache Pulsar function runtime. Only applicable to process and Kubernetes runtimes.

schemaType

No

The schema type used to encode messages emitted from a Pulsar source. Accepts either a built-in schema like avro or json, or a custom Schema class name.

secrets

No

If security is enabled on your function workers, you can provide a map of secret names (secretName) to secrets provider configuration objects. The secretName is used by the connector to reference the secret. The mapped object contains the required properties to fetch the secret from the secrets provider.

To get the types for the values in this mapping, use the SecretProviderConfigurator.getSecretObjectType() method.

This is separate from connector-specific security settings in configs and the Pulsar authentication token used to connect to your Pulsar cluster.

serdeClassName

No

The SerDe classname for the source.

tenant

Yes

The Pulsar tenant where you want to create the connector.

topicName

Yes

An existing topic in your Astra Streaming tenant where messages are published. Must be formatted as a full persistent or non-persistent topic name, such as persistent://$TENANT/$NAMESPACE/$TOPIC.

typeClassName

Yes

The class reference for the connector type. Begins with org.apache.pulsar, such as org.apache.pulsar.common.schema.KeyValue or org.apache.pulsar.io.debezium.mysql.DebeziumMysqlSource.

Kinesis source connector properties (configs)

Set these properties in the configs section of the connector configuration.

Generally, all properties provided in the OSS Apache Pulsar Kinesis source connector are supported. Exceptions include properties that aren’t relevant to Astra Streaming and properties that aren’t present in Apache Pulsar 3.1.

Was this helpful?

Edit this page

Give Feedback

How can we improve the documentation?

© Copyright IBM Corporation 2026 | Privacy policy | Terms of use |  Manage Privacy Choices

Apache, Apache Cassandra, Cassandra, Apache Tomcat, Tomcat, Apache Lucene, Apache Solr, Apache Hadoop, Hadoop, Apache Pulsar, Pulsar, Apache Spark, Spark, Apache TinkerPop, TinkerPop, Apache Kafka and Kafka are either registered trademarks or trademarks of the Apache Software Foundation or its subsidiaries in Canada, the United States and/or other countries. Kubernetes is the registered trademark of the Linux Foundation.

General Inquiries: Contact IBM