pulsar.Client

Set the authentication provider to be used with the broker. Supported methods:

• AuthenticationTLS
• AuthenticationToken
• AuthenticationAthenz
• AuthenticationOauth2

Set a Python logger for this Pulsar client. Should be an instance of logging.Logger. It should be noted that if the Python logger is configured, during the termination of the Python interpreter, the Python logger will be unavailable and the default logger will be used for logging. To avoid strange behavior, you'd better delete all instances explicitly before exiting.

import logging
client = Client(service_url, logger=logging.getLogger('pulsar'))
producer = client.create_producer(topic)
# ...
del producer
del client

Define the schema of the data that will be published by this producer, e.g, schema=JsonSchema(MyRecordClass).

The schema will be used for two purposes:

• Validate the data format against the topic defined schema
• Perform serialization/deserialization between data and objects

Set the compression type for the producer. By default, message payloads are not compressed.

Supported compression types:

• CompressionType.LZ4
• CompressionType.ZLib
• CompressionType.ZSTD
• CompressionType.SNAPPY

ZSTD is supported since Pulsar 2.3. Consumers will need to be at least at that release in order to be able to receive messages compressed with ZSTD.

SNAPPY is supported since Pulsar 2.4. Consumers will need to be at least at that release in order to be able to receive messages compressed with SNAPPY.

Set the message routing mode for the partitioned producer.

Supported modes:

• PartitionsRoutingMode.RoundRobinDistribution
• PartitionsRoutingMode.UseSinglePartition

This config affects producers of partitioned topics only. It controls whether producers register and connect immediately to the owner broker of each partition or start lazily on demand. The internal producer of one partition is always started eagerly, chosen by the routing policy, but the internal producers of any additional partitions are started on demand, upon receiving their first message.

Using this mode can reduce the strain on brokers for topics with large numbers of partitions and when the SinglePartition routing policy is used without keyed messages. Because producer connection can be on demand, this can produce extra send latency for the first messages of a given partition.

Sets the batching type for the producer.

There are two batching type: DefaultBatching and KeyBasedBatching.

DefaultBatching will batch single messages:

(k1, v1), (k2, v1), (k3, v1), (k1, v2), (k2, v2), (k3, v2), (k1, v3), (k2, v3), (k3, v3)

... into single batch message:

[(k1, v1), (k2, v1), (k3, v1), (k1, v2), (k2, v2), (k3, v2), (k1, v3), (k2, v3), (k3, v3)]

KeyBasedBatching will batch incoming single messages:

(k1, v1), (k2, v1), (k3, v1), (k1, v2), (k2, v2), (k3, v2), (k1, v3), (k2, v3), (k3, v3)

... into single batch message:

[(k1, v1), (k1, v2), (k1, v3)], [(k2, v1), (k2, v2), (k2, v3)], [(k3, v1), (k3, v2), (k3, v3)]

Set the type of access mode that the producer requires on the topic.

Supported modes:

• Shared: By default multiple producers can publish on a topic.

• Exclusive: Require exclusive access for producer.

  Fail immediately if there's already a producer connected.

• WaitForExclusive: Producer creation is pending until it can acquire exclusive access.

• ExclusiveWithFencing: Acquire exclusive access for the producer.

  Any existing producer will be removed and invalidated immediately.

The initial reader positioning is done by specifying a message id. The options are:

• MessageId.earliest:

Start reading from the earliest message available in the topic

• MessageId.latest:

Start reading from the end topic, only getting messages published after the reader was created

• MessageId:

When passing a particular message id, the reader will position itself on that specific position. The first message to be read will be the message next to the specified messageId. Message id can be serialized into a string and deserialized back into a MessageId object:

# Serialize to string
s = msg.message_id().serialize()

# Deserialize from string
msg_id = MessageId.deserialize(s)

Sets a message listener for the reader. When the listener is set, the application will receive messages through it. Calls to reader.read_next() will not be allowed. The listener function needs to accept (reader, message), for example:

def my_listener(reader, message):
    # process message
    pass

Sets a message listener for the consumer. When the listener is set, the application will receive messages through it. Calls to consumer.receive() will not be allowed. The listener function needs to accept (consumer, message), for example:

def my_listener(consumer, message):
    # process message
    consumer.acknowledge(message)

Sets the size of the consumer receive queue. The consumer receive queue controls how many messages can be accumulated by the consumer before the application calls receive(). Using a higher value could potentially increase the consumer throughput at the expense of higher memory utilization. Setting the consumer queue size to zero decreases the throughput of the consumer by disabling pre-fetching of messages.

This approach improves the message distribution on shared subscription by pushing messages only to those consumers that are ready to process them. Neither receive with timeout nor partitioned topics can be used if the consumer queue size is zero. The receive() function call should not be interrupted when the consumer queue size is zero. The default value is 1000 messages and should work well for most use cases.

Consumer buffers chunk messages into memory until it receives all the chunks of the original message. While consuming chunk-messages, chunks from same message might not be contiguous in the stream, and they might be mixed with other messages' chunks. so, consumer has to maintain multiple buffers to manage chunks coming from different messages. This mainly happens when multiple publishers are publishing messages on the topic concurrently or publisher failed to publish all chunks of the messages.

If it's zero, the pending chunked messages will not be limited.

Set the regex subscription mode for use when the topic is a regex pattern.

Supported modes:

• PersistentOnly: By default only subscribe to persistent topics.
• NonPersistentOnly: Only subscribe to non-persistent topics.
• AllTopics: Subscribe to both persistent and non-persistent topics.