[improve][client] Add schema cache to improve performance

[yunmaoQu]

close #23707

Motivation

Schema creation (e.g., Schema.AVRO(SomeClass.class)) is fairly CPU intensive. It would be useful it there would be a weak reference cache for caching the schema instance.

Modifications

• Add SchemaCache implementation using WeakHashMap for schema instance caching
• Add cache configuration and metrics for monitoring
• Add cleanup strategy for expired cache entries
• Modify Schema creation methods (AVRO/JSON/PROTOBUF) to use cache
• Add cloning mechanism to maintain schema immutability

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

Matching PR in forked repository

N/A

[github-actions]

@yunmaoQu Please add the following content to your PR description and select a checkbox:

- [ ] `doc` <!-- Your PR contains doc changes -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [ ] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->

[lhotari]

There are several inconsistencies in this PR. For example, the class names and the class file names don't match. Please test this PR in your own fork first to ensure that it passes tests.
It seems that this PR contains a lot of features related to the schema caching. Instead of adding a lot of features, it would be better to keep the implementation to the minimum.
I'm surprised by competing implementations for implementing the schema cache. There's currently already an open PR #23777.

[yunmaoQu]

ok,i test all and could you review it and give me some suggestions

[lhotari]

ok,i test all and could you review it and give me some suggestions

Instead of adding more code to test everything, please reduce to a minimal implementation. This means to remove features to track cache metrics. That's not something that is needed. For the cache implementation, I'd suggest using a ConcurrentMap created with Guava's MapMaker. Instead of adding yet another abstraction, I'd suggest modifying the PulsarClientImplementationBinding interface and adding a new interface method <T extends com.google.protobuf.GeneratedMessageV3> Schema<T> newProtobufSchema(Class<T> clazz). Then we could keep the cache as an implementation level detail.

example of minimal implementation for newProtobufSchema using Guava's MapMaker with weak keys:

    private static final ConcurrentMap<Class<?>, Schema<?>> PROTOBUF_CACHE = new MapMaker().weakKeys().makeMap();

    public <T extends com.google.protobuf.GeneratedMessageV3> Schema<T> newProtobufSchema(Class<T> clazz) {
        return (Schema<T>) PROTOBUF_CACHE.computeIfAbsent(clazz,
                k -> ProtobufSchema.of(SchemaDefinition.builder().withPojo(clazz).build())).clone();
    }

There shouldn't be a need to ever clear the cache since it's bounded by the number of classes with strong references. It won't consume a significant amount of memory in the first place.

[yunmaoQu]

OK.Should i implement it based on the pre commit or what?

[lhotari]

OK.Should i implement it based on the pre commit or what?

That's something you can decide. Please read my previous message and draw your conclusions.

[walkinggo]

It looks like we're working on similar tasks. I've already created a pull request #23777 to complete this task. Should we work together to finish it, or what do you suggest? @yunmaoQu

[yunmaoQu]

Yes. We can work it together.@walkinggo

[yunmaoQu]

@lhotari

OK.Should i implement it based on the pre commit or what?

That's something you can decide. Please read my previous message and draw your conclusions.

I implement a minimal version. Could you review it and give me some suggestion. Thanks for your previous guide.