Pulsar

Pulsar sink connector

Support Those Engines

Spark
Flink
SeaTunnel Zeta

Key features

• exactly-once
• support multiple table write
• timer flush

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

Datasource	Supported Versions
Pulsar	Universal

Sink Options

Name	Type	Required	Default	Description
topic	String	No	-	Target Pulsar topic. Required for normal single-table writes. Optional only when multi-table rows carry table ids.
client.service-url	String	Yes	-	Pulsar client service URL, for example pulsar://localhost:6650.
admin.service-url	String	Yes	-	Pulsar admin HTTP URL, for example http://localhost:8080.
auth.plugin-class	String	No	-	Pulsar authentication plugin class name.
auth.params	String	No	-	Parameters for the authentication plugin. Configure it together with auth.plugin-class.
format	String	No	json	Data format. The default format is json. Optional text and avro format.
field_delimiter	String	No	,	Field delimiter used when format = "text".
semantics	Enum	No	AT_LEAST_ONCE	Write consistency. Valid values: NON, AT_LEAST_ONCE, EXACTLY_ONCE.
transaction_timeout	Int	No	600	Pulsar transaction timeout in seconds. Used by EXACTLY_ONCE.
pulsar.config	Map	No	-	Extra producer properties passed to the Pulsar producer client.
message.routing.mode	Enum	No	RoundRobinPartition	Routing mode for partitioned topics. Valid values: SinglePartition, RoundRobinPartition.
partition_key_fields	Array	No	-	Fields used to build the Pulsar message key.
multi_table_sink_replica	Int	No	1	Writer replica count for multi-table writes.
common-options	Config	No	-	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
  }
}

Write Text Messages With a Custom Delimiter

Set format = text and configure field_delimiter to serialize each row into a delimited text payload. Useful when the downstream consumer expects a simple flat format.

sink {
  Pulsar {
    topic = "text_events"
    client.service-url = "pulsar://localhost:6650"
    admin.service-url = "http://localhost:8080"
    format = text
    field_delimiter = "|"
  }
}

Write Avro Messages

Set format = avro. The connector derives the Avro schema from the upstream row type, so no sink-side schema option is required.

sink {
  Pulsar {
    topic = "test_avro_topic_fake_source"
    client.service-url = "pulsar://localhost:6650"
    admin.service-url = "http://localhost:8080"
    format = avro
  }
}

Write With Pulsar Producer Properties

Pass extra producer properties via pulsar.config. This is forwarded to the Pulsar producer client and can be used for tuning (timeouts, batching, compression, etc.).

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

Changelog

Change Log

Change	Commit	Version
[Improve][Common] Add HashUtils.bucketIndex for hash-to-bucket routing (#11987)	https://github.com/apache/seatunnel/commit/be53a1d3d	3.0.0
[Improve][Connector-V2][Pulsar] Migrate validation to declarative OptionRule (#11985)	https://github.com/apache/seatunnel/commit/5dbfb374f	3.0.0
[Fix][Connector-V2] Support text format in Pulsar source (#11792)	https://github.com/apache/seatunnel/commit/244e58044	3.0.0
[Feature][Connector-V2] Support Avro format for Pulsar connector (#11230)	https://github.com/apache/seatunnel/commit/3ba11c718	3.0.0
[Fix][Connector-V2] Run Pulsar cleanup with connector classloader (#11170)	https://github.com/apache/seatunnel/commit/4f41c4888	3.0.0
[Fix][Connector-V2] Add Pulsar multi-table sink replica option (#11168)	https://github.com/apache/seatunnel/commit/e7309efb4	3.0.0
[Fix][Connector-V2] Add missing MULTI_TABLE_SINK_REPLICA option to PulsarSinkFactory (#11160)	https://github.com/apache/seatunnel/commit/fd2a5963a	3.0.0
[Feature][Connector-V2] Support Pulsar sink multi-table writes (#10670)	https://github.com/apache/seatunnel/commit/d1701ee6e	3.0.0
[Feature][Connector-V2][Pulsar] Support multi-table read for Pulsar source (#10425) (#10825)	https://github.com/apache/seatunnel/commit/8055ed396	3.0.0
[Improve][API] Optimize the enumerator API semantics and reduce lock calls at the connector level (#9671)	https://github.com/apache/seatunnel/commit/9212a77140	2.3.12
[improve] pulsar options (#9180)	https://github.com/apache/seatunnel/commit/26a2160c80	2.3.12
[Feature][Checkpoint] Add check script for source/sink state class serialVersionUID missing (#9118)	https://github.com/apache/seatunnel/commit/4f5adeb1c7	2.3.11
[Improve] restruct connector common options (#8634)	https://github.com/apache/seatunnel/commit/f3499a6eeb	2.3.10
[Improve][dist]add shade check rule (#8136)	https://github.com/apache/seatunnel/commit/51ef800016	2.3.9
[Feature][Restapi] Allow metrics information to be associated to logical plan nodes (#7786)	https://github.com/apache/seatunnel/commit/6b7c53d03c	2.3.9
[Improve][API] Make sure the table name in TablePath not be null (#7252)	https://github.com/apache/seatunnel/commit/764d8b0bc8	2.3.7
[Feature][Kafka] Support multi-table source read (#5992)	https://github.com/apache/seatunnel/commit/60104602d1	2.3.6
[PulsarSource]Improve pulsar throughput performance. (#6234)	https://github.com/apache/seatunnel/commit/37461f4f3e	2.3.4
[Feature][Connector-v2][PulsarSink]Add Pulsar Sink Connector. (#4382)	https://github.com/apache/seatunnel/commit/543d2c5086	2.3.4
[Chore] Remove useless DeserializationFormatFactory and its implement (#5880)	https://github.com/apache/seatunnel/commit/f0511544ff	2.3.4
fix: update IDENTIFIER = Pulsar for pulsar-datasource on project:seatunnel-web (#5852)	https://github.com/apache/seatunnel/commit/3b6de3743e	2.3.4
[Improve][Common] Introduce new error define rule (#5793)	https://github.com/apache/seatunnel/commit/9d1b2582b2	2.3.4
Support config column/primaryKey/constraintKey in schema (#5564)	https://github.com/apache/seatunnel/commit/eac76b4e50	2.3.4
[Improve][CheckStyle] Remove useless 'SuppressWarnings' annotation of checkstyle. (#5260)	https://github.com/apache/seatunnel/commit/51c0d709ba	2.3.4
[Hotfix] Fix com.google.common.base.Preconditions to seatunnel shade one (#5284)	https://github.com/apache/seatunnel/commit/ed5eadcf73	2.3.3
[Feature][Json-format] support read format for pulsar (#4111)	https://github.com/apache/seatunnel/commit/7d61ae93e7	2.3.2
[hotfix][pulsar] Fix the bug that can't consume messages all the time. (#4125)	https://github.com/apache/seatunnel/commit/a6705cc5bf	2.3.2
[Feature] add cdc multiple table support & fix zeta bug	https://github.com/apache/seatunnel/commit/533ff2c2fa	2.3.1
[hotfix][pulsar] PulsarSource consumer ack exception. (#4237)	https://github.com/apache/seatunnel/commit/9725d675da	2.3.1
Merge branch 'dev' into merge/cdc	https://github.com/apache/seatunnel/commit/4324ee1912	2.3.1
[Improve][Project] Code format with spotless plugin.	https://github.com/apache/seatunnel/commit/423b583038	2.3.1
[Improve][Connector-v2][Pulsar] Set the name of the pulsar consumption thread. (#4182)	https://github.com/apache/seatunnel/commit/e567203f7d	2.3.1
[improve][api] Refactoring schema parse (#4157)	https://github.com/apache/seatunnel/commit/b2f573a13e	2.3.1
[Improve][build] Give the maven module a human readable name (#4114)	https://github.com/apache/seatunnel/commit/d7cd601051	2.3.1
[Improve][Project] Code format with spotless plugin. (#4101)	https://github.com/apache/seatunnel/commit/a2ab166561	2.3.1
[Bug][Connector-v2][PulsarSource]Fix pulsar option topic-pattern bug. (#3989)	https://github.com/apache/seatunnel/commit/aee2c580ea	2.3.1
[Feature][Connector] add get source method to all source connector (#3846)	https://github.com/apache/seatunnel/commit/417178fb84	2.3.1
[Feature][API & Connector & Doc] add parallelism and column projection interface (#3829)	https://github.com/apache/seatunnel/commit/b9164b8ba1	2.3.1
[Improve][Connector-V2][Pulsar] Unified exception for Pulsar source &… (#3590)	https://github.com/apache/seatunnel/commit/4fe9323419	2.3.0
[Hotfix][OptionRule] Fix option rule about all connectors (#3592)	https://github.com/apache/seatunnel/commit/226dc6a119	2.3.0
[Hotfix][Connector-V2][Pulsar] fix conditional options (#3504)	https://github.com/apache/seatunnel/commit/0066affacf	2.3.0
[Feature][Connector][pulsar] expose configurable options in Pulsar (#3341)	https://github.com/apache/seatunnel/commit/200faa7c29	2.3.0
[Connector][Dependency] Add Miss Dependency Cassandra And Change Kudu Plugin Name (#3432)	https://github.com/apache/seatunnel/commit/6ac6a0a0cd	2.3.0
[chore] fix pulsar consumer comment error (#3356)	https://github.com/apache/seatunnel/commit/91e632c526	2.3.0
[Connector-V2][ElasticSearch] Add ElasticSearch Source/Sink Factory (#3325)	https://github.com/apache/seatunnel/commit/38254e3f26	2.3.0
[hotfix][connector][pulsar] Fix not being able to mark #noMoreNewSplits when restoring (#2945)	https://github.com/apache/seatunnel/commit/5ad69076b3	2.3.0-beta
Move Handover to common module (#2877)	https://github.com/apache/seatunnel/commit/d94a874bcb	2.3.0-beta
[hotfix][connector-v2] fix pulsar source exceptions (#2820)	https://github.com/apache/seatunnel/commit/8ff0ba7015	2.2.0-beta
[#2606]Dependency management split (#2630)	https://github.com/apache/seatunnel/commit/fc047be69b	2.2.0-beta
[SeaTunnel]Simply seatunnel package pipeline. (#2563)	https://github.com/apache/seatunnel/commit/9d88b6221a	2.2.0-beta
[Improve][Connector-V2] Pulsar support user-defined schema (#2436)	https://github.com/apache/seatunnel/commit/16cabe6a35	2.2.0-beta
[improve][UT] Upgrade junit to 5.+ (#2305)	https://github.com/apache/seatunnel/commit/362319ff3e	2.2.0-beta
StateT of SeaTunnelSource should extend Serializable (#2214)	https://github.com/apache/seatunnel/commit/8c426ef850	2.2.0-beta
[doc][connector-v2] pulsar source options doc (#2128)	https://github.com/apache/seatunnel/commit/59ce8a2b32	2.2.0-beta
[api-draft][Optimize] Optimize module name (#2062)	https://github.com/apache/seatunnel/commit/f79e3112b1	2.2.0-beta