Skip to main content
Version: Next

Pulsar

Pulsar sink connector

Support Those Engines​

Spark
Flink
SeaTunnel Zeta

Key features​

Description​

The Pulsar sink writes SeaTunnel rows to Apache Pulsar topics. It can write to one configured topic, or route multi-table records by the table id carried in each row.

Supported DataSource Info​

DatasourceSupported Versions
PulsarUniversal

Sink Options​

NameTypeRequiredDefaultDescription
topicStringNo-Target Pulsar topic. Required for normal single-table writes. Optional only when multi-table rows carry table ids.
client.service-urlStringYes-Pulsar client service URL, for example pulsar://localhost:6650.
admin.service-urlStringYes-Pulsar admin HTTP URL, for example http://localhost:8080.
auth.plugin-classStringNo-Pulsar authentication plugin class name.
auth.paramsStringNo-Parameters for the authentication plugin. Configure it together with auth.plugin-class.
formatStringNojsonData format. The default format is json. Optional text and avro format.
field_delimiterStringNo,Field delimiter used when format = "text".
semanticsEnumNoAT_LEAST_ONCEWrite consistency. Valid values: NON, AT_LEAST_ONCE, EXACTLY_ONCE.
transaction_timeoutIntNo600Pulsar transaction timeout in seconds. Used by EXACTLY_ONCE.
pulsar.configMapNo-Extra producer properties passed to the Pulsar producer client.
message.routing.modeEnumNoRoundRobinPartitionRouting mode for partitioned topics. Valid values: SinglePartition, RoundRobinPartition.
partition_key_fieldsArrayNo-Fields used to build the Pulsar message key.
multi_table_sink_replicaIntNo1Writer replica count for multi-table writes.
common-optionsConfigNo-Sink plugin common parameters. See Sink Common Options.

Parameter Interpretation​

topic [String]​

The default Pulsar topic used by the sink.

For single-table pipelines, configure topic. For multi-table pipelines, the sink uses SeaTunnelRow.getTableId() as the target topic when the row has a table id, and falls back to topic only when the row does not carry one.

If neither the row table id nor topic is available, the sink fails fast with a configuration error.

client.service-url [String]​

Pulsar client service URL. Use the Pulsar protocol, for example pulsar://localhost:6650.

admin.service-url [String]​

Pulsar admin HTTP URL.

For example, http://my-broker.example.com:8080, or https://my-broker.example.com:8443 for TLS.

auth.plugin-class [String]​

Name of the authentication plugin.

auth.params [String]​

Parameters for the authentication plugin.

For example, key1:val1,key2:val2.

format [String]​

Data format. The default format is json. Optional text and avro format. You can also use text. When using text, configure field_delimiter if the default comma delimiter is not suitable. When using avro format, the Avro schema is derived from the upstream row type; no sink-side schema option is required.

field_delimiter [String]​

Field delimiter used by the text format. The default value is ,.

semantics [Enum]​

Write consistency semantics.

  • AT_LEAST_ONCE: the default. Messages may be duplicated after job restart or retry.
  • EXACTLY_ONCE: writes messages through Pulsar transactions. The Pulsar cluster must enable transaction support, and transaction_timeout should be greater than the checkpoint interval.
  • NON: sends messages directly without checkpoint coordination. Data may be duplicated or lost after restart, retry, or network errors.

transaction_timeout [Int]​

Pulsar transaction timeout in seconds. The default value is 600.

This option is only used when semantics = "EXACTLY_ONCE". If a transaction is not committed before the timeout, Pulsar aborts it automatically.

pulsar.config [Map]​

Extra Pulsar producer properties. These properties are passed to the Pulsar producer client.

message.routing.mode [Enum]​

Routing mode for partitioned topics.

  • SinglePartition: without a message key, one partition is selected and all messages are sent to it. With a key, Pulsar hashes the key and sends the message to the selected partition.
  • RoundRobinPartition: without a message key, messages are sent across partitions in round-robin order. Pulsar applies round-robin routing at the batching-delay boundary, not per individual message.

partition_key_fields [Array]​

Fields used to build the Pulsar message key.

For example, if the upstream row has name and age, setting partition_key_fields = ["name"] builds a JSON key such as {"name":"Jack"}. The selected fields must exist in the upstream schema.

multi_table_sink_replica [Int]​

Replica count for multi-table sink writers. It applies to all target topics in the multi-table sink.

common options​

Sink plugin common parameters, please refer to Sink Common Options for details.

Task Example​

Write FakeSource To Pulsar​

env {
  parallelism = 1
  job.mode = "BATCH"
}

source {
  FakeSource {
    plugin_output = "fake"
    row.num = 10
    schema = {
      fields {
        c_string = string
        c_int = int
        c_bigint = bigint
        c_double = double
        c_timestamp = timestamp
      }
    }
  }
}

sink {
  Pulsar {
    topic = "topic_test"
    client.service-url = "pulsar://localhost:6650"
    admin.service-url = "http://localhost:8080"
    format = json
    pulsar.config = {
      sendTimeoutMs = 30000
    }
  }
}

Write With Message Key​

sink {
  Pulsar {
    topic = "orders"
    client.service-url = "pulsar://localhost:6650"
    admin.service-url = "http://localhost:8080"
    partition_key_fields = ["order_id"]
    message.routing.mode = "SinglePartition"
    format = json
  }
}

Exactly-Once Write​

env {
  parallelism = 1
  job.mode = "STREAMING"
  checkpoint.interval = 10000
}

sink {
  Pulsar {
    topic = "orders"
    client.service-url = "pulsar://localhost:6650"
    admin.service-url = "http://localhost:8080"
    semantics = "EXACTLY_ONCE"
    transaction_timeout = 600
    format = json
  }
}

Multi-Table Write​

When upstream rows carry table ids, the sink can route each row to the topic with the same table id. In this mode, topic can be omitted.

sink {
  Pulsar {
    client.service-url = "pulsar://localhost:6650"
    admin.service-url = "http://localhost:8080"
    multi_table_sink_replica = 2
    format = json
  }
}

Changelog​

Change Log
ChangeCommitVersion
[Improve][API] Optimize the enumerator API semantics and reduce lock calls at the connector level (#9671)https://github.com/apache/seatunnel/commit/9212a771402.3.12
[improve] pulsar options (#9180)https://github.com/apache/seatunnel/commit/26a2160c802.3.12
[Feature][Checkpoint] Add check script for source/sink state class serialVersionUID missing (#9118)https://github.com/apache/seatunnel/commit/4f5adeb1c72.3.11
[Improve] restruct connector common options (#8634)https://github.com/apache/seatunnel/commit/f3499a6eeb2.3.10
[Improve][dist]add shade check rule (#8136)https://github.com/apache/seatunnel/commit/51ef8000162.3.9
[Feature][Restapi] Allow metrics information to be associated to logical plan nodes (#7786)https://github.com/apache/seatunnel/commit/6b7c53d03c2.3.9
[Improve][API] Make sure the table name in TablePath not be null (#7252)https://github.com/apache/seatunnel/commit/764d8b0bc82.3.7
[Feature][Kafka] Support multi-table source read (#5992)https://github.com/apache/seatunnel/commit/60104602d12.3.6
[PulsarSource]Improve pulsar throughput performance. (#6234)https://github.com/apache/seatunnel/commit/37461f4f3e2.3.4
[Feature][Connector-v2][PulsarSink]Add Pulsar Sink Connector. (#4382)https://github.com/apache/seatunnel/commit/543d2c50862.3.4
[Chore] Remove useless DeserializationFormatFactory and its implement (#5880)https://github.com/apache/seatunnel/commit/f0511544ff2.3.4
fix: update IDENTIFIER = Pulsar for pulsar-datasource on project:seatunnel-web (#5852)https://github.com/apache/seatunnel/commit/3b6de3743e2.3.4
[Improve][Common] Introduce new error define rule (#5793)https://github.com/apache/seatunnel/commit/9d1b2582b22.3.4
Support config column/primaryKey/constraintKey in schema (#5564)https://github.com/apache/seatunnel/commit/eac76b4e502.3.4
[Improve][CheckStyle] Remove useless 'SuppressWarnings' annotation of checkstyle. (#5260)https://github.com/apache/seatunnel/commit/51c0d709ba2.3.4
[Hotfix] Fix com.google.common.base.Preconditions to seatunnel shade one (#5284)https://github.com/apache/seatunnel/commit/ed5eadcf732.3.3
[Feature][Json-format] support read format for pulsar (#4111)https://github.com/apache/seatunnel/commit/7d61ae93e72.3.2
[hotfix][pulsar] Fix the bug that can't consume messages all the time. (#4125)https://github.com/apache/seatunnel/commit/a6705cc5bf2.3.2
[Feature] add cdc multiple table support & fix zeta bughttps://github.com/apache/seatunnel/commit/533ff2c2fa2.3.1
[hotfix][pulsar] PulsarSource consumer ack exception. (#4237)https://github.com/apache/seatunnel/commit/9725d675da2.3.1
Merge branch 'dev' into merge/cdchttps://github.com/apache/seatunnel/commit/4324ee19122.3.1
[Improve][Project] Code format with spotless plugin.https://github.com/apache/seatunnel/commit/423b5830382.3.1
[Improve][Connector-v2][Pulsar] Set the name of the pulsar consumption thread. (#4182)https://github.com/apache/seatunnel/commit/e567203f7d2.3.1
[improve][api] Refactoring schema parse (#4157)https://github.com/apache/seatunnel/commit/b2f573a13e2.3.1
[Improve][build] Give the maven module a human readable name (#4114)https://github.com/apache/seatunnel/commit/d7cd6010512.3.1
[Improve][Project] Code format with spotless plugin. (#4101)https://github.com/apache/seatunnel/commit/a2ab1665612.3.1
[Bug][Connector-v2][PulsarSource]Fix pulsar option topic-pattern bug. (#3989)https://github.com/apache/seatunnel/commit/aee2c580ea2.3.1
[Feature][Connector] add get source method to all source connector (#3846)https://github.com/apache/seatunnel/commit/417178fb842.3.1
[Feature][API & Connector & Doc] add parallelism and column projection interface (#3829)https://github.com/apache/seatunnel/commit/b9164b8ba12.3.1
[Improve][Connector-V2][Pulsar] Unified exception for Pulsar source &… (#3590)https://github.com/apache/seatunnel/commit/4fe93234192.3.0
[Hotfix][OptionRule] Fix option rule about all connectors (#3592)https://github.com/apache/seatunnel/commit/226dc6a1192.3.0
[Hotfix][Connector-V2][Pulsar] fix conditional options (#3504)https://github.com/apache/seatunnel/commit/0066affacf2.3.0
[Feature][Connector][pulsar] expose configurable options in Pulsar (#3341)https://github.com/apache/seatunnel/commit/200faa7c292.3.0
[Connector][Dependency] Add Miss Dependency Cassandra And Change Kudu Plugin Name (#3432)https://github.com/apache/seatunnel/commit/6ac6a0a0cd2.3.0
[chore] fix pulsar consumer comment error (#3356)https://github.com/apache/seatunnel/commit/91e632c5262.3.0
[Connector-V2][ElasticSearch] Add ElasticSearch Source/Sink Factory (#3325)https://github.com/apache/seatunnel/commit/38254e3f262.3.0
[hotfix][connector][pulsar] Fix not being able to mark #noMoreNewSplits when restoring (#2945)https://github.com/apache/seatunnel/commit/5ad69076b32.3.0-beta
Move Handover to common module (#2877)https://github.com/apache/seatunnel/commit/d94a874bcb2.3.0-beta
[hotfix][connector-v2] fix pulsar source exceptions (#2820)https://github.com/apache/seatunnel/commit/8ff0ba70152.2.0-beta
[#2606]Dependency management split (#2630)https://github.com/apache/seatunnel/commit/fc047be69b2.2.0-beta
[SeaTunnel]Simply seatunnel package pipeline. (#2563)https://github.com/apache/seatunnel/commit/9d88b6221a2.2.0-beta
[Improve][Connector-V2] Pulsar support user-defined schema (#2436)https://github.com/apache/seatunnel/commit/16cabe6a352.2.0-beta
[improve][UT] Upgrade junit to 5.+ (#2305)https://github.com/apache/seatunnel/commit/362319ff3e2.2.0-beta
StateT of SeaTunnelSource should extend Serializable (#2214)https://github.com/apache/seatunnel/commit/8c426ef8502.2.0-beta
[doc][connector-v2] pulsar source options doc (#2128)https://github.com/apache/seatunnel/commit/59ce8a2b322.2.0-beta
[api-draft][Optimize] Optimize module name (#2062)https://github.com/apache/seatunnel/commit/f79e3112b12.2.0-beta