Producer Cache becomes ineffective when ProducerBuilderCustomizer is configured

[RungePei]

When using PulsarTemplate to send messages, the following code snippet is used to generate a Producer:

producerCache uses producerCacheKey as the key. The hashCode of producerCacheKey is as follows:

The customizers are a list whose contents are a lambda,so the hashCode is uncertain.
As a result, the producer cannot be obtained from the producerCache, which causes the producerCache to expand rapidly.
The problem I encountered was when using ProducerInterceptor. If the ProducerInterceptor is configured, it will be last configured to the customizers

[jonas-grgt]

Thanks for submitting this issue and great find! You provided half of the solution already.

One way of solving this is by determining the customizers when the intercepters are passed in, in the constructor of CachingPulsarProducerFactory.
WDYT @onobc ?

[onobc]

Hello @RungePei,

Thanks for submitting this issue and great find!

I could not agree more. Also, the details and imagery are quite helpful and very much appreciated.

This also reveals a hole in our cache key tests for this case. We do quite a bit of cache key testing at the caching factory level but not w/ the template + caching factory w/ this interceptor combination. This will be added as part of the solution.

Background

Before we talk about interceptors lets first talk about the use of Lambdas as a cache key, in general.

A couple of facts:

• We let users pass in random customizers (which are likely lambdas).
• We need to include these "wild/random" customizers when caching as they likely affect the producer.

Without enforcing some eq/hc required contract on the customizers, the best option we have is to take a chance and use eq/hc on the customizer. This works in many cases, but not all.

Same lambda instance (static)

The same Lambda instance will match on eq/hc if it has no variable argument passed into it.

@RestController
class ControllerConfig {

	@Autowired
	private PulsarTemplate<User> template;

	@GetMapping("/send1")
	String send1() {
		var user = new User("u1-" + UUID.randomUUID());
		var msgId = template.newMessage(user)
				.withTopic("user-topic")
				.withProducerCustomizer((pb) -> pb.producerName("p1"))
				.send();
		return "Send1 %s -> %s".formatted(user, msgId);
	}

	@GetMapping("/send2")
	String send2() {
		var user = new User("u2-" + UUID.randomUUID());
		var msgId = template.newMessage(user)
				.withTopic("user-topic")
				.withProducerCustomizer((pb) -> pb.producerName("p2"))
				.send();
		return "Send2 %s -> %s".formatted(user, msgId);
	}
}

Each endpoint will create its own producer since it is a different Lambda instance.

• Calls to /send1 will create "p1" producer and then subsequent calls cache as expected.
• Calls to /send2 will create "p2" producer and then subsequent calls cache as expected.

If the customizer was shared amongst them such as:

@RestController
class ControllerConfig {

	@Autowired
	private PulsarTemplate<User> template;

	private ProducerBuilderCustomizer<User> fooCustomizer = (pb) -> pb.producerName("foo");
	
	@GetMapping("/send1")
	String send1() {
		var user = new User("u1-" + UUID.randomUUID());
		var msgId = template.newMessage(user)
				.withTopic("user-topic")
				.withProducerCustomizer(fooCustomizer)
				.send();
		return "Send1 %s -> %s".formatted(user, msgId);
	}

	@GetMapping("/send2")
	String send2() {
		var user = new User("u2-" + UUID.randomUUID());
		var msgId = template.newMessage(user)
				.withTopic("user-topic")
				.withProducerCustomizer(fooCustomizer)
				.send();
		return "Send2 %s -> %s".formatted(user, msgId);
	}
}

Each endpoint will share the same "foo" producer since its the same Lambda instance.

• Calls to /send1 will create "foo" producer and subsequent calls cache as expected.
• Calls to /send2 will use "foo" cached producer.

Same Lambda instance (dynamic)

Warning

If the lambda takes in a variable argument, then it does not match on eq/hc.

If we take the same Lambda from above, but instead use a variable to set the producer name:

@RestController
class ControllerConfig {

	@Autowired
	private PulsarTemplate<User> template;

	@GetMapping("/send1/{pName}")
	String send1(@PathVariable pName) {
		var user = new User("u1-" + UUID.randomUUID());
		var msgId = template.newMessage(user)
				.withTopic("user-topic")
				.withProducerCustomizer((b) -> b.producerName(pName))
				.send();
		return "Send1 %s -> %s".formatted(user, msgId);
	}
}

This will NEVER cache. This is a big limitation and in cases like this requires an eq/hc implementation.

CLICK HERE TO SEE "Goodbye my beautiful baby Lambdas 😿"

class ProducerNameBuilderCustomizer implements ProducerBuilderCustomizer<User> {

    private final String name;

    public ProducerNameBuilderCustomizer(String name) {
        this.name = name;
    }

    @Override
    public void customize(ProducerBuilder<User> producerBuilder) {
        producerBuilder.producerName(this.name);
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;
        ProducerNameBuilderCustomizer that = (ProducerNameBuilderCustomizer) o;
        return name.equals(that.name);
    }

    @Override
    public int hashCode() {
        return Objects.hash(name);
    }
}

Solution

Document the limitiation

Document the use of Lambda customizers wrt caching (summarize all of my rambling above into a legible section in the reference guide).

Add test case(s)

A bit of TDD...
Add case(s) for the interceptor use case(s) and watch the test turn 🟥 . Once fixed, it will go 🟢 .

Fix interceptors issue

One way of solving this is by determining the customizers when the intercepters are passed in, in the constructor of CachingPulsarProducerFactory.
WDYT @onobc ?

Agreed @jonas-grgt . The producer interceptors are just a symptom of the use of Lambdas. If we created them once rather than on every call the eq/hc would be fine. As you suggested, we can "cache" the list as instance var in the PulsarTemplate.<init>.

CLICK HERE TO SEE "This did the trick locally 😸 "

	public PulsarTemplate(PulsarProducerFactory<T> producerFactory, List<ProducerInterceptor> interceptors,
			SchemaResolver schemaResolver, TopicResolver topicResolver, boolean observationEnabled) {
		this.producerFactory = producerFactory;
		this.schemaResolver = schemaResolver;
		this.topicResolver = topicResolver;
		this.observationEnabled = observationEnabled;
		this.interceptors = interceptors;
		if (!CollectionUtils.isEmpty(this.interceptors)) {
			this.interceptorsCustomizers = this.interceptors.stream().map(this::adaptInterceptorToCustomizer).toList();
		}
		else {
			this.interceptorsCustomizers = null;
		}
	}

	private ProducerBuilderCustomizer<T> adaptInterceptorToCustomizer(ProducerInterceptor i) {
		return b -> b.intercept(i);
	}

and then in the prepareForSend:

	private Producer<T> prepareProducerForSend(@Nullable String topic, @Nullable T message, @Nullable Schema<T> schema,
			@Nullable Collection<String> encryptionKeys, @Nullable ProducerBuilderCustomizer<T> producerCustomizer) {
		Schema<T> resolvedSchema = schema == null ? this.schemaResolver.resolveSchema(message).orElseThrow() : schema;
		List<ProducerBuilderCustomizer<T>> customizers = new ArrayList<>();
		if (!CollectionUtils.isEmpty(this.interceptorsCustomizers)) {
			customizers.addAll(this.interceptorsCustomizers);
		}
		if (producerCustomizer != null) {
			customizers.add(producerCustomizer);
		}
		return this.producerFactory.createProducer(resolvedSchema, topic, encryptionKeys, customizers);
	}

When will this be fixed?

We will get the above action items in the 1.0.4 and 1.1.0-M2 releases on March 18 and likely in a SNAPSHOT in the next 1-2 days.

What's the workaround until then?

I think the only option to prevent a continuous cacheMiss->cacheFull->cacheTruncated->cacheMiss loop is to disable caching via setting the spring.pulsar.producer.cache.enabled property to false.

[onobc]

Leaving open until the docs are updated.