[fix][broker] Avoid storing MessageMetadata instances returned by peekMessageMetadata

[BewareMyPower]

Motivation

#15967 removed the EntryWrapper,
which holds an Entry instance that is never used. Instead, after the
refactoring, the MessageMetadata array is useless, see

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java

Lines 517 to 519 in 298a573

	final MessageMetadata[] metadataArray = entries.stream()
	.map(entry -> Commands.peekMessageMetadata(entry.getDataBuffer(), subscription.toString(), -1))
	.toArray(MessageMetadata[]::new);

Each MessageMetadata instance in the array is returned by
peekMessageMetadata, whose returned value references a thread local
object Commands#LOCAL_MESSAGE_METADATA. It brings a problem that if
multiple entries were read, all MessageMetadata elements in the array
reference the same object.

However, accidentally, the wrong invocation of Optional#orElse saves
it. See

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java

Lines 133 to 134 in 298a573

	final MessageMetadata msgMetadata = optMetadataArray.map(metadataArray -> metadataArray[metadataIndex])
	.orElse(Commands.peekMessageMetadata(metadataAndPayload, subscription.toString(), -1));

Each time peekMessageMetadata is called, the thread local message
metadata will be updated. Unlike orElseGet, the expression in orElse
is always called no matter if the optional is empty.

This behavior change increases the invocations count of
peekMessageMetadata and the metadataArray cache became redundant.

Modifications

• Use orElseGet instead of orElse in AbstractBaseDispatcher.
• Add a new static method Commands#peekAndCopyMessageMetadata that
  returns a MessageMetadata instance allocated from heap memory.
• Call peekAndCopyMessageMetadata to cache all message metadata
  instances in PersistentDispatcherMultipleConsumers.

Verifying this change

• Make sure that the change passes the CI checks.

It's hard to add tests. As I've explained before, #15967 only degrades the
performance and doesn't affect the correctness.

Documentation

Check the box below or label this PR directly.

Need to update docs?

• doc-required
  (Your PR needs to update docs and you will update later)

• doc-not-needed
  (Please explain why)

• doc
  (Your PR contains doc changes)

• doc-complete

[BewareMyPower]

I tried to add the test, but it looks harder than I thought. The only way is to verify the invocation count of the staticCommands#peekMessageMetadata method.

A simple way to expose the problem is reverting the changes of PersistentDispatcherMultipleConsumers.java in this PR and running the following test in FilterEntryTest.

    @Test
    public void testMultiEntriesInMultiConsumerDispatcher() throws Exception {
        String topic = "persistent://prop/ns-abc/topic" + UUID.randomUUID();
        String subName = "sub";

        Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
                .topic(topic)
                // PersistentDispatcherMultipleConsumers will be used for Shared subscription
                .subscriptionType(SubscriptionType.Shared)
                .subscriptionName(subName)
                .subscribe();
        consumer.close();
        PersistentSubscription subscription = (PersistentSubscription) pulsar.getBrokerService()
                .getTopicReference(topic).get().getSubscription(subName);
        Dispatcher dispatcher = subscription.getDispatcher();
        Field field = AbstractBaseDispatcher.class.getDeclaredField("entryFilters");
        field.setAccessible(true);
        NarClassLoader narClassLoader = mock(NarClassLoader.class);

        final List<Long> sequenceIdList = Collections.synchronizedList(new ArrayList<>());
        EntryFilter filter = new EntryFilter() {

            @Override
            public FilterResult filterEntry(Entry entry, FilterContext context) {
                sequenceIdList.add(context.getMsgMetadata().getSequenceId());
                return FilterResult.ACCEPT;
            }

            @Override
            public void close() {
            }
        };
        EntryFilterWithClassLoader loader = spyWithClassAndConstructorArgs(
                EntryFilterWithClassLoader.class, filter, narClassLoader);
        field.set(dispatcher, ImmutableList.of(loader));

        final int numMessages = 10;

        @Cleanup
        Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
                .enableBatching(false)
                .topic(topic)
                .create();
        for (int i = 0; i < numMessages; i++) {
            producer.newMessage().value("msg-" + i).send();
        }

        consumer = pulsarClient.newConsumer(Schema.STRING)
                .topic(topic)
                .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
                // PersistentDispatcherMultipleConsumers will be used for Shared subscription
                .subscriptionType(SubscriptionType.Shared)
                .subscriptionName(subName)
                .subscribe();
        while (true) {
            Message<String> msg = consumer.receive(1, TimeUnit.SECONDS);
            if (msg == null) {
                break;
            }
        }

        assertEquals(LongStream.range(0, numMessages).boxed().collect(Collectors.toList()), sequenceIdList);
    }