Skip to main content
Version: Next

Work with producer

After setting up your clients, you can explore more to start working with producers.

Create the producer​

This example shows how to create a producer.

Create a producer synchronously:

Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
              .topic("my-topic")
              .create();

Create a producer asynchronously:

pulsarClient.newProducer(Schema.STRING)
              .topic("my-topic")
              .createAsync()
              .thenAccept(p -> {
                  log.info("Producer created: {}", p.getProducerName());
              });

Publish messages​

Pulsar supports both synchronous and asynchronous publishing of messages in most clients. In some language-specific clients, such as Node.js and C#, you can publish messages synchronously based on the asynchronous method using language-specific mechanisms (like await).

With async publishment, the producer puts the message in a blocking queue and returns it immediately. Then the client library sends the message to the broker in the background. If the queue is full (max size configurable), the producer is blocked or fails immediately when calling the API, depending on arguments passed to the producer.

This example shows how to publish messages using producers. The publish operation is done until the broker tells you the message has been successfully published. The broker returns the message ID after the message is published successfully.

Publish messages synchronously:

MessageId messageId = producer.newMessage()
                  .value("my-sync-message")
                  .send();

Publish messages asynchronously:

producer.newMessage()
        .value("my-sync-message")
        .sendAsync()
        .thenAccept(messageId -> {
            log.info("Message ID: {}", messageId);
        });

Configure messages​

You can set various properties of Pulsar's messages. The values of these properties are stored in the metadata of a message.

producer.newMessage()
              .key("my-key") // Set the message key
              .eventTime(System.currentTimeMillis()) // Set the event time
              .sequenceId(1203) // Set the sequenceId for the deduplication purposes
              .deliverAfter(1, TimeUnit.HOURS) // Delay message delivery for 1 hour
              .property("my-key", "my-value") // Set the customized metadata
              .property("my-other-key", "my-other-value")
              .replicationClusters(
                      Lists.newArrayList("r1", "r2")) // Set the geo-replication clusters for this message.
              .value("content")
              .send();

For the Java client, you can also use loadConf to configure the message metadata. Here is an example:

Map<String, Object> conf = new HashMap<>();
conf.put("key", "my-key");
conf.put("eventTime", System.currentTimeMillis());
producer.newMessage()
        .value("my-message")
        .loadConf(conf)
        .send();

Publish messages to partitioned topics​

By default, Pulsar topics are served by a single broker, which limits the maximum throughput of a topic. Partitioned topics can span multiple brokers and thus allow for higher throughput.

You can publish messages to partitioned topics using Pulsar client libraries. When publishing messages to partitioned topics, you must specify a routing mode. If you do not specify any routing mode when you create a new producer, the round-robin routing mode is used.

Use built-in message router​

The routing mode determines which partition (internal topic) each message should be published to.

The following is an example:

Producer<byte[]> producer = pulsarClient.newProducer()
   .topic("my-topic")
   .messageRoutingMode(MessageRoutingMode.SinglePartition)
   .create();

Customize message router​

To use a custom message router, you need to provide an implementation of the MessageRouter interface, which has just one choosePartition method:

public interface MessageRouter extends Serializable {
    int choosePartition(Message msg);
}

The following router routes every message to partition 10:

public class AlwaysTenRouter implements MessageRouter {
    public int choosePartition(Message msg) {
        return 10;
    }
}

With that implementation, you can send messages to partitioned topics as below.

Producer<byte[]> producer = pulsarClient.newProducer()
        .topic("my-topic")
        .messageRouter(new AlwaysTenRouter())
        .create();
producer.send("Partitioned topic message".getBytes());

Choose partitions when using a key​

If a message has a key, it supersedes the round robin routing policy. The following java example code illustrates how to choose the partition when using a key.

// If the message has a key, it supersedes the round robin routing policy
if (msg.hasKey()) {
    return signSafeMod(hash.makeHash(msg.getKey()), topicMetadata.numPartitions());
}

if (isBatchingEnabled) { // if batching is enabled, choose partition on `partitionSwitchMs` boundary.
    long currentMs = clock.millis();
    return signSafeMod(currentMs / partitionSwitchMs + startPtnIdx, topicMetadata.numPartitions());
} else {
    return signSafeMod(PARTITION_INDEX_UPDATER.getAndIncrement(this), topicMetadata.numPartitions());
}

Enable chunking​

Message chunking enables Pulsar to process large payload messages by splitting the message into chunks at the producer side and aggregating chunked messages on the consumer side.

The message chunking feature is OFF by default. The following is an example of how to enable message chunking when creating a producer.

  Producer<byte[]> producer = client.newProducer()
     .topic(topic)
     .enableChunking(true)
     .enableBatching(false)
     .create();

By default, producer chunks the large message based on max message size (maxMessageSize) configured at broker (eg: 5MB). However, client can also configure max chunked size using producer configuration chunkMaxMessageSize.

note

To enable chunking, you need to disable batching (enableBatching=false) concurrently.

Intercept messages​

ProducerInterceptor intercepts and possibly mutates messages received by the producer before they are published to the brokers.

The interface has three main events:

  • eligible checks if the interceptor can be applied to the message.
  • beforeSend is triggered before the producer sends the message to the broker. You can modify messages within this event.
  • onSendAcknowledgement is triggered when the message is acknowledged by the broker or the sending failed.

To intercept messages, you can add a ProducerInterceptor or multiple ones when creating a Producer as follows.

Producer<byte[]> producer = client.newProducer()
     .topic(topic)
     .intercept(new ProducerInterceptor {
 		@Override
 		boolean eligible(Message message) {
 		    return true;  // process all messages
 		}

 		@Override
 		Message beforeSend(Producer producer, Message message) {
 		    // user-defined processing logic
 		}

 		@Override
 		void onSendAcknowledgement(Producer producer, Message message, MessageId msgId, Throwable exception) {
 		    // user-defined processing logic
 		}
     })
     .create();
note

Multiple interceptors apply in the order they are passed to the intercept method.

Configure encryption policies​

The Pulsar C# client supports four kinds of encryption policies:

  • EnforceUnencrypted: always use unencrypted connections.
  • EnforceEncrypted: always use encrypted connections)
  • PreferUnencrypted: use unencrypted connections, if possible.
  • PreferEncrypted: use encrypted connections, if possible.

This example shows how to set the EnforceUnencrypted encryption policy.

using DotPulsar;

var client = PulsarClient.Builder()
                      .ConnectionSecurity(EncryptionPolicy.EnforceEncrypted)
                      .Build();

Configure access mode​

Access mode allows applications to require exclusive producer access on a topic to achieve a "single-writer" situation.

This example shows how to set producer access mode.

note

This feature is supported in Java client 2.8.0 or later versions.

Producer<byte[]> producer = client.newProducer()
     .topic(topic)
     .accessMode(ProducerAccessMode.Exclusive)
     .create();
Was this helpful?
💡 Suggest changes🛟 Get support