@RunLLM it appears the recommendation when using the Java pulsar client is to not create a new producer for each message send, but rather to keep them around in some sort of map/LRU cache. Are there any existing tools/libraries/utilities that help achieve this, or any further information/recommendations?

Yes, the recommended practice is to create a Pulsar producer once and reuse it for sending multiple messages, rather than creating a new producer for each send. This is because producer creation is resource-intensive and can lead to performance issues if done repeatedly. The Pulsar Java client does not provide a built-in producer cache, so you need to manage producer instances yourself, typically using a map or LRU cache keyed by topic or producer configuration. For more details, see the discussion in apache/pulsar-client-go#614 and apache/pulsar-client-reactive AdaptedReactiveMessageSenderTests.java. There are no official Java utilities or libraries from Pulsar for producer caching, but you can implement your own using standard Java collections (e.g., ConcurrentHashMap or Guava's Cache). The Pulsar Reactive Client demonstrates a cache pattern for producers. Would you like more detail or example code for implementing such a cache? Sources: - apache/pulsar-client-go#614 - apache/pulsar-client-reactive AdaptedReactiveMessageSenderTests.java 👍 1 button 👎 0 button 💡 Hint: Mention @RunLLM in the thread for followups.

@RunLLM yes, please provide more details and example code for implementing such a cache

Certainly! To implement a producer cache in Java for Pulsar, you typically use a thread-safe map (like ConcurrentHashMap) to store and reuse Producer instances, keyed by topic or configuration. This avoids the overhead of creating a new producer for every message. You can also use a cache with expiration or size limits (e.g., Caffeine or Guava) if you expect many topics. Example using ConcurrentHashMap:

import org.apache.pulsar.client.api.*;
import java.util.concurrent.*;

public class ProducerCache implements AutoCloseable {
    private final PulsarClient client;
    private final ConcurrentHashMap<String, Producer<byte[]>> cache = new ConcurrentHashMap<>();

    public ProducerCache(PulsarClient client) {
        this.client = client;
    }

    public Producer<byte[]> getOrCreateProducer(String topic) throws PulsarClientException {
        return cache.computeIfAbsent(topic, t -> {
            try {
                return client.newProducer().topic(t).create();
            } catch (PulsarClientException e) {
                throw new RuntimeException(e);
            }
        });
    }

    @Override
    public void close() {
        cache.values().forEach(producer -> {
            try {
                producer.close();
            } catch (Exception ignored) {}
        });
        cache.clear();
    }
}

Usage:

ProducerCache producerCache = new ProducerCache(pulsarClient);
Producer<byte[]> producer = producerCache.getOrCreateProducer("my-topic");
producer.send("message".getBytes());
producerCache.close();

This pattern is similar to what is used internally in Pulsar Functions and the Pulsar Reactive Client. For more advanced caching (with expiration or size limits), consider using Caffeine as shown in the Pulsar codebase. Would you like an example using Caffeine or Guava for LRU/expiring cache? Sources: - ConcurrentHashMapProducerCacheProvider.java - ProducerCache.java 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

@RunLLM is there any case when a producer becomes invalid or needs to be reconstructed? if i create a cache to store producers, should entries ever expire? or is it safe to hold on to a producer indefinitely?

Yes, there are cases when a Pulsar producer becomes invalid and needs to be reconstructed. Producers can become invalid due to broker restarts, topic unloads, network failures, or unrecoverable errors (e.g., topic deletion, fencing, or quota exceeded). In such cases, the producer may be closed or disconnected, and attempts to use it will fail, requiring recreation. See ProducerImpl.java and ProducerCache.java. Therefore, if you implement a producer cache, entries should be checked for validity before reuse, and expired or closed producers should be removed and recreated as needed. It is not always safe to hold on to a producer indefinitely without such checks. Would you like more detail on how to handle producer invalidation in a cache? Sources: - ProducerImpl.java - ProducerCache.java 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

@RunLLM yes, please provide more detail on how to handle producer invalidation in a cache, specifically for some of those cases you mentioned: "Producers can become invalid due to broker restarts, topic unloads, network failures, or unrecoverable errors (e.g., topic deletion, fencing, or quota exceeded)."

It seems likely that to handle producer invalidation in a cache, you should check the producer’s state before reuse and recreate it if it is closed or disconnected. For example, in the Pulsar Java client, you can use

producer.isConnected()

to verify if the producer is still valid. If not, remove it from the cache, close it, and create a new one. This approach is used in the Pulsar Functions and Reactive Client caches, which also handle async close and flush on removal (ProducerCache.java, ProducerCacheEntry.java). For unrecoverable errors (e.g., topic deletion, fencing, quota exceeded), the producer will throw exceptions on send or may transition to a closed/terminated state. In such cases, catch the exception, remove the producer from the cache, and attempt to create a new one if needed. This ensures your cache does not hold onto invalid producers indefinitely. Would you like a code example for this pattern? Sources: - ProducerCache.java - ProducerCacheEntry.java 👍 1 button 👎 0 button 💡 Hint: Mention @RunLLM in the thread for followups.

@RunLLM yes, please provide a code example for that pattern

Certainly! Here’s a code example for a Pulsar producer cache that checks producer validity before reuse and recreates the producer if it is closed or disconnected. This pattern is based on the logic in ProducerCache.java and ProducerCacheEntry.java:

import org.apache.pulsar.client.api.*;
import java.util.concurrent.*;

public class ValidatingProducerCache implements AutoCloseable {
    private final PulsarClient client;
    private final ConcurrentHashMap<String, Producer<byte[]>> cache = new ConcurrentHashMap<>();

    public ValidatingProducerCache(PulsarClient client) {
        this.client = client;
    }

    public Producer<byte[]> getOrCreateProducer(String topic) throws PulsarClientException {
        Producer<byte[]> producer = cache.get(topic);
        if (producer == null || !producer.isConnected()) {
            // Remove and close invalid producer if present
            if (producer != null) {
                cache.remove(topic, producer);
                try { producer.close(); } catch (Exception ignored) {}
            }
            // Create and cache a new producer
            producer = client.newProducer().topic(topic).create();
            cache.put(topic, producer);
        }
        return producer;
    }

    public void send(String topic, byte[] data) throws PulsarClientException {
        try {
            getOrCreateProducer(topic).send(data);
        } catch (PulsarClientException e) {
            // Handle unrecoverable errors by removing the producer from cache
            cache.remove(topic);
            throw e;
        }
    }

    @Override
    public void close() {
        cache.values().forEach(producer -> {
            try { producer.close(); } catch (Exception ignored) {}
        });
        cache.clear();
    }
}

This pattern ensures that only valid, connected producers are reused, and invalid ones are closed and replaced as needed. Sources: - ProducerCache.java - ProducerCacheEntry.java 👍 1 button 👎 0 button 💡 Hint: Mention @RunLLM in the thread for followups.

@RunLLM you recommended i check if

producer.isConnected

before using the producer. looking at the code, it seems this method only returns

true

if the producer's state is

Ready

or

RegisteringSchema

. are freshly created producers guaranteed to be one of these two states? it seems the flow of states is

Uninitialized

->

Connecting

->

RegisteringSchema

->

Ready

. Is it possible for the producer to still be in the

Uninitialized

or

Connecting

states, thus failing

isConnected

?

Yes, you are correct:

producer.isConnected()

only returns

true

if the producer's state is

Ready

or

RegisteringSchema

. A freshly created producer may still be in

Uninitialized

or

Connecting

states immediately after creation, especially if you use asynchronous creation or access the producer before the connection is fully established. This means

isConnected()

could return

false

until the producer transitions to

Ready

or

RegisteringSchema

(ProducerImpl.java). To ensure the producer is usable, you should wait for the producer to reach the

Ready

state before sending messages. If you use the synchronous

create()

method, the producer will be in a usable state after creation. If you use

createAsync()

, you must wait for the future to complete. Would you like more detail on handling producer state transitions? Sources: - ProducerImpl.java 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

@RunLLM can you clarify what you mean by "If you use

createAsync()

, you must wait for the future to complete."? are you saying that if

createAsync

completes successfully, the producer will always be in a

Ready

state? and yes, please include some more detail on handling producer state transitions

If you use

createAsync()

, you must wait for the returned

CompletableFuture<Producer<T>>

to complete successfully before using the producer. When the future completes, the producer is guaranteed to be in a usable state—specifically, either

Ready

or, in some schema cases,

RegisteringSchema

(ProducerImpl.java). This means you should not use the producer (e.g., call

send

) until the future has completed. The producer state transitions are:

Uninitialized

→

Connecting

→

RegisteringSchema

(if schema is involved) →

Ready

. The synchronous

create()

method blocks until the producer is ready, while

createAsync()

completes the future only after the producer is ready for use. If the future completes exceptionally, the producer is not usable and should not be cached or used. Would you like a code example for safe async producer creation and usage? Sources: - ProducerImpl.java 👍 1 button 👎 0 button 💡 Hint: Mention @RunLLM in the thread for followups.

@RunLLM yes, please provide a code example for safe async producer creation and usage

Certainly! Here’s a code example for safe async producer creation and usage in Java. The key is to use the

CompletableFuture<Producer<T>>

returned by

createAsync()

and only use the producer after the future completes successfully. This ensures the producer is in a usable (

Ready

or

RegisteringSchema

) state (ProducerImpl.java).

import org.apache.pulsar.client.api.*;
import java.util.concurrent.CompletableFuture;

PulsarClient client = PulsarClient.builder()
    .serviceUrl("pulsar://localhost:6650")
    .build();

CompletableFuture<Producer<byte[]>> futureProducer = client.newProducer()
    .topic("my-topic")
    .createAsync();

futureProducer.thenAccept(producer -> {
    // Safe to use the producer here
    try {
        producer.send("hello".getBytes());
    } catch (PulsarClientException e) {
        // Handle send error
    }
}).exceptionally(ex -> {
    // Handle producer creation failure
    ex.printStackTrace();
    return null;
});

This pattern ensures you never use the producer before it is fully initialized and ready. Sources: - ProducerImpl.java 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

@RunLLM is it idiomatic to catch

PulsarClientException

on publish, or to check

isConnected

before publishing and also to catch exceptions on publish?

It seems likely that the idiomatic approach is to always catch

PulsarClientException

when publishing, regardless of the result of

isConnected()

. Checking

isConnected()

before publishing can help avoid some errors, but it is not sufficient because the connection state can change between the check and the publish operation. The Pulsar Java client and official docs recommend handling exceptions on publish, as network failures, topic unloads, or broker errors can occur at any time (ProducerImpl.java, client-libraries-producers). So, the best practice is to always catch

PulsarClientException

on publish, and optionally check

isConnected()

if you want to avoid obvious failures, but never rely on it alone. Sources: - ProducerImpl.java - client-libraries-producers 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

@RunLLM are all

PulsarClientException

terminal/indicate that the client needs to be recycled? on the flip side, will a client in a bad state ever have

isConnected = true

?

Not all

PulsarClientException

errors are terminal or require recycling the client or producer. Many exceptions (e.g., transient network issues, timeouts) are retriable, and the client can recover automatically. Only unrecoverable exceptions—such as

TopicDoesNotExistException

,

ProducerFencedException

, or `AlreadyClosedException`—indicate that the producer or client should be closed and recreated (ProducerImpl.java, PulsarClientException.java). A client or producer in a bad state (e.g., after unrecoverable errors) will generally have

isConnected = false

. However, there can be brief periods where

isConnected

returns

true

even as a failure is about to be detected, due to asynchronous state changes. Therefore, always handle exceptions on publish, and use

isConnected

as a hint, not a guarantee. Sources: - ProducerImpl.java - PulsarClientException.java 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

@RunLLM how do i determine if a

PulsarClientException

is recoverable or not?

You can determine if a

PulsarClientException

is recoverable by using the static method

PulsarClientException.isRetriableError(Throwable t)

. This method returns

true

for exceptions that are considered retriable (recoverable), and

false

for unrecoverable errors such as authorization failures, topic not found, or producer fenced (PulsarClientException.java). Would you like more detail or a code example? Sources: - PulsarClientException.java 👍 1 button 👎 0 button 💡 Hint: Mention @RunLLM in the thread for followups.

@RunLLM if publishing with a producer fails with an unrecoverable error, do i still need to explicitly flush and/or close the producer?

If publishing with a producer fails with an unrecoverable error, you should explicitly close the producer to release resources, but flushing is not necessary—pending messages will not be retried and are typically failed immediately (ProducerImpl.java, Producer.java). The Pulsar client handles failing pending messages on close after unrecoverable errors. Would you like more detail on the close behavior or error handling? Sources: - ProducerImpl.java - Producer.java 👍 1 button 👎 0 button 💡 Hint: Mention @RunLLM in the thread for followups.

@RunLLM in the ProducerCache.java example you linked above (in the

pulsar-functions

project) there's no handling that supports invalid producers as far as i can tell. can you clarify if

pulsar-functions

has any code to handle unrecoverable/unretriable errors on publish?

It seems likely that the

ProducerCache.java

in the

pulsar-functions

project does not itself handle invalid or unrecoverable producers; it simply caches and closes producers (ProducerCache.java). The cache does not check producer state or remove producers on unrecoverable errors. Handling of unrecoverable/unretriable errors on publish is typically done at a higher level, such as in the function or sink logic that uses the producer. For example, in

PulsarSink.java

, errors during publishing are logged and counted, but there is no automatic removal or recreation of producers in the cache when an unrecoverable error occurs (PulsarSink.java). Would you like more detail on how error handling is structured in Pulsar Functions? Sources: - ProducerCache.java - PulsarSink.java 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

@RunLLM yes, please give me more detail on how error handling is structured in Pulsar Functions. if all the error handler does is just log the error, doesn't it mean that the application is effectively "stuck", and no future publishes will ever succeed?

It seems likely that in Pulsar Functions, error handling for producer publish failures is mostly limited to logging the error and incrementing exception counters, as seen in PulsarSink.java. The default error handler does not attempt to remove or recreate a failed producer from the cache, so if a producer encounters an unrecoverable error (e.g., topic deletion, fencing), future publish attempts using that cached producer will continue to fail, effectively leaving the function "stuck" for that topic. This means that, unless the function or sink is restarted (which would recreate the producer cache), the application will not recover from unrecoverable producer errors on its own. If you need automatic recovery, you would have to implement custom logic to detect unrecoverable errors and refresh the producer in the cache. Sources: - PulsarSink.java 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.