[FEATURE] Upgrade pulsar with LightProto support

[BewareMyPower]

Is your feature request related to a problem? Please describe.
PIP-75 introduced LightProto for serialization/deserialization to replace Protobuf. After this change, some imports are changed. We should update these imports for Pulsar 2.8.0 in future.

Describe the solution you'd like
Upgrade pulsar version to StreamNative's daily build that contains apache/pulsar#9046 and fix the build/test error.

[BewareMyPower]

After upgrading the pulsar version, the KoP cannot start now. It can be reproduced with running any test, then the test block at setup phase:

[broker-topic-workers-OrderedScheduler-4-0:org.apache.pulsar.common.protocol.Commands@1658] ERROR org.apache.pulsar.common.protocol.Commands - [PersistentSubscription{topic=persistent://public/default/__consumer_offsets-partition-0, name=reader-e1f0c31277}] [-1] Failed to parse message metadata
java.lang.IllegalArgumentException: Invalid unknonwn tag type: 6
	at org.apache.pulsar.common.api.proto.LightProtoCodec.skipUnknownField(LightProtoCodec.java:270) ~[pulsar-common-2.8.0-rc-202101192246.jar:2.8.0-rc-202101192246]
	at org.apache.pulsar.common.api.proto.MessageMetadata.parseFrom(MessageMetadata.java:1370) ~[pulsar-common-2.8.0-rc-202101192246.jar:2.8.0-rc-202101192246]
	at org.apache.pulsar.common.protocol.Commands.parseMessageMetadata(Commands.java:425) ~[pulsar-common-2.8.0-rc-202101192246.jar:2.8.0-rc-202101192246]
	at org.apache.pulsar.common.protocol.Commands.parseMessageMetadata(Commands.java:415) ~[pulsar-common-2.8.0-rc-202101192246.jar:2.8.0-rc-202101192246]
	at org.apache.pulsar.common.protocol.Commands.peekMessageMetadata(Commands.java:1653) ~[pulsar-common-2.8.0-rc-202101192246.jar:2.8.0-rc-202101192246]
	at org.apache.pulsar.broker.service.AbstractBaseDispatcher.filterEntriesForConsumer(AbstractBaseDispatcher.java:82) ~[pulsar-broker-2.8.0-rc-202101192246.jar:2.8.0-rc-202101192246]
	at org.apache.pulsar.broker.service.persistent.PersistentDispatcherSingleActiveConsumer.internalReadEntriesComplete(PersistentDispatcherSingleActiveConsumer.java:232) ~[pulsar-broker-2.8.0-rc-202101192246.jar:2.8.0-rc-202101192246]
	at org.apache.pulsar.broker.service.persistent.PersistentDispatcherSingleActiveConsumer.lambda$readEntriesComplete$1(PersistentDispatcherSingleActiveConsumer.java:178) ~[pulsar-broker-2.8.0-rc-202101192246.jar:2.8.0-rc-202101192246]
	at org.apache.bookkeeper.mledger.util.SafeRun$1.safeRun(SafeRun.java:32) [managed-ledger-2.8.0-rc-202101192246.jar:2.8.0-rc-202101192246]
	at org.apache.bookkeeper.common.util.SafeRunnable.run(SafeRunnable.java:36) [bookkeeper-common-4.12.1.jar:4.12.1]
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [?:1.8.0_261]
	at java.util.concurrent.FutureTask.run(FutureTask.java:266) [?:1.8.0_261]
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) [?:1.8.0_261]
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) [?:1.8.0_261]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [?:1.8.0_261]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [?:1.8.0_261]
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) [netty-common-4.1.51.Final.jar:4.1.51.Final]
	at java.lang.Thread.run(Thread.java:748) [?:1.8.0_261]

[BewareMyPower]

I wrote a test class to debug

import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.SubscriptionInitialPosition;

import java.util.concurrent.TimeUnit;

@Slf4j
public class Test {
    public static void run() {
        try (PulsarClient client = PulsarClient.builder().serviceUrl("pulsar://localhost:6650").build()) {
            Producer<byte[]> producer = client.newProducer()
                    .maxPendingMessages(100000)
                    .topic("public/default/__consumer_offsets-partition-0")
                    .create();
            MessageId id = producer.newMessage().value("hello".getBytes()).send();
            log.info("Send to {}", id);
            org.apache.pulsar.client.api.Consumer<byte[]> consumer = client.newConsumer()
                    .topic("public/default/__consumer_offsets")
                    .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
                    .subscriptionName("xyz-test")
                    .subscribe();
            Message<byte[]> message = consumer.receive(3, TimeUnit.SECONDS);
            if (message != null) {
                log.info("Receive message: {}", message.getValue().length);
            }
        } catch (PulsarClientException e) {
            log.error("{}", e.getMessage());
        }
    }
}

Because the ports are hard coded, we need to change KopProtocolHandlerTestBase:

diff --git a/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KopProtocolHandlerTestBase.java b/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KopProtocolHandlerTestBase.java
index c12f1ab..733bba7 100644
--- a/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KopProtocolHandlerTestBase.java
+++ b/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KopProtocolHandlerTestBase.java
@@ -91,10 +91,10 @@ public abstract class KopProtocolHandlerTestBase {
     protected URI lookupUrl;
     protected PulsarClient pulsarClient;
 
-    protected int brokerWebservicePort = PortManager.nextFreePort();
+    protected int brokerWebservicePort = 8080;
     protected int brokerWebservicePortTls = PortManager.nextFreePort();
     @Getter
-    protected int brokerPort = PortManager.nextFreePort();
+    protected int brokerPort = 6650;
     @Getter
     protected int kafkaBrokerPort = PortManager.nextFreePort();
     @Getter

After these preparations are done, add a line in KafkaProtocolHandler#initCoordinator:

    public void initGroupCoordinator(BrokerService service) throws Exception {
        /* ... */
        MetadataUtils.createKafkaMetadataIfMissing(pulsarAdmin, kafkaConfig);
        // we should put the line after metadata is created
        Test.run(null);

Before upgrading the pulsar, the test run well that consumer could receive the message. However, after the upgrade, the consumer could not receive any message and the java.lang.IllegalArgumentException was thrown like I've mentioned before.