Updating the DataStax Apache Pulsar™ Connector configuration

To update the configuration of an active DataStax Apache Pulsar™ Connector sink, use the Apache Pulsar™ pulsar-admin sinks update command:

pulsar-admin sinks update --name sink_name [options]

where

sink-name - The sink name associated with the DataStax Pulsar Connector.
[options] - Apache Pulsar configuration options. For a complete list, see pulsar-admin sinks update

Example configuration update

Increase a sink’s parallelism factor to 3:

bin/pulsar-admin sinks update --name dse-sink-kv --parallelism 3
"Updated successfully"

View the updated sink configuration:

bin/pulsar-admin sinks get --name dse-sink-kv
{
  "tenant": "public",
  "namespace": "default",
  "name": "cass-sink-kv",
  "className": "com.datastax.oss.sink.pulsar.StringCassandraSinkTask",
  "inputSpecs": {
    "persistent://public/default/example_topic": {
      "isRegexPattern": false,
      "schemaProperties": {},
      "consumerProperties": {}
    }
  },
  "configs": {
    "loadBalancing.localDc": "Cassandra",
    "queryExecutionTimeout": 30,
    "auth": {
      "provider": "None",
      "gssapi": {
        "service": "dse"
      }
    },
    "tasks.max": 1,
    "topics": "example_topic",
    "contactPoints": "localhost",
    "batchFlushTimeoutMs": 1000,
    "maxConcurrentRequests": 500,
    "ignoreErrors": "None",
    "ssl": {
      "provider": "None",
      "hostnameValidation": true,
      "keystore": {},
      "openssl": {},
      "truststore": {}
    },
    "verbose": false,
    "connectionPoolLocalSize": 4,
    "jmx": true,
    "port": 9042,
    "maxNumberOfRecordsInBatch": 32,
    "topic": {
      "example_topic": {
        "pulsar_qs": {
          "pulsar_kv": {
            "mapping": "key\u003dkey,content\u003dvalue",
            "consistencyLevel": "LOCAL_ONE",
            "ttl": -1,
            "ttlTimeUnit": "SECONDS",
            "timestampTimeUnit": "MICROSECONDS",
            "nullToUnset": true,
            "deletesEnabled": true
          }
        },
        "codec": {
          "locale": "en_US",
          "timeZone": "UTC",
          "timestamp": "CQL_TIMESTAMP",
          "date": "ISO_LOCAL_DATE",
          "time": "ISO_LOCAL_TIME",
          "unit": "MILLISECONDS"
        }
      }
    },
    "batchSize": 3000,
    "compression": "None"
  },
  "parallelism": 3,
  "processingGuarantees": "ATLEAST_ONCE",
  "retainOrdering": false,
  "autoAck": true,
  "archive": "builtin://cassandra-enhanced"
}

Was this helpful?

Edit this page