Skip to content

docs: document remaining pulsar_source / pulsar_sink arguments in README #205

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
david-streamlio wants to merge 2 commits into from
+72 −2
Closed

docs: document remaining pulsar_source / pulsar_sink arguments in README #205

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
52f1c02
docs: document secrets and custom_runtime_options on source/sink
david-streamlio Jun 5, 2026
124fb72
docs: document remaining source/sink resource arguments in README
david-streamlio Jun 5, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
53 changes: 52 additions & 1 deletion README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -605,9 +605,29 @@ resource "pulsar_source" "source-1" {
cpu = 2
disk_mb = 20480
ram_mb = 2048

// Keep credentials out of `configs` (which is stored in plaintext in the
// function metadata topic, `pulsar-admin sources get` output, and state).
// Reference them through the worker's secrets provider instead.
secrets = jsonencode(
{
"gsaKey": {
"path": "my-k8s-secret"
"key": "gsa-key.json"
}
})
}
```

> **Handling credentials:** values placed in `configs` are stored and returned
> in plaintext (function metadata topic, `pulsar-admin sources get`, Terraform
> state). For API keys, passwords, and service-account keys use the `secrets`
> argument instead. Each entry maps a secret name (the key the connector reads)
> to a `{ "path": ..., "key": ... }` reference that the worker's configured
> `SecretsProvider` (for example the `KubernetesSecretsProviderConfigurator`)
> resolves at runtime, so only the reference — never the value — lands in
> config, metadata, or state.

#### Properties

| Property | Description | Required |
Expand All @@ -626,6 +646,19 @@ resource "pulsar_source" "source-1" {
| `ram_mb` | The RAM that need to be allocated per source instance (applicable only to the process and Docker runtimes) | False |
| `disk_mb` | The disk that need to be allocated per source instance (applicable only to Docker runtime) | False |
| `runtime_flags` | User defined configs key/values (JSON string) | False |
| `schema_type` | The schema type (either a builtin schema like 'avro', 'json', etc.. or custom Schema class name to be used to encode messages emitted from the source | False |
| `custom_runtime_options` | A string that encodes options to customize the runtime, see docs for configured runtime for details | False |
| `secrets` | The map of secretName to an object that encapsulates how the secret is fetched by the underlying secrets provider | False |
| `max_pending_messages` | The maximum size of a queue holding pending messages | False |
| `max_pending_messages_across_partitions` | The maximum number of pending messages across partitions | False |
| `use_thread_local_producers` | Whether to use thread local producers | False |
| `batch_builder` | BatchBuilder provides two types of batch construction methods, DEFAULT and KEY_BASED. | False |
| `compression_type` | Set the compression type for the producer. By default, message payloads are not compressed. Supported compression types are: LZ4, ZLIB, ZSTD, SNAPPY and NONE | False |
| `crypto_key_reader_classname` | The classname for the crypto key reader that can be used to access the keys in the keystore | False |
| `crypto_key_reader_config` | The config for the crypto key reader that can be used to access the keys in the keystore | False |
| `encryption_keys` | One or more public keys to encrypt data key. It can be used to encrypt data key with multiple keys. | False |
| `producer_crypto_failure_action` | The desired action if producer fail to encrypt data, one of FAIL, SEND | False |
| `consumer_crypto_failure_action` | The desired action if consumer fail to decrypt data, one of FAIL, DISCARD, CONSUME | False |

### `pulsar_sink`

Expand Down Expand Up @@ -653,7 +686,18 @@ resource "pulsar_sink" "sample-sink-1" {
processing_guarantees = "EFFECTIVELY_ONCE"

archive = "testdata/pulsar-io/pulsar-io-jdbc-postgres-2.10.4.nar"
configs = "{\"jdbcUrl\":\"jdbc:clickhouse://localhost:8123/pulsar_clickhouse_jdbc_sink\",\"password\":\"password\",\"tableName\":\"pulsar_clickhouse_jdbc_sink\",\"userName\":\"clickhouse\"}"
configs = "{\"jdbcUrl\":\"jdbc:clickhouse://localhost:8123/pulsar_clickhouse_jdbc_sink\",\"tableName\":\"pulsar_clickhouse_jdbc_sink\",\"userName\":\"clickhouse\"}"

// Reference the database password through the secrets provider rather than
// embedding it in `configs` (see the credential-handling note under
// `pulsar_source`). The sink reads it under the `password` secret name.
secrets = jsonencode(
{
"password": {
"path": "clickhouse-credentials"
"key": "password"
}
})
}
```

Expand All @@ -669,6 +713,7 @@ resource "pulsar_sink" "sample-sink-1" {
| `input_specs` | The map of input topics specs | False |
| `configs` | User defined configs key/values (JSON string) | False |
| `archive` | Path to the archive file for the sink. It also supports url-path [http/https/file (file protocol assumes that file already exists on worker host)] from which worker can download the package | True |
| `classname` | The sink's class name if archive is file-url-path (file://) | False |
| `subscription_name` | Pulsar source subscription name if user wants a specific subscription-name for input-topic consumer | False |
| `subscription_position` | Pulsar source subscription position if user wants to consume messages from the specified location (Latest, Earliest) | False |
| `cleanup_subscription` | Whether the subscriptions the functions created/used should be deleted when the functions was deleted | True |
Expand All @@ -683,6 +728,12 @@ resource "pulsar_sink" "sample-sink-1" {
| `custom_schema_inputs` | The map of input topics to Schema types or class names (as a JSON string) | False |
| `custom_serde_inputs` | The map of input topics to SerDe class names (as a JSON string) | False |
| `custom_runtime_options` | A string that encodes options to customize the runtime | False |
| `secrets` | The map of secretName to an object that encapsulates how the secret is fetched by the underlying secrets provider | False |
| `dead_letter_topic` | Name of the dead topic where the failing messages will be sent | False |
| `max_redeliver_count` | Maximum number of times that a message will be redelivered before being sent to the dead letter topic | False |
| `negative_ack_redelivery_delay_ms` | The negative ack message redelivery delay in milliseconds | False |
| `retain_key_ordering` | Sink consumes and processes messages in key order | False |
| `sink_type` | The sinks's connector provider | False |

### `pulsar_subscription`

Expand Down
12 changes: 11 additions & 1 deletion examples/sinks/main.tf
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,5 +48,15 @@ resource "pulsar_sink" "sink-1" {
disk_mb = 102400

archive = "https://www.apache.org/dyn/mirrors/mirrors.cgi?action=download&filename=pulsar/pulsar-2.10.4/connectors/pulsar-io-jdbc-postgres-2.10.4.nar"
configs = "{\"jdbcUrl\":\"jdbc:postgresql://localhost:5432/pulsar_postgres_jdbc_sink\",\"password\":\"password\",\"tableName\":\"pulsar_postgres_jdbc_sink\",\"userName\":\"postgres\"}"
configs = "{\"jdbcUrl\":\"jdbc:postgresql://localhost:5432/pulsar_postgres_jdbc_sink\",\"tableName\":\"pulsar_postgres_jdbc_sink\",\"userName\":\"postgres\"}"

// Reference the database password through the secrets provider rather than
// embedding it in `configs` (stored in plaintext). The sink reads it under
// the `password` secret name.
secrets = jsonencode({
"password" = {
"path" = "postgres-credentials"
"key" = "password"
}
})
}
9 changes: 9 additions & 0 deletions examples/sources/main.tf
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,4 +46,13 @@ resource "pulsar_source" "source-1" {
cpu = 2
disk_mb = 20480
ram_mb = 2048

// Keep credentials out of `configs` (stored in plaintext). Reference them via
// the worker's secrets provider instead; only the reference is persisted.
secrets = jsonencode({
"gsaKey" = {
"path" = "my-k8s-secret"
"key" = "gsa-key.json"
}
})
}
Loading