Apache Pulsar Connector
The Pulsar connector allows you to stream, filter, and transform events between Hazelcast clusters and Pulsar.
Apache Pulsar is a popular distributed streaming and messaging system. Data in Pulsar is structured as topics and each topic consists of one or more partitions, stored in the Pulsar cluster. Each topic is assigned to one tenant and namespace.
Installing the Connector
This connector is included in the full distribution of Hazelcast.
If you're using the slim distribution, you must add the hazelcast-jet-pulsar module to your member's classpath.
Permissions
Enterprise EditionThis feature requires Hazelcast Enterprise, a licensed edition of the Hazelcast Platform.If security is enabled, your clients may need permissions to use this connector. For details, see Securing Jobs.
Configuration Options
To read from Pulsar as a source you must provide at least Schema, which Pulsar uses for given topic:
Pipeline p = Pipeline.reate();
p.readFrom(PulsarSources.pulsarConsumerBuilder(json(MyPojo.class))
.connectionSupplier(() -> PulsarClient.builder().serviceUrl("pulsar://localhost:6650").build())
.topic("myTopic")
.build())
.withNativeTimestamps(0)
.writeTo(Sinks.logger());
Each member of Hazelcast cluster will create a Pulsar Consumer with the same subscription name and subscription type "Shared", meaning that the reading will be spread across Hazelcast cluster.
You can also use PulsarSources#pulsarConsumer or PulsarSources#pulsarReader, which are the same as in old version from hazelcast-jet-contrib. We kept those for easy migration, but please be aware, that they will be removed at some point.
When used as a sink, then the only requirement is again the Schema of target topic:
Pipeline p = Pipeline.create();
p.readFrom(Sources.files("home/logs"))
.map(line -> LogParser.parse(line))
.map(log -> new LogEntry(log))
.writeTo(PulsarSinks.builder(json(LogEntry.class))
.connectionSupplier(() -> PulsarClient.builder().serviceUrl("pulsar://localhost:6650").build())
.topic("logs")
.build());`
Fault Tolerance
One of the most important features of using Pulsar as a source is that it's possible to replay data, which enables fault-tolerance. If the job has a processing guarantee configured, Hazelcast will periodically save the current offsets internally and then replay from the saved offset when the job is restarted. In this mode, Hazelcast will commit messages, keeping track of their offsets.
If the processing guarantees are disabled, the source will start reading from earliest messages in the topic.
Pulsar source is a stateful source—it remembers the last processed offset. To prevent data loss when using at-least-once or exactly-once job guarantees and Pulsar consumer configured to start from the latest offset, enable setRequireSnapshotBeforeProcessing in the job configuration. See additional details xref:hazelcast:fault-tolerance:fault-tolerance.adoc#_fault_tolerance.
Transactional Guarantees
As a sink, the Pulsar connector provides exactly-once guarantees at the cost of using Pulsar transactions, which must be enabled on the broker and in DataConnecion/provided PulsarClient. Hazelcast commits the produced records messages each snapshot is completed. This greatly increases the latency because consumers see the records only after they are committed.
If you use at-least-once guarantee or user calls useTransactions(false) in sink builder, records are visible immediately, but in the case of a failure some records could be duplicated. You can also configure jobs in exactly-once mode and decrease the guarantee just for a particular Pulsar sink.
It is important to enable deduplication on Pulsar sink topic, because in some very rare occasions Pulsar Client may retry message delivery (e.g. network issue caused client to not receive acknowledgment). Sink will automatically check if deduplication is enabled and warn if it isn't or if it is not possible to check.
Version Compatibility
The Pulsar sink and source are based on version 5.x while keeping support of 4.x.